0009. Captura centralizada de auditoría en el backend (FastAPI)

Fecha: 2026-09-01 Última actualización: 2026-09-10 Estado: Propuesto Módulo(s) afectado(s): todos — transversal (FastAPI middleware, servicios, persistencia)

Contexto

El ADR 0008 definió el modelo de auditoría (audit_log, cfg_menu, cfg_menu_acciones) y dónde se persiste (una tabla por producto). El ADR 0013 definió el modelo de navegación (cat_aplicaciones, cfg_modulos, cfg_menu, cfg_menu_acciones). Falta definir cómo se captura el movimiento en el backend de forma centralizada y con garantías:

• Que no dependa de que cada desarrollador recuerde insertar uno a uno.
• Que no rompa el patrón Router → Service → Repository (ADR 0001).
• Que no bloquee la transacción de negocio.
• Que cubra el 100% de los endpoints sin código manual.
• Que resuelva automáticamente el módulo, la página y la acción.

Problema del enfoque por decorador: un decorador @audit(menu_id=..., accion_id=...) en cada endpoint obliga al desarrollador a conocer los códigos del catálogo, escribirlo en cada ruta, y es propenso a olvidos. Si el endpoint no lleva el decorador, no se registra nada → huecos de auditoría.

Decisión

La auditoría se captura desde un middleware HTTP global que intercepta todas las peticiones y registra automáticamente: método, endpoint, actor, empresa, módulo, página y acción. El desarrollador no escribe código de auditoría en el 90% de los casos (CRUD estándar). Solo usa un decorador opcional cuando necesita enriquecer el registro (acción especial, lectura sensible, datos de negocio).

Flujo de captura

Request HTTP → MiddlewareAudit (global, ANTES del Router)
    │
    ├─ 1. CAPTURA AUTOMÁTICA (sin código del desarrollador):
    │    • metodo_http       (del Request: GET, POST, PUT, DELETE, PATCH)
    │    • endpoint          (del Request: /api/v1/inventario/ingresos)
    │    • actor_user_id     (del JWT: sub)
    │    • actor_username    (del JWT: username)
    │    • actor_nombre      (del JWT: nombre)
    │    • actor_email       (del JWT: email)
    │    • empresa_id        (del JWT: empresa_id)
    │    • fecha             (now())
    │    • trace_id          (del header X-Request-ID o generado)
    │
    ├─ 2. RESOLUCIÓN DEL MÓDULO (mapping por servicio):
    │    • Mapping cargado al startup desde MODULE_MAP
    │    • /api/v1/facturas → modulo_codigo="contabilidad"
    │    • /api/v1/pedidos  → modulo_codigo="pedidos"
    │    • Resultado: modulo_id, modulo_codigo, modulo_nombre
    │
    ├─ 3. RESOLUCIÓN DEL CATÁLOGO (caché en memoria):
    │    • Al startup carga cfg_menu + cfg_menu_acciones de core_sigfa
    │    • Mapea (method, endpoint) → menu_id, accion_id, menu_codigo, etc.
    │    • POST /api/v1/facturas → accion_codigo="crear"
    │    • Refresh periódico cada 5 minutos
    │
    ├─ 4. DECORADOR OPCIONAL (solo para enriquecer):
    │    • @audit(accion_codigo="aprobar")   ← override de POST→crear
    │    • @audit(referencia_tipo="pedido")   ← tipo de entidad
    │    • @audit(sensitive=True)             ← lectura sensible
    │    • @audit(detalle="aprobó por $500")  ← contexto de negocio
    │
    ├─ 5. EJECUTA EL REQUEST (Router → Service → Repository)
    │
    └─ 6. AuditService.grabar():
         • construye el registro audit_log completo
         • encola evento (outbox, ADR 0003) — no bloquea la respuesta
         • un consumidor persiste en audit_log de la BD del producto

El middleware (captura automática)

El middleware se registra una vez en main.py de cada servicio y captura todas las peticiones HTTP entrantes. No conoce lógica de negocio; solo extrae datos del request, del JWT y del catálogo cacheado.

# svc-contable/app/main.py
from fastapi import FastAPI
from app.audit.middleware import AuditMiddleware

app = FastAPI()

# Se registra ANTES de los routers
app.add_middleware(AuditMiddleware)

# svc-contable/app/audit/middleware.py
import uuid
from datetime import datetime, timezone
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from app.audit.module_map import MODULE_MAP
from app.audit.catalog_cache import CatalogCache
from app.audit.service import AuditService


class AuditMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, catalog_cache: CatalogCache | None = None):
        super().__init__(app)
        self.module_map = MODULE_MAP
        self.catalog = catalog_cache or CatalogCache()

    async def dispatch(self, request: Request, call_next) -> Response:
        # ── 1. Datos automáticos del request ──
        audit_data = {
            "metodo_http": request.method,
            "endpoint": str(request.url.path),
            "fecha": datetime.now(timezone.utc),
            "trace_id": self._extract_trace_id(request),
        }

        # ── 2. Datos del JWT (ya decodificado en el middleware de auth) ──
        token = getattr(request.state, "token_payload", None)
        if token:
            audit_data.update({
                "actor_user_id": uuid.UUID(token["sub"]),
                "actor_username": token.get("username"),
                "actor_nombre": token.get("nombre"),
                "actor_email": token.get("email"),
                "empresa_id": token.get("empresa_id"),
            })

        # ── 3. Resolución del módulo (mapping del servicio) ──
        modulo = self._resolve_modulo(request.url.path)
        if modulo:
            audit_data.update({
                "modulo_id": modulo["id"],
                "modulo_codigo": modulo["codigo"],
                "modulo_nombre": modulo["nombre"],
            })

        # ── 4. Resolución del catálogo (menú y acción) ──
        catalog_entry = self.catalog.resolve(request.method, request.url.path)
        if catalog_entry:
            audit_data.update(catalog_entry)

        # ── 5. Ejecuta el request ──
        response = await call_next(request)

        # ── 6. Graba auditoría (async, no bloquea) ──
        audit_data["status_code"] = response.status_code
        await AuditService.grabar(audit_data)

        return response

    def _resolve_modulo(self, path: str) -> dict | None:
        """Match por prefijo más largo: /api/v1/facturas/123 → /api/v1/facturas"""
        sorted_prefixes = sorted(self.module_map.keys(), key=len, reverse=True)
        for prefix in sorted_prefixes:
            if path.startswith(prefix):
                return self.module_map[prefix]
        return None

    def _extract_trace_id(self, request: Request) -> uuid.UUID | None:
        header = request.headers.get("X-Request-ID")
        if header:
            try:
                return uuid.UUID(header)
            except ValueError:
                pass
        return uuid.uuid4()

Mapping por servicio (MODULE_MAP)

Cada microservicio declara una vez qué módulo(s) maneja. Este mapping conecta los prefijos de los endpoints con los códigos de cfg_modulos (el mismo codigo del seed definido en ADR 0013 de svc-identidad).

¿Por qué un mapping y no auto-detectar del path? Porque:

• El path del API (/api/v1/facturas) no siempre coincide con el código del módulo en cfg_modulos (contabilidad).
• Un servicio puede manejar múltiples módulos (svc-rrhh maneja rrhh).
• Es explícito: si agregás un endpoint nuevo, el mapping te obliga a declararlo — sin olvidos.

# svc-contable/app/audit/module_map.py
"""
Mapping de endpoints → módulos para svc-contable.

Claves: prefijo del path (lo que viene después de /api/v1)
Valores: {id, codigo, nombre} del módulo en cfg_modulos

El id se carga al startup desde core_sigfa (caché en memoria).
El código es el mismo del seed (ADR 0013 de svc-identidad).
"""

MODULE_MAP = {
    "/api/v1/facturas": {
        "id": None,  # se carga al startup desde core_sigfa
        "codigo": "contabilidad",
        "nombre": "Contabilidad",
    },
    "/api/v1/cuentas_por_cobrar": {
        "id": None,
        "codigo": "contabilidad",
        "nombre": "Contabilidad",
    },
    "/api/v1/caja": {
        "id": None,
        "codigo": "contabilidad",
        "nombre": "Contabilidad",
    },
}

# svc-operaciones/app/audit/module_map.py
MODULE_MAP = {
    "/api/v1/inventario": {
        "id": None,
        "codigo": "inventario",
        "nombre": "Inventario",
    },
    "/api/v1/produccion": {
        "id": None,
        "codigo": "produccion",
        "nombre": "Producción",
    },
    "/api/v1/calidad": {
        "id": None,
        "codigo": "calidad",
        "nombre": "Calidad",
    },
}

# svc-pedidos/app/audit/module_map.py
MODULE_MAP = {
    "/api/v1/pedidos": {
        "id": None,
        "codigo": "pedidos",
        "nombre": "Pedidos",
    },
    "/api/v1/cotizaciones": {
        "id": None,
        "codigo": "ventas",
        "nombre": "Ventas",
    },
}

# svc-rrhh/app/audit/module_map.py
MODULE_MAP = {
    "/api/v1/nomina": {
        "id": None,
        "codigo": "rrhh",
        "nombre": "RRHH",
    },
    "/api/v1/marcaciones": {
        "id": None,
        "codigo": "rrhh",
        "nombre": "RRHH",
    },
    "/api/v1/asistencia": {
        "id": None,
        "codigo": "rrhh",
        "nombre": "RRHH",
    },
    "/api/v1/faena_pesquera": {
        "id": None,
        "codigo": "rrhh",
        "nombre": "RRHH",
    },
}

# svc-clientes/app/audit/module_map.py
MODULE_MAP = {
    "/api/v1/clientes": {
        "id": None,
        "codigo": "clientes",
        "nombre": "Clientes",
    },
    "/api/v1/contactos": {
        "id": None,
        "codigo": "clientes",
        "nombre": "Clientes",
    },
}

# svc-fuerza-ventas/app/audit/module_map.py
MODULE_MAP = {
    "/api/v1/promotores": {
        "id": None,
        "codigo": "promotores",
        "nombre": "Promotores",
    },
    "/api/v1/presupuesto_cliente": {
        "id": None,
        "codigo": "ventas",
        "nombre": "Ventas",
    },
}

Resolución del catálogo (CatalogCache)

El CatalogCache carga al startup las tablas cfg_menu y cfg_menu_acciones de core_sigfa y construye un mapping in-memory que resuelve (method, endpoint) → datos de auditoría.

# svc-contable/app/audit/catalog_cache.py
import uuid
from datetime import datetime, timezone, timedelta
from sqlalchemy import text
from sqlalchemy.orm import Session


class CatalogCache:
    """
    Caché en memoria del catálogo de navegación (cfg_menu + cfg_menu_acciones).

    Se carga al startup desde core_sigfa y se refresca cada 5 minutos.
    Mapea (HTTP method, endpoint path) → {menu_id, accion_id, menu_codigo,
    menu_nombre, accion_codigo, accion_nombre, modulo_id}.

    NO es una tabla local: es una vista cacheada de core_sigfa que se
    reconstruye periódicamente. Si el catálogo cambia, el próximo refresh lo
    refleja.
    """

    REFRESH_INTERVAL = timedelta(minutes=5)

    def __init__(self):
        self._cache: dict[tuple[str, str], dict] = {}
        self._last_refresh: datetime = datetime.min
        self._module_ids: dict[str, uuid.UUID] = {}

    def resolve(self, method: str, path: str) -> dict | None:
        """Resuelve un endpoint a sus datos de auditoría del catálogo."""
        self._maybe_refresh()
        key = (method, self._normalize(path))
        return self._cache.get(key)

    def _normalize(self, path: str) -> str:
        """
        Normaliza el path para matching:
        /api/v1/facturas/123  →  /api/v1/facturas/{id}
        /api/v1/facturas      →  /api/v1/facturas
        """
        parts = path.strip("/").split("/")
        normalized = []
        for i, part in enumerate(parts):
            if i >= 3:  # después de /api/v1/
                if part.isdigit() or (len(part) == 36 and "-" in part):
                    normalized.append("{id}")
                else:
                    normalized.append(part)
            else:
                normalized.append(part)
        return "/" + "/".join(normalized)

    def _maybe_refresh(self):
        now = datetime.now(timezone.utc)
        if now - self._last_refresh > self.REFRESH_INTERVAL:
            self._refresh()

    def _refresh(self):
        """
        Consulta core_sigfa para reconstruir el caché.

        JOIN: cfg_menu (para modulo_id, codigo, nombre)
              + cfg_menu_acciones (para accion)
              + cfg_modulos (para nombre del módulo)

        Filtra por is_visible=true, is_activo=true en ambas tablas.
        """
        # Implementación que consulta core_sigfa vía HTTP o conexión directa
        # y reconstruye self._cache y self._module_ids
        ...
        self._last_refresh = datetime.now(timezone.utc)

Resolución automática de acciones por HTTP method

La regla por defecto (sin decorador) asigna la acción según el HTTP method:

HTTP Method	accion_codigo	accion_nombre	Cuándo
POST	crear	Crear	Cualquier POST
PUT	editar	Editar	Actualización completa
PATCH	editar	Editar	Actualización parcial
DELETE	eliminar	Eliminar	Cualquier DELETE
GET (sin {id})	listar	Listar	Listados/paginación
GET (con {id})	ver	Ver	Consulta de un registro

¿Y las acciones especiales? (aprobar, exportar, conciliar, etc.) Se usan cuando el endpoint no es CRUD estándar:

# POST /api/v1/pedidos/{id}/aprobar → no es "crear", es "aprobar"
@router.post("/api/v1/pedidos/{pedido_id}/aprobar")
@audit(accion_codigo="aprobar", accion_nombre="Aprobar pedido")
def aprobar_pedido(pedido_id: str, ...):
    ...

# GET /api/v1/reportes/gerencial → no es "listar", es "exportar"
@router.get("/api/v1/reportes/gerencial")
@audit(accion_codigo="exportar", accion_nombre="Exportar reporte gerencial",
       sensitive=True)
def exportar_gerencial(...):
    ...

El decorador (opcional — solo enriquecimiento)

El decorador no es obligatorio. Solo se usa cuando el endpoint necesita algo que el middleware no puede deducir automáticamente:

from app.audit import audit

# CASO 1: Acción especial (override de POST→crear)
@router.post("/api/v1/pedidos/{pedido_id}/aprobar")
@audit(accion_codigo="aprobar", accion_nombre="Aprobar pedido")
async def aprobar_pedido(pedido_id: str): ...

# CASO 2: Lectura sensible (GET que normalmente no audita)
@router.get("/api/v1/cartera")
@audit(sensitive=True)
async def ver_cartera(): ...

# CASO 3: Contexto de negocio adicional
@router.put("/api/v1/inventario/productos/{producto_id}")
@audit(
    referencia_tipo="producto",
    detalle="Actualización de precio de 10.00 a 12.50"
)
async def actualizar_producto(producto_id: str, data: ProductoUpdate): ...

# CASO 4: Multiple archivos
@router.post("/api/v1/facturas/{factura_id}/archivos")
@audit(referencia_tipo="factura")
async def subir_archivos(factura_id: str, archivos: list[UploadFile]): ...

Lo que hace el decorador:

• No conoce la lógica de negocio (no sabe qué cambió exactamente).
• Sobreescribe los valores resueltos por el middleware (solo los campos que se especifican; los demás quedan los del middleware).
• Si el endpoint falla con error HTTP, registra el intento con el estado fallido (ej. código de error y mensaje) salvo que el error sea un 401/403 de autenticación (no auditar los rechazos de auth por separado, ya los maneja la capa de seguridad).

# svc-contable/app/audit/__init__.py
import functools
from app.audit.service import AuditService


def audit(
    accion_codigo: str | None = None,
    accion_nombre: str | None = None,
    referencia_tipo: str | None = None,
    sensitive: bool = False,
    detalle: str | None = None,
):
    """
    Decorador opcional para enriquecer la auditoría del middleware.

    Solo overridea los campos que se especifican. Los demás quedan con
    los valores resueltos automáticamente por AuditMiddleware.

    Uso:
        @audit(accion_codigo="aprobar", sensitive=True)
    """
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            request = kwargs.get("request") or _get_request(args)
            if request:
                # Guarda los overrides en request.state para que
                # AuditMiddleware los lea después del request
                request.state.audit_overrides = {
                    k: v for k, v in {
                        "accion_codigo": accion_codigo,
                        "accion_nombre": accion_nombre,
                        "referencia_tipo": referencia_tipo,
                        "sensitive": sensitive,
                        "detalle": detalle,
                    }.items() if v is not None
                }
            return await func(*args, **kwargs)
        return wrapper
    return decorator

Subida de archivos (AuditContext)

Problema: el middleware corre antes del endpoint, pero la información de archivos (nombre, ruta, hash, tamaño) se conoce después de que el endpoint procesa el upload. Además, el middleware no puede leer el body multipart porque lo consumiría y el endpoint no podría recibirlo.

Solución: un AuditContext que el endpoint llena después de procesar el request. El middleware lo lee al final y lo incluye en el registro.

# svc-contable/app/audit/context.py
from dataclasses import dataclass, field
import uuid


@dataclass
class AuditFile:
    """
    Representa un archivo en el contexto de auditoría.

    Solo `nombre` es requerido. Los demás campos son opcionales y dependen
    de lo que el endpoint quiera registrar. El campo `meta` es un dict
    libre para metadata adicional del servicio (ej. versiones, estados,
    lotes, etc.).
    """
    nombre: str
    ruta: str | None = None           # ruta local o URL de S3
    tipo: str | None = None           # MIME type
    hash: str | None = None           # sha256:...
    tamano_bytes: int | None = None
    descripcion: str | None = None    # descripción individual del archivo
    meta: dict | None = None          # metadata libre adicional


@dataclass
class AuditContext:
    """
    Contexto de auditoría que el endpoint llena después de procesar.

    Accesible desde request.state.audit_ctx (inyectado por el middleware).
    El middleware lee este contexto al final y lo incluye en audit_log.

    Ejemplos de uso:
    - Endpoint de upload: ctx.add_archivo(nombre="f-001.pdf", ruta="s3://...")
    - Endpoint de update: ctx.datos_antes = {...}, ctx.datos_despues = {...}
    - Endpoint de delete: ctx.detalle = "Se eliminó el registro X"
    """
    archivos: list[AuditFile] = field(default_factory=list)
    referencia_tipo: str | None = None
    referencia_id: uuid.UUID | None = None
    detalle: str | None = None
    datos_antes: dict | None = None
    datos_despues: dict | None = None

    def add_archivo(
        self,
        nombre: str,
        ruta: str | None = None,
        tipo: str | None = None,
        hash: str | None = None,
        tamano_bytes: int | None = None,
        descripcion: str | None = None,
        meta: dict | None = None,
    ):
        """Helper para agregar un archivo al contexto."""
        self.archivos.append(AuditFile(
            nombre=nombre, ruta=ruta, tipo=tipo,
            hash=hash, tamano_bytes=tamano_bytes,
            descripcion=descripcion, meta=meta,
        ))

    def add_archivos_s3(self, archivos_s3: list[dict]):
        """
        Helper para agregar múltiples archivos desde S3.

        Cada dict debe tener al menos 'nombre' y 'ruta' (URL de S3).
        Ejemplo:
            ctx.add_archivos_s3([
                {"nombre": "f-001.pdf", "ruta": "https://s3.amazonaws.com/...", "descripcion": "Factura"},
                {"nombre": "nc-001.pdf", "ruta": "https://s3.amazonaws.com/...", "descripcion": "Nota de crédito"},
            ])
        """
        for item in archivos_s3:
            self.archivos.append(AuditFile(
                nombre=item["nombre"],
                ruta=item.get("ruta"),
                tipo=item.get("tipo"),
                hash=item.get("hash"),
                tamano_bytes=item.get("tamano_bytes"),
                descripcion=item.get("descripcion"),
                meta=item.get("meta"),
            ))

El middleware inyecta el contexto en request.state y lo lee al final:

# En AuditMiddleware.dispatch(), después del request:
audit_ctx = getattr(request.state, "audit_ctx", AuditContext())
if audit_ctx.archivos:
    audit_data["archivos"] = [
        {k: v for k, v in {
            "nombre": f.nombre,
            "ruta": f.ruta,
            "tipo": f.tipo,
            "hash": f.hash,
            "tamano_bytes": f.tamano_bytes,
            "descripcion": f.descripcion,
            "meta": f.meta,
        }.items() if v is not None}
        for f in audit_ctx.archivos
    ]
if audit_ctx.referencia_tipo:
    audit_data["referencia_tipo"] = audit_ctx.referencia_tipo
if audit_ctx.referencia_id:
    audit_data["referencia_id"] = audit_ctx.referencia_id
if audit_ctx.detalle:
    audit_data["detalle"] = audit_ctx.detalle
if audit_ctx.datos_antes:
    audit_data["datos_antes"] = audit_ctx.datos_antes
if audit_ctx.datos_despues:
    audit_data["datos_despues"] = audit_ctx.datos_despues

El endpoint llena el contexto después de procesar el upload:

# CASO 1: Subida simple — 1 archivo con descripción
@router.post("/api/v1/facturas/{factura_id}/archivos")
@audit(referencia_tipo="factura")
async def subir_archivo(
    factura_id: str,
    archivo: UploadFile,
    request: Request
):
    # 1. Guarda en S3
    url_s3 = await s3_service.subir(archivo, carpeta="facturas")

    # 2. Llena el contexto de auditoría
    ctx: AuditContext = request.state.audit_ctx
    ctx.add_archivo(
        nombre=archivo.filename,
        ruta=url_s3,                              # URL de S3
        tipo=archivo.content_type,
        tamano_bytes=archivo.size,
        descripcion="Factura original del proveedor"  # descripción individual
    )
    ctx.referencia_id = uuid.UUID(factura_id)

    return {"ok": True, "url": url_s3}

# CASO 2: Múltiples archivos, cada uno con su descripción
@router.post("/api/v1/pedidos/{pedido_id}/documentos")
@audit(referencia_tipo="pedido", accion_codigo="subir_archivo")
async def subir_documentos(
    pedido_id: str,
    archivos: list[UploadFile],
    descripciones: list[str],     # el frontend manda las descripciones
    request: Request
):
    ctx: AuditContext = request.state.audit_ctx
    for archivo, desc in zip(archivos, descripciones):
        url_s3 = await s3_service.subir(archivo, carpeta="pedidos")
        ctx.add_archivo(
            nombre=archivo.filename,
            ruta=url_s3,
            tipo=archivo.content_type,
            tamano_bytes=archivo.size,
            descripcion=desc         # cada archivo con su descripción
        )
    ctx.referencia_id = uuid.UUID(pedido_id)
    return {"ok": True}

# CASO 3: Archivos + metadata libre via campo meta
@router.post("/api/v1/calidad/inspecciones/{inspeccion_id}/evidencias")
@audit(referencia_tipo="inspeccion")
async def subir_evidencias(
    inspeccion_id: str,
    archivos: list[UploadFile],
    request: Request
):
    ctx: AuditContext = request.state.audit_ctx
    for archivo in archivos:
        url_s3 = await s3_service.subir(archivo, carpeta="calidad")
        ctx.add_archivo(
            nombre=archivo.filename,
            ruta=url_s3,
            tipo=archivo.content_type,
            tamano_bytes=archivo.size,
            meta={                                    # metadata libre
                "tipo_evidencia": "foto",
                "lote": "L-2026-001",
                "estado_calidad": "pendiente_revision"
            }
        )
    ctx.referencia_id = uuid.UUID(inspeccion_id)
    return {"ok": True}

# CASO 4: Helper para múltiples archivos desde S3
@router.post("/api/v1/compras/{compra_id}/adjuntos")
@audit(referencia_tipo="compra")
async def subir_adjuntos(
    compra_id: str,
    archivos: list[UploadFile],
    request: Request
):
    ctx: AuditContext = request.state.audit_ctx

    # Sube todos a S3 y agrega de una
    archivos_s3 = []
    for archivo in archivos:
        url = await s3_service.subir(archivo, carpeta="compras")
        archivos_s3.append({
            "nombre": archivo.filename,
            "ruta": url,
            "tipo": archivo.content_type,
            "tamano_bytes": archivo.size,
        })

    ctx.add_archivos_s3(archivos_s3)   # helper para bulk
    ctx.referencia_id = uuid.UUID(compra_id)
    return {"ok": True}

# CASO 3: Endpoint SIN archivos (el middleware no pone archivos en audit_log)
@router.put("/api/v1/inventario/productos/{producto_id}")
@audit(referencia_tipo="producto")
async def actualizar_producto(
    producto_id: str,
    data: ProductoUpdate,
    request: Request
):
    # 1. Captura estado anterior
    ctx: AuditContext = request.state.audit_ctx
    anterior = await service.obtener(producto_id)
    ctx.datos_antes = {"precio": anterior.precio}

    # 2. Actualiza
    await service.actualizar(producto_id, data)

    # 3. Captura estado posterior
    ctx.datos_despues = {"precio": data.precio}
    ctx.detalle = f"Precio de {anterior.precio} a {data.precio}"
    ctx.referencia_id = uuid.UUID(producto_id)

    return {"ok": True}

# CASO 5: Service que maneja archivos internamente
# (el repository reporta al contexto vía el service)

# service.py
class FacturaService:
    def __init__(self, db: Session, audit_ctx: AuditContext):
        self.db = db
        self.audit_ctx = audit_ctx

    def subir_adjunto(self, factura_id: str, archivo: UploadFile):
        url_s3 = self._subir_a_s3(archivo, carpeta="facturas")
        # Reporta al contexto de auditoría
        self.audit_ctx.add_archivo(
            nombre=archivo.filename,
            ruta=url_s3,
            tipo=archivo.content_type,
            tamano_bytes=archivo.size,
            descripcion="Adjunto de factura"
        )
        return url_s3

# router.py
@router.post("/api/v1/facturas/{factura_id}/adjuntos")
@audit(referencia_tipo="factura")
async def subir_adjunto(
    factura_id: str,
    archivo: UploadFile,
    request: Request,
    db: Session = Depends(get_db),
):
    ctx: AuditContext = request.state.audit_ctx
    service = FacturaService(db, ctx)
    service.subir_adjunto(factura_id, archivo)
    ctx.referencia_id = uuid.UUID(factura_id)
    return {"ok": True, "url": url_s3}

Regla: si el endpoint no llena el contexto (no hay archivos, ni detalle, ni datos antes/después), el middleware igualmente registra la operación — solo sin los campos de enriquecimiento. La auditoría nunca deja de grabarse por falta de contexto.

Garantías

• Cobertura total: el middleware captura el 100% de las peticiones HTTP. No se escapa ningún endpoint, aunque sea nuevo y no tenga decorator.
• Escritura asíncrona: la persistencia se hace por trabajo/outbox (regla 4 del ADR 0003) para no sumar latencia a la transacción de negocio. El AuditService.grabar() encola un evento; un consumidor lo persiste en audit_log con empresa_id y RLS (ADR 0002).
• Lecturas sensibles: solo ciertos GET se marcan como sensibles (el decorador acepta sensitive=True); el listado común no audita lecturas.
• Sin lógica en cada módulo: el Service de negocio no inserta en audit_log; solo entrega al AuditService qué cambió (detalle, args). La capa de persistencia real del audit es única.
• Los IDs vienen del catálogo (cfg_menu, cfg_menu_acciones del ADR 0008/0013) — nunca se construyen en el endpoint ni se guardan como texto. Los nombres se copian en el registro como snapshot del momento (ADR 0014): el historial se lee con un solo SELECT local, sin consultar core_sigfa (ADR 0003).
• Snapshot del módulo: modulo_id, modulo_codigo y modulo_nombre se copian del mapping al momento de grabar, permitiendo filtrar por módulo sin JOIN a core_sigfa.
• Separación por producto: AuditService escribe solo en la BD del producto que lo invoca (ERP o CRM), sin cruces (ADR 0003).
• Participación del patrón Router → Service → Repository (ADR 0001): el middleware es parte de la capa de infraestructura (antes del Router); el AuditService es un Service; la persistencia ocurre vía Repository del dominio de auditoría.

Ejemplo completo: endpoint CRUD SIN decorador

POST /api/v1/facturas (svc-contable, empresa_id=1)

Middleware resuelve automáticamente:
  ├─ del Request:      metodo_http=POST, endpoint=/api/v1/facturas
  ├─ del JWT:          actor_user_id=uuid, username="d.alcivar",
  │                    nombre="Duval Alcivar", email="d@sigfa.com",
  │                    empresa_id=1
  ├─ del MODULE_MAP:   modulo_codigo="contabilidad",
  │                    modulo_nombre="Contabilidad"
  ├─ del catálogo:     menu_id=uuid("facturas"), accion_id=uuid("crear"),
  │                    menu_codigo="facturas", menu_nombre="Facturación",
  │                    accion_codigo="crear", accion_nombre="Crear
  └─ post-request:     status_code=201

Registro en audit_log:
  modulo_codigo     = "contabilidad"
  modulo_nombre     = "Contabilidad"
  menu_codigo       = "facturas"
  menu_nombre       = "Facturación"
  accion_codigo     = "crear"
  accion_nombre     = "Crear"
  actor_username    = "d.alcivar"
  actor_nombre      = "Duval Alcivar"
  metodo_http       = POST
  endpoint          = /api/v1/facturas
  empresa_id        = 1
  fecha             = 2026-09-10T14:30:00Z

Ejemplo completo: endpoint CON decorador (acción especial)

POST /api/v1/pedidos/abc-123/aprobar (svc-pedidos, empresa_id=1)

Decorador declara:  accion_codigo="aprobar", accion_nombre="Aprobar pedido"
Middleware resuelve:
  ├─ del Request:      metodo_http=POST, endpoint=/api/v1/pedidos/{id}/aprobar
  ├─ del JWT:          actor_user_id=uuid, username="j.garcia", ...
  ├─ del MODULE_MAP:   modulo_codigo="pedidos", modulo_nombre="Pedidos"
  ├─ del catálogo:     menu_id=uuid("pedidos"), menu_codigo="pedidos"
  ├─ del decorador:    accion_codigo="aprobar" (override de POST→crear)
  └─ post-request:     status_code=200

Registro en audit_log:
  modulo_codigo     = "pedidos"
  menu_codigo       = "pedidos"
  accion_codigo     = "aprobar"      ← del decorador, NO "crear"
  accion_nombre     = "Aprobar pedido" ← del decorador
  actor_username    = "j.garcia"
  metodo_http       = POST
  endpoint          = /api/v1/pedidos/abc-123/aprobar
  empresa_id        = 1

Ejemplo completo: lectura sensible

GET /api/v1/cartera (svc-contable, empresa_id=1)

Decorador declara:  sensitive=True
Middleware resuelve:
  ├─ del Request:      metodo_http=GET, endpoint=/api/v1/cartera
  ├─ del JWT:          actor_user_id=uuid, username="admin", ...
  ├─ del MODULE_MAP:   modulo_codigo="contabilidad"
  ├─ del catálogo:     menu_id=uuid("cartera"), menu_codigo="cartera"
  │                    accion_codigo="ver", accion_nombre="Ver"
  ├─ del decorador:    sensitive=True
  └─ post-request:     status_code=200

Resultado: SÍ se registra en audit_log (porque sensitive=True).
Sin el decorador, GET /api/v1/cartera NO se auditaría (lectura común).

Cómo identifica el módulo: el problema del path ambiguo

El path del endpoint no siempre coincide con el módulo en cfg_modulos:

Endpoint	Path	Módulo en cfg_modulos
svc-contable	/api/v1/facturas	contabilidad
svc-contable	/api/v1/cuentas_por_cobrar	contabilidad
svc-rrhh	/api/v1/nomina	rrhh
svc-rrhh	/api/v1/marcaciones	rrhh
svc-operaciones	/api/v1/productos	inventario
svc-pedidos	/api/v1/cotizaciones	ventas

El mapping por servicio resuelve esto: cada servicio sabe a qué módulo pertenece cada endpoint. Es un archivo pequeño (~5-15 líneas) que se carga una vez al startup.

¿Qué pasa si un endpoint no está en el mapping? Se registra sin datos de módulo (modulo_codigo=NULL, modulo_nombre=NULL). El registro de auditoría se crea igual (con actor, endpoint, método, fecha), pero sin la dimensión de módulo. Esto permite que la auditoría nunca deje de funcionar, aunque el mapping esté incompleto.

Cómo se identifica el producto (ERP vs CRM)

Cada servicio ya sabe en qué base de datos escribe (lo tiene en su config.py). El AuditService usa la misma conexión:

# svc-contable/app/audit/service.py
class AuditService:
    @staticmethod
    async def grabar(data: dict):
        """
        Escribe en erp_sigfa (svc-contable siempre es ERP).
        svc-clientes escribe en crm_sigfapro.
        """
        # El servicio ya tiene configurada su BD
        # AuditService usa la misma conexión/replica
        ...

No hace falta identificar el producto en el middleware; cada servicio escribe en su propia base por diseño (ADR 0003).

Alternativas descartadas

• Decorador @audit obligatorio en cada endpoint — descartado: tedioso, propenso a olvidos, obliga al desarrollador a conocer los códigos del catálogo. El middleware automático cubre el 100% sin código manual.
• Middleware HTTP global sin mapping de módulos — descartado: captura actor y endpoint, pero no puede resolver a qué módulo pertenece sin un mapping explícito. El mapping por servicio resuelve esto con un costo mínimo.
• Insertar audit_log manualmente en cada método del Service — descartado: se olvida, se duplica lógica y viola la responsabilidad única.
• Persistir en la misma transacción del negocio — descartado: suma latencia y riesgos de un fail conjunto; la escritura asíncrona aísla el fallo de auditoría del resultado del negocio.
• Triggers de BD — descartado (mismo criterio del ADR 0008: sin contexto de usuario/acción funcional ni página).
• Auto-resolución del módulo por parsing del path — descartado: el path del API no siempre coincide con el código del módulo en cfg_modulos. Un mapping explícito es más confiable y auditado.

Consecuencias

• Ganancia: cobertura total de auditoría sin código manual en el 90% de los endpoints. El middleware captura todo; el decorador solo enriquece.
• Ganancia: una sola forma de auditar en todo el backend → consistencia y regla de revisión en PR: "todo endpoint nuevo debe estar en el MODULE_MAP de su servicio".
• Ganancia: el desarrollador no necesita conocer los códigos del catálogo de navegación para auditar un CRUD estándar.
• Costo: cada servicio debe mantener un MODULE_MAP (~5-15 líneas) con sus endpoints → módulos. Este mapping se revisa en PR como cualquier configuración.
• Costo: el CatalogCache consulta core_sigfa al startup y cada 5 minutos. Si el catálogo cambia mientras el servicio está corriendo, tarda hasta 5 minutos en reflejarlo (aceptable para configuración que cambia raramente).
• Costo de implementar el middleware + AuditService una vez en el core shared y consumirlo desde cada servicio; la política de lecturas sensibles y el catálogo inicial de menu_id/accion_id quedan para definición por módulo.
• El fallo del consumidor de auditoría no hace fallar la operación de negocio (cola tolerante; se monitorea la cola como un contrato más).