feat(fin): entrega al expediente EFC los XML enviado y recibido del timbrado

Al timbrar con éxito, el CFDI transmitido al PAC y el que contestó se encolan hacia
el expediente electrónico. El expediente ya existe: nace con la oportunidad y la
factura lo hereda vía fin.invoices.case_id.

Reusa el carril que ya estaba armado (outbox transaccional + worker con reintentos);
este es su primer consumidor en producción.

Decisiones:
- efc_tipo = factura_venta. La lista de tipos está duplicada a mano en este repo y en
  EFC (TIPOS_DOCUMENTO_CRM), y una clave que solo exista de este lado se rechaza allá.
  Los tres archivos se distinguen por nombre: FAC-*.pdf, CFDI-<uuid>-envio.xml,
  CFDI-<uuid>-respuesta.xml.
- Dos kinds (cfdi_request / cfdi_response) y no uno: la guarda de idempotencia es
  (source_table, source_id, kind) y los dos XML comparten source_id —el id del
  intento—, así que un kind común dejaría el par a medias en silencio.
- delete_local=False, a diferencia de los documentos que sube el usuario: el XML
  timbrado es el comprobante fiscal y el CRM lo sirve por /stamp/xml-url.
- Solo el intento que obtuvo timbre. Los rechazos quedan en el CRM y se consultan por
  /stamp/attempts/{id}/xml-url.

El encolado va en la transacción del timbre y el despacho después del commit. Nada de
esto puede propagar: un CFDI ya válido ante el SAT no se cae porque EFC esté apagado.

Ajusta test_fin_sat_catalogs al nuevo conteo de regímenes (19 -> 22) y fija ahí que el
616 es de persona física.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/api/v1/modules/crm/expediente_gateway/models.py

@@ -25,6 +25,12 @@ KIND_COMPLETAR = "completar"
 
 # Tipos del outbox de ARCHIVOS (efc_file_outbox).
 FILE_KIND_DOCUMENTO = "documento"
+# Los dos XML del timbrado. Son kinds SEPARADOS y no un solo 'cfdi', porque la guarda de
+# idempotencia es (source_table, source_id, kind): los dos XML de un mismo timbre comparten
+# source_id —el id del intento—, así que con un kind común la entrega del segundo se saltaría
+# para siempre en cuanto el primero quedara 'sent'.
+FILE_KIND_CFDI_REQUEST = "cfdi_request"
+FILE_KIND_CFDI_RESPONSE = "cfdi_response"
 
 # Tablas de origen posibles de un archivo. El CRM tiene DOS tablas de documentos con secuencias
 # independientes, así que `source_id` por sí solo es ambiguo: crm.documents.id = 5 y
@@ -32,6 +38,9 @@ FILE_KIND_DOCUMENTO = "documento"
 SOURCE_CRM_DOCUMENTS = "crm.documents"
 SOURCE_OPS_SHIPMENT_DOCUMENTS = "ops.shipment_documents"
 SOURCE_FIN_INVOICES = "fin.invoices"
+# El origen de los XML del timbrado es el INTENTO (fin.invoice_stamps), no la factura: una
+# factura puede acumular varios intentos y el par enviado/recibido pertenece a uno concreto.
+SOURCE_FIN_INVOICE_STAMPS = "fin.invoice_stamps"
 
 # Estados (columna status).
 STATUS_PENDING = "pending"

backend/api/v1/modules/crm/expediente_gateway/service.py

@@ -420,6 +420,19 @@ def _dispatch_file_delivery(outbox_id: int, tenant_id: int, company_id: int) ->
         )
 
 
+def dispatch_file_delivery(outbox_id: int, tenant_id: int, company_id: int) -> None:
+    """Despacha la entrega de una fila ya COMMITEADA del outbox de archivos.
+
+    Está separado de ``enqueue_file_best_effort`` porque esa función no commitea: la fila viaja en
+    la transacción de quien la origina, y despachar antes del commit haría que el worker buscara
+    una fila que todavía no existe. El orden es siempre encolar → commit → despachar.
+
+    No despachar no pierde nada: ``sweep_file_outbox`` recoge lo que quede en ``pending``. Esto
+    solo acelera la entrega del caso normal.
+    """
+    _dispatch_file_delivery(outbox_id, tenant_id, company_id)
+
+
 def deliver_file_row(db: Session, row: EfcFileOutbox, client: Optional[EfcClient] = None) -> None:
     """Sube el archivo de ``row.s3_key`` al expediente de EFC y, si ``delete_local``, borra la copia.
 

backend/api/v1/modules/fin/stamping/service.py

@@ -44,6 +44,12 @@ logger = logging.getLogger(__name__)
 CFDI_NS = "http://www.sat.gob.mx/cfd/4"
 TFD_NS = "http://www.sat.gob.mx/TimbreFiscalDigital"
 
+# Tipo con el que los XML del timbrado entran al expediente de EFC. Se reusa el de la factura en
+# vez de inventar uno: la lista de tipos está duplicada a mano en este repo y en EFC
+# (``TIPOS_DOCUMENTO_CRM``), y una clave que solo exista de este lado se rechaza allá. Los tres
+# archivos del CFDI —PDF, XML enviado y XML recibido— se distinguen por su nombre de archivo.
+_EFC_TIPO_CFDI = "factura_venta"
+
 
 # --------------------------------------------------------------------------------------
 # Lectura
@@ -455,11 +461,119 @@ def stamp_invoice(
         # timbre ni provocar un retimbrado. Se guarda el registro sin la clave y se anota.
         stamp.error_message = f"Timbrado correcto, pero no se pudo guardar el XML: {exc}"
 
+    # Las filas del outbox van en ESTA transacción, junto con el timbre: así no puede quedar un
+    # CFDI timbrado sin su intención de entrega al expediente, ni al revés.
+    pendientes = _encolar_xml_al_expediente(db, invoice, stamp)
+
     db.commit()
     db.refresh(stamp)
+
+    # Después del commit: el worker necesita encontrar las filas ya existentes.
+    _despachar_xml_al_expediente(pendientes, tenant_id, company_id)
     return stamp
 
 
+def _encolar_xml_al_expediente(db: Session, invoice: Invoice, stamp: InvoiceStamp) -> list[int]:
+    """Encola hacia el expediente de EFC el XML transmitido al PAC y el que contestó.
+
+    Devuelve los ids de las filas del outbox, para despacharlas después del commit.
+
+    Se entrega el par del intento que SÍ obtuvo timbre; los rechazados quedan en el CRM y se
+    consultan por ``/stamp/attempts/{id}/xml-url``. Un expediente fiscal con los comprobantes que
+    el PAC rechazó no aporta respaldo, solo ruido.
+
+    ``delete_local=False`` a diferencia de los documentos que sube el usuario: el XML timbrado es
+    el comprobante fiscal y el CRM lo sirve por ``/stamp/xml-url``. El corte directo que borra la
+    copia local aplica a un documento cuya única razón de existir es vivir en el expediente; aquí
+    dejaría esos endpoints apuntando a un objeto inexistente.
+
+    Best-effort de punta a punta: si EFC está apagado o el encolado falla, el timbre ya es válido
+    ante el SAT y no puede caerse por esto. Por eso NUNCA propaga: ``enqueue_file_best_effort`` ya
+    se protege con un SAVEPOINT, pero todo lo que rodea a la llamada —resolver el expediente,
+    armar los nombres— también tiene que ser incapaz de tumbar un CFDI ya timbrado.
+    """
+    try:
+        return _encolar_xml_al_expediente_inner(db, invoice, stamp)
+    except Exception:  # noqa: BLE001
+        logger.exception(
+            "timbrado: falló el encolado de los XML del intento %s hacia EFC; el timbre no se toca",
+            stamp.id,
+        )
+        return []
+
+
+def _encolar_xml_al_expediente_inner(
+    db: Session, invoice: Invoice, stamp: InvoiceStamp
+) -> list[int]:
+    """Cuerpo de ``_encolar_xml_al_expediente``; ver ahí el contrato y el porqué."""
+    from ...crm.expediente_gateway import service as gateway  # noqa: PLC0415
+    from ...crm.expediente_gateway.doc_types import is_valid_doc_type  # noqa: PLC0415
+    from ...crm.expediente_gateway.models import (  # noqa: PLC0415
+        FILE_KIND_CFDI_REQUEST,
+        FILE_KIND_CFDI_RESPONSE,
+        SOURCE_FIN_INVOICE_STAMPS,
+    )
+
+    # El expediente nace con la oportunidad y se hereda vía ``case_id``. Una factura suelta —
+    # capturada sin pasar por el ciclo comercial— no tiene a dónde entregar, y eso no es un error.
+    if not invoice.case_id:
+        logger.info(
+            "timbrado: la factura %s no tiene expediente (case_id nulo); no se entregan los XML a EFC",
+            invoice.id,
+        )
+        return []
+
+    # El tipo viaja al catálogo GLOBAL de EFC, compartido por todas las organizaciones. Se valida
+    # contra el set cerrado para no crear ahí un tipo basura que nadie limpia después.
+    if not is_valid_doc_type(_EFC_TIPO_CFDI):
+        logger.error(
+            "timbrado: %r no está en el catálogo de tipos que EFC acepta; no se entregan los XML",
+            _EFC_TIPO_CFDI,
+        )
+        return []
+
+    partes = (
+        (FILE_KIND_CFDI_REQUEST, stamp.request_xml_file_key, "envio", "CFDIREQ"),
+        (FILE_KIND_CFDI_RESPONSE, stamp.response_xml_file_key, "respuesta", "CFDIRES"),
+    )
+
+    filas: list[int] = []
+    for kind, s3_key, sufijo, prefijo_ref in partes:
+        if not s3_key:
+            # El almacenamiento falló al guardar el intento: no hay objeto que entregar.
+            logger.warning(
+                "timbrado: el intento %s no tiene XML de %s guardado; no se entrega a EFC",
+                stamp.id, sufijo,
+            )
+            continue
+        row = gateway.enqueue_file_best_effort(
+            db,
+            kind=kind,
+            s3_key=s3_key,
+            file_name=f"CFDI-{stamp.uuid}-{sufijo}.xml",
+            content_type="application/xml",
+            efc_tipo=_EFC_TIPO_CFDI,
+            source_table=SOURCE_FIN_INVOICE_STAMPS,
+            source_id=stamp.id,
+            crm_document_ref=f"{prefijo_ref}-{stamp.company_id}-{stamp.id}",
+            expediente_ref=invoice.case_id,
+            tenant_id=stamp.tenant_id,
+            company_id=stamp.company_id,
+            delete_local=False,
+        )
+        if row is not None:
+            filas.append(row.id)
+    return filas
+
+
+def _despachar_xml_al_expediente(outbox_ids: list[int], tenant_id: int, company_id: int) -> None:
+    """Despacha las filas ya commiteadas. Lo que no se despache lo recoge el sweep del beat."""
+    from ...crm.expediente_gateway import service as gateway  # noqa: PLC0415
+
+    for outbox_id in outbox_ids:
+        gateway.dispatch_file_delivery(outbox_id, tenant_id, company_id)
+
+
 def _store_attempt_xml(stamp: InvoiceStamp, sent: bytes, received: str) -> None:
     """Guarda el par enviado/recibido del intento y anota sus claves en ``stamp``.
 

backend/tests/test_efc_cfdi_xml.py

@@ -0,0 +1,215 @@
+"""Entrega al expediente de EFC de los dos XML del timbrado.
+
+Lo que se fija aquí es el encolado: qué se manda, con qué tipo, y —lo más frágil— que los DOS
+XML del mismo timbre lleguen. Comparten ``source_id`` (el id del intento), así que un solo
+``kind`` para ambos haría que la guarda de idempotencia se comiera el segundo.
+
+No se prueba la subida a EFC: eso es ``deliver_file_row``, ya cubierto en las pruebas del outbox.
+"""
+
+import pytest
+
+from api.v1.modules.crm.cases import service as cases_service
+from api.v1.modules.crm.expediente_gateway.models import (
+    FILE_KIND_CFDI_REQUEST,
+    FILE_KIND_CFDI_RESPONSE,
+    SOURCE_FIN_INVOICE_STAMPS,
+    STATUS_SENT,
+    EfcFileOutbox,
+)
+from api.v1.modules.fin.invoices.models import Invoice
+from api.v1.modules.fin.stamping import service as stamping
+from api.v1.modules.fin.stamping.models import STATUS_STAMPED, InvoiceStamp
+from tests.conftest import COMPANY_ID, TENANT_ID
+
+UUID = "11111111-2222-3333-4444-555555555555"
+KEY_REQUEST = "tenants/1/companies/1/fin-invoices/7/stamps/attempts/3-request.xml"
+KEY_RESPONSE = "tenants/1/companies/1/fin-invoices/7/stamps/attempts/3-response.xml"
+
+
+@pytest.fixture()
+def efc_encendido(monkeypatch):
+    from core.config import settings
+
+    monkeypatch.setattr(settings, "EFC_API_URL", "https://efc.example.test/", raising=False)
+    return settings
+
+
+def _expediente(db):
+    """Expediente como el que nace con la oportunidad, sin tocar el carril hacia EFC."""
+    from api.v1.modules.crm.expediente_gateway import service as gateway
+
+    # ``create_case`` replica el expediente a EFC; aquí solo interesa la fila local.
+    original = gateway.replicate_expediente_best_effort
+    gateway.replicate_expediente_best_effort = lambda *a, **k: None
+    try:
+        case = cases_service.create_case(
+            db, TENANT_ID, COMPANY_ID, title="Importación de prueba", user_id="user-1"
+        )
+    finally:
+        gateway.replicate_expediente_best_effort = original
+    db.commit()
+    return case
+
+
+def _factura_timbrada(db, *, case_id, request_key=KEY_REQUEST, response_key=KEY_RESPONSE):
+    invoice = Invoice(
+        reference="FAC-001", case_id=case_id, currency="MXN", status="emitida",
+        tenant_id=TENANT_ID, company_id=COMPANY_ID,
+    )
+    db.add(invoice)
+    db.flush()
+    stamp = InvoiceStamp(
+        invoice_id=invoice.id, mode="pruebas", status=STATUS_STAMPED, uuid=UUID,
+        request_xml_file_key=request_key, response_xml_file_key=response_key,
+        tenant_id=TENANT_ID, company_id=COMPANY_ID,
+    )
+    db.add(stamp)
+    db.flush()
+    return invoice, stamp
+
+
+def _filas(db):
+    return db.query(EfcFileOutbox).order_by(EfcFileOutbox.id).all()
+
+
+# ── El caso normal ───────────────────────────────────────────────────────────
+
+def test_encola_los_dos_xml_del_timbre(db, efc_encendido):
+    case = _expediente(db)
+    invoice, stamp = _factura_timbrada(db, case_id=case.id)
+
+    ids = stamping._encolar_xml_al_expediente(db, invoice, stamp)
+    db.commit()
+
+    assert len(ids) == 2
+    filas = _filas(db)
+    assert [f.kind for f in filas] == [FILE_KIND_CFDI_REQUEST, FILE_KIND_CFDI_RESPONSE]
+    assert [f.s3_key for f in filas] == [KEY_REQUEST, KEY_RESPONSE]
+    assert [f.file_name for f in filas] == [
+        f"CFDI-{UUID}-envio.xml",
+        f"CFDI-{UUID}-respuesta.xml",
+    ]
+    for fila in filas:
+        assert fila.expediente_ref == case.id
+        assert fila.source_table == SOURCE_FIN_INVOICE_STAMPS
+        assert fila.source_id == stamp.id
+        assert fila.content_type == "application/xml"
+        assert fila.efc_tipo == "factura_venta"
+
+
+def test_el_tipo_es_uno_que_efc_acepta(db, efc_encendido):
+    """El tipo viaja al catálogo GLOBAL de EFC: una clave inventada lo contamina."""
+    from api.v1.modules.crm.expediente_gateway.doc_types import is_valid_doc_type
+
+    assert is_valid_doc_type(stamping._EFC_TIPO_CFDI)
+
+
+def test_no_borra_la_copia_local(db, efc_encendido):
+    """El XML timbrado es el comprobante fiscal y el CRM lo sirve por /stamp/xml-url.
+
+    Con ``delete_local=True`` la entrega borraría el objeto de MinIO y esos endpoints quedarían
+    apuntando a algo inexistente.
+    """
+    case = _expediente(db)
+    invoice, stamp = _factura_timbrada(db, case_id=case.id)
+
+    stamping._encolar_xml_al_expediente(db, invoice, stamp)
+    db.commit()
+
+    assert [f.delete_local for f in _filas(db)] == [False, False]
+
+
+def test_referencias_distintas_por_xml(db, efc_encendido):
+    """``crm_document_ref`` es la idempotencia del lado de EFC: un UNIQUE por (org, ref).
+
+    Si los dos XML compartieran la referencia, EFC devolvería el primero al entregar el segundo
+    y el expediente se quedaría con la mitad del par.
+    """
+    case = _expediente(db)
+    invoice, stamp = _factura_timbrada(db, case_id=case.id)
+
+    stamping._encolar_xml_al_expediente(db, invoice, stamp)
+    db.commit()
+
+    refs = [f.crm_document_ref for f in _filas(db)]
+    assert refs == [f"CFDIREQ-{COMPANY_ID}-{stamp.id}", f"CFDIRES-{COMPANY_ID}-{stamp.id}"]
+    assert len(set(refs)) == 2
+
+
+# ── La colisión de kinds ─────────────────────────────────────────────────────
+
+def test_el_xml_ya_entregado_no_bloquea_al_otro(db, efc_encendido):
+    """La razón de que los kinds sean dos y no uno.
+
+    La guarda es ``(source_table, source_id, kind)`` y los dos XML comparten source_id. Con un
+    kind común, tener el de envío en 'sent' haría que el de respuesta no se encolara nunca — y
+    el expediente se quedaría a medias sin un solo error visible.
+    """
+    case = _expediente(db)
+    invoice, stamp = _factura_timbrada(db, case_id=case.id)
+
+    db.add(
+        EfcFileOutbox(
+            kind=FILE_KIND_CFDI_REQUEST, s3_key=KEY_REQUEST, file_name="x.xml",
+            content_type="application/xml", efc_tipo="factura_venta",
+            source_table=SOURCE_FIN_INVOICE_STAMPS, source_id=stamp.id,
+            expediente_ref=case.id, status=STATUS_SENT, delete_local=False,
+            tenant_id=TENANT_ID, company_id=COMPANY_ID,
+        )
+    )
+    db.commit()
+
+    stamping._encolar_xml_al_expediente(db, invoice, stamp)
+    db.commit()
+
+    nuevas = [f for f in _filas(db) if f.status != STATUS_SENT]
+    assert [f.kind for f in nuevas] == [FILE_KIND_CFDI_RESPONSE]
+
+
+# ── Los casos en que no hay nada que entregar ────────────────────────────────
+
+def test_factura_sin_expediente_no_encola(db, efc_encendido):
+    """Una factura capturada fuera del ciclo comercial no tiene expediente. No es un error."""
+    invoice, stamp = _factura_timbrada(db, case_id=None)
+
+    assert stamping._encolar_xml_al_expediente(db, invoice, stamp) == []
+    assert _filas(db) == []
+
+
+def test_sin_xml_guardado_se_encola_solo_el_que_existe(db, efc_encendido):
+    """Si el almacenamiento falló al guardar el intento, no hay objeto que subir."""
+    case = _expediente(db)
+    invoice, stamp = _factura_timbrada(db, case_id=case.id, response_key=None)
+
+    ids = stamping._encolar_xml_al_expediente(db, invoice, stamp)
+    db.commit()
+
+    assert len(ids) == 1
+    assert [f.kind for f in _filas(db)] == [FILE_KIND_CFDI_REQUEST]
+
+
+def test_un_fallo_del_carril_no_tumba_el_timbre(db, efc_encendido, monkeypatch):
+    """El CFDI ya es válido ante el SAT: nada del carril hacia EFC puede propagar."""
+    from api.v1.modules.crm.expediente_gateway import service as gateway
+
+    def _explota(*_a, **_k):
+        raise RuntimeError("EFC en llamas")
+
+    monkeypatch.setattr(gateway, "enqueue_file_best_effort", _explota)
+    case = _expediente(db)
+    invoice, stamp = _factura_timbrada(db, case_id=case.id)
+
+    assert stamping._encolar_xml_al_expediente(db, invoice, stamp) == []
+
+
+def test_con_efc_apagado_no_encola(db, monkeypatch):
+    """Sin ``EFC_API_URL`` el carril es un no-op y el timbrado sigue igual."""
+    from core.config import settings
+
+    monkeypatch.setattr(settings, "EFC_API_URL", "", raising=False)
+    case = _expediente(db)
+    invoice, stamp = _factura_timbrada(db, case_id=case.id)
+
+    assert stamping._encolar_xml_al_expediente(db, invoice, stamp) == []
+    assert _filas(db) == []

backend/tests/test_fin_sat_catalogs.py

@@ -68,7 +68,7 @@ def _concept_payload(db, code: str = "FLETE-MAR", ps_code: str = "78101600") ->
 # ---------- Catálogos del SAT: lectura ----------
 
 CATALOG_EXPECTATIONS = [
-    ("tax-regimes", 19, "601"),
+    ("tax-regimes", 22, "601"),
     ("taxes", 3, "002"),
     ("payment-forms", 22, "03"),
     ("units-of-measure", 21, "H87"),
@@ -111,6 +111,15 @@ def test_tax_regimes_person_type_excludes_individual_only(client):
     fisica_codes = [r["code"] for r in fisica]
     assert "605" in fisica_codes and "601" not in fisica_codes
 
+    # Claves con vigencia 2024: el 628 es solo de personas morales y el 630 solo de físicas.
+    assert "628" in codes and "628" not in fisica_codes
+    assert "630" in fisica_codes and "630" not in codes
+
+    # 616 «Sin obligaciones fiscales» es de persona física en el catálogo del SAT: una cuenta
+    # marcada como moral NO debe ofrecerlo. Se fija porque es justo lo que se reporta como
+    # «falta el 616» cuando en realidad está bien filtrado.
+    assert "616" in fisica_codes and "616" not in codes
+
 
 def test_products_services_limit_caps_results(client):
     assert len(client.get("/fin/catalogs/products-services", params={"limit": 3}).json()) == 3