feat(ops,fin,crm): reglas de negocio del PDF (decisiones, cierre, facturación, continuidad, RBAC)

Cierra los huecos de la auditoría contra "SOFTWARE PARA AGENTES DE CARGA":

- ops (Diag. 2/3): bitácora con puntos de decisión (kind=decision) y ciclo de
  corrección (parent_event_id/attempt) para ¿Cut Off? y ¿despacho autorizado?
  (R-E-05/13, R-I-06). Reprogramación de salida (previous_etd, R-E-06). Hitos
  operativos completos export/import. Cierre operativo con costos finales
  (close_shipment, R-E-22).
- fin (Diag. 4): facturación con gate por cierre operativo y sin duplicar
  (R-F-01), costos de operación arrastrados (ops_cost_total, R-F-02), envío con
  PDF generado y guardado en MinIO (send_invoice + pdf.py sin dependencias,
  R-F-05) y revisión del cliente (en_revision_cliente + aprobación, R-F-06).
- crm (Diag. 1): opportunity_id enlaza embudo→RFQ (R-C-02), contacto como etapa
  (first_contact_at, R-C-04), re-cotización (clone_quote + reopen, R-C-12).
- transversal: catálogo de Incoterms y participantes/actores incl. autoridad
  aduanera (R-T-01/10), enforcement de permisos por carril (RBAC) con roles
  sembrados y dependencias dev-safe (R-T-07).
- Migración d5e6f7a8b9c0 con downgrade. Seed extendido. 70 tests (12 nuevos).

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

backend/alembic/versions/d5e6f7a8b9c0_pdf_compliance.py

@@ -0,0 +1,92 @@
+"""Cumplimiento PDF agente de carga: decisiones/costos en ops, envío/revisión en fin,
+continuidad comercial en crm (opportunity_id, contacto).
+
+Revision ID: d5e6f7a8b9c0
+Revises: c4d5e6f7a8b9
+Create Date: 2026-07-15 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "d5e6f7a8b9c0"
+down_revision: Union[str, None] = "c4d5e6f7a8b9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ---------- ops.shipments: transporte terrestre, reprogramación y cierre operativo ----------
+    op.add_column("shipments", sa.Column("ground_carrier_supplier_id", sa.Integer(), nullable=True), schema="ops")
+    op.add_column("shipments", sa.Column("pickup_at", sa.DateTime(), nullable=True), schema="ops")
+    op.add_column("shipments", sa.Column("previous_etd", sa.Date(), nullable=True), schema="ops")
+    op.add_column("shipments", sa.Column("actual_cost_total", sa.Numeric(precision=14, scale=2), nullable=True), schema="ops")
+    op.add_column("shipments", sa.Column("cost_currency", sa.String(length=3), nullable=True), schema="ops")
+    op.add_column("shipments", sa.Column("closed_at", sa.DateTime(), nullable=True), schema="ops")
+    op.add_column("shipments", sa.Column("closed_by", sa.String(length=64), nullable=True), schema="ops")
+    op.create_foreign_key(
+        "fk_ops_shipments_ground_carrier_supplier_id", "shipments", "suppliers",
+        ["ground_carrier_supplier_id"], ["id"], source_schema="ops", referent_schema="crm",
+    )
+
+    # ---------- ops.shipment_events: puntos de decisión y ciclo de corrección ----------
+    op.add_column("shipment_events", sa.Column("kind", sa.String(length=20), nullable=False, server_default=sa.text("'hito'")), schema="ops")
+    op.add_column("shipment_events", sa.Column("outcome", sa.String(length=20), nullable=True), schema="ops")
+    op.add_column("shipment_events", sa.Column("parent_event_id", sa.Integer(), nullable=True), schema="ops")
+    op.add_column("shipment_events", sa.Column("attempt", sa.Integer(), nullable=False, server_default=sa.text("1")), schema="ops")
+    op.create_foreign_key(
+        "fk_ops_shipment_events_parent_event_id", "shipment_events", "shipment_events",
+        ["parent_event_id"], ["id"], source_schema="ops", referent_schema="ops",
+    )
+
+    # ---------- fin.invoices: costos de operación, PDF y revisión del cliente ----------
+    op.add_column("invoices", sa.Column("ops_cost_total", sa.Numeric(precision=14, scale=2), nullable=True), schema="fin")
+    op.add_column("invoices", sa.Column("pdf_file_key", sa.String(length=512), nullable=True), schema="fin")
+    op.add_column("invoices", sa.Column("client_reviewed_at", sa.DateTime(), nullable=True), schema="fin")
+    op.add_column("invoices", sa.Column("client_approved", sa.Boolean(), nullable=True), schema="fin")
+    op.add_column("invoices", sa.Column("review_notes", sa.Text(), nullable=True), schema="fin")
+
+    # ---------- crm.service_requests: continuidad comercial y contacto ----------
+    op.add_column("service_requests", sa.Column("opportunity_id", sa.Integer(), nullable=True), schema="crm")
+    op.add_column("service_requests", sa.Column("first_contact_at", sa.DateTime(), nullable=True), schema="crm")
+    op.add_column("service_requests", sa.Column("first_contact_notes", sa.Text(), nullable=True), schema="crm")
+    op.create_index("ix_crm_service_requests_opportunity_id", "service_requests", ["opportunity_id"], schema="crm")
+    op.create_foreign_key(
+        "fk_crm_service_requests_opportunity_id", "service_requests", "opportunities",
+        ["opportunity_id"], ["id"], source_schema="crm", referent_schema="crm",
+    )
+
+
+def downgrade() -> None:
+    # crm.service_requests
+    op.drop_constraint("fk_crm_service_requests_opportunity_id", "service_requests", schema="crm", type_="foreignkey")
+    op.drop_index("ix_crm_service_requests_opportunity_id", table_name="service_requests", schema="crm")
+    op.drop_column("service_requests", "first_contact_notes", schema="crm")
+    op.drop_column("service_requests", "first_contact_at", schema="crm")
+    op.drop_column("service_requests", "opportunity_id", schema="crm")
+
+    # fin.invoices
+    op.drop_column("invoices", "review_notes", schema="fin")
+    op.drop_column("invoices", "client_approved", schema="fin")
+    op.drop_column("invoices", "client_reviewed_at", schema="fin")
+    op.drop_column("invoices", "pdf_file_key", schema="fin")
+    op.drop_column("invoices", "ops_cost_total", schema="fin")
+
+    # ops.shipment_events
+    op.drop_constraint("fk_ops_shipment_events_parent_event_id", "shipment_events", schema="ops", type_="foreignkey")
+    op.drop_column("shipment_events", "attempt", schema="ops")
+    op.drop_column("shipment_events", "parent_event_id", schema="ops")
+    op.drop_column("shipment_events", "outcome", schema="ops")
+    op.drop_column("shipment_events", "kind", schema="ops")
+
+    # ops.shipments
+    op.drop_constraint("fk_ops_shipments_ground_carrier_supplier_id", "shipments", schema="ops", type_="foreignkey")
+    op.drop_column("shipments", "closed_by", schema="ops")
+    op.drop_column("shipments", "closed_at", schema="ops")
+    op.drop_column("shipments", "cost_currency", schema="ops")
+    op.drop_column("shipments", "actual_cost_total", schema="ops")
+    op.drop_column("shipments", "previous_etd", schema="ops")
+    op.drop_column("shipments", "pickup_at", schema="ops")
+    op.drop_column("shipments", "ground_carrier_supplier_id", schema="ops")

backend/api/v1/modules/core/permissions/dependencies.py

@@ -3,14 +3,19 @@ Dependencias de FastAPI para verificación de permisos multi-tenant.
 Proporciona decoradores y funciones para proteger rutas con permisos específicos.
 """
 
+import logging
 from typing import List, Optional, Callable
 from fastapi import Depends, HTTPException, status, Header
 from sqlalchemy.orm import Session
 from functools import wraps
+from core.config import settings
 from core.database import get_core_db
 from core.security import get_current_user  # Asumiendo que existe esta función
 from .service import PermissionService
 
+logger = logging.getLogger(__name__)
+
+
 # Dependencia para obtener el servicio de permisos
 def get_permission_service(db: Session = Depends(get_core_db)) -> PermissionService:
     """
@@ -19,6 +24,34 @@ def get_permission_service(db: Session = Depends(get_core_db)) -> PermissionServ
     return PermissionService(db)
 
 
+def _authorize(
+    permission_service: PermissionService,
+    user_id: str,
+    company_id: int,
+    codes: List[str],
+    require_all: bool,
+) -> bool:
+    """Verifica permisos y, en desarrollo, aplica auto-bootstrap si el acceso falla.
+
+    Replica el bootstrap perezoso de ``core.security.validate_access_to_resource``:
+    en ``development`` el usuario (incluido el dev local) obtiene el rol super_admin
+    con todos los permisos la primera vez que lo necesita, para no bloquear el
+    entorno de desarrollo al activar el enforcement de permisos por área/carril.
+    """
+    check = permission_service.has_all_permissions if require_all else permission_service.has_any_permission
+    if check(user_id=user_id, company_id=company_id, permission_codes=codes):
+        return True
+    if settings.ENVIRONMENT == "development":
+        try:
+            permission_service.bootstrap_super_admin(user_id, company_id)
+            if check(user_id=user_id, company_id=company_id, permission_codes=codes):
+                logger.info("Auto-bootstrap de permisos en dev: user_id=%s company_id=%s", user_id, company_id)
+                return True
+        except Exception as exc:  # el bootstrap nunca debe escalar como acceso concedido
+            logger.warning("Auto-bootstrap de permisos falló: %s", exc)
+    return False
+
+
 # Clase para verificación de permisos (puede usarse como dependencia)
 class PermissionChecker:
     """
@@ -61,21 +94,8 @@ class PermissionChecker:
                 detail="User ID not found in token",
             )
 
-        # Verificar permisos sobre la compañía
+        # Verificar permisos sobre la compañía (con auto-bootstrap en desarrollo)
-        if self.require_all:
+        if not _authorize(permission_service, user_id, company_id, self.required_permissions, self.require_all):
-            has_access = permission_service.has_all_permissions(
-                user_id=user_id,
-                company_id=company_id,
-                permission_codes=self.required_permissions,
-            )
-        else:
-            has_access = permission_service.has_any_permission(
-                user_id=user_id,
-                company_id=company_id,
-                permission_codes=self.required_permissions,
-            )
-
-        if not has_access:
             raise HTTPException(
                 status_code=status.HTTP_403_FORBIDDEN,
                 detail=f"Missing required permissions: {', '.join(self.required_permissions)}",
@@ -114,13 +134,7 @@ class RequirePermission:
                 detail="User ID not found in token",
             )
 
-        has_permission = permission_service.has_permission(
+        if not _authorize(permission_service, user_id, company_id, [self.permission_code], require_all=True):
-            user_id=user_id,
-            company_id=company_id,
-            permission_code=self.permission_code,
-        )
-
-        if not has_permission:
             raise HTTPException(
                 status_code=status.HTTP_403_FORBIDDEN,
                 detail=f"Missing required permission: {self.permission_code}",

backend/api/v1/modules/crm/catalogs/data.py

@@ -0,0 +1,37 @@
+"""Catálogos de referencia del dominio (Incoterms y actores/participantes).
+
+Se centralizan aquí para administrarlos en un solo lugar y validarlos desde el
+levantamiento de requerimientos (R-T-10) y modelar los participantes del proceso
+(R-T-01), incluyendo la autoridad aduanera.
+"""
+
+# Incoterms 2020 (R-T-10)
+INCOTERMS: list[dict] = [
+    {"code": "EXW", "name": "Ex Works — En fábrica"},
+    {"code": "FCA", "name": "Free Carrier — Franco transportista"},
+    {"code": "FAS", "name": "Free Alongside Ship — Franco al costado del buque"},
+    {"code": "FOB", "name": "Free On Board — Franco a bordo"},
+    {"code": "CFR", "name": "Cost and Freight — Costo y flete"},
+    {"code": "CIF", "name": "Cost, Insurance and Freight — Costo, seguro y flete"},
+    {"code": "CPT", "name": "Carriage Paid To — Transporte pagado hasta"},
+    {"code": "CIP", "name": "Carriage and Insurance Paid To — Transporte y seguro pagados hasta"},
+    {"code": "DAP", "name": "Delivered At Place — Entregado en lugar"},
+    {"code": "DPU", "name": "Delivered At Place Unloaded — Entregado en lugar descargado"},
+    {"code": "DDP", "name": "Delivered Duty Paid — Entregado con derechos pagados"},
+]
+
+INCOTERM_CODES: set[str] = {i["code"] for i in INCOTERMS}
+
+# Roles/actores del proceso (R-T-01). Los actores externos se administran como
+# proveedores (crm.suppliers) vía su clasificación; el cliente/prospecto como cuenta.
+PARTICIPANT_ROLES: list[dict] = [
+    {"code": "exportador", "label": "Exportador", "source": "account"},
+    {"code": "importador", "label": "Importador", "source": "account"},
+    {"code": "agente_carga", "label": "Agente de carga", "source": "supplier"},
+    {"code": "agente_aduanal", "label": "Agente aduanal", "source": "supplier"},
+    {"code": "naviera", "label": "Naviera", "source": "supplier"},
+    {"code": "aerolinea", "label": "Aerolínea", "source": "supplier"},
+    {"code": "transportista_terrestre", "label": "Transportista terrestre", "source": "supplier"},
+    {"code": "agente_corresponsal", "label": "Agente corresponsal", "source": "supplier"},
+    {"code": "autoridad_aduanera", "label": "Autoridad aduanera", "source": "supplier"},
+]

backend/api/v1/modules/crm/catalogs/routes.py

@@ -0,0 +1,91 @@
+"""Endpoints de catálogos de referencia y participantes del proceso (R-T-01, R-T-10)."""
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from core.database import get_core_db
+from core.security import get_current_user
+
+from ..accounts.models import Account
+from ..suppliers.models import Supplier
+from .data import INCOTERMS, PARTICIPANT_ROLES
+
+router = APIRouter()
+
+
+@router.get("/catalogs/incoterms")
+def list_incoterms(
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+):
+    """Catálogo de Incoterms 2020 (R-T-10)."""
+    return INCOTERMS
+
+
+@router.get("/catalogs/participant-roles")
+def list_participant_roles(
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+):
+    """Catálogo de roles/actores del proceso, incluida la autoridad aduanera (R-T-01)."""
+    return PARTICIPANT_ROLES
+
+
+@router.get("/participants")
+def list_participants(
+    company_id: int = Query(..., description="Company ID"),
+    role: str | None = Query(None, description="Filtra por rol/clasificación"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Vista unificada de participantes del proceso: clientes/prospectos (cuentas) y
+    actores externos (proveedores por clasificación), en un solo catálogo (R-T-01)."""
+    tenant_id = current_user["tenant_id"]
+    result: list[dict] = []
+
+    accounts = (
+        db.query(Account)
+        .filter(
+            Account.tenant_id == tenant_id,
+            Account.company_id == company_id,
+            Account.deleted_at.is_(None),
+        )
+        .all()
+    )
+    for acc in accounts:
+        acc_role = getattr(acc, "record_type", None) or "cliente"
+        if role and role not in (acc_role, "exportador", "importador"):
+            # Las cuentas representan exportador/importador/cliente; sólo se omiten
+            # cuando el filtro pide explícitamente un rol de proveedor.
+            if role not in ("exportador", "importador", "cliente", "prospecto"):
+                continue
+        result.append({
+            "id": acc.id,
+            "source": "account",
+            "name": acc.name,
+            "role": acc_role,
+            "roles": [acc_role],
+        })
+
+    suppliers = (
+        db.query(Supplier)
+        .filter(
+            Supplier.tenant_id == tenant_id,
+            Supplier.company_id == company_id,
+            Supplier.deleted_at.is_(None),
+        )
+        .all()
+    )
+    for sup in suppliers:
+        classifications = sup.classifications or []
+        if role and role not in classifications:
+            continue
+        result.append({
+            "id": sup.id,
+            "source": "supplier",
+            "name": sup.name,
+            "role": classifications[0] if classifications else "proveedor",
+            "roles": classifications,
+        })
+
+    return result

backend/api/v1/modules/crm/quotes/routes.py

@@ -98,6 +98,17 @@ def reject_quote(
     return service.reject_quote(db, quote_id, current_user["tenant_id"], company_id)
 
 
+@router.post("/quotes/{quote_id}/clone", response_model=QuoteResponse, status_code=status.HTTP_201_CREATED)
+def clone_quote(
+    quote_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Clona la cotización como borrador para re-cotizar (R-C-12)."""
+    return service.clone_quote(db, quote_id, current_user["tenant_id"], company_id, _user_id(current_user))
+
+
 @router.delete("/quotes/{quote_id}", status_code=status.HTTP_204_NO_CONTENT)
 def delete_quote(
     quote_id: int,

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

@@ -158,6 +158,56 @@ def reject_quote(db: Session, quote_id: int, tenant_id: int, company_id: int) ->
     return quote
 
 
+def clone_quote(
+    db: Session, quote_id: int, tenant_id: int, company_id: int, user_id: str | None = None
+) -> Quote:
+    """Clona una cotización (y sus conceptos) como borrador para re-cotizar (R-C-12).
+
+    Si la cotización origen fue rechazada, reabre su solicitud a 'en_analisis' para
+    cerrar el ciclo de reintento del Diagrama 1.
+    """
+    src = get_quote(db, quote_id, tenant_id, company_id)
+    new_quote = Quote(
+        reference=(f"{src.reference}-R" if src.reference else None),
+        service_request_id=src.service_request_id,
+        account_id=src.account_id,
+        currency=src.currency,
+        status="borrador",
+        valid_until=src.valid_until,
+        notes=src.notes,
+        terms=src.terms,
+        owner_user_id=src.owner_user_id,
+        tenant_id=tenant_id,
+        company_id=company_id,
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(new_quote)
+    db.flush()
+    src_items = (
+        db.query(QuoteItem)
+        .filter(QuoteItem.quote_id == src.id, QuoteItem.deleted_at.is_(None))
+        .all()
+    )
+    for it in src_items:
+        db.add(QuoteItem(
+            quote_id=new_quote.id, concept=it.concept, description=it.description,
+            supplier_id=it.supplier_id, quantity=it.quantity, unit_cost=it.unit_cost,
+            unit_sale=it.unit_sale, currency=it.currency,
+            tenant_id=tenant_id, company_id=company_id,
+        ))
+    db.flush()
+    _recompute_totals(db, new_quote)
+    # Reabre la solicitud origen para el ciclo de re-cotización
+    if src.service_request_id:
+        sr = db.query(ServiceRequest).filter(ServiceRequest.id == src.service_request_id).first()
+        if sr and sr.status in ("rechazada", "cotizada"):
+            sr.status = "en_analisis"
+    db.commit()
+    db.refresh(new_quote)
+    return new_quote
+
+
 # ----- Quote items -----
 
 def get_quote_items(db: Session, quote_id: int, tenant_id: int, company_id: int) -> list[QuoteItem]:

backend/api/v1/modules/crm/router.py

@@ -5,12 +5,15 @@ Importar este módulo también registra los permisos del CRM (side-effect de
 ``permissions``), siguiendo el patrón del ``PermissionRegistry``.
 """
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
+
+from api.v1.modules.core.permissions.dependencies import PermissionChecker
 
 from . import permissions  # noqa: F401  (side-effect: registra permisos del CRM)
 from .accounts.routes import router as accounts_router
 from .activities.routes import router as activities_router
 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 .leads.routes import router as leads_router
@@ -22,7 +25,10 @@ from .service_requests.routes import router as service_requests_router
 from .suppliers.routes import router as suppliers_router
 from .uploads.routes import router as uploads_router
 
-router = APIRouter()
+# Enforcement por área/carril (R-T-07): se exige el permiso crm.access para tocar
+# cualquier endpoint del módulo. En desarrollo el usuario se auto-bootstrapea a
+# super_admin (ver PermissionChecker) para no bloquear el entorno.
+router = APIRouter(dependencies=[Depends(PermissionChecker(["crm.access"]))])
 
 router.include_router(accounts_router)
 router.include_router(suppliers_router)
@@ -36,4 +42,5 @@ router.include_router(pipelines_router)
 router.include_router(opportunities_router)
 router.include_router(activities_router)
 router.include_router(metrics_router)
+router.include_router(catalogs_router)
 router.include_router(uploads_router)

backend/api/v1/modules/crm/service_requests/dto.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field
 class ServiceRequestBase(BaseModel):
     reference: str | None = Field(None, max_length=40)
     account_id: int | None = None
+    opportunity_id: int | None = None
     operation_type: str = Field(..., max_length=20)  # importacion | exportacion
     transport_mode: str | None = Field(None, max_length=20)
     service_type: str | None = Field(None, max_length=20)
@@ -31,9 +32,26 @@ class ServiceRequestCreate(ServiceRequestBase):
     pass
 
 
+class ServiceRequestContactInput(BaseModel):
+    """Registro del contacto al cliente como etapa del flujo comercial (R-C-04)."""
+    notes: str | None = None
+
+
+class ServiceRequestFromOpportunityInput(BaseModel):
+    """Datos para convertir una oportunidad del embudo en solicitud/RFQ (R-C-02)."""
+    operation_type: str = Field(..., max_length=20)  # importacion | exportacion
+    transport_mode: str | None = Field(None, max_length=20)
+    service_type: str | None = Field(None, max_length=20)
+    incoterm: str | None = Field(None, max_length=10)
+    origin: str | None = Field(None, max_length=160)
+    destination: str | None = Field(None, max_length=160)
+    notes: str | None = None
+
+
 class ServiceRequestUpdate(BaseModel):
     reference: str | None = Field(None, max_length=40)
     account_id: int | None = None
+    opportunity_id: int | None = None
     operation_type: str | None = Field(None, max_length=20)
     transport_mode: str | None = Field(None, max_length=20)
     service_type: str | None = Field(None, max_length=20)
@@ -58,6 +76,8 @@ class ServiceRequestResponse(ServiceRequestBase):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
+    first_contact_at: datetime | None = None
+    first_contact_notes: str | None = None
     tenant_id: int
     company_id: int
     created_by: str | None = None

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

@@ -1,6 +1,6 @@
-from datetime import date
+from datetime import date, datetime
 
-from sqlalchemy import Date, ForeignKey, Integer, Numeric, String, Text, text
+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
@@ -22,6 +22,10 @@ class ServiceRequest(Base, TenantScopedMixin, TimestampMixin):
     account_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("crm.accounts.id"), nullable=True, index=True
     )
+    # Oportunidad de origen: enlaza el embudo (primer contacto) con la cadena RFQ→cotización (R-C-02)
+    opportunity_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("crm.opportunities.id"), nullable=True, index=True
+    )
     # importacion | exportacion
     operation_type: Mapped[str] = mapped_column(String(20), nullable=False)
     # maritimo | aereo | terrestre | ferroviario | multimodal
@@ -43,7 +47,10 @@ class ServiceRequest(Base, TenantScopedMixin, TimestampMixin):
         Integer, ForeignKey("crm.suppliers.id"), nullable=True
     )
     requirements: Mapped[str | None] = mapped_column(Text, nullable=True)  # otros requerimientos
-    # nueva | en_analisis | cotizada | aceptada | rechazada | liberada
+    # Contacto al cliente como etapa del flujo comercial (Diagrama 1, paso 2 — R-C-04)
+    first_contact_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    first_contact_notes: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # nueva | contacto | en_analisis | cotizada | aceptada | rechazada | liberada
     status: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'nueva'"), index=True)
     notes: Mapped[str | None] = mapped_column(Text, nullable=True)
     owner_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)

backend/api/v1/modules/crm/service_requests/routes.py

@@ -9,7 +9,9 @@ from .dto import (
     RateRequestCreate,
     RateRequestResponse,
     RateRequestUpdate,
+    ServiceRequestContactInput,
     ServiceRequestCreate,
+    ServiceRequestFromOpportunityInput,
     ServiceRequestResponse,
     ServiceRequestUpdate,
 )
@@ -71,6 +73,44 @@ def update_service_request(
     return service.update_service_request(db, request_id, payload, tenant_id, company_id, _user_id(current_user))
 
 
+@router.post("/service-requests/from-opportunity", response_model=ServiceRequestResponse, status_code=status.HTTP_201_CREATED)
+def create_from_opportunity(
+    payload: ServiceRequestFromOpportunityInput,
+    opportunity_id: int = Query(..., description="Oportunidad a convertir en solicitud"),
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Convierte una oportunidad del embudo en solicitud/RFQ enlazada (R-C-02)."""
+    tenant_id = current_user["tenant_id"]
+    return service.create_from_opportunity(db, opportunity_id, payload, tenant_id, company_id, _user_id(current_user))
+
+
+@router.post("/service-requests/{request_id}/contact", response_model=ServiceRequestResponse)
+def register_contact(
+    request_id: int,
+    payload: ServiceRequestContactInput,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Registra el contacto al cliente como etapa del flujo comercial (R-C-04)."""
+    tenant_id = current_user["tenant_id"]
+    return service.register_contact(db, request_id, payload, tenant_id, company_id, _user_id(current_user))
+
+
+@router.post("/service-requests/{request_id}/requote", response_model=ServiceRequestResponse)
+def reopen_for_requote(
+    request_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Reabre una solicitud rechazada para volver a cotizar (R-C-12)."""
+    tenant_id = current_user["tenant_id"]
+    return service.reopen_for_requote(db, request_id, tenant_id, company_id, _user_id(current_user))
+
+
 @router.delete("/service-requests/{request_id}", status_code=status.HTTP_204_NO_CONTENT)
 def delete_service_request(
     request_id: int,

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

@@ -4,8 +4,17 @@ from fastapi import HTTPException, status
 from sqlalchemy.orm import Session
 
 from ..accounts.models import Account
+from ..catalogs.data import INCOTERM_CODES
+from ..opportunities.models import Opportunity
 from ..suppliers.models import Supplier
-from .dto import RateRequestCreate, RateRequestUpdate, ServiceRequestCreate, ServiceRequestUpdate
+from .dto import (
+    RateRequestCreate,
+    RateRequestUpdate,
+    ServiceRequestContactInput,
+    ServiceRequestCreate,
+    ServiceRequestFromOpportunityInput,
+    ServiceRequestUpdate,
+)
 from .models import RateRequest, ServiceRequest
 
 
@@ -30,6 +39,14 @@ def _validate_request_refs(db: Session, data: dict, tenant_id: int, company_id:
         raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="El cliente asociado no existe")
     if not _exists(db, Supplier, data.get("destination_agent_id"), tenant_id, company_id):
         raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="El agente en destino no existe")
+    if not _exists(db, Opportunity, data.get("opportunity_id"), tenant_id, company_id):
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="La oportunidad asociada no existe")
+    incoterm = data.get("incoterm")
+    if incoterm and incoterm not in INCOTERM_CODES:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Incoterm inválido: usa uno del catálogo ({', '.join(sorted(INCOTERM_CODES))})",
+        )
 
 
 # ----- Service requests (RFQ) -----
@@ -112,6 +129,83 @@ def delete_service_request(db: Session, request_id: int, tenant_id: int, company
     db.commit()
 
 
+def register_contact(
+    db: Session, request_id: int, payload: ServiceRequestContactInput, tenant_id: int, company_id: int,
+    user_id: str | None = None,
+) -> ServiceRequest:
+    """Registra el contacto al cliente como etapa del flujo comercial (R-C-04)."""
+    obj = get_service_request(db, request_id, tenant_id, company_id)
+    obj.first_contact_at = datetime.now(timezone.utc)
+    obj.first_contact_notes = payload.notes
+    if obj.status == "nueva":
+        obj.status = "contacto"
+    obj.updated_by = user_id
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def create_from_opportunity(
+    db: Session, opportunity_id: int, payload: ServiceRequestFromOpportunityInput,
+    tenant_id: int, company_id: int, user_id: str | None = None,
+) -> ServiceRequest:
+    """Convierte una oportunidad del embudo en una solicitud/RFQ enlazada (R-C-02).
+
+    Da continuidad al hilo comercial: el embudo (primer contacto) queda ligado a la
+    cadena RFQ → cotización → embarque vía ``opportunity_id``.
+    """
+    opp = (
+        db.query(Opportunity)
+        .filter(
+            Opportunity.id == opportunity_id,
+            Opportunity.tenant_id == tenant_id,
+            Opportunity.company_id == company_id,
+            Opportunity.deleted_at.is_(None),
+        )
+        .first()
+    )
+    if not opp:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Oportunidad no encontrada")
+    obj = ServiceRequest(
+        account_id=opp.account_id,
+        opportunity_id=opp.id,
+        operation_type=payload.operation_type,
+        transport_mode=payload.transport_mode,
+        service_type=payload.service_type,
+        incoterm=payload.incoterm,
+        origin=payload.origin,
+        destination=payload.destination,
+        status="nueva",
+        notes=payload.notes,
+        owner_user_id=opp.owner_user_id,
+        tenant_id=tenant_id,
+        company_id=company_id,
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def reopen_for_requote(
+    db: Session, request_id: int, tenant_id: int, company_id: int, user_id: str | None = None
+) -> ServiceRequest:
+    """Reabre una solicitud rechazada para volver a cotizar (R-C-12)."""
+    obj = get_service_request(db, request_id, tenant_id, company_id)
+    if obj.status not in ("rechazada", "cotizada"):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Solo una solicitud rechazada o cotizada puede reabrirse para re-cotizar",
+        )
+    obj.status = "en_analisis"
+    obj.updated_by = user_id
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
 # ----- Rate requests -----
 
 def get_rate_requests(

backend/api/v1/modules/fin/invoices/dto.py

@@ -4,6 +4,12 @@ from decimal import Decimal
 from pydantic import BaseModel, ConfigDict, Field, computed_field
 
 
+class InvoiceClientReviewInput(BaseModel):
+    """Resultado de la revisión de la factura por el cliente (R-F-06)."""
+    approved: bool
+    notes: str | None = None
+
+
 class InvoiceItemBase(BaseModel):
     concept: str = Field(..., max_length=60)
     description: str | None = Field(None, max_length=255)
@@ -100,8 +106,13 @@ class InvoiceResponse(InvoiceBase):
     total: Decimal
     paid_amount: Decimal
     balance: Decimal
+    ops_cost_total: Decimal | None = None
     sent_at: datetime | None = None
     paid_at: datetime | None = None
+    pdf_file_key: str | None = None
+    client_reviewed_at: datetime | None = None
+    client_approved: bool | None = None
+    review_notes: str | None = None
     created_by: str | None = None
     updated_by: str | None = None
     tenant_id: int

backend/api/v1/modules/fin/invoices/models.py

@@ -1,6 +1,6 @@
 from datetime import date, datetime
 
-from sqlalchemy import Date, DateTime, ForeignKey, Integer, Numeric, String, Text, text
+from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Integer, Numeric, String, Text, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
@@ -25,7 +25,7 @@ class Invoice(Base, TenantScopedMixin, TimestampMixin):
         Integer, ForeignKey("crm.accounts.id"), nullable=True, index=True
     )
     currency: Mapped[str] = mapped_column(String(3), nullable=False, server_default=text("'MXN'"))
-    # borrador | emitida | enviada | pagada | cancelada
+    # borrador | emitida | enviada | en_revision_cliente | pagada | cancelada
     status: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'borrador'"), index=True)
     issue_date: Mapped[date | None] = mapped_column(Date, nullable=True)
     due_date: Mapped[date | None] = mapped_column(Date, nullable=True)
@@ -35,10 +35,18 @@ class Invoice(Base, TenantScopedMixin, TimestampMixin):
     total: Mapped[float] = mapped_column(Numeric(14, 2), nullable=False, server_default=text("0"))
     paid_amount: Mapped[float] = mapped_column(Numeric(14, 2), nullable=False, server_default=text("0"))
     balance: Mapped[float] = mapped_column(Numeric(14, 2), nullable=False, server_default=text("0"))
+    # Costos reales de la operación traídos de Operaciones al cierre (R-F-02)
+    ops_cost_total: Mapped[float | None] = mapped_column(Numeric(14, 2), nullable=True)
     bank_info: Mapped[str | None] = mapped_column(Text, nullable=True)  # datos bancarios
     notes: Mapped[str | None] = mapped_column(Text, nullable=True)
     sent_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
     paid_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    # ----- Envío al cliente (R-F-05): PDF almacenado en MinIO -----
+    pdf_file_key: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    # ----- Revisión del cliente (R-F-06) -----
+    client_reviewed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    client_approved: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
+    review_notes: Mapped[str | None] = mapped_column(Text, nullable=True)
     owner_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
     updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)

backend/api/v1/modules/fin/invoices/pdf.py

@@ -0,0 +1,186 @@
+"""Generador de PDF de factura sin dependencias externas.
+
+Se evita ``pdfkit`` (requiere el binario ``wkhtmltopdf``, ausente en el contenedor)
+y librerías extra. Produce un PDF válido de una o varias páginas con la fuente
+estándar Helvetica (no requiere incrustar fuentes). El texto se codifica en
+WinAnsi/Latin-1; los caracteres fuera de ese rango se sustituyen para no romper
+el flujo de contenido.
+"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from typing import Sequence
+
+_PAGE_W = 612  # carta (8.5in) en puntos
+_PAGE_H = 792  # carta (11in)
+_MARGIN = 56
+_LINE_H = 16
+_LINES_PER_PAGE = 42
+
+
+def _esc(text: str) -> str:
+    """Escapa y codifica una cadena para un literal de texto PDF (WinAnsi)."""
+    out = (text or "").encode("latin-1", "replace").decode("latin-1")
+    return out.replace("\\", r"\\").replace("(", r"\(").replace(")", r"\)")
+
+
+def _money(value, currency: str) -> str:
+    d = Decimal(str(value or 0)).quantize(Decimal("0.01"))
+    return f"{currency} {d:,.2f}"
+
+
+def _wrap(text: str, width: int) -> list[str]:
+    text = text or ""
+    words = text.split()
+    if not words:
+        return [""]
+    lines: list[str] = []
+    current = ""
+    for word in words:
+        candidate = f"{current} {word}".strip()
+        if len(candidate) > width and current:
+            lines.append(current)
+            current = word
+        else:
+            current = candidate
+    if current:
+        lines.append(current)
+    return lines
+
+
+def _build_lines(
+    *,
+    folio: str,
+    issue_date: str,
+    due_date: str,
+    account_name: str,
+    currency: str,
+    items: Sequence[dict],
+    subtotal,
+    tax_rate,
+    tax_amount,
+    total,
+    paid,
+    balance,
+    bank_info: str | None,
+    notes: str | None,
+) -> list[tuple[str, int]]:
+    """Devuelve una lista de (texto, tamaño_fuente) que compone el cuerpo."""
+    L: list[tuple[str, int]] = []
+    L.append(("FACTURA", 20))
+    L.append((f"Folio: {folio or 's/f'}", 11))
+    L.append((f"Fecha de emision: {issue_date or '-'}    Vencimiento: {due_date or '-'}", 11))
+    L.append(("", 11))
+    L.append((f"Cliente: {account_name or '-'}", 12))
+    L.append(("", 11))
+    L.append(("Conceptos", 13))
+    L.append(("-" * 78, 10))
+    L.append(("Cant.  Concepto                                   P. unitario        Importe", 10))
+    L.append(("-" * 78, 10))
+    for it in items:
+        concept = str(it.get("concept") or "")
+        desc = str(it.get("description") or "")
+        qty = Decimal(str(it.get("quantity") or 0))
+        unit = Decimal(str(it.get("unit_amount") or 0))
+        amount = (qty * unit).quantize(Decimal("0.01"))
+        label = concept if not desc else f"{concept} — {desc}"
+        label = label[:42].ljust(42)
+        row = f"{qty:>5.2f}  {label}  {unit:>12,.2f}  {amount:>12,.2f}"
+        L.append((row, 10))
+    L.append(("-" * 78, 10))
+    L.append(("", 11))
+    L.append((f"Subtotal: {_money(subtotal, currency)}", 11))
+    L.append((f"IVA ({Decimal(str(tax_rate or 0)):.2f}%): {_money(tax_amount, currency)}", 11))
+    L.append((f"Total: {_money(total, currency)}", 13))
+    L.append((f"Pagado: {_money(paid, currency)}", 11))
+    L.append((f"Saldo: {_money(balance, currency)}", 12))
+    if bank_info:
+        L.append(("", 11))
+        L.append(("Datos bancarios / de pago", 12))
+        for line in _wrap(bank_info, 90):
+            L.append((line, 10))
+    if notes:
+        L.append(("", 11))
+        L.append(("Notas", 12))
+        for line in _wrap(notes, 90):
+            L.append((line, 10))
+    return L
+
+
+def build_invoice_pdf(**kwargs) -> bytes:
+    """Construye el PDF de la factura y devuelve los bytes."""
+    lines = _build_lines(**kwargs)
+
+    # Paginar el cuerpo
+    pages: list[list[tuple[str, int]]] = []
+    for i in range(0, len(lines), _LINES_PER_PAGE):
+        pages.append(lines[i : i + _LINES_PER_PAGE])
+    if not pages:
+        pages = [[("FACTURA", 20)]]
+
+    # Un content stream por página
+    content_streams: list[bytes] = []
+    for page_lines in pages:
+        parts = ["BT", f"/F1 11 Tf", f"1 0 0 1 {_MARGIN} {_PAGE_H - _MARGIN} Tm", f"{_LINE_H} TL"]
+        first = True
+        for text, size in page_lines:
+            parts.append(f"/F1 {size} Tf")
+            if first:
+                parts.append(f"({_esc(text)}) Tj")
+                first = False
+            else:
+                parts.append(f"T* ({_esc(text)}) Tj")
+        parts.append("ET")
+        content_streams.append("\n".join(parts).encode("latin-1", "replace"))
+
+    # Ensamblado de objetos PDF
+    objects: list[bytes] = []
+
+    def add(obj: bytes) -> int:
+        objects.append(obj)
+        return len(objects)  # número de objeto (1-indexado)
+
+    # Reservamos números: catalog(1), pages(2), font(3), luego páginas y streams
+    font_obj_num = 3
+    page_obj_nums: list[int] = []
+    content_obj_nums: list[int] = []
+    # Precalcular números de páginas y streams
+    next_num = 4
+    for _ in pages:
+        page_obj_nums.append(next_num)
+        next_num += 1
+    for _ in pages:
+        content_obj_nums.append(next_num)
+        next_num += 1
+
+    kids = " ".join(f"{n} 0 R" for n in page_obj_nums)
+    add(f"<< /Type /Catalog /Pages 2 0 R >>".encode("latin-1"))
+    add(f"<< /Type /Pages /Kids [{kids}] /Count {len(pages)} >>".encode("latin-1"))
+    add(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >>")
+    for i, _ in enumerate(pages):
+        page_dict = (
+            f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 {_PAGE_W} {_PAGE_H}] "
+            f"/Resources << /Font << /F1 {font_obj_num} 0 R >> >> "
+            f"/Contents {content_obj_nums[i]} 0 R >>"
+        )
+        add(page_dict.encode("latin-1"))
+    for stream in content_streams:
+        obj = b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + stream + b"\nendstream"
+        add(obj)
+
+    # Serialización con tabla xref
+    out = bytearray()
+    out += b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n"
+    offsets: list[int] = []
+    for i, obj in enumerate(objects, start=1):
+        offsets.append(len(out))
+        out += f"{i} 0 obj\n".encode("latin-1") + obj + b"\nendobj\n"
+    xref_pos = len(out)
+    n = len(objects) + 1
+    out += f"xref\n0 {n}\n".encode("latin-1")
+    out += b"0000000000 65535 f \n"
+    for off in offsets:
+        out += f"{off:010d} 00000 n \n".encode("latin-1")
+    out += f"trailer\n<< /Size {n} /Root 1 0 R >>\nstartxref\n{xref_pos}\n%%EOF".encode("latin-1")
+    return bytes(out)

backend/api/v1/modules/fin/invoices/routes.py

@@ -6,6 +6,7 @@ from core.security import get_current_user
 
 from . import service
 from .dto import (
+    InvoiceClientReviewInput,
     InvoiceCreate,
     InvoiceItemCreate,
     InvoiceItemResponse,
@@ -62,7 +63,26 @@ def emit_invoice(invoice_id: int, company_id: int = Query(...), current_user: di
 
 @router.patch("/invoices/{invoice_id}/send", response_model=InvoiceResponse)
 def send_invoice(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
-    return service.send_invoice(db, invoice_id, current_user["tenant_id"], company_id)
+    """Genera el PDF, lo guarda en MinIO y marca la factura como enviada (R-F-05)."""
+    return service.send_invoice(db, invoice_id, current_user["tenant_id"], company_id, _uid(current_user))
+
+
+@router.get("/invoices/{invoice_id}/pdf-url")
+def get_invoice_pdf_url(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
+    """URL firmada fresca del PDF de la factura (R-F-05)."""
+    return {"url": service.get_invoice_pdf_url(db, invoice_id, current_user["tenant_id"], company_id)}
+
+
+@router.patch("/invoices/{invoice_id}/client-review", response_model=InvoiceResponse)
+def mark_client_review(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
+    """Marca la factura en revisión del cliente (R-F-06)."""
+    return service.mark_client_review(db, invoice_id, current_user["tenant_id"], company_id, _uid(current_user))
+
+
+@router.patch("/invoices/{invoice_id}/client-decision", response_model=InvoiceResponse)
+def client_review_decision(invoice_id: int, payload: InvoiceClientReviewInput, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
+    """Registra la decisión del cliente sobre la factura: aprobada o con observaciones (R-F-06)."""
+    return service.client_review_decision(db, invoice_id, payload, current_user["tenant_id"], company_id, _uid(current_user))
 
 
 @router.patch("/invoices/{invoice_id}/cancel", response_model=InvoiceResponse)

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

@@ -9,8 +9,16 @@ from api.v1.modules.crm.accounts.models import Account
 from api.v1.modules.crm.quotes.models import Quote, QuoteItem
 from api.v1.modules.ops.shipments.models import Shipment
 
-from .dto import InvoiceCreate, InvoiceItemCreate, InvoiceItemUpdate, InvoiceUpdate, PaymentCreate
+from .dto import (
+    InvoiceClientReviewInput,
+    InvoiceCreate,
+    InvoiceItemCreate,
+    InvoiceItemUpdate,
+    InvoiceUpdate,
+    PaymentCreate,
+)
 from .models import Invoice, InvoiceItem, Payment
+from .pdf import build_invoice_pdf
 
 
 def _exists(db: Session, model, _id, tenant_id, company_id) -> bool:
@@ -52,7 +60,7 @@ def _recompute(db: Session, invoice: Invoice) -> None:
     invoice.paid_amount = paid
     invoice.balance = total - paid
     # Estado de cobranza (no toca borrador ni cancelada)
-    if invoice.status in ("emitida", "enviada", "pagada"):
+    if invoice.status in ("emitida", "enviada", "en_revision_cliente", "pagada"):
         if total > 0 and invoice.balance <= 0:
             invoice.status = "pagada"
             invoice.paid_at = datetime.now(timezone.utc)
@@ -131,8 +139,108 @@ def emit_invoice(db, invoice_id, tenant_id, company_id) -> Invoice:
     return _set_status(db, invoice_id, tenant_id, company_id, "emitida", set_issue=True)
 
 
-def send_invoice(db, invoice_id, tenant_id, company_id) -> Invoice:
+def _build_pdf_bytes(db, invoice: Invoice, tenant_id, company_id) -> bytes:
-    return _set_status(db, invoice_id, tenant_id, company_id, "enviada", set_issue=True)
+    """Arma los bytes del PDF de la factura a partir de sus datos y conceptos."""
+    items = get_items(db, invoice.id, tenant_id, company_id)
+    account_name = None
+    if invoice.account_id:
+        acc = db.query(Account).filter(Account.id == invoice.account_id).first()
+        account_name = acc.name if acc else None
+    return build_invoice_pdf(
+        folio=invoice.reference or f"FAC-{invoice.id}",
+        issue_date=str(invoice.issue_date or ""),
+        due_date=str(invoice.due_date or ""),
+        account_name=account_name or "Cliente",
+        currency=invoice.currency or "MXN",
+        items=[
+            {"concept": it.concept, "description": it.description, "quantity": it.quantity, "unit_amount": it.unit_amount}
+            for it in items
+        ],
+        subtotal=invoice.subtotal,
+        tax_rate=invoice.tax_rate,
+        tax_amount=invoice.tax_amount,
+        total=invoice.total,
+        paid=invoice.paid_amount,
+        balance=invoice.balance,
+        bank_info=invoice.bank_info,
+        notes=invoice.notes,
+    )
+
+
+def send_invoice(db, invoice_id, tenant_id, company_id, user_id=None) -> Invoice:
+    """Envía la factura al cliente: genera el PDF, lo guarda en MinIO y marca 'enviada' (R-F-05)."""
+    from core.storage_s3 import put_object_bytes  # import diferido: evita conectar en tests
+
+    obj = get_invoice(db, invoice_id, tenant_id, company_id)
+    if obj.status in ("borrador", "cancelada"):
+        # La factura debe estar emitida antes de enviarse al cliente
+        if obj.status == "cancelada":
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="La factura está cancelada")
+        obj.status = "emitida"
+        if not obj.issue_date:
+            obj.issue_date = date.today()
+        db.flush()
+    pdf_bytes = _build_pdf_bytes(db, obj, tenant_id, company_id)
+    key = f"tenants/{tenant_id}/companies/{company_id}/fin-invoices/{obj.id}/factura-{obj.reference or obj.id}.pdf"
+    put_object_bytes(key, pdf_bytes, content_type="application/pdf")
+    obj.pdf_file_key = key
+    obj.status = "enviada"
+    obj.sent_at = datetime.now(timezone.utc)
+    if not obj.issue_date:
+        obj.issue_date = date.today()
+    obj.updated_by = user_id
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def get_invoice_pdf_url(db, invoice_id, tenant_id, company_id) -> str:
+    """Devuelve una URL firmada fresca del PDF de la factura (las presignadas expiran)."""
+    from core.storage_s3 import presigned_get_url
+
+    obj = get_invoice(db, invoice_id, tenant_id, company_id)
+    if not obj.pdf_file_key:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="La factura aún no tiene PDF; envíala al cliente para generarlo",
+        )
+    return presigned_get_url(obj.pdf_file_key)
+
+
+def mark_client_review(db, invoice_id, tenant_id, company_id, user_id=None) -> Invoice:
+    """Pone la factura en revisión del cliente (R-F-06)."""
+    obj = get_invoice(db, invoice_id, tenant_id, company_id)
+    if obj.status not in ("enviada", "en_revision_cliente"):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Solo una factura enviada puede pasar a revisión del cliente",
+        )
+    obj.status = "en_revision_cliente"
+    obj.updated_by = user_id
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def client_review_decision(
+    db, invoice_id, payload: InvoiceClientReviewInput, tenant_id, company_id, user_id=None
+) -> Invoice:
+    """Registra la decisión de revisión del cliente: aprobada o con observaciones (R-F-06)."""
+    obj = get_invoice(db, invoice_id, tenant_id, company_id)
+    if obj.status not in ("enviada", "en_revision_cliente"):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="La factura debe estar enviada o en revisión para registrar la decisión del cliente",
+        )
+    obj.client_reviewed_at = datetime.now(timezone.utc)
+    obj.client_approved = payload.approved
+    obj.review_notes = payload.notes
+    # Aprobada → lista para cobranza (enviada). Con observaciones → regresa a emitida para corregir.
+    obj.status = "enviada" if payload.approved else "emitida"
+    obj.updated_by = user_id
+    db.commit()
+    db.refresh(obj)
+    return obj
 
 
 def cancel_invoice(db, invoice_id, tenant_id, company_id) -> Invoice:
@@ -140,12 +248,32 @@ def cancel_invoice(db, invoice_id, tenant_id, company_id) -> Invoice:
 
 
 def generate_from_shipment(db, shipment_id, tenant_id, company_id, user_id=None) -> Invoice:
-    """Genera la factura de un embarque, tomando los conceptos (venta) de su cotización."""
+    """Genera la factura de un embarque, tomando los conceptos (venta) de su cotización.
+
+    El disparador válido de la facturación es el cierre operativo del embarque
+    (R-F-01): solo se factura un embarque en estado 'cerrada'. Los costos reales de
+    la operación se arrastran a la factura (R-F-02).
+    """
     shipment = db.query(Shipment).filter(
         Shipment.id == shipment_id, Shipment.tenant_id == tenant_id, Shipment.company_id == company_id, Shipment.deleted_at.is_(None)
     ).first()
     if not shipment:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Embarque no encontrado")
+    if shipment.status != "cerrada":
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="El embarque debe estar cerrado (cierre operativo) para facturarse",
+        )
+    existing = db.query(Invoice).filter(
+        Invoice.shipment_id == shipment_id, Invoice.tenant_id == tenant_id,
+        Invoice.company_id == company_id, Invoice.deleted_at.is_(None),
+        Invoice.status != "cancelada",
+    ).first()
+    if existing:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="El embarque ya tiene una factura vigente",
+        )
 
     quote = None
     if shipment.quote_id:
@@ -156,7 +284,8 @@ def generate_from_shipment(db, shipment_id, tenant_id, company_id, user_id=None)
         shipment_id=shipment.id,
         quote_id=shipment.quote_id,
         account_id=shipment.account_id,
-        currency=quote.currency if quote else "MXN",
+        currency=(shipment.cost_currency or (quote.currency if quote else "MXN")),
+        ops_cost_total=shipment.actual_cost_total,
         status="borrador",
         tenant_id=tenant_id,
         company_id=company_id,

backend/api/v1/modules/fin/router.py

@@ -1,9 +1,12 @@
 """Router agregador del módulo Facturación (Diagrama 4). Prefijo ``/fin``."""
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
+
+from api.v1.modules.core.permissions.dependencies import PermissionChecker
 
 from . import permissions  # noqa: F401  (side-effect: registra permisos)
 from .invoices.routes import router as invoices_router
 
-router = APIRouter()
+# Enforcement por área/carril (R-T-07): se exige fin.access para el módulo.
+router = APIRouter(dependencies=[Depends(PermissionChecker(["fin.access"]))])
 router.include_router(invoices_router)

backend/api/v1/modules/ops/router.py

@@ -3,11 +3,14 @@
 Se monta bajo el prefijo ``/ops`` en ``api/v1/router.py``.
 """
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
+
+from api.v1.modules.core.permissions.dependencies import PermissionChecker
 
 from . import permissions  # noqa: F401  (side-effect: registra permisos de ops)
 from .shipments.routes import router as shipments_router
 
-router = APIRouter()
+# Enforcement por área/carril (R-T-07): se exige ops.access para el módulo.
+router = APIRouter(dependencies=[Depends(PermissionChecker(["ops.access"]))])
 
 router.include_router(shipments_router)

backend/api/v1/modules/ops/shipments/dto.py

@@ -1,4 +1,6 @@
 from datetime import date, datetime
+from decimal import Decimal
+from typing import Literal
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -17,14 +19,19 @@ class ShipmentBase(BaseModel):
     status: str = Field("abierta", max_length=20)
     booking_number: str | None = Field(None, max_length=60)
     carrier_supplier_id: int | None = None
+    ground_carrier_supplier_id: int | None = None
     customs_agent_id: int | None = None
     destination_agent_id: int | None = None
     cutoff_date: datetime | None = None
+    pickup_at: datetime | None = None
     etd: date | None = None
+    previous_etd: date | None = None
     eta: date | None = None
     vessel_flight: str | None = Field(None, max_length=120)
     container_number: str | None = Field(None, max_length=60)
     notes: str | None = None
+    actual_cost_total: Decimal | None = None
+    cost_currency: str | None = Field(None, max_length=3)
     owner_user_id: str | None = Field(None, max_length=64)
 
 
@@ -44,9 +51,11 @@ class ShipmentUpdate(BaseModel):
     status: str | None = Field(None, max_length=20)
     booking_number: str | None = Field(None, max_length=60)
     carrier_supplier_id: int | None = None
+    ground_carrier_supplier_id: int | None = None
     customs_agent_id: int | None = None
     destination_agent_id: int | None = None
     cutoff_date: datetime | None = None
+    pickup_at: datetime | None = None
     etd: date | None = None
     eta: date | None = None
     vessel_flight: str | None = Field(None, max_length=120)
@@ -55,10 +64,26 @@ class ShipmentUpdate(BaseModel):
     owner_user_id: str | None = Field(None, max_length=64)
 
 
+class ShipmentRescheduleInput(BaseModel):
+    """Reprogramación de salida cuando no se alcanza el Cut Off (R-E-06)."""
+    etd: date | None = None
+    cutoff_date: datetime | None = None
+    reason: str | None = None
+
+
+class ShipmentCloseInput(BaseModel):
+    """Cierre operativo del embarque con costos finales (R-E-22)."""
+    actual_cost_total: Decimal = Field(..., ge=0)
+    cost_currency: str = Field("MXN", max_length=3)
+    notes: str | None = None
+
+
 class ShipmentResponse(ShipmentBase):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
+    closed_at: datetime | None = None
+    closed_by: str | None = None
     created_by: str | None = None
     updated_by: str | None = None
     tenant_id: int
@@ -71,7 +96,11 @@ class ShipmentEventBase(BaseModel):
     shipment_id: int
     event_type: str | None = Field(None, max_length=60)
     title: str = Field(..., min_length=1, max_length=160)
+    kind: str = Field("hito", max_length=20)  # hito | decision
     status: str = Field("pendiente", max_length=20)
+    outcome: str | None = Field(None, max_length=20)  # autorizado | rechazado
+    parent_event_id: int | None = None
+    attempt: int = Field(1, ge=1)
     position: int = Field(0, ge=0)
     planned_date: datetime | None = None
     actual_date: datetime | None = None
@@ -85,6 +114,7 @@ class ShipmentEventCreate(ShipmentEventBase):
 class ShipmentEventUpdate(BaseModel):
     event_type: str | None = Field(None, max_length=60)
     title: str | None = Field(None, min_length=1, max_length=160)
+    kind: str | None = Field(None, max_length=20)
     status: str | None = Field(None, max_length=20)
     position: int | None = Field(None, ge=0)
     planned_date: datetime | None = None
@@ -92,6 +122,12 @@ class ShipmentEventUpdate(BaseModel):
     notes: str | None = None
 
 
+class ShipmentEventDecisionInput(BaseModel):
+    """Resultado de un punto de decisión del flujo (R-E-13, R-E-05, R-I-06)."""
+    outcome: Literal["autorizado", "rechazado"]
+    notes: str | None = None
+
+
 class ShipmentEventResponse(ShipmentEventBase):
     model_config = ConfigDict(from_attributes=True)
 

backend/api/v1/modules/ops/shipments/models.py

@@ -1,6 +1,6 @@
 from datetime import date, datetime
 
-from sqlalchemy import Date, DateTime, ForeignKey, Integer, String, Text, text
+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
@@ -35,19 +35,29 @@ class Shipment(Base, TenantScopedMixin, TimestampMixin):
     booking_number: Mapped[str | None] = mapped_column(String(60), nullable=True)
     carrier_supplier_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("crm.suppliers.id"), nullable=True
-    )  # naviera / aerolínea / transportista
+    )  # naviera / aerolínea / transportista principal
+    ground_carrier_supplier_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("crm.suppliers.id"), nullable=True
+    )  # transporte terrestre / recolección (R-E-07)
     customs_agent_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("crm.suppliers.id"), nullable=True
     )  # agente aduanal
     destination_agent_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("crm.suppliers.id"), nullable=True
-    )  # agente en destino
+    )  # agente corresponsal en destino
     cutoff_date: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)  # Cut Off
+    pickup_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)  # cita/ventana de recolección (R-E-07)
     etd: Mapped[date | None] = mapped_column(Date, nullable=True)  # salida estimada
+    previous_etd: Mapped[date | None] = mapped_column(Date, nullable=True)  # salida previa tras reprogramación (R-E-06)
     eta: Mapped[date | None] = mapped_column(Date, nullable=True)  # llegada estimada
     vessel_flight: Mapped[str | None] = mapped_column(String(120), nullable=True)  # buque / vuelo
     container_number: Mapped[str | None] = mapped_column(String(60), nullable=True)
     notes: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # ----- Cierre operativo (R-E-22 / disparador de facturación R-F-01) -----
+    actual_cost_total: Mapped[float | None] = mapped_column(Numeric(14, 2), nullable=True)  # costos finales reales
+    cost_currency: Mapped[str | None] = mapped_column(String(3), nullable=True)
+    closed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)  # cierre operativo
+    closed_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
     owner_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
     updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
@@ -65,8 +75,17 @@ class ShipmentEvent(Base, TenantScopedMixin, TimestampMixin):
     )
     event_type: Mapped[str | None] = mapped_column(String(60), nullable=True)  # clave del hito
     title: Mapped[str] = mapped_column(String(160), nullable=False)
-    # pendiente | completado | omitido
+    # hito | decision  — un 'decision' es un punto de decisión del diagrama (rombo)
+    kind: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'hito'"))
+    # pendiente | completado | omitido | rechazado | en_correccion
     status: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'pendiente'"))
+    # Resultado de un punto de decisión: autorizado | rechazado (NULL mientras está pendiente)
+    outcome: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    # Ciclo de corrección: el hito de re-trámite apunta a la decisión rechazada que lo originó
+    parent_event_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ops.shipment_events.id"), nullable=True
+    )
+    attempt: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("1"))  # número de intento
     position: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
     planned_date: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
     actual_date: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

backend/api/v1/modules/ops/shipments/routes.py

@@ -6,13 +6,16 @@ from core.security import get_current_user
 
 from . import service
 from .dto import (
+    ShipmentCloseInput,
     ShipmentCreate,
     ShipmentDocumentCreate,
     ShipmentDocumentResponse,
     ShipmentDocumentUpdate,
     ShipmentEventCreate,
+    ShipmentEventDecisionInput,
     ShipmentEventResponse,
     ShipmentEventUpdate,
+    ShipmentRescheduleInput,
     ShipmentResponse,
     ShipmentUpdate,
 )
@@ -69,6 +72,32 @@ def create_shipment_from_quote(
     return service.create_shipment_from_quote(db, quote_id, tenant_id, company_id, _user_id(current_user))
 
 
+@router.post("/shipments/{shipment_id}/reschedule", response_model=ShipmentResponse)
+def reschedule_shipment(
+    shipment_id: int,
+    payload: ShipmentRescheduleInput,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Reprograma la salida cuando no se alcanza el Cut Off (R-E-06)."""
+    tenant_id = current_user["tenant_id"]
+    return service.reschedule_departure(db, shipment_id, payload, tenant_id, company_id, _user_id(current_user))
+
+
+@router.post("/shipments/{shipment_id}/close", response_model=ShipmentResponse)
+def close_shipment(
+    shipment_id: int,
+    payload: ShipmentCloseInput,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Cierre operativo del embarque con costos finales (R-E-22, dispara facturación R-F-01)."""
+    tenant_id = current_user["tenant_id"]
+    return service.close_shipment(db, shipment_id, payload, tenant_id, company_id, _user_id(current_user))
+
+
 @router.patch("/shipments/{shipment_id}", response_model=ShipmentResponse)
 def update_shipment(
     shipment_id: int,
@@ -189,6 +218,18 @@ def complete_shipment_event(
     return service.complete_shipment_event(db, event_id, current_user["tenant_id"], company_id)
 
 
+@router.patch("/shipment-events/{event_id}/decision", response_model=ShipmentEventResponse)
+def decide_shipment_event(
+    event_id: int,
+    payload: ShipmentEventDecisionInput,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Resuelve un punto de decisión: autorizado o rechazado (abre corrección). R-E-13/R-I-06."""
+    return service.decide_shipment_event(db, event_id, payload, current_user["tenant_id"], company_id)
+
+
 @router.delete("/shipment-events/{event_id}", status_code=status.HTTP_204_NO_CONTENT)
 def delete_shipment_event(
     event_id: int,

backend/api/v1/modules/ops/shipments/service.py

@@ -1,6 +1,7 @@
 from datetime import datetime, timezone
 
 from fastapi import HTTPException, status
+from sqlalchemy import func
 from sqlalchemy.orm import Session
 
 from api.v1.modules.crm.accounts.models import Account
@@ -9,33 +10,57 @@ from api.v1.modules.crm.service_requests.models import ServiceRequest
 from api.v1.modules.crm.suppliers.models import Supplier
 
 from .dto import (
+    ShipmentCloseInput,
     ShipmentCreate,
     ShipmentDocumentCreate,
     ShipmentDocumentUpdate,
     ShipmentEventCreate,
+    ShipmentEventDecisionInput,
     ShipmentEventUpdate,
+    ShipmentRescheduleInput,
     ShipmentUpdate,
 )
 from .models import Shipment, ShipmentDocument, ShipmentEvent
 
-# Hitos por defecto según el tipo de operación (Diagramas 2 y 3)
+# Hitos por defecto según el tipo de operación (Diagramas 2 y 3).
+# Tupla: (event_type, título, kind). kind="decision" son puntos de decisión (rombos)
+# que se resuelven con autorizado/rechazado y disparan el ciclo de corrección.
 _DEFAULT_MILESTONES = {
+    # Diagrama 2 — Proceso operativo de exportación
     "exportacion": [
-        ("recoleccion", "Recolección de mercancía"),
+        ("coordinacion_fecha_cliente", "Coordinar fecha de operación con el cliente", "hito"),
-        ("despacho_exportacion", "Despacho de exportación"),
+        ("revision_salidas", "Revisar disponibilidad de salidas del transporte", "hito"),
-        ("embarque", "Embarque"),
+        ("validacion_cutoff", "Validar Cut Off del transportista", "hito"),
-        ("zarpe", "Zarpe / Salida del transporte"),
+        ("decision_cutoff", "¿Se alcanza el Cut Off?", "decision"),
-        ("arribo", "Arribo a destino"),
+        ("programacion_transporte_terrestre", "Programar transporte terrestre y recolección", "hito"),
-        ("entrega", "Entrega al consignatario"),
+        ("recoleccion", "Recolección de mercancía", "hito"),
+        ("traslado_puerto", "Trasladar la mercancía al puerto / aeropuerto", "hito"),
+        ("entrega_terminal", "Entregar la mercancía en la terminal", "hito"),
+        ("entrega_docs_agente", "Entregar documentación al agente aduanal", "hito"),
+        ("despacho_exportacion", "Despacho de exportación", "hito"),
+        ("decision_despacho_exportacion", "¿Despacho de exportación autorizado?", "decision"),
+        ("emision_docs_internacionales", "Emitir documentación internacional (MBL/HBL, MAWB/HAWB, CMR)", "hito"),
+        ("embarque", "Embarque", "hito"),
+        ("zarpe", "Zarpe / Salida del transporte", "hito"),
+        ("coordinacion_corresponsal", "Coordinar con el agente corresponsal en destino", "hito"),
+        ("arribo", "Arribo a destino", "hito"),
+        ("despacho_destino", "Despacho de importación en destino (corresponsal)", "hito"),
+        ("entrega", "Entrega al consignatario", "hito"),
+        ("cierre_operativo", "Cierre operativo (registrar costos finales)", "hito"),
     ],
+    # Diagrama 3 — Proceso de importación
     "importacion": [
-        ("aviso_llegada", "Aviso de llegada"),
+        ("aviso_llegada", "Aviso de llegada", "hito"),
-        ("recepcion_docs", "Recepción de documentos (MBL/MAWB)"),
+        ("recepcion_docs", "Recepción de documentos (MBL/MAWB)", "hito"),
-        ("despacho_importacion", "Despacho de importación"),
+        ("coordinacion_agente_aduanal", "Coordinar con el agente aduanal el despacho", "hito"),
-        ("liberacion", "Liberación de mercancía"),
+        ("entrega_docs_agente", "Entregar documentos y requisitos al agente aduanal", "hito"),
-        ("retiro", "Retiro en puerto / aeropuerto"),
+        ("despacho_importacion", "Despacho de importación", "hito"),
-        ("traslado", "Traslado a bodega del importador"),
+        ("decision_despacho_importacion", "¿Despacho de importación autorizado?", "decision"),
-        ("entrega", "Entrega final al cliente"),
+        ("liberacion", "Liberación de mercancía", "hito"),
+        ("retiro", "Retiro en puerto / aeropuerto", "hito"),
+        ("traslado", "Traslado a bodega del importador", "hito"),
+        ("entrega", "Entrega final al cliente", "hito"),
+        ("cierre_operativo", "Cierre operativo (registrar costos finales)", "hito"),
     ],
 }
 
@@ -62,6 +87,7 @@ def _validate_refs(db: Session, data: dict, tenant_id: int, company_id: int) ->
         ("quote_id", Quote, "La cotización asociada no existe"),
         ("service_request_id", ServiceRequest, "La solicitud asociada no existe"),
         ("carrier_supplier_id", Supplier, "El transportista/naviera no existe"),
+        ("ground_carrier_supplier_id", Supplier, "El transportista terrestre no existe"),
         ("customs_agent_id", Supplier, "El agente aduanal no existe"),
         ("destination_agent_id", Supplier, "El agente en destino no existe"),
     ]
@@ -307,6 +333,12 @@ def update_shipment_event(db: Session, event_id: int, payload: ShipmentEventUpda
 
 def complete_shipment_event(db: Session, event_id: int, tenant_id: int, company_id: int) -> ShipmentEvent:
     obj = _get_event(db, event_id, tenant_id, company_id)
+    # Un punto de decisión no se "completa" a mano: se resuelve con decide_shipment_event
+    if obj.kind == "decision":
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Este hito es un punto de decisión: resuélvelo como autorizado o rechazado",
+        )
     obj.status = "completado"
     obj.actual_date = datetime.now(timezone.utc)
     db.commit()
@@ -314,12 +346,158 @@ def complete_shipment_event(db: Session, event_id: int, tenant_id: int, company_
     return obj
 
 
+def decide_shipment_event(
+    db: Session, event_id: int, payload: ShipmentEventDecisionInput, tenant_id: int, company_id: int
+) -> ShipmentEvent:
+    """Resuelve un punto de decisión del flujo (Cut Off / despacho autorizado).
+
+    - autorizado  → la decisión queda completada y el flujo continúa.
+    - rechazado   → la decisión queda 'rechazada' y se genera automáticamente un hito
+      de corrección (rehacer trámite) que apunta a esta decisión, implementando el
+      ciclo de corrección de los diagramas 2 (R-E-14) y 3 (R-I-07).
+    """
+    obj = _get_event(db, event_id, tenant_id, company_id)
+    if obj.kind != "decision":
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Solo los puntos de decisión aceptan un resultado (autorizado/rechazado)",
+        )
+    obj.outcome = payload.outcome
+    obj.actual_date = datetime.now(timezone.utc)
+    if payload.notes:
+        obj.notes = payload.notes
+
+    if payload.outcome == "autorizado":
+        obj.status = "completado"
+        db.commit()
+        db.refresh(obj)
+        return obj
+
+    # Rechazado: se abre el ciclo de corrección
+    obj.status = "rechazado"
+    correction = ShipmentEvent(
+        shipment_id=obj.shipment_id,
+        event_type=f"{obj.event_type or 'tramite'}_correccion",
+        title=f"Corrección: rehacer trámite — {obj.title}",
+        kind="hito",
+        status="en_correccion",
+        parent_event_id=obj.id,
+        attempt=(obj.attempt or 1) + 1,
+        # Se inserta justo después de la decisión rechazada para conservar el orden del flujo
+        position=obj.position,
+        tenant_id=tenant_id,
+        company_id=company_id,
+    )
+    # Empuja una posición los hitos posteriores para dejar hueco a la corrección
+    db.query(ShipmentEvent).filter(
+        ShipmentEvent.shipment_id == obj.shipment_id,
+        ShipmentEvent.tenant_id == tenant_id,
+        ShipmentEvent.company_id == company_id,
+        ShipmentEvent.deleted_at.is_(None),
+        ShipmentEvent.position > obj.position,
+    ).update({ShipmentEvent.position: ShipmentEvent.position + 1})
+    correction.position = obj.position + 1
+    db.add(correction)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
 def delete_shipment_event(db: Session, event_id: int, tenant_id: int, company_id: int) -> None:
     obj = _get_event(db, event_id, tenant_id, company_id)
     obj.deleted_at = datetime.now(timezone.utc)
     db.commit()
 
 
+def reschedule_departure(
+    db: Session, shipment_id: int, payload: ShipmentRescheduleInput, tenant_id: int, company_id: int,
+    user_id: str | None = None,
+) -> Shipment:
+    """Reprograma la salida cuando no se alcanza el Cut Off (R-E-06).
+
+    Conserva la salida anterior en ``previous_etd`` y deja constancia en la bitácora.
+    """
+    shipment = get_shipment(db, shipment_id, tenant_id, company_id)
+    if payload.etd is not None:
+        shipment.previous_etd = shipment.etd
+        shipment.etd = payload.etd
+    if payload.cutoff_date is not None:
+        shipment.cutoff_date = payload.cutoff_date
+    shipment.updated_by = user_id
+
+    last_pos = (
+        db.query(func.max(ShipmentEvent.position))
+        .filter(
+            ShipmentEvent.shipment_id == shipment_id,
+            ShipmentEvent.tenant_id == tenant_id,
+            ShipmentEvent.company_id == company_id,
+            ShipmentEvent.deleted_at.is_(None),
+        )
+        .scalar()
+    )
+    detail = payload.reason or "Reprogramación de salida por Cut Off no alcanzado"
+    db.add(
+        ShipmentEvent(
+            shipment_id=shipment_id,
+            event_type="reprogramacion",
+            title="Reprogramación de salida (nuevo Cut Off / ETD)",
+            kind="hito",
+            status="completado",
+            actual_date=datetime.now(timezone.utc),
+            position=(last_pos or 0) + 1,
+            notes=detail,
+            tenant_id=tenant_id,
+            company_id=company_id,
+        )
+    )
+    db.commit()
+    db.refresh(shipment)
+    return shipment
+
+
+def close_shipment(
+    db: Session, shipment_id: int, payload: ShipmentCloseInput, tenant_id: int, company_id: int,
+    user_id: str | None = None,
+) -> Shipment:
+    """Cierre operativo del embarque con costos finales (R-E-22).
+
+    Marca el embarque como 'cerrada' y registra los costos reales; el cierre es el
+    disparador válido de la facturación (R-F-01). No permite cerrar si quedan puntos
+    de decisión sin resolver.
+    """
+    shipment = get_shipment(db, shipment_id, tenant_id, company_id)
+    if shipment.status == "cancelada":
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="El embarque está cancelado"
+        )
+    pending_decision = (
+        db.query(ShipmentEvent.id)
+        .filter(
+            ShipmentEvent.shipment_id == shipment_id,
+            ShipmentEvent.tenant_id == tenant_id,
+            ShipmentEvent.company_id == company_id,
+            ShipmentEvent.deleted_at.is_(None),
+            ShipmentEvent.kind == "decision",
+            ShipmentEvent.status.in_(["pendiente", "rechazado", "en_correccion"]),
+        )
+        .first()
+    )
+    if pending_decision:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="No se puede cerrar: hay puntos de decisión pendientes o en corrección",
+        )
+    shipment.actual_cost_total = payload.actual_cost_total
+    shipment.cost_currency = payload.cost_currency
+    shipment.status = "cerrada"
+    shipment.closed_at = datetime.now(timezone.utc)
+    shipment.closed_by = user_id
+    shipment.updated_by = user_id
+    db.commit()
+    db.refresh(shipment)
+    return shipment
+
+
 def seed_default_milestones(db: Session, shipment_id: int, tenant_id: int, company_id: int) -> list[ShipmentEvent]:
     """Crea los hitos por defecto del embarque según su tipo de operación (import/export)."""
     shipment = get_shipment(db, shipment_id, tenant_id, company_id)
@@ -333,10 +511,10 @@ def seed_default_milestones(db: Session, shipment_id: int, tenant_id: int, compa
             detail="Define el tipo de operación (importación/exportación) para generar los hitos",
         )
     created = []
-    for position, (event_type, title) in enumerate(milestones):
+    for position, (event_type, title, kind) in enumerate(milestones):
         ev = ShipmentEvent(
-            shipment_id=shipment_id, event_type=event_type, title=title, status="pendiente",
+            shipment_id=shipment_id, event_type=event_type, title=title, kind=kind,
-            position=position, tenant_id=tenant_id, company_id=company_id,
+            status="pendiente", position=position, tenant_id=tenant_id, company_id=company_id,
         )
         db.add(ev)
         created.append(ev)

backend/seed_crm.py

@@ -25,11 +25,14 @@ from api.v1.modules.crm.pipelines.models import Pipeline, PipelineStage
 from api.v1.modules.crm.quotes.models import Quote, QuoteItem
 from api.v1.modules.crm.service_requests.models import ServiceRequest
 from api.v1.modules.crm.suppliers.models import Supplier
-from api.v1.modules.ops.shipments.models import Shipment, ShipmentDocument
+from api.v1.modules.ops.shipments.models import Shipment, ShipmentDocument, ShipmentEvent
 from api.v1.modules.ops.shipments import service as shipments_service
+from api.v1.modules.ops.shipments.dto import ShipmentCloseInput, ShipmentEventDecisionInput
 from api.v1.modules.fin.invoices.models import Invoice
 from api.v1.modules.fin.invoices import service as invoices_service
 from api.v1.modules.fin.invoices.dto import PaymentCreate
+from api.v1.modules.core.permissions.models import CompanyRole, Permission, RolePermission
+from api.v1.modules.core.permissions.service import PermissionService
 from core.database import CoreSessionLocal
 
 # Deben coincidir con DEV_LOCAL_AUTH_TENANT_ID / DEV_LOCAL_AUTH_COMPANY_ID
@@ -347,17 +350,81 @@ def seed_invoicing_and_events(db) -> None:
         print("• Sin embarque; se omite facturación")
         return
 
-    # Bitácora de hitos (según tipo de operación)
+    # Bitácora de hitos (según tipo de operación, incluye puntos de decisión)
-    shipments_service.seed_default_milestones(db, shipment.id, TENANT_ID, COMPANY_ID)
+    events = shipments_service.seed_default_milestones(db, shipment.id, TENANT_ID, COMPANY_ID)
 
-    # Factura generada desde el embarque (toma conceptos de la cotización)
+    # Resolver los puntos de decisión como autorizados y completar los hitos simples,
+    # para poder cerrar operativamente el embarque.
+    for ev in events:
+        if ev.kind == "decision":
+            shipments_service.decide_shipment_event(
+                db, ev.id, ShipmentEventDecisionInput(outcome="autorizado"), TENANT_ID, COMPANY_ID
+            )
+        else:
+            shipments_service.complete_shipment_event(db, ev.id, TENANT_ID, COMPANY_ID)
+
+    # Cierre operativo con costos finales (dispara la facturación — R-F-01/R-E-22)
+    shipments_service.close_shipment(
+        db, shipment.id,
+        ShipmentCloseInput(actual_cost_total=Decimal("2350"), cost_currency="USD", notes="Cierre demo"),
+        TENANT_ID, COMPANY_ID,
+    )
+
+    # Factura generada desde el embarque cerrado (toma conceptos de la cotización)
     invoice = invoices_service.generate_from_shipment(db, shipment.id, TENANT_ID, COMPANY_ID)
     invoices_service.emit_invoice(db, invoice.id, TENANT_ID, COMPANY_ID)
     invoices_service.create_payment(
         db, PaymentCreate(invoice_id=invoice.id, amount=Decimal("1000"), method="transferencia", reference="SPEI-001"),
         TENANT_ID, COMPANY_ID,
     )
-    print("✓ Factura emitida desde el embarque con pago parcial + hitos de la bitácora")
+    print("✓ Embarque cerrado, factura emitida con pago parcial + bitácora con decisiones resueltas")
+
+
+# Carriles del proceso (R-T-07): a qué módulos/acciones puede acceder cada rol.
+# clave = (code, nombre); valor = función que decide si un permiso pertenece al rol.
+def _carril_roles() -> dict:
+    return {
+        ("ventas", "Ventas"): lambda p: p.module == "crm" or p in {"ops.access", "ops.shipment.view"},
+        ("operaciones", "Operaciones"): lambda p: p.module == "ops"
+            or p in {"crm.access", "crm.account.view", "crm.quote.view", "crm.service_request.view", "fin.access", "fin.invoice.view"},
+        ("facturacion", "Facturación"): lambda p: p.module == "fin"
+            or p in {"ops.access", "ops.shipment.view", "crm.access", "crm.account.view"},
+        ("consulta", "Consulta"): lambda p: p.endswith(".access") or p.endswith(".view"),
+    }
+
+
+def seed_carril_roles(db) -> None:
+    """Crea los roles por carril del proceso con sus permisos (R-T-07). Idempotente."""
+    # Poblar el catálogo de permisos desde el registry de código
+    PermissionService(db).sync_permissions()
+    all_perms = db.query(Permission).filter(Permission.is_active == True).all()  # noqa: E712
+
+    created = 0
+    for (code, name), belongs in _carril_roles().items():
+        role = (
+            db.query(CompanyRole)
+            .filter(CompanyRole.company_id == COMPANY_ID, CompanyRole.tenant_id == TENANT_ID, CompanyRole.code == code)
+            .first()
+        )
+        if not role:
+            role = CompanyRole(
+                company_id=COMPANY_ID, tenant_id=TENANT_ID, name=name, code=code,
+                description=f"Rol de carril: {name}", is_active=True,
+            )
+            db.add(role)
+            db.flush()
+            created += 1
+        existing_perm_ids = {
+            pid for (pid,) in db.query(RolePermission.permission_id).filter(RolePermission.company_role_id == role.id)
+        }
+        for perm in all_perms:
+            if belongs(perm.code) and perm.id not in existing_perm_ids:
+                db.add(RolePermission(
+                    company_role_id=role.id, permission_id=perm.id,
+                    tenant_id=TENANT_ID, company_id=COMPANY_ID,
+                ))
+    db.commit()
+    print(f"✓ Roles por carril sembrados (nuevos: {created}) — Ventas, Operaciones, Facturación, Consulta")
 
 
 def main() -> None:
@@ -369,6 +436,7 @@ def main() -> None:
         seed_suppliers_and_related(db)
         seed_commercial_and_ops(db)
         seed_invoicing_and_events(db)
+        seed_carril_roles(db)
         print("\nSeed CRM completado.")
     finally:
         db.close()

backend/tests/test_compliance.py

@@ -0,0 +1,169 @@
+"""Pruebas de las reglas de negocio agregadas para cumplir el PDF de agente de carga.
+
+Cubren: puntos de decisión y ciclo de corrección de la bitácora (R-E-13/14, R-I-06/07),
+cierre operativo y gate de facturación (R-E-22 / R-F-01), envío y revisión de factura
+(R-F-05/06) y continuidad comercial (R-C-02/04/12) + catálogo de Incoterms (R-T-10).
+"""
+from decimal import Decimal
+
+import pytest
+from fastapi import HTTPException
+
+from api.v1.modules.crm.accounts import service as accounts_service
+from api.v1.modules.crm.accounts.dto import AccountCreate
+from api.v1.modules.crm.opportunities import service as opps_service
+from api.v1.modules.crm.opportunities.dto import OpportunityCreate
+from api.v1.modules.crm.quotes import service as quotes_service
+from api.v1.modules.crm.quotes.dto import QuoteCreate, QuoteItemCreate
+from api.v1.modules.crm.service_requests import service as sr_service
+from api.v1.modules.crm.service_requests.dto import (
+    ServiceRequestContactInput,
+    ServiceRequestCreate,
+    ServiceRequestFromOpportunityInput,
+)
+from api.v1.modules.fin.invoices import service as inv_service
+from api.v1.modules.fin.invoices.dto import InvoiceClientReviewInput, InvoiceCreate, InvoiceItemCreate
+from api.v1.modules.ops.shipments import service as ops_service
+from api.v1.modules.ops.shipments.dto import (
+    ShipmentCloseInput,
+    ShipmentCreate,
+    ShipmentEventDecisionInput,
+    ShipmentRescheduleInput,
+)
+
+T, C = 1, 1
+
+
+def _shipment(db, op_type="exportacion"):
+    return ops_service.create_shipment(db, ShipmentCreate(reference="EMB-T", operation_type=op_type), T, C)
+
+
+# ---------- Bitácora: decisiones y ciclo de corrección ----------
+
+def test_seed_milestones_include_decision_points(db):
+    sh = _shipment(db, "exportacion")
+    events = ops_service.seed_default_milestones(db, sh.id, T, C)
+    decisions = [e for e in events if e.kind == "decision"]
+    assert any(e.event_type == "decision_cutoff" for e in decisions)
+    assert any(e.event_type == "decision_despacho_exportacion" for e in decisions)
+
+
+def test_decision_autorizado_completa(db):
+    sh = _shipment(db, "importacion")
+    events = ops_service.seed_default_milestones(db, sh.id, T, C)
+    decision = next(e for e in events if e.kind == "decision")
+    updated = ops_service.decide_shipment_event(db, decision.id, ShipmentEventDecisionInput(outcome="autorizado"), T, C)
+    assert updated.status == "completado" and updated.outcome == "autorizado"
+
+
+def test_decision_rechazado_abre_correccion(db):
+    sh = _shipment(db, "importacion")
+    events = ops_service.seed_default_milestones(db, sh.id, T, C)
+    before = len(ops_service.get_shipment_events(db, T, C, sh.id))
+    decision = next(e for e in events if e.kind == "decision")
+    ops_service.decide_shipment_event(db, decision.id, ShipmentEventDecisionInput(outcome="rechazado", notes="Docs incompletos"), T, C)
+    after = ops_service.get_shipment_events(db, T, C, sh.id)
+    assert len(after) == before + 1  # se creó el hito de corrección
+    correction = next(e for e in after if e.parent_event_id == decision.id)
+    assert correction.status == "en_correccion" and correction.attempt == 2
+
+
+def test_complete_on_decision_falla(db):
+    sh = _shipment(db, "importacion")
+    events = ops_service.seed_default_milestones(db, sh.id, T, C)
+    decision = next(e for e in events if e.kind == "decision")
+    with pytest.raises(HTTPException) as exc:
+        ops_service.complete_shipment_event(db, decision.id, T, C)
+    assert exc.value.status_code == 422
+
+
+# ---------- Cierre operativo y gate de facturación ----------
+
+def test_close_blocked_with_pending_decision(db):
+    sh = _shipment(db, "importacion")
+    ops_service.seed_default_milestones(db, sh.id, T, C)  # deja decisiones pendientes
+    with pytest.raises(HTTPException) as exc:
+        ops_service.close_shipment(db, sh.id, ShipmentCloseInput(actual_cost_total=Decimal("100")), T, C)
+    assert exc.value.status_code == 409
+
+
+def test_generate_invoice_requires_closed_shipment(db):
+    sh = _shipment(db, "exportacion")  # abierta, sin cerrar
+    with pytest.raises(HTTPException) as exc:
+        inv_service.generate_from_shipment(db, sh.id, T, C)
+    assert exc.value.status_code == 409
+    # Tras cerrar (sin hitos → sin decisiones pendientes) sí factura
+    ops_service.close_shipment(db, sh.id, ShipmentCloseInput(actual_cost_total=Decimal("500"), cost_currency="MXN"), T, C)
+    inv = inv_service.generate_from_shipment(db, sh.id, T, C)
+    assert inv.shipment_id == sh.id
+
+
+def test_reschedule_keeps_previous_etd(db):
+    from datetime import date
+    sh = ops_service.create_shipment(db, ShipmentCreate(reference="EMB-R", operation_type="exportacion", etd=date(2026, 1, 10)), T, C)
+    ops_service.reschedule_departure(db, sh.id, ShipmentRescheduleInput(etd=date(2026, 1, 20), reason="Cut Off perdido"), T, C)
+    sh = ops_service.get_shipment(db, sh.id, T, C)
+    assert str(sh.previous_etd) == "2026-01-10" and str(sh.etd) == "2026-01-20"
+
+
+# ---------- Envío y revisión de factura ----------
+
+def test_send_invoice_generates_pdf_and_reviews(db, monkeypatch):
+    stored = {}
+    monkeypatch.setattr("core.storage_s3.put_object_bytes", lambda key, body, content_type="": stored.update({"key": key, "len": len(body)}))
+    acc = accounts_service.create_account(db, AccountCreate(name="Cliente PDF"), T, C)
+    inv = inv_service.create_invoice(db, InvoiceCreate(reference="F-PDF", account_id=acc.id, tax_rate=Decimal("16")), T, C)
+    inv_service.create_item(db, InvoiceItemCreate(invoice_id=inv.id, concept="flete_internacional", quantity=1, unit_amount=1000), T, C)
+    inv_service.emit_invoice(db, inv.id, T, C)
+    inv = inv_service.send_invoice(db, inv.id, T, C)
+    assert inv.status == "enviada" and inv.pdf_file_key and stored["len"] > 0
+    # Revisión del cliente: aprobada
+    inv = inv_service.mark_client_review(db, inv.id, T, C)
+    assert inv.status == "en_revision_cliente"
+    inv = inv_service.client_review_decision(db, inv.id, InvoiceClientReviewInput(approved=True, notes="OK"), T, C)
+    assert inv.status == "enviada" and inv.client_approved is True and inv.client_reviewed_at is not None
+
+
+# ---------- Continuidad comercial ----------
+
+def test_register_contact_sets_stage(db):
+    acc = accounts_service.create_account(db, AccountCreate(name="Prospecto"), T, C)
+    sr = sr_service.create_service_request(db, ServiceRequestCreate(account_id=acc.id, operation_type="exportacion"), T, C)
+    sr = sr_service.register_contact(db, sr.id, ServiceRequestContactInput(notes="Primer contacto"), T, C)
+    assert sr.status == "contacto" and sr.first_contact_at is not None
+
+
+def test_service_request_from_opportunity_links_back(db):
+    acc = accounts_service.create_account(db, AccountCreate(name="Cliente Op"), T, C)
+    opp = opps_service.create_opportunity(db, OpportunityCreate(name="Oportunidad X", account_id=acc.id), T, C)
+    sr = sr_service.create_from_opportunity(
+        db, opp.id, ServiceRequestFromOpportunityInput(operation_type="importacion"), T, C
+    )
+    assert sr.opportunity_id == opp.id and sr.account_id == acc.id
+
+
+def test_clone_quote_reopens_request(db):
+    acc = accounts_service.create_account(db, AccountCreate(name="Cliente Q"), T, C)
+    sr = sr_service.create_service_request(db, ServiceRequestCreate(account_id=acc.id, operation_type="exportacion"), T, C)
+    q = quotes_service.create_quote(db, QuoteCreate(reference="COT-1", account_id=acc.id, service_request_id=sr.id, currency="USD"), T, C)
+    quotes_service.create_quote_item(db, QuoteItemCreate(quote_id=q.id, concept="flete_internacional", quantity=1, unit_cost=100, unit_sale=200), T, C)
+    quotes_service.reject_quote(db, q.id, T, C)
+    clone = quotes_service.clone_quote(db, q.id, T, C)
+    assert clone.id != q.id and clone.status == "borrador"
+    items = quotes_service.get_quote_items(db, clone.id, T, C)
+    assert len(items) == 1
+    sr = sr_service.get_service_request(db, sr.id, T, C)
+    assert sr.status == "en_analisis"  # reabierta para re-cotizar
+
+
+def test_incoterm_catalog_validation(db):
+    acc = accounts_service.create_account(db, AccountCreate(name="Cliente Inc"), T, C)
+    with pytest.raises(HTTPException) as exc:
+        sr_service.create_service_request(
+            db, ServiceRequestCreate(account_id=acc.id, operation_type="exportacion", incoterm="XXX"), T, C
+        )
+    assert exc.value.status_code == 422
+    ok = sr_service.create_service_request(
+        db, ServiceRequestCreate(account_id=acc.id, operation_type="exportacion", incoterm="FOB"), T, C
+    )
+    assert ok.incoterm == "FOB"

backend/tests/test_invoices.py

@@ -7,7 +7,7 @@ from api.v1.modules.crm.quotes.dto import QuoteCreate, QuoteItemCreate
 from api.v1.modules.fin.invoices import service
 from api.v1.modules.fin.invoices.dto import InvoiceCreate, InvoiceItemCreate, PaymentCreate
 from api.v1.modules.ops.shipments import service as shipments_service
-from api.v1.modules.ops.shipments.dto import ShipmentCreate
+from api.v1.modules.ops.shipments.dto import ShipmentCloseInput, ShipmentCreate
 
 T, C = 1, 1
 
@@ -42,11 +42,16 @@ def test_generate_from_shipment_copies_quote_items(db):
     quotes_service.create_quote_item(db, QuoteItemCreate(quote_id=quote.id, concept="flete_internacional", quantity=1, unit_cost=1000, unit_sale=1500), T, C)
     quotes_service.accept_quote(db, quote.id, T, C)
     shipment = shipments_service.create_shipment_from_quote(db, quote.id, T, C)
+    # El embarque debe cerrarse operativamente antes de facturar (R-F-01)
+    shipments_service.close_shipment(
+        db, shipment.id, ShipmentCloseInput(actual_cost_total=Decimal("1000"), cost_currency="USD"), T, C
+    )
 
     inv = service.generate_from_shipment(db, shipment.id, T, C)
     assert inv.shipment_id == shipment.id
     assert inv.account_id == acc.id
     assert inv.currency == "USD"
+    assert float(inv.ops_cost_total) == 1000.0  # costos de operación arrastrados (R-F-02)
     items = service.get_items(db, inv.id, T, C)
     assert len(items) == 1 and float(items[0].unit_amount) == 1500.0
     assert float(inv.subtotal) == 1500.0