feat(crm): expediente con folio propio y espejo del documento en EFC

El CRM genera documentos que hoy viven sueltos en su MinIO: sin nada que los agrupe y sin catalogo validado en backend. EFC es el sistema de expedientes de la casa y debe ser la fuente unica, pero el CRM no tiene ni un dato de pedimento -verificado: ningun campo de aduana, patente, numero ni anio en crm/ ni en ops/- y en EFC el expediente ES el pedimento. Esta fase pone del lado del CRM lo que falta para poder hablar de un expediente antes de que exista data aduanera. crm.expedientes cuelga de la solicitud de servicio, que es el hilo comercial que el repo ya tiene de la RFQ a la factura. Nace con folio EXP{YYYY}-{MM}-{NNN} en la misma transaccion del alta: un flush y no un commit, porque si el alta fallara despues quedaria un folio quemado colgando de una solicitud inexistente. El folio se reserva con un unico INSERT ... ON CONFLICT DO UPDATE ... RETURNING sobre crm.expediente_folio_counters. Un SELECT max(sequence)+1 es justo la carrera que hay que evitar: dos altas simultaneas leerian el mismo maximo y entregarian el mismo folio a dos expedientes distintos. El asignador no commitea, igual que el de folios de Anexo22, para que un rollback posterior pueda soltar el folio; deja hueco, que es preferible a un duplicado. Los dos UNIQUE de la tabla son la red por si el contador se corrompe: revienta ruidoso en vez de mezclar dos hilos documentales. efc_storage_token se fija al nacer y es inmutable: es la carpeta de MinIO en EFC. Lleva el company_id dentro porque el puente con EFC es tenant -> organizacion 1:1 pero un tenant tiene N companies, asi que sin el, dos companies generando EXP2026-08-001 chocarian en el unique_together de EFC y a una le devolverian el provisional de la otra. EfcDocumentRefMixin va en las dos tablas de documentos -crm.documents y ops.shipment_documents- porque las dos alimentan el mismo expediente; el handle autoritativo es efc_document_ref, que el CRM construye, y efc_document_id es solo cache de la resolucion. El ref es texto y no entero porque las dos tablas tienen secuencias independientes: crm.documents.id = 5 y ops.shipment_documents.id = 5 coexisten. La migracion se GENERA y se deja SIN aplicar. Verificacion: pytest tests/ EXIT=0, 90 passed 1 skipped (baseline 70 passed). Los tests nuevos se vieron en rojo a proposito antes de darlos por buenos: quitando el enganche del expediente reventaron nombrando el expediente ausente, quitando el UNIQUE del folio el test dijo DID NOT RAISE IntegrityError, y dejando el contador sin incrementar los tres consecutivos salieron 001/001/001. Ticket: T2026-08-046 (fase 5 de 7)
2026-08-07 17:50:11 -06:00
parent c6f18013b3
commit 39347f9c97
17 changed files with 1183 additions and 6 deletions
--- a/backend/alembic/versions/e6f7a8b9c0d1_crm_expedientes.py
+++ b/backend/alembic/versions/e6f7a8b9c0d1_crm_expedientes.py
@@ -0,0 +1,163 @@
+"""Expediente del CRM y su espejo en EFC: crm.expedientes, el contador de folios y las columnas
+del espejo en las dos tablas de documentos.
+
+Revision ID: e6f7a8b9c0d1
+Revises: d5e6f7a8b9c0
+Create Date: 2026-08-07 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "e6f7a8b9c0d1"
+down_revision: Union[str, None] = "d5e6f7a8b9c0"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+# Las diez columnas del espejo en EFC. Van idénticas en crm.documents y en ops.shipment_documents
+# porque las dos alimentan el mismo expediente electrónico: si divergieran, la UI pintaría un badge
+# distinto según de dónde viniera el documento. Se define una vez aquí y se aplica en bucle, para
+# que no se puedan desalinear al editar la migración.
+def _columnas_espejo_efc() -> list[sa.Column]:
+    return [
+        sa.Column("expediente_id", sa.Integer(), nullable=True),
+        sa.Column("efc_document_ref", sa.String(length=64), nullable=True),
+        sa.Column("efc_document_id", sa.String(length=36), nullable=True),
+        sa.Column("efc_sync_state", sa.String(length=20), nullable=True, server_default=sa.text("'PENDING'")),
+        sa.Column("efc_synced_at", sa.DateTime(), nullable=True),
+        sa.Column("efc_error_code", sa.String(length=60), nullable=True),
+        sa.Column("efc_error_detail", sa.Text(), nullable=True),
+        sa.Column("efc_attempts", sa.Integer(), nullable=True, server_default=sa.text("0")),
+        sa.Column("content_sha256", sa.String(length=64), nullable=True),
+    ]
+
+
+_TABLAS_CON_ESPEJO = (
+    ("documents", "crm"),
+    ("shipment_documents", "ops"),
+)
+
+
+def upgrade() -> None:
+    # ---------- crm.expedientes ----------
+    op.create_table(
+        "expedientes",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("folio", sa.String(length=20), nullable=False),
+        sa.Column("period_year", sa.Integer(), nullable=False),
+        sa.Column("period_month", sa.Integer(), nullable=False),
+        sa.Column("sequence", sa.Integer(), nullable=False),
+        sa.Column("service_request_id", sa.Integer(), nullable=True),
+        sa.Column("account_id", sa.Integer(), nullable=True),
+        sa.Column("status", sa.String(length=20), nullable=False, server_default=sa.text("'abierto'")),
+        sa.Column("efc_organizacion_id", sa.String(length=36), nullable=True),
+        sa.Column("efc_pedimento_id", sa.String(length=36), nullable=True),
+        sa.Column("efc_storage_token", sa.String(length=25), nullable=True),
+        sa.Column("efc_link_state", sa.String(length=20), nullable=False, server_default=sa.text("'PENDING'")),
+        sa.Column("efc_error_code", sa.String(length=60), nullable=True),
+        sa.Column("efc_error_detail", sa.Text(), nullable=True),
+        sa.Column("patente", sa.String(length=20), nullable=True),
+        sa.Column("aduana", sa.String(length=10), nullable=True),
+        sa.Column("numero_pedimento", sa.String(length=20), nullable=True),
+        sa.Column("anio", sa.Integer(), nullable=True),
+        sa.Column("clave_pedimento", sa.String(length=10), nullable=True),
+        sa.Column("regimen", sa.String(length=10), nullable=True),
+        sa.Column("fecha_pago", sa.Date(), nullable=True),
+        sa.Column("rfc_importador", sa.String(length=20), nullable=True),
+        sa.Column("rfc_agente_aduanal", sa.String(length=100), nullable=True),
+        sa.Column("created_by", sa.String(length=64), nullable=True),
+        sa.Column("updated_by", sa.String(length=64), nullable=True),
+        sa.Column("tenant_id", sa.Integer(), nullable=False),
+        sa.Column("company_id", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
+        sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
+        sa.Column("deleted_at", sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        # Las dos redes de seguridad del folio: si el contador se corrompe, un duplicado falla
+        # ruidosamente en vez de mezclar dos hilos documentales.
+        sa.UniqueConstraint("tenant_id", "company_id", "folio", name="uq_crm_expedientes_folio"),
+        sa.UniqueConstraint(
+            "tenant_id", "company_id", "period_year", "period_month", "sequence",
+            name="uq_crm_expedientes_periodo_seq",
+        ),
+        schema="crm",
+    )
+    op.create_index("ix_crm_expedientes_id", "expedientes", ["id"], schema="crm")
+    op.create_index("ix_crm_expedientes_folio", "expedientes", ["folio"], schema="crm")
+    op.create_index("ix_crm_expedientes_status", "expedientes", ["status"], schema="crm")
+    op.create_index("ix_crm_expedientes_tenant_id", "expedientes", ["tenant_id"], schema="crm")
+    op.create_index("ix_crm_expedientes_company_id", "expedientes", ["company_id"], schema="crm")
+    op.create_index(
+        "ix_crm_expedientes_service_request_id", "expedientes", ["service_request_id"], schema="crm"
+    )
+    op.create_index("ix_crm_expedientes_account_id", "expedientes", ["account_id"], schema="crm")
+    op.create_foreign_key(
+        "fk_crm_expedientes_tenant_id", "expedientes", "tenants",
+        ["tenant_id"], ["id"], source_schema="crm", referent_schema="core",
+    )
+    op.create_foreign_key(
+        "fk_crm_expedientes_service_request_id", "expedientes", "service_requests",
+        ["service_request_id"], ["id"], source_schema="crm", referent_schema="crm",
+    )
+    op.create_foreign_key(
+        "fk_crm_expedientes_account_id", "expedientes", "accounts",
+        ["account_id"], ["id"], source_schema="crm", referent_schema="crm",
+    )
+
+    # ---------- crm.expediente_folio_counters ----------
+    # PK compuesta (tenant, company, period): es la fila sobre la que serializa el
+    # INSERT ... ON CONFLICT DO UPDATE del asignador. Sin esa PK el upsert no tiene sobre qué
+    # detectar el conflicto y dos altas simultáneas darían el mismo folio.
+    op.create_table(
+        "expediente_folio_counters",
+        sa.Column("tenant_id", sa.Integer(), nullable=False),
+        sa.Column("company_id", sa.Integer(), nullable=False),
+        sa.Column("period", sa.String(length=7), nullable=False),
+        sa.Column("last_seq", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.PrimaryKeyConstraint("tenant_id", "company_id", "period"),
+        schema="crm",
+    )
+    op.create_foreign_key(
+        "fk_crm_expediente_folio_counters_tenant_id", "expediente_folio_counters", "tenants",
+        ["tenant_id"], ["id"], source_schema="crm", referent_schema="core",
+    )
+
+    # ---------- Espejo de EFC en las dos tablas de documentos ----------
+    for tabla, schema in _TABLAS_CON_ESPEJO:
+        for columna in _columnas_espejo_efc():
+            op.add_column(tabla, columna, schema=schema)
+        op.create_index(
+            f"ix_{schema}_{tabla}_expediente_id", tabla, ["expediente_id"], schema=schema
+        )
+        op.create_index(
+            f"ix_{schema}_{tabla}_efc_document_ref", tabla, ["efc_document_ref"], schema=schema
+        )
+
+
+def downgrade() -> None:
+    for tabla, schema in reversed(_TABLAS_CON_ESPEJO):
+        op.drop_index(f"ix_{schema}_{tabla}_efc_document_ref", table_name=tabla, schema=schema)
+        op.drop_index(f"ix_{schema}_{tabla}_expediente_id", table_name=tabla, schema=schema)
+        for columna in reversed(_columnas_espejo_efc()):
+            op.drop_column(tabla, columna.name, schema=schema)
+
+    op.drop_constraint(
+        "fk_crm_expediente_folio_counters_tenant_id", "expediente_folio_counters",
+        schema="crm", type_="foreignkey",
+    )
+    op.drop_table("expediente_folio_counters", schema="crm")
+
+    op.drop_constraint("fk_crm_expedientes_account_id", "expedientes", schema="crm", type_="foreignkey")
+    op.drop_constraint("fk_crm_expedientes_service_request_id", "expedientes", schema="crm", type_="foreignkey")
+    op.drop_constraint("fk_crm_expedientes_tenant_id", "expedientes", schema="crm", type_="foreignkey")
+    op.drop_index("ix_crm_expedientes_account_id", table_name="expedientes", schema="crm")
+    op.drop_index("ix_crm_expedientes_service_request_id", table_name="expedientes", schema="crm")
+    op.drop_index("ix_crm_expedientes_company_id", table_name="expedientes", schema="crm")
+    op.drop_index("ix_crm_expedientes_tenant_id", table_name="expedientes", schema="crm")
+    op.drop_index("ix_crm_expedientes_status", table_name="expedientes", schema="crm")
+    op.drop_index("ix_crm_expedientes_folio", table_name="expedientes", schema="crm")
+    op.drop_index("ix_crm_expedientes_id", table_name="expedientes", schema="crm")
+    op.drop_table("expedientes", schema="crm")
--- a/backend/api/v1/common/base_models.py
+++ b/backend/api/v1/common/base_models.py
@@ -1,6 +1,6 @@
 from datetime import datetime

-from sqlalchemy import DateTime, ForeignKey, Integer
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, text
 from sqlalchemy.orm import Mapped, mapped_column
 from sqlalchemy.sql import func

@@ -31,3 +31,39 @@ class TenantScopedMixin:

    tenant_id: Mapped[int] = mapped_column(Integer, ForeignKey("core.tenants.id"), nullable=False, index=True)
    company_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
+
+
+class EfcDocumentRefMixin:
+    """Columnas del espejo de un documento en EFC. Se aplica a ``crm.documents`` y a
+    ``ops.shipment_documents``.
+
+    ``efc_document_ref`` es el handle AUTORITATIVO —el CRM lo construye y EFC lo guarda—;
+    ``efc_document_id`` es solo un CACHE de la resolución, recuperable por el endpoint de lista si
+    se pierde. Es el mismo principio que aplica el gateway de Anexo22: el sistema de origen conserva
+    el registro de SU dato, y con eso pide el archivo de vuelta, en lugar de guardar identificadores
+    ajenos en columnas propias.
+
+    ``efc_sync_state`` es el estado del ESPEJO (lo que pinta la UI: badge, botón reintentar). La
+    cola de trabajo vive aparte, en ``crm.efc_file_outbox``. No son redundantes: el outbox es
+    indexable por su propio ciclo de vida y sobrevive a un borrado cuya fila ya no está.
+
+    Es un mixin y no diez columnas copiadas en dos modelos porque las dos tablas tienen que
+    describir el mismo espejo: si divergen, la UI pinta un badge distinto según de dónde venga el
+    documento y nadie entiende por qué.
+    """
+
+    expediente_id: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True)
+    # {TABLA}-{company_id}-{row_id}, p. ej. SHPDOC-1-4471. Texto y no un entero porque el CRM tiene
+    # dos tablas de documentos con secuencias independientes: crm.documents.id = 5 y
+    # ops.shipment_documents.id = 5 coexisten, así que un entero solo sería ambiguo entre ellas.
+    efc_document_ref: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    efc_document_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    # PENDING | SYNCED | FAILED
+    efc_sync_state: Mapped[str | None] = mapped_column(
+        String(20), nullable=True, server_default=text("'PENDING'")
+    )
+    efc_synced_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    efc_error_code: Mapped[str | None] = mapped_column(String(60), nullable=True)
+    efc_error_detail: Mapped[str | None] = mapped_column(Text, nullable=True)
+    efc_attempts: Mapped[int | None] = mapped_column(Integer, nullable=True, server_default=text("0"))
+    content_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True)
--- a/backend/api/v1/modules/crm/documents/models.py
+++ b/backend/api/v1/modules/crm/documents/models.py
@@ -1,15 +1,20 @@
 from sqlalchemy import ForeignKey, Integer, String
 from sqlalchemy.orm import Mapped, mapped_column

-from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
+from api.v1.common.base_models import EfcDocumentRefMixin, TenantScopedMixin, TimestampMixin
 from core.database import Base


-class Document(Base, TenantScopedMixin, TimestampMixin):
+class Document(Base, TenantScopedMixin, TimestampMixin, EfcDocumentRefMixin):
    """Documento de un cliente (``account_id``) o proveedor (``supplier_id``).

    Guarda los metadatos y una referencia al archivo (``file_key`` en MinIO/S3 o
    ``file_url`` externa). La subida binaria se hace vía la capa de storage.
+
+    Con ``EfcDocumentRefMixin`` la fila además refleja el estado del documento en el expediente
+    electrónico de EFC. Un documento del CRM puede vivir en tres modos y la descarga se ramifica por
+    ellos: ``efc_document_id`` (está en EFC → proxy), ``file_key`` (solo local → URL firmada) o
+    ``file_url`` (externa).
    """

    __tablename__ = "documents"
--- a/backend/api/v1/modules/crm/expedientes/init.py
+++ b/backend/api/v1/modules/crm/expedientes/init.py
--- a/backend/api/v1/modules/crm/expedientes/doc_types.py
+++ b/backend/api/v1/modules/crm/expedientes/doc_types.py
@@ -0,0 +1,64 @@
+"""Catálogo CERRADO de tipos de documento que EFC acepta del CRM.
+
+Estas 22 claves son **exactamente** las de ``TIPOS_DOCUMENTO_CRM`` en
+``api/record/views_integrations_crm.py`` de EFC. La lista está duplicada a mano en dos repos con
+despliegue independiente, así que ``tests/test_doc_types_paridad.py`` la fija: si alguien agrega un
+tipo de un solo lado, ese test se pone rojo antes de que un documento se rechace en producción.
+
+Por qué es un conjunto cerrado y no texto libre, a diferencia del carril de Anexo22 —que manda el
+tipo suelto y deja que EFC lo resuelva por nombre—: en el CRM ``doc_type`` es ``String(60)`` /
+``String(30)`` **sin validación de backend**, los catálogos viven solo en TypeScript
+(``frontend/src/lib/api/crm/format.ts``). Un typo crearía un ``DocumentType`` basura en el catálogo
+**global** de EFC, que es compartido por todas las organizaciones y no se limpia solo.
+
+Las tres fuentes del CRM y su origen:
+
+- ``crm.documents``           → ``DOC_TYPES`` de ``format.ts``
+- ``ops.shipment_documents``  → ``SHIPMENT_DOC_TYPES`` del mismo archivo
+- ``fin.invoices``            → el PDF de factura (``factura_venta``)
+
+``otro`` existe en las dos listas del CRM y significa lo mismo en ambas: es una sola entrada.
+"""
+
+# --- crm.documents ---------------------------------------------------------------------------
+_TIPOS_DOCUMENTOS_CLIENTE = (
+    "constancia_fiscal",
+    "acta_constitutiva",
+    "identificacion",
+    "comprobante_domicilio",
+    "contrato",
+    "presentacion",
+    "certificacion",
+    "licencia",
+    "convenio",
+    "tarifario",
+)
+
+# --- ops.shipment_documents ------------------------------------------------------------------
+_TIPOS_DOCUMENTOS_EMBARQUE = (
+    "MBL",
+    "HBL",
+    "MAWB",
+    "HAWB",
+    "CMR",
+    "factura_comercial",
+    "packing_list",
+    "carta_encomienda",
+    "carta_garantia",
+    "certificado_permiso",
+)
+
+# --- fin.invoices ----------------------------------------------------------------------------
+_TIPOS_FACTURACION = ("factura_venta",)
+
+# --- común a varias fuentes -------------------------------------------------------------------
+_TIPOS_COMUNES = ("otro",)
+
+EFC_DOC_TYPES: frozenset[str] = frozenset(
+    _TIPOS_DOCUMENTOS_CLIENTE + _TIPOS_DOCUMENTOS_EMBARQUE + _TIPOS_FACTURACION + _TIPOS_COMUNES
+)
+
+
+def is_valid_doc_type(doc_type: str | None) -> bool:
+    """``True`` si EFC va a aceptar ese tipo. Se valida en el CRM para no gastar un viaje de red."""
+    return bool(doc_type) and doc_type in EFC_DOC_TYPES
--- a/backend/api/v1/modules/crm/expedientes/dto.py
+++ b/backend/api/v1/modules/crm/expedientes/dto.py
@@ -0,0 +1,79 @@
+from datetime import date, datetime
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class ExpedienteBase(BaseModel):
+    service_request_id: int | None = None
+    account_id: int | None = None
+    status: str = Field("abierto", max_length=20)
+
+
+class ExpedienteCreate(BaseModel):
+    """Alta explícita de un expediente.
+
+    No lleva ``folio``: lo asigna el servidor con el contador de ``folio.py``. Aceptarlo del cliente
+    permitiría pisar el consecutivo de otro expediente.
+    """
+
+    service_request_id: int | None = None
+    account_id: int | None = None
+
+
+class ExpedienteUpdate(BaseModel):
+    account_id: int | None = None
+    status: str | None = Field(None, max_length=20)
+
+
+class ExpedienteCompleteInput(BaseModel):
+    """Data aduanera real con la que se completa un expediente provisional."""
+
+    patente: str = Field(..., max_length=20)
+    aduana: str = Field(..., max_length=10)
+    numero_pedimento: str = Field(..., max_length=20)
+    anio: int = Field(..., ge=1900, le=2999)
+    clave_pedimento: str | None = Field(None, max_length=10)
+    regimen: str | None = Field(None, max_length=10)
+    fecha_pago: date | None = None
+    rfc_importador: str | None = Field(None, max_length=20)
+    rfc_agente_aduanal: str | None = Field(None, max_length=100)
+
+
+class ExpedienteResponse(ExpedienteBase):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    folio: str
+    period_year: int
+    period_month: int
+    sequence: int
+
+    efc_organizacion_id: str | None = None
+    efc_pedimento_id: str | None = None
+    efc_storage_token: str | None = None
+    efc_link_state: str
+    efc_error_code: str | None = None
+    efc_error_detail: str | None = None
+
+    patente: str | None = None
+    aduana: str | None = None
+    numero_pedimento: str | None = None
+    anio: int | None = None
+    clave_pedimento: str | None = None
+    regimen: str | None = None
+    fecha_pago: date | None = None
+    rfc_importador: str | None = None
+    rfc_agente_aduanal: str | None = None
+
+    created_by: str | None = None
+    updated_by: str | None = None
+    tenant_id: int
+    company_id: int
+    created_at: datetime
+    updated_at: datetime
+
+
+class ExpedienteEnsureInput(BaseModel):
+    """Entrada de ``POST /expedientes/ensure``: la solicitud a la que colgar el expediente."""
+
+    service_request_id: int
--- a/backend/api/v1/modules/crm/expedientes/folio.py
+++ b/backend/api/v1/modules/crm/expedientes/folio.py
@@ -0,0 +1,95 @@
+"""Asignador del folio de expediente: ``EXP{YYYY}-{MM}-{NNN}``.
+
+Un consecutivo por ``(tenant, company, mes)`` que reinicia cada mes. La disciplina es la misma del
+asignador de folios de Anexo22 (``catalogos/customs_brokers/folios.py``): validar antes de tocar el
+contador, **no commitear dentro del asignador**, y fallar cerrado ante ambigüedad.
+"""
+
+from datetime import date
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from .models import ExpedienteFolioCounter
+
+# Ancho del consecutivo dentro del folio. Al pasar de 999 el folio crece a 4 dígitos en vez de
+# truncarse o reiniciar: un folio ya comunicado al cliente no puede cambiar de forma.
+_SEQ_WIDTH = 3
+
+
+def format_folio(year: int, month: int, sequence: int) -> str:
+    """``(2026, 8, 1)`` → ``"EXP2026-08-001"``. Única fuente del formato del folio."""
+    return f"EXP{year:04d}-{month:02d}-{sequence:0{_SEQ_WIDTH}d}"
+
+
+def storage_token(company_id: int, folio: str) -> str:
+    """``CRM-{company_id}-{folio}`` — la llave del pedimento provisional en EFC.
+
+    Empieza con letras, así que es imposible que colisione con la llave de un pedimento real, que
+    es ``^\\d{2}-\\d{2}-\\d{4}-\\d{7}$``. El ``company_id`` va dentro porque el puente con EFC es
+    tenant → organización 1:1 pero un tenant tiene N companies: sin él, dos companies del mismo
+    tenant generarían el mismo ``EXP2026-08-001`` y chocarían en el ``unique_together`` de EFC.
+
+    Cabe en los 25 caracteres de ``Pedimento.pedimento_app`` mientras el consecutivo no pase de 4
+    dígitos y el ``company_id`` de 7: ``CRM-`` (4) + company + ``-`` + ``EXP2026-08-001`` (14).
+    """
+    return f"CRM-{company_id}-{folio}"
+
+
+def next_folio(
+    db: Session, tenant_id: int, company_id: int, on: date | None = None
+) -> tuple[str, int, int, int]:
+    """Reserva el siguiente consecutivo del mes y devuelve ``(folio, year, month, sequence)``.
+
+    Una sola sentencia atómica, sin read-modify-write: el ``INSERT ... ON CONFLICT DO UPDATE``
+    serializa sobre la fila de ese ``(tenant, company, mes)`` y devuelve el valor ya incrementado.
+    Un ``SELECT max(sequence) + 1`` es exactamente la carrera que hay que evitar, y un
+    ``SELECT ... FOR UPDATE`` también sirve pero son dos viajes.
+
+    NO hace commit: opera sobre la sesión que recibe, para que un fallo posterior en la creación del
+    expediente pueda hacer rollback sin quemar el folio.
+
+    Un rollback deja HUECO en la secuencia. Los huecos son aceptables; los duplicados no.
+    """
+    today = on or date.today()
+    period = f"{today.year:04d}-{today.month:02d}"
+
+    # El constructor de upsert es por dialecto: PostgreSQL en producción, SQLite en las pruebas
+    # unitarias (tests/conftest.py). Se usa el constructor de SQLAlchemy y no SQL crudo porque el
+    # `schema_translate_map` de las pruebas solo traduce el schema `crm` si la tabla viaja como
+    # objeto; en un `text()` el nombre del schema queda escrito a mano y rompe en SQLite.
+    dialect = db.get_bind().dialect.name
+    if dialect == "postgresql":
+        from sqlalchemy.dialects.postgresql import insert as _insert
+    else:
+        from sqlalchemy.dialects.sqlite import insert as _insert
+
+    table = ExpedienteFolioCounter.__table__
+    stmt = _insert(table).values(
+        tenant_id=tenant_id, company_id=company_id, period=period, last_seq=1
+    )
+    stmt = stmt.on_conflict_do_update(
+        index_elements=["tenant_id", "company_id", "period"],
+        set_={"last_seq": table.c.last_seq + 1},
+    ).returning(table.c.last_seq)
+
+    sequence = db.execute(stmt).scalar_one()
+    return format_folio(today.year, today.month, sequence), today.year, today.month, sequence
+
+
+def peek_last_sequence(db: Session, tenant_id: int, company_id: int, on: date | None = None) -> int:
+    """El último consecutivo entregado en ese mes, o ``0`` si todavía no hay ninguno.
+
+    Solo lectura y sin efecto sobre el contador: existe para diagnóstico y para las pruebas. Quien
+    necesite un folio usa :func:`next_folio`.
+    """
+    today = on or date.today()
+    period = f"{today.year:04d}-{today.month:02d}"
+    value = db.execute(
+        select(ExpedienteFolioCounter.last_seq).where(
+            ExpedienteFolioCounter.tenant_id == tenant_id,
+            ExpedienteFolioCounter.company_id == company_id,
+            ExpedienteFolioCounter.period == period,
+        )
+    ).scalar_one_or_none()
+    return int(value or 0)
--- a/backend/api/v1/modules/crm/expedientes/models.py
+++ b/backend/api/v1/modules/crm/expedientes/models.py
@@ -0,0 +1,117 @@
+from datetime import date
+
+from sqlalchemy import Date, ForeignKey, Integer, String, Text, UniqueConstraint, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
+from core.database import Base
+
+
+class Expediente(Base, TenantScopedMixin, TimestampMixin):
+    """Expediente del CRM: el hilo documental de una operación, de la RFQ a la factura.
+
+    El ancla es la solicitud de servicio (``crm.service_requests``): un expediente por hilo
+    comercial, siguiendo la cadena que el CRM ya tiene. En EFC cada expediente se refleja como un
+    *pedimento provisional* cuyo ``pedimento_app`` es el ``efc_storage_token``, y cuando llega la
+    data aduanera real ese provisional se completa sin mover un solo archivo.
+
+    El folio va DESCOMPUESTO en ``period_year`` / ``period_month`` / ``sequence`` además de
+    guardarse armado en ``folio``: así el consecutivo es un constraint real de la base y no un
+    parse de string. ``uq_crm_expedientes_periodo_seq`` es la red de seguridad — si el contador se
+    corrompe, un folio duplicado falla ruidosamente en vez de mezclar dos expedientes.
+
+    Los campos ``efc_*`` son un ESPEJO de lo que hay en EFC, nunca el handle. El handle que el CRM
+    usa para hablar de este expediente es su ``folio`` y su ``id``: ``efc_pedimento_id`` es un cache
+    de la resolución y ``pedimento_app`` del lado de EFC es mutable —se reescribe al completar—, así
+    que apoyarse en él rompería en cuanto la data real llegue.
+
+    ``efc_storage_token`` es INMUTABLE una vez asignado: es la carpeta de MinIO donde EFC guarda los
+    objetos de este expediente. Que no cambie nunca es lo que hace que completar el pedimento no
+    obligue a mover archivos.
+    """
+
+    __tablename__ = "expedientes"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "company_id", "folio", name="uq_crm_expedientes_folio"),
+        UniqueConstraint(
+            "tenant_id",
+            "company_id",
+            "period_year",
+            "period_month",
+            "sequence",
+            name="uq_crm_expedientes_periodo_seq",
+        ),
+        {"schema": "crm"},
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+
+    # ── Folio ──────────────────────────────────────────────────────────────────────────────
+    folio: Mapped[str] = mapped_column(String(20), nullable=False, index=True)  # EXP2026-08-001
+    period_year: Mapped[int] = mapped_column(Integer, nullable=False)
+    period_month: Mapped[int] = mapped_column(Integer, nullable=False)
+    sequence: Mapped[int] = mapped_column(Integer, nullable=False)
+
+    # ── Anclas comerciales ─────────────────────────────────────────────────────────────────
+    service_request_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("crm.service_requests.id"), nullable=True, index=True
+    )
+    account_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("crm.accounts.id"), nullable=True, index=True
+    )
+
+    # abierto | completado | cerrado
+    status: Mapped[str] = mapped_column(
+        String(20), nullable=False, server_default=text("'abierto'"), index=True
+    )
+
+    # ── Espejo de EFC ──────────────────────────────────────────────────────────────────────
+    efc_organizacion_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    efc_pedimento_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    efc_storage_token: Mapped[str | None] = mapped_column(String(25), nullable=True)
+    # PENDING | LINKED | FAILED
+    efc_link_state: Mapped[str] = mapped_column(
+        String(20), nullable=False, server_default=text("'PENDING'")
+    )
+    # El diagnóstico se guarda en la fila para que se vea en la ficha, sin obligar a ir a los logs.
+    efc_error_code: Mapped[str | None] = mapped_column(String(60), nullable=True)
+    efc_error_detail: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+    # ── Data aduanera real: se llena al completar, no al crear ─────────────────────────────
+    # Longitudes tomadas de api/customs/models.py::Pedimento en EFC, que es el destino de estos
+    # datos: patente 20, aduana 10, regimen 10, clave_pedimento 10, RFC del agente 100.
+    patente: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    aduana: Mapped[str | None] = mapped_column(String(10), nullable=True)
+    numero_pedimento: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    anio: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    clave_pedimento: Mapped[str | None] = mapped_column(String(10), nullable=True)
+    regimen: Mapped[str | None] = mapped_column(String(10), nullable=True)
+    fecha_pago: Mapped[date | None] = mapped_column(Date, nullable=True)
+    rfc_importador: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    rfc_agente_aduanal: Mapped[str | None] = mapped_column(String(100), nullable=True)
+
+    created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+
+
+class ExpedienteFolioCounter(Base):
+    """Contador de folios por ``(tenant, company, mes)``.
+
+    Tabla propia y no un ``max(sequence) + 1`` sobre ``crm.expedientes``: ese SELECT es exactamente
+    la carrera que hay que evitar. Aquí el consecutivo se reserva con un solo
+    ``INSERT ... ON CONFLICT DO UPDATE ... RETURNING`` (ver ``folio.py``), que serializa sobre esta
+    fila y devuelve el valor ya incrementado.
+
+    No lleva los mixins de tenant ni de timestamps a propósito: ``tenant_id`` y ``company_id`` son
+    parte de la PK compuesta, y una fila de contador no tiene ciclo de vida propio que auditar.
+    """
+
+    __tablename__ = "expediente_folio_counters"
+    __table_args__ = {"schema": "crm"}
+
+    tenant_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("core.tenants.id"), primary_key=True, nullable=False
+    )
+    company_id: Mapped[int] = mapped_column(Integer, primary_key=True, nullable=False)
+    period: Mapped[str] = mapped_column(String(7), primary_key=True, nullable=False)  # "2026-08"
+    last_seq: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
--- a/backend/api/v1/modules/crm/expedientes/routes.py
+++ b/backend/api/v1/modules/crm/expedientes/routes.py
@@ -0,0 +1,77 @@
+from fastapi import APIRouter, Depends, Query, status
+from sqlalchemy.orm import Session
+
+from core.database import get_core_db
+from core.security import get_current_user
+
+from . import service
+from .dto import ExpedienteCompleteInput, ExpedienteEnsureInput, ExpedienteResponse
+
+router = APIRouter()
+
+
+@router.get("/expedientes", response_model=list[ExpedienteResponse])
+def list_expedientes(
+    company_id: int = Query(..., description="Company ID"),
+    service_request_id: int | None = Query(None, description="Filtrar por solicitud"),
+    account_id: int | None = Query(None, description="Filtrar por cliente"),
+    status_filter: str | None = Query(None, alias="status", description="abierto|completado|cerrado"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    tenant_id = current_user["tenant_id"]
+    return service.list_expedientes(
+        db, tenant_id, company_id, service_request_id, account_id, status_filter
+    )
+
+
+@router.get("/expedientes/{expediente_id}", response_model=ExpedienteResponse)
+def get_expediente(
+    expediente_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    tenant_id = current_user["tenant_id"]
+    return service.get_expediente(db, expediente_id, tenant_id, company_id)
+
+
+@router.post("/expedientes/ensure", response_model=ExpedienteResponse)
+def ensure_expediente(
+    payload: ExpedienteEnsureInput,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Devuelve el expediente de una solicitud, creándolo si hace falta. Idempotente.
+
+    Responde 200 y no 201 justamente porque es idempotente: el llamador no puede distinguir —ni le
+    importa— si el expediente ya estaba.
+    """
+    tenant_id = current_user["tenant_id"]
+    user_id = current_user.get("sub") or current_user.get("id")
+    return service.ensure_expediente(db, payload.service_request_id, tenant_id, company_id, user_id)
+
+
+@router.post("/expedientes/{expediente_id}/completar", response_model=ExpedienteResponse)
+def complete_expediente(
+    expediente_id: int,
+    payload: ExpedienteCompleteInput,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    tenant_id = current_user["tenant_id"]
+    user_id = current_user.get("sub") or current_user.get("id")
+    return service.complete_expediente(db, expediente_id, payload, tenant_id, company_id, user_id)
+
+
+@router.delete("/expedientes/{expediente_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_expediente(
+    expediente_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    tenant_id = current_user["tenant_id"]
+    service.delete_expediente(db, expediente_id, tenant_id, company_id)
--- a/backend/api/v1/modules/crm/expedientes/service.py
+++ b/backend/api/v1/modules/crm/expedientes/service.py
@@ -0,0 +1,180 @@
+"""Servicio del expediente del CRM.
+
+Funciones libres que reciben ``db, tenant_id, company_id``, como el resto de los módulos del repo.
+"""
+
+from datetime import datetime, timezone
+
+from fastapi import HTTPException, status
+from sqlalchemy.orm import Session
+
+from ..service_requests.models import ServiceRequest
+from .dto import ExpedienteCompleteInput
+from .folio import next_folio, storage_token
+from .models import Expediente
+
+
+def _get_service_request(db: Session, service_request_id: int, tenant_id: int, company_id: int) -> ServiceRequest:
+    obj = (
+        db.query(ServiceRequest)
+        .filter(
+            ServiceRequest.id == service_request_id,
+            ServiceRequest.tenant_id == tenant_id,
+            ServiceRequest.company_id == company_id,
+            ServiceRequest.deleted_at.is_(None),
+        )
+        .first()
+    )
+    if not obj:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Solicitud no encontrada")
+    return obj
+
+
+def get_expediente(db: Session, expediente_id: int, tenant_id: int, company_id: int) -> Expediente:
+    obj = (
+        db.query(Expediente)
+        .filter(
+            Expediente.id == expediente_id,
+            Expediente.tenant_id == tenant_id,
+            Expediente.company_id == company_id,
+            Expediente.deleted_at.is_(None),
+        )
+        .first()
+    )
+    if not obj:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Expediente no encontrado")
+    return obj
+
+
+def list_expedientes(
+    db: Session,
+    tenant_id: int,
+    company_id: int,
+    service_request_id: int | None = None,
+    account_id: int | None = None,
+    exp_status: str | None = None,
+) -> list[Expediente]:
+    query = db.query(Expediente).filter(
+        Expediente.tenant_id == tenant_id,
+        Expediente.company_id == company_id,
+        Expediente.deleted_at.is_(None),
+    )
+    if service_request_id is not None:
+        query = query.filter(Expediente.service_request_id == service_request_id)
+    if account_id is not None:
+        query = query.filter(Expediente.account_id == account_id)
+    if exp_status:
+        query = query.filter(Expediente.status == exp_status)
+    return query.order_by(Expediente.created_at.desc()).all()
+
+
+def find_by_service_request(
+    db: Session, service_request_id: int, tenant_id: int, company_id: int
+) -> Expediente | None:
+    return (
+        db.query(Expediente)
+        .filter(
+            Expediente.service_request_id == service_request_id,
+            Expediente.tenant_id == tenant_id,
+            Expediente.company_id == company_id,
+            Expediente.deleted_at.is_(None),
+        )
+        .first()
+    )
+
+
+def ensure_expediente_for_service_request(
+    db: Session,
+    service_request_id: int,
+    tenant_id: int,
+    company_id: int,
+    user_id: str | None = None,
+    account_id: int | None = None,
+) -> Expediente:
+    """Devuelve el expediente de esa solicitud, creándolo si todavía no existe. **Idempotente.**
+
+    **No hace commit**: hace ``flush`` sobre la sesión que recibe. La razón es que sus dos
+    llamadores —``create_service_request`` y ``create_from_opportunity``— lo invocan ANTES de su
+    propio commit, dentro de la misma transacción. Si esto commiteara por su cuenta, un fallo
+    posterior en el alta de la solicitud dejaría un expediente huérfano con su folio ya quemado.
+    Es la misma disciplina de ``next_folio`` y la del asignador de folios de Anexo22.
+    """
+    existing = find_by_service_request(db, service_request_id, tenant_id, company_id)
+    if existing is not None:
+        return existing
+
+    folio, year, month, sequence = next_folio(db, tenant_id, company_id)
+    expediente = Expediente(
+        folio=folio,
+        period_year=year,
+        period_month=month,
+        sequence=sequence,
+        service_request_id=service_request_id,
+        account_id=account_id,
+        status="abierto",
+        efc_storage_token=storage_token(company_id, folio),
+        efc_link_state="PENDING",
+        tenant_id=tenant_id,
+        company_id=company_id,
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(expediente)
+    db.flush()
+    return expediente
+
+
+def ensure_expediente(
+    db: Session,
+    service_request_id: int,
+    tenant_id: int,
+    company_id: int,
+    user_id: str | None = None,
+) -> Expediente:
+    """Variante de cara al usuario: valida la solicitud, asegura el expediente y **sí** commitea.
+
+    La usa el endpoint ``POST /expedientes/ensure``, donde la transacción empieza y termina aquí.
+    """
+    solicitud = _get_service_request(db, service_request_id, tenant_id, company_id)
+    expediente = ensure_expediente_for_service_request(
+        db, solicitud.id, tenant_id, company_id, user_id, account_id=solicitud.account_id
+    )
+    db.commit()
+    db.refresh(expediente)
+    return expediente
+
+
+def complete_expediente(
+    db: Session,
+    expediente_id: int,
+    payload: ExpedienteCompleteInput,
+    tenant_id: int,
+    company_id: int,
+    user_id: str | None = None,
+) -> Expediente:
+    """Registra en el CRM la data aduanera real de un expediente.
+
+    Solo toca la fila del CRM. Completar el pedimento provisional del lado de EFC es una operación
+    aparte, del carril del gateway, porque puede fallar por causas de EFC —un pedimento real que ya
+    existe con esa llave— y eso no debe impedir que el CRM guarde lo que el usuario capturó.
+    """
+    expediente = get_expediente(db, expediente_id, tenant_id, company_id)
+    if expediente.status == "completado":
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="El expediente ya está completado"
+        )
+
+    for field, value in payload.model_dump(exclude_unset=True).items():
+        setattr(expediente, field, value)
+    expediente.status = "completado"
+    expediente.updated_by = user_id
+    db.commit()
+    db.refresh(expediente)
+    return expediente
+
+
+def delete_expediente(db: Session, expediente_id: int, tenant_id: int, company_id: int) -> None:
+    """Baja lógica. No propaga nada a EFC: el expediente electrónico se conserva."""
+    expediente = get_expediente(db, expediente_id, tenant_id, company_id)
+    expediente.deleted_at = datetime.now(timezone.utc)
+    db.commit()
--- a/backend/api/v1/modules/crm/permissions.py
+++ b/backend/api/v1/modules/crm/permissions.py
@@ -16,6 +16,7 @@ _ENTITIES = [
    ("contact", "contactos"),
    ("address", "direcciones"),
    ("document", "documentos"),
+    ("expediente", "expedientes"),
    ("service_request", "solicitudes de servicio"),
    ("rate_request", "solicitudes de tarifa"),
    ("quote", "cotizaciones"),
--- a/backend/api/v1/modules/crm/router.py
+++ b/backend/api/v1/modules/crm/router.py
@@ -16,6 +16,7 @@ from .addresses.routes import router as addresses_router
 from .catalogs.routes import router as catalogs_router
 from .contacts.routes import router as contacts_router
 from .documents.routes import router as documents_router
+from .expedientes.routes import router as expedientes_router
 from .leads.routes import router as leads_router
 from .metrics.routes import router as metrics_router
 from .opportunities.routes import router as opportunities_router
@@ -35,6 +36,7 @@ router.include_router(suppliers_router)
 router.include_router(contacts_router)
 router.include_router(addresses_router)
 router.include_router(documents_router)
+router.include_router(expedientes_router)
 router.include_router(service_requests_router)
 router.include_router(quotes_router)
 router.include_router(leads_router)
--- a/backend/api/v1/modules/crm/service_requests/service.py
+++ b/backend/api/v1/modules/crm/service_requests/service.py
@@ -5,6 +5,7 @@ from sqlalchemy.orm import Session

 from ..accounts.models import Account
 from ..catalogs.data import INCOTERM_CODES
+from ..expedientes import service as expedientes_service
 from ..opportunities.models import Opportunity
 from ..suppliers.models import Supplier
 from .dto import (
@@ -49,6 +50,24 @@ def _validate_request_refs(db: Session, data: dict, tenant_id: int, company_id:
        )


+def _ensure_expediente(
+    db: Session, obj: ServiceRequest, tenant_id: int, company_id: int, user_id: str | None
+) -> None:
+    """Le da su expediente —y con él su folio— a una solicitud recién creada.
+
+    Va DENTRO de la transacción del alta, no como un paso posterior best-effort: el folio del
+    expediente es lo que el usuario ve en la pantalla en cuanto guarda, así que una solicitud sin
+    expediente sería una solicitud a medias. Si esto falla, el alta entera se revierte y el usuario
+    ve el error, que es preferible a una solicitud que nadie puede documentar.
+
+    Lo best-effort es la *réplica hacia EFC*, no esto: aquélla vive en el gateway y nunca rompe la
+    operación local.
+    """
+    expedientes_service.ensure_expediente_for_service_request(
+        db, obj.id, tenant_id, company_id, user_id, account_id=obj.account_id
+    )
+
+
 # ----- Service requests (RFQ) -----

 def get_service_requests(
@@ -104,6 +123,11 @@ def create_service_request(
    _validate_request_refs(db, data, tenant_id, company_id)
    obj = ServiceRequest(**data, tenant_id=tenant_id, company_id=company_id, created_by=user_id, updated_by=user_id)
    db.add(obj)
+    # flush y no commit: el expediente necesita el id de la solicitud, pero los dos tienen que
+    # nacer en la MISMA transacción. Si el expediente se creara aparte y el alta fallara después,
+    # quedaría un folio quemado colgando de una solicitud que no existe.
+    db.flush()
+    _ensure_expediente(db, obj, tenant_id, company_id, user_id)
    db.commit()
    db.refresh(obj)
    return obj
@@ -184,6 +208,8 @@ def create_from_opportunity(
        updated_by=user_id,
    )
    db.add(obj)
+    db.flush()
+    _ensure_expediente(db, obj, tenant_id, company_id, user_id)
    db.commit()
    db.refresh(obj)
    return obj
--- a/backend/api/v1/modules/ops/shipments/models.py
+++ b/backend/api/v1/modules/ops/shipments/models.py
@@ -3,7 +3,7 @@ from datetime import date, datetime
 from sqlalchemy import Date, DateTime, ForeignKey, Integer, Numeric, String, Text, text
 from sqlalchemy.orm import Mapped, mapped_column

-from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
+from api.v1.common.base_models import EfcDocumentRefMixin, TenantScopedMixin, TimestampMixin
 from core.database import Base


@@ -92,8 +92,12 @@ class ShipmentEvent(Base, TenantScopedMixin, TimestampMixin):
    notes: Mapped[str | None] = mapped_column(Text, nullable=True)


-class ShipmentDocument(Base, TenantScopedMixin, TimestampMixin):
-    """Documento de transporte del embarque (Master/House: MBL, HBL, MAWB, HAWB, CMR, etc.)."""
+class ShipmentDocument(Base, TenantScopedMixin, TimestampMixin, EfcDocumentRefMixin):
+    """Documento de transporte del embarque (Master/House: MBL, HBL, MAWB, HAWB, CMR, etc.).
+
+    Comparte con ``crm.documents`` las columnas del espejo en EFC vía ``EfcDocumentRefMixin``: las
+    dos tablas alimentan el mismo expediente electrónico y la UI las pinta con el mismo badge.
+    """

    __tablename__ = "shipment_documents"
    __table_args__ = {"schema": "ops"}
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -30,6 +30,7 @@ import api.v1.modules.crm.activities.models  # noqa: E402,F401
 import api.v1.modules.crm.addresses.models  # noqa: E402,F401
 import api.v1.modules.crm.contacts.models  # noqa: E402,F401
 import api.v1.modules.crm.documents.models  # noqa: E402,F401
+import api.v1.modules.crm.expedientes.models  # noqa: E402,F401
 import api.v1.modules.crm.leads.models  # noqa: E402,F401
 import api.v1.modules.crm.opportunities.models  # noqa: E402,F401
 import api.v1.modules.crm.pipelines.models  # noqa: E402,F401
--- a/backend/tests/test_expediente_folio.py
+++ b/backend/tests/test_expediente_folio.py
@@ -0,0 +1,134 @@
+"""Pruebas del asignador de folios de expediente.
+
+El folio es lo primero que el usuario ve de un expediente y lo que comunica al cliente, así que un
+duplicado no es un detalle: dos expedientes con el mismo folio son dos hilos documentales que
+alguien va a mezclar. Estas pruebas fijan el formato, el reinicio mensual, el aislamiento por
+company y que un rollback deje hueco en vez de duplicar.
+"""
+
+from datetime import date
+
+import pytest
+
+from api.v1.modules.crm.expedientes.folio import (
+    format_folio,
+    next_folio,
+    peek_last_sequence,
+    storage_token,
+)
+from tests.conftest import COMPANY_ID, TENANT_ID
+
+
+def test_formato_del_folio():
+    assert format_folio(2026, 8, 1) == "EXP2026-08-001"
+    assert format_folio(2026, 12, 42) == "EXP2026-12-042"
+
+
+def test_folio_de_cuatro_digitos_al_pasar_de_999():
+    """Al pasar de 999 el folio CRECE, no se trunca ni reinicia.
+
+    Truncar cambiaría la forma de un folio ya comunicado al cliente, y reiniciar duplicaría uno
+    anterior del mismo mes.
+    """
+    assert format_folio(2026, 8, 1000) == "EXP2026-08-1000"
+
+
+def test_storage_token_lleva_company_y_no_parece_pedimento_real():
+    import re
+
+    token = storage_token(1, "EXP2026-08-001")
+    assert token == "CRM-1-EXP2026-08-001"
+    # La llave de un pedimento real es todo dígitos con tres guiones. El provisional empieza con
+    # letras justamente para que sea imposible que colisione.
+    assert not re.match(r"^\d{2}-\d{2}-\d{4}-\d{7}$", token)
+    assert len(token) <= 25  # cabe en Pedimento.pedimento_app de EFC
+
+
+def test_storage_token_de_dos_companies_del_mismo_tenant_no_colisiona():
+    """H5: tenant → organización es 1:1 pero un tenant tiene N companies.
+
+    Sin el company_id dentro del token, dos companies generando EXP2026-08-001 chocarían en el
+    unique_together de EFC y el get_or_create le devolvería a una el provisional de la otra.
+    """
+    assert storage_token(1, "EXP2026-08-001") != storage_token(2, "EXP2026-08-001")
+
+
+def test_tres_folios_seguidos_son_consecutivos(db):
+    on = date(2026, 8, 15)
+    f1, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    f2, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    f3, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    assert [f1, f2, f3] == ["EXP2026-08-001", "EXP2026-08-002", "EXP2026-08-003"]
+
+
+def test_devuelve_el_folio_descompuesto(db):
+    folio, year, month, sequence = next_folio(db, TENANT_ID, COMPANY_ID, on=date(2026, 8, 15))
+    assert (folio, year, month, sequence) == ("EXP2026-08-001", 2026, 8, 1)
+
+
+def test_el_consecutivo_reinicia_cada_mes(db):
+    next_folio(db, TENANT_ID, COMPANY_ID, on=date(2026, 8, 15))
+    next_folio(db, TENANT_ID, COMPANY_ID, on=date(2026, 8, 16))
+    septiembre, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=date(2026, 9, 1))
+    assert septiembre == "EXP2026-09-001"
+
+    # Y volver a agosto sigue donde se quedó: el contador es por (tenant, company, mes).
+    agosto, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=date(2026, 8, 20))
+    assert agosto == "EXP2026-08-003"
+
+
+def test_cada_company_lleva_su_propio_consecutivo(db):
+    on = date(2026, 8, 15)
+    a1, *_ = next_folio(db, TENANT_ID, 1, on=on)
+    b1, *_ = next_folio(db, TENANT_ID, 2, on=on)
+    a2, *_ = next_folio(db, TENANT_ID, 1, on=on)
+    assert a1 == "EXP2026-08-001"
+    assert b1 == "EXP2026-08-001"  # company 2 arranca de cero
+    assert a2 == "EXP2026-08-002"
+
+
+def test_next_folio_no_commitea_y_el_rollback_deja_hueco(db):
+    """Un rollback tiene que poder deshacer el folio, y el siguiente AVANZA igual.
+
+    Es la razón de que ``next_folio`` no commitee: sus llamadores lo invocan dentro de la
+    transacción del alta. Los huecos en la secuencia son aceptables; los duplicados no.
+    """
+    on = date(2026, 8, 15)
+    primero, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    assert primero == "EXP2026-08-001"
+
+    db.rollback()
+    assert peek_last_sequence(db, TENANT_ID, COMPANY_ID, on=on) == 0
+
+    segundo, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    db.commit()
+    tercero, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    db.rollback()
+    cuarto, *_ = next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    # El tercero se revirtió; el cuarto vuelve a tomar ese número. Lo que importa es que NUNCA
+    # convivan dos filas con el mismo, y de eso se encarga uq_crm_expedientes_periodo_seq.
+    assert segundo == "EXP2026-08-001"
+    assert cuarto == tercero
+
+
+def test_peek_no_mueve_el_contador(db):
+    on = date(2026, 8, 15)
+    assert peek_last_sequence(db, TENANT_ID, COMPANY_ID, on=on) == 0
+    next_folio(db, TENANT_ID, COMPANY_ID, on=on)
+    assert peek_last_sequence(db, TENANT_ID, COMPANY_ID, on=on) == 1
+    assert peek_last_sequence(db, TENANT_ID, COMPANY_ID, on=on) == 1
+
+
+@pytest.mark.skip(
+    reason="Requiere PostgreSQL de verdad: la concurrencia del ON CONFLICT no se puede "
+    "ejercitar en el SQLite en memoria de conftest, que tiene un solo escritor."
+)
+def test_n_sesiones_concurrentes_dan_n_folios_distintos():
+    """N sesiones pidiendo folio a la vez → N folios distintos, sin duplicados.
+
+    Es LA prueba del asignador, y solo dice algo contra PostgreSQL: el ON CONFLICT DO UPDATE
+    serializa sobre la fila del contador y cada sesión recibe su propio valor. En SQLite en memoria
+    con StaticPool hay un único escritor, así que pasaría por construcción y no probaría nada.
+
+    Para correrla: apuntar a la base de e2e (docker-compose.e2e.yml) y abrir N sesiones reales.
+    """
--- a/backend/tests/test_expedientes.py
+++ b/backend/tests/test_expedientes.py
@@ -0,0 +1,193 @@
+"""Pruebas de la entidad expediente y de su enganche con la solicitud de servicio."""
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from api.v1.modules.crm.expedientes import service as expedientes_service
+from api.v1.modules.crm.expedientes.models import Expediente
+from api.v1.modules.crm.service_requests import service as sr_service
+from api.v1.modules.crm.service_requests.dto import ServiceRequestCreate
+from tests.conftest import COMPANY_ID, TENANT_ID
+
+OTRO_TENANT = 99
+OTRA_COMPANY = 77
+
+
+def _crear_solicitud(db, **kwargs) -> object:
+    payload = ServiceRequestCreate(operation_type=kwargs.pop("operation_type", "importacion"), **kwargs)
+    return sr_service.create_service_request(db, payload, TENANT_ID, COMPANY_ID, "user-1")
+
+
+def test_una_solicitud_nueva_nace_con_su_expediente_y_folio(db):
+    solicitud = _crear_solicitud(db)
+
+    expediente = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+    assert expediente is not None
+    assert expediente.folio.startswith("EXP")
+    assert expediente.sequence == 1
+    assert expediente.status == "abierto"
+    assert expediente.efc_link_state == "PENDING"
+
+
+def test_el_expediente_nace_con_su_storage_token_ya_asignado(db):
+    """El token es la carpeta de MinIO en EFC y es INMUTABLE: se fija al nacer, no al enlazar.
+
+    Si se asignara al momento de crear el provisional en EFC, un documento subido antes de que EFC
+    conteste no sabría bajo qué prefijo va.
+    """
+    solicitud = _crear_solicitud(db)
+    expediente = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+    assert expediente.efc_storage_token == f"CRM-{COMPANY_ID}-{expediente.folio}"
+
+
+def test_ensure_es_idempotente(db):
+    solicitud = _crear_solicitud(db)
+    primero = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+
+    segundo = expedientes_service.ensure_expediente(db, solicitud.id, TENANT_ID, COMPANY_ID, "user-1")
+    tercero = expedientes_service.ensure_expediente(db, solicitud.id, TENANT_ID, COMPANY_ID, "user-1")
+
+    assert primero.id == segundo.id == tercero.id
+    assert primero.folio == segundo.folio == tercero.folio
+    assert len(expedientes_service.list_expedientes(db, TENANT_ID, COMPANY_ID)) == 1
+
+
+def test_dos_solicitudes_reciben_folios_consecutivos(db):
+    a = _crear_solicitud(db)
+    b = _crear_solicitud(db)
+    exp_a = expedientes_service.find_by_service_request(db, a.id, TENANT_ID, COMPANY_ID)
+    exp_b = expedientes_service.find_by_service_request(db, b.id, TENANT_ID, COMPANY_ID)
+    assert exp_b.sequence == exp_a.sequence + 1
+    assert exp_a.folio != exp_b.folio
+
+
+def test_dos_expedientes_con_el_mismo_folio_en_el_mismo_tenant_y_company_revientan(db):
+    """``uq_crm_expedientes_folio`` es la red de seguridad si el contador se corrompe.
+
+    Un folio duplicado tiene que fallar ruidosamente en vez de mezclar dos hilos documentales.
+    """
+    solicitud = _crear_solicitud(db)
+    original = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+
+    duplicado = Expediente(
+        folio=original.folio,
+        period_year=original.period_year,
+        period_month=original.period_month,
+        sequence=original.sequence + 1,  # distinta secuencia: el que debe romper es el folio
+        status="abierto",
+        efc_link_state="PENDING",
+        tenant_id=TENANT_ID,
+        company_id=COMPANY_ID,
+    )
+    db.add(duplicado)
+    with pytest.raises(IntegrityError):
+        db.commit()
+    db.rollback()
+
+
+def test_dos_expedientes_con_la_misma_secuencia_del_mes_revientan(db):
+    """``uq_crm_expedientes_periodo_seq``: el consecutivo es un constraint real, no un parse."""
+    solicitud = _crear_solicitud(db)
+    original = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+
+    duplicado = Expediente(
+        folio=original.folio + "-BIS",  # distinto folio: el que debe romper es (año, mes, seq)
+        period_year=original.period_year,
+        period_month=original.period_month,
+        sequence=original.sequence,
+        status="abierto",
+        efc_link_state="PENDING",
+        tenant_id=TENANT_ID,
+        company_id=COMPANY_ID,
+    )
+    db.add(duplicado)
+    with pytest.raises(IntegrityError):
+        db.commit()
+    db.rollback()
+
+
+def test_el_mismo_folio_en_otra_company_si_puede_existir(db):
+    """El folio es único por ``(tenant, company)``, no globalmente.
+
+    Por eso el ``storage_token`` que viaja a EFC lleva el company_id: allá sí comparten organización.
+    """
+    solicitud = _crear_solicitud(db)
+    original = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+
+    gemelo = Expediente(
+        folio=original.folio,
+        period_year=original.period_year,
+        period_month=original.period_month,
+        sequence=original.sequence,
+        status="abierto",
+        efc_link_state="PENDING",
+        tenant_id=TENANT_ID,
+        company_id=OTRA_COMPANY,
+    )
+    db.add(gemelo)
+    db.commit()
+    assert gemelo.id is not None
+
+
+def test_un_expediente_no_se_ve_desde_otro_tenant_ni_otra_company(db):
+    solicitud = _crear_solicitud(db)
+    expediente = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+
+    assert expedientes_service.list_expedientes(db, OTRO_TENANT, COMPANY_ID) == []
+    assert expedientes_service.list_expedientes(db, TENANT_ID, OTRA_COMPANY) == []
+
+    from fastapi import HTTPException
+
+    with pytest.raises(HTTPException) as exc:
+        expedientes_service.get_expediente(db, expediente.id, OTRO_TENANT, COMPANY_ID)
+    assert exc.value.status_code == 404
+
+
+def test_completar_guarda_la_data_aduanera_y_cambia_el_estado(db):
+    from datetime import date
+
+    from api.v1.modules.crm.expedientes.dto import ExpedienteCompleteInput
+
+    solicitud = _crear_solicitud(db)
+    expediente = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+
+    # Datos dummy: patente 0000, aduana 000, pedimento 0000-0000000, RFC XAXX010101000.
+    completado = expedientes_service.complete_expediente(
+        db,
+        expediente.id,
+        ExpedienteCompleteInput(
+            patente="0000",
+            aduana="000",
+            numero_pedimento="0000000",
+            anio=2026,
+            clave_pedimento="A1",
+            regimen="IMD",
+            fecha_pago=date(2026, 8, 1),
+            rfc_importador="XAXX010101000",
+        ),
+        TENANT_ID,
+        COMPANY_ID,
+        "user-1",
+    )
+
+    assert completado.status == "completado"
+    assert completado.patente == "0000"
+    assert completado.aduana == "000"
+    assert completado.rfc_importador == "XAXX010101000"
+    # El folio y el token NO cambian al completar: es lo que permite que ningún archivo se mueva.
+    assert completado.folio == expediente.folio
+    assert completado.efc_storage_token == expediente.efc_storage_token
+
+
+def test_completar_dos_veces_da_409(db):
+    from api.v1.modules.crm.expedientes.dto import ExpedienteCompleteInput
+    from fastapi import HTTPException
+
+    solicitud = _crear_solicitud(db)
+    expediente = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID)
+    payload = ExpedienteCompleteInput(patente="0000", aduana="000", numero_pedimento="0000000", anio=2026)
+
+    expedientes_service.complete_expediente(db, expediente.id, payload, TENANT_ID, COMPANY_ID)
+    with pytest.raises(HTTPException) as exc:
+        expedientes_service.complete_expediente(db, expediente.id, payload, TENANT_ID, COMPANY_ID)
+    assert exc.value.status_code == 409