Cápsula 05: Namespaces de Pinecone — el aislamiento multi-tenant nativo y por qué es mejor que metadata filtering

Descripción de la cápsula

En el M06 implementaste aislamiento multi-tenant con where={"workspace_id": "X"}. Funciona — siempre que el filter se aplique correctamente. El problema: en sistemas grandes, basta un endpoint nuevo que olvide el filter para tener data leak. La defensa es código + tests + linters, todo correcto, pero todo defensa por convención.

Pinecone tiene una alternativa estructural: namespaces. Cada tenant vive en un espacio lógico separado dentro del mismo índice. Una query a namespace="acme" físicamente NO puede ver vectores de namespace="other_tenant" — la separación está en el motor, no en el filter. Es el equivalente a tener bases de datos separadas por tenant pero pagando una sola.

Esta cápsula te enseña cómo usar namespaces como capa primaria de aislamiento, cómo combinarlos con metadata filtering para sub-segmentación dentro del tenant, y por qué para sistemas multi-tenant productivos es siempre la elección correcta.

Al finalizar esta cápsula serás capaz de:

• ✅ Explicar la diferencia conceptual entre namespace y metadata filter
• ✅ Implementar wrapper que fuerza el uso de namespace correcto por tenant
• ✅ Combinar namespaces (tenant) + metadata filters (sub-segmentación)
• ✅ Diseñar tests de aislamiento que validen físicamente la separación
• ✅ Decidir cuándo usar namespace vs metadata vs ambos
• ✅ Anticipar las trampas operativas: namespace inexistente, naming inconsistente, queries cross-namespace mal diseñadas

Tiempo estimado: 30-35 minutos

La diferencia conceptual: filter vs namespace

                        ENFOQUE 1: METADATA FILTER (M06)

     Pinecone Index
     ┌──────────────────────────────────────────┐
     │  Vectors mezclados de TODOS los tenants  │
     │                                           │
     │  vec1 (tenant=A)  vec2 (tenant=B)  vec3  │ ← todos en mismo espacio
     │  vec4 (tenant=A)  vec5 (tenant=C)  ...   │
     │                                           │
     │  Query: filter={"tenant": "A"}            │
     │     ↓                                     │
     │  Pinecone busca en TODOS los vectores,   │
     │  descarta los que no son tenant A         │
     │                                           │
     │  Si filter falla → DATA LEAK              │
     └──────────────────────────────────────────┘


                        ENFOQUE 2: NAMESPACES

     Pinecone Index
     ┌────────────────┬────────────────┬─────────────────┐
     │  namespace=A   │  namespace=B   │  namespace=C    │
     │                │                │                  │
     │  vec1, vec4    │  vec2          │  vec5           │
     │  ...           │  ...           │  ...            │
     └────────────────┴────────────────┴─────────────────┘

     Query con namespace="A":
       → Pinecone SOLO busca en namespace=A
       → Físicamente imposible ver namespace=B/C
       → Data leak imposible aún si código tiene bug

     Mismo índice, mismo costo, separación garantizada por motor

Diferencia clave: con filter, la seguridad depende de que el código siempre incluya el filter correcto. Con namespace, la seguridad depende del motor. Si la query se ejecuta sin namespace o con namespace incorrecto, devuelve vacío — no puede leak datos.

Implementación correcta

Patrón 1: función única para construir namespace

# pinecone_namespaces.py


def tenant_namespace(workspace_id: str) -> str:
    """
    Convención única para namespace por workspace.
    Usar en TODO el código para evitar inconsistencias.
    """
    if not workspace_id:
        raise ValueError("workspace_id requerido")

    # Sanitizar: lowercase, sin caracteres raros
    sanitized = workspace_id.lower().replace(" ", "-").strip()

    return f"ws-{sanitized}"

Por qué función única:

• Si después decides cambiar la convención (ej: agregar prefijo de ambiente), un solo lugar.
• Imposible que un dev escriba f"workspace-{wid}" y otro f"ws-{wid}" — todos usan la misma función.

Patrón 2: secure_query con namespace obligatorio

# secure_query_pinecone.py
from pinecone import Pinecone, Index


class TenantIsolationError(Exception):
    pass


def secure_query_pinecone(
    index: Index,
    query_vector: list[float],
    tenant: TenantContext,
    additional_filters: dict = None,
    top_k: int = 5,
):
    """
    Wrapper que SIEMPRE usa namespace correcto.
    Imposible saltar este wrapper sin obtener namespace=None → query inválida.
    """
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("workspace_id obligatorio")

    namespace = tenant_namespace(tenant.workspace_id)

    response = index.query(
        vector=query_vector,
        top_k=top_k,
        namespace=namespace,                    # ← obligatorio
        filter=additional_filters,              # opcional, sub-segmentación
        include_metadata=True,
    )

    # Audit log
    log_query_audit(tenant, namespace, top_k)

    return response


def secure_upsert_pinecone(
    index: Index,
    vectors: list,
    tenant: TenantContext,
):
    """Wrapper para upserts con namespace forzado."""
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("workspace_id obligatorio")

    namespace = tenant_namespace(tenant.workspace_id)
    return index.upsert(vectors=vectors, namespace=namespace)

Patrón 3: tests automatizados de aislamiento

# tests/test_namespace_isolation.py
import pytest
from secure_query_pinecone import secure_query_pinecone, TenantIsolationError


@pytest.fixture
def index_with_two_tenants():
    """Setup con vectores en namespaces distintos."""
    index = pinecone_client.Index("test-isolation")

    # Tenant A: 100 vectors
    vectors_a = [
        {"id": f"a_{i}", "values": [0.1] * 1536, "metadata": {"text": f"doc A {i}"}}
        for i in range(100)
    ]
    index.upsert(vectors=vectors_a, namespace="ws-tenant-a")

    # Tenant B: 100 vectors
    vectors_b = [
        {"id": f"b_{i}", "values": [0.2] * 1536, "metadata": {"text": f"doc B {i}"}}
        for i in range(100)
    ]
    index.upsert(vectors=vectors_b, namespace="ws-tenant-b")

    return index


def test_tenant_a_cannot_see_namespace_b(index_with_two_tenants):
    """Tenant A NO puede ver vectors de tenant B."""
    ctx = TenantContext(workspace_id="tenant-a", user_id="u1")
    response = secure_query_pinecone(
        index_with_two_tenants,
        query_vector=[0.1] * 1536,
        tenant=ctx,
        top_k=20,
    )

    for match in response["matches"]:
        assert match["id"].startswith("a_"), (
            f"LEAK: tenant-a vio {match['id']} (esperado solo a_*)"
        )


def test_nonexistent_namespace_returns_empty(index_with_two_tenants):
    """Query a namespace que no existe NO leakea — devuelve vacío."""
    ctx = TenantContext(workspace_id="tenant-c", user_id="u1")  # tenant-c no tiene vectors
    response = secure_query_pinecone(
        index_with_two_tenants,
        query_vector=[0.1] * 1536,
        tenant=ctx,
        top_k=20,
    )

    assert len(response["matches"]) == 0, (
        f"LEAK: namespace inexistente devolvió matches: {response['matches']}"
    )


def test_workspace_id_is_required(index_with_two_tenants):
    """Sin workspace_id no se puede hacer query."""
    with pytest.raises(TenantIsolationError):
        secure_query_pinecone(
            index_with_two_tenants,
            query_vector=[0.1] * 1536,
            tenant=None,
            top_k=5,
        )


def test_concurrent_tenants_no_leakage(index_with_two_tenants):
    """Queries concurrentes de tenants distintos están aisladas."""
    import threading

    leaks = []

    def query_as_tenant(tenant_id: str):
        ctx = TenantContext(workspace_id=tenant_id, user_id=f"u_{tenant_id}")
        response = secure_query_pinecone(
            index_with_two_tenants,
            query_vector=[0.1] * 1536,
            tenant=ctx,
            top_k=20,
        )
        for match in response["matches"]:
            expected_prefix = "a_" if tenant_id == "tenant-a" else "b_"
            if not match["id"].startswith(expected_prefix):
                leaks.append((tenant_id, match["id"]))

    threads = [
        threading.Thread(target=query_as_tenant, args=("tenant-a",))
        for _ in range(10)
    ] + [
        threading.Thread(target=query_as_tenant, args=("tenant-b",))
        for _ in range(10)
    ]

    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not leaks, f"Concurrent leakage: {leaks}"

Combinar namespaces + metadata filtering

Namespace para tenant. Metadata para todo lo demás (visibility, type, fecha, tags).

def search_with_full_security(
    index: Index,
    query_vector: list[float],
    tenant: TenantContext,
    optional_filters: dict = None,
    top_k: int = 5,
):
    """
    Pipeline completo de seguridad:
    - Namespace: aislamiento de tenant (motor)
    - Filter: visibility, type, fecha, tags (metadata)
    """
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("workspace_id obligatorio")

    namespace = tenant_namespace(tenant.workspace_id)

    # Construir filter de visibility según rol
    visibility_filter = build_visibility_filter(tenant, optional_filters)

    return index.query(
        vector=query_vector,
        top_k=top_k,
        namespace=namespace,           # ← aislamiento principal
        filter=visibility_filter,      # ← sub-segmentación
        include_metadata=True,
    )


def build_visibility_filter(tenant: TenantContext, additional: dict = None):
    """Construir filter de metadata para visibility según rol."""
    visibility_clauses = []

    # Public to tenant
    visibility_clauses.append({"visibility": "public"})

    # Team docs si user es del team
    if tenant.project_ids:
        visibility_clauses.append({
            "$and": [
                {"visibility": "team"},
                {"project_id": {"$in": tenant.project_ids}},
            ]
        })

    # Personal solo del owner
    visibility_clauses.append({
        "$and": [
            {"visibility": "personal"},
            {"owner_id": {"$eq": tenant.user_id}},
        ]
    })

    # Admin docs solo si es admin
    if tenant.user_role in ("admin", "owner"):
        visibility_clauses.append({"visibility": "admin_only"})

    base_filter = {"$or": visibility_clauses}

    if additional:
        return {"$and": [base_filter, additional]}
    return base_filter

Resultado: dos capas de seguridad. Aunque el filter de visibility tenga un bug, el namespace garantiza aislamiento mínimo entre tenants.

Cuándo usar namespace, metadata, o ambos

Regla: namespace para la dimensión de aislamiento más fuerte (típicamente tenant). Metadata para sub-segmentación dentro de eso.

Patrones avanzados

Patrón A: dos niveles de namespace

Para corpus grandes con muchos tenants y proyectos:

def project_namespace(workspace_id: str, project_id: str) -> str:
    """Namespace de granularidad fina: tenant + project."""
    return f"ws-{workspace_id}__proj-{project_id}"


# Uso
namespace = project_namespace("acme", "marketing")
# Resultado: "ws-acme__proj-marketing"

Trade-off: mejor aislamiento pero queries cross-project son más complejas (necesitas multiple namespaces).

Patrón B: namespace por ambiente

def env_namespace(workspace_id: str, env: str = None) -> str:
    """Namespace que incluye ambiente para staging/prod separation."""
    env = env or os.getenv("APP_ENV", "prod")
    return f"{env}__ws-{workspace_id}"

Útil cuando un tenant tiene datos de staging que NO deben mezclarse con prod.

Patrón C: lista de namespaces existentes

def list_active_tenants(index: Index) -> list[str]:
    """Lista todos los tenants con datos en el índice."""
    stats = index.describe_index_stats()
    namespaces = stats["namespaces"]

    return [
        ns_name.replace("ws-", "")
        for ns_name in namespaces.keys()
        if ns_name.startswith("ws-") and namespaces[ns_name]["vector_count"] > 0
    ]


# Útil para admin operations (analytics, audit, cleanup)
active_tenants = list_active_tenants(index)
print(f"Active tenants: {len(active_tenants)}")

Trampas y errores comunes

Trampa 1: olvidar namespace en query

El error:

response = index.query(vector=v, top_k=5)  # ← sin namespace

Síntoma: Pinecone busca en namespace "" (default). Si nadie escribió ahí, devuelve vacío. Pero si alguien guardó datos en default por error, leak entre tenants.

Cómo prevenir: SIEMPRE usar secure_query_pinecone que enforca namespace.

Trampa 2: naming inconsistente entre devs

El error: dev A escribe f"workspace-{wid}", dev B escribe f"ws-{wid}". Mismo tenant tiene datos en dos namespaces distintos.

Síntoma: queries del dev A no ven docs ingestados por el dev B. Datos "fragmentados".

Cómo prevenir: función única tenant_namespace() usada en TODO el código.

Trampa 3: namespace de un test queda en producción

El error: dev hace test en producción con namespace="my-test". Test termina, namespace queda con basura.

Síntoma: después de meses, hay 50 namespaces de tests viejos. Audits ven docs sin owner claro.

Cómo prevenir:

• Tests siempre en índice de staging, nunca prod.
• Si tests en prod son inevitables, usar namespace prefix test- y cleanup script periódico.

Trampa 4: queries cross-namespace mal diseñadas

El error: admin necesita estadísticas de todos los tenants. Hace loop sobre todos los namespaces:

for ns in active_tenants:
    response = index.query(vector=v, top_k=5, namespace=ns)
    # ... procesar ...

Síntoma: funciona pero es lento (N queries) y puede romper rate limits.

Cómo prevenir: para operaciones admin que requieren cross-tenant, considerar:

• Materialized views agregadas (cómputo offline).
• Índice separado para datos cross-tenant aprobados.
• Queries selectivas (solo top tenants en lugar de todos).

Trampa 5: namespace con caracteres inválidos

El error: tenant_namespace("Acme Corp / Brazil") → "ws-Acme Corp / Brazil".

Síntoma: Pinecone puede aceptar (con limitaciones) o rechazar según versión.

Cómo prevenir: función de namespace sanitiza:

import re

def tenant_namespace(workspace_id: str) -> str:
    sanitized = re.sub(r'[^a-z0-9-_]', '-', workspace_id.lower())
    return f"ws-{sanitized}"

Trampa 6: pensar que namespaces son gratis

El error: crear namespace por cada user (en lugar de por tenant).

Síntoma: índice con 1M de namespaces. Performance del describe_index_stats degrada.

Cómo prevenir: namespaces son granularidad de tenant (típicamente 50-10K). Para granularidad finer, metadata filter es mejor.

Ejercicio aplicado

Escenario: eres AI Engineer en una empresa SaaS de marketing. Datos:

• 200 clientes (tenants), cada uno con 1-50 proyectos
• Total: ~5000 proyectos
• Cada proyecto tiene 5K-100K docs
• Compliance: GDPR + SOC2 audit cada 6 meses
• Queries típicas: dentro de un proyecto específico

Tu trabajo:

1. Decide estrategia de namespaces (uno por cliente o uno por proyecto).
2. Diseña secure_query y secure_upsert apropiados.
3. Plan de tests automatizados de aislamiento para CI.

# secure_marketing_search.py
from typing import Literal


def tenant_namespace(workspace_id: str) -> str:
    sanitized = workspace_id.lower().replace(" ", "-").strip()
    return f"ws-{sanitized}"


@dataclass(frozen=True)
class MarketingTenantContext:
    workspace_id: str
    user_id: str
    user_role: Literal["admin", "manager", "team_member"]
    project_ids: list[str]  # proyectos a los que el user pertenece

    def __post_init__(self):
        if not self.workspace_id or not self.user_id:
            raise TenantIsolationError("workspace_id y user_id obligatorios")


def secure_marketing_query(
    index,
    query_vector: list[float],
    tenant: MarketingTenantContext,
    project_id: str = None,
    additional_filters: dict = None,
    top_k: int = 5,
):
    """
    Pipeline de seguridad:
    - Namespace: aislamiento por tenant
    - Filter: project_id (si se pasa) + visibility según rol
    """
    namespace = tenant_namespace(tenant.workspace_id)

    # Validar acceso al proyecto
    if project_id and project_id not in tenant.project_ids:
        raise PermissionError(f"User no pertenece al proyecto {project_id}")

    # Construir filter
    filter_clauses = []

    if project_id:
        # Query a proyecto específico
        filter_clauses.append({"project_id": {"$eq": project_id}})
    else:
        # Query a todos los proyectos del user (manager view)
        filter_clauses.append({"project_id": {"$in": tenant.project_ids}})

    # Visibility según rol
    if tenant.user_role == "team_member":
        filter_clauses.append({"visibility": {"$ne": "admin_only"}})
    # admin y manager ven todo

    if additional_filters:
        filter_clauses.append(additional_filters)

    final_filter = {"$and": filter_clauses}

    return index.query(
        vector=query_vector,
        top_k=top_k,
        namespace=namespace,
        filter=final_filter,
        include_metadata=True,
    )


def secure_marketing_upsert(
    index,
    vectors: list,
    tenant: MarketingTenantContext,
):
    """Upsert con namespace forzado."""
    namespace = tenant_namespace(tenant.workspace_id)

    # Validar que cada vector tiene project_id en metadata
    for v in vectors:
        if "project_id" not in v["metadata"]:
            raise ValueError("Cada vector debe tener project_id en metadata")
        if v["metadata"]["project_id"] not in tenant.project_ids:
            raise PermissionError(
                f"User no pertenece al proyecto {v['metadata']['project_id']}"
            )

    return index.upsert(vectors=vectors, namespace=namespace)

# tests/test_marketing_isolation.py
import pytest


@pytest.fixture
def index_multi_tenant():
    """Setup: 3 tenants × 2 proyectos cada uno."""
    index = pinecone_client.Index("test-marketing")

    for tenant in ["acme", "globex", "wonka"]:
        for project in ["mkt", "sales"]:
            vectors = [
                {
                    "id": f"{tenant}_{project}_{i}",
                    "values": [0.1] * 1536,
                    "metadata": {
                        "project_id": f"{tenant}_{project}",
                        "visibility": "public",
                    },
                }
                for i in range(100)
            ]
            index.upsert(vectors=vectors, namespace=f"ws-{tenant}")

    return index


def test_cross_tenant_isolation(index_multi_tenant):
    """Acme NO puede ver docs de Globex."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="manager",
        project_ids=["acme_mkt", "acme_sales"],
    )

    response = secure_marketing_query(
        index_multi_tenant,
        [0.1] * 1536,
        ctx,
        top_k=50,
    )

    for match in response["matches"]:
        assert match["id"].startswith("acme_"), (
            f"LEAK: acme vio {match['id']}"
        )


def test_cross_project_within_tenant(index_multi_tenant):
    """Acme team_member solo del proyecto mkt NO ve sales."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="team_member",
        project_ids=["acme_mkt"],  # SOLO mkt
    )

    response = secure_marketing_query(
        index_multi_tenant,
        [0.1] * 1536,
        ctx,
        top_k=50,
    )

    for match in response["matches"]:
        assert "_mkt_" in match["id"], (
            f"LEAK: user solo de mkt vio {match['id']}"
        )


def test_cannot_query_unauthorized_project():
    """User no puede pasar project_id de proyecto al que no pertenece."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="team_member",
        project_ids=["acme_mkt"],
    )

    with pytest.raises(PermissionError):
        secure_marketing_query(
            index,
            [0.1] * 1536,
            ctx,
            project_id="acme_sales",  # ← no pertenece
        )


def test_admin_sees_all_visibilities():
    """Admin ve todos los visibility levels."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="admin",
        project_ids=["acme_mkt", "acme_sales"],
    )

    # Insertar doc admin_only en proyecto acme_mkt
    # Admin debe verlo
    response = secure_marketing_query(
        index,
        [0.1] * 1536,
        ctx,
        project_id="acme_mkt",
    )

    has_admin_only = any(
        m["metadata"]["visibility"] == "admin_only"
        for m in response["matches"]
    )
    assert has_admin_only, "Admin no vio docs admin_only"


def test_team_member_cannot_see_admin_only():
    """team_member NO ve admin_only docs."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="team_member",
        project_ids=["acme_mkt"],
    )

    response = secure_marketing_query(
        index,
        [0.1] * 1536,
        ctx,
        project_id="acme_mkt",
    )

    for match in response["matches"]:
        assert match["metadata"]["visibility"] != "admin_only", (
            f"LEAK: team_member vio admin_only doc {match['id']}"
        )


# Configurar en CI
# .github/workflows/ci.yml
# - run: pytest tests/test_marketing_isolation.py

# Cada query loggea:
{
    "timestamp": "2026-05-15T...",
    "workspace_id": tenant.workspace_id,
    "user_id": tenant.user_id,
    "user_role": tenant.user_role,
    "namespace_used": namespace,
    "project_id": project_id,
    "query_hash": hash(query),  # NO el query plaintext
    "results_count": len(response["matches"]),
}

# Retention 7 años para SOC2
# Almacenamiento: S3 con object lock (write-once)

Resumen y siguiente paso

Lo que aprendiste:

• Namespaces de Pinecone son aislamiento estructural — el motor garantiza separación, no el código.
• Mejor que metadata filter para multi-tenant: aún si filter falla, namespace previene leak.
• Combinar: namespace para tenant, metadata filter para sub-segmentación (visibility, project_id, fecha).
• Función única tenant_namespace() previene inconsistencia entre devs.
• secure_query_pinecone wrapper obligatorio que enforca namespace correcto.
• Tests de aislamiento automatizados son críticos para producción.
• Trampas: olvidar namespace en query (default), naming inconsistente, queries cross-namespace lentas, namespaces con caracteres inválidos.

Checkpoint: antes de avanzar, deberías poder:

• Diseñar tenant_namespace() con sanitización.
• Implementar secure_query_pinecone con namespace forzado.
• Escribir tests automatizados de aislamiento que corren en CI.

Siguiente cápsula: 06 — Metadata filtering en Pinecone.

Tienes namespaces para tenant. Ahora cubrimos metadata filtering nativo de Pinecone — sintaxis levemente distinta de ChromaDB, mismos conceptos, mejor performance. Vas a aprender a migrar tus filters de M06 a Pinecone correctamente.

Recursos

1. Pinecone — Multitenancy Guide — Patrón oficial
2. Pinecone — Namespaces — Documentación de feature
3. OWASP API Security — Threat model
4. Microsoft — Multi-tenant SaaS Patterns — Decisión arquitectónica
5. SOC2 Trust Services Criteria — Compliance
6. Anthropic — Contextual Retrieval — Patrón complementario

Tiempo estimado: 30-35 minutos Siguiente: 06-metadata-filtering-in-pinecone.md