fix(crm): correcciones de revisión adversarial en servicios de dominio

- convert_lead valida embudo/etapa en el scope tenant/company (evita fuga multi-tenant y 500)
- oportunidades: estado/probabilidad/cierre se derivan de la etapa también en create/update (no solo move)
- oportunidades: valida que la etapa pertenezca al embudo indicado
- cuentas: country por defecto 'MX' (el server_default no aplicaba con NULL explícito)
- convert_lead: contact_name se normaliza (evita first_name vacío)
- +4 tests que fijan el comportamiento corregido (28 en verde)

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

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

@@ -15,7 +15,9 @@ class AccountCreate(BaseModel):
     address: str | None = None
     city: str | None = Field(None, max_length=120)
     state: str | None = Field(None, max_length=120)
-    country: str | None = Field(None, max_length=2)
+    # País por defecto MX (CRM aduanero). Evita depender del server_default,
+    # que no aplica cuando model_dump envía la columna como NULL explícito.
+    country: str | None = Field("MX", max_length=2)
     patente_aduanal: str | None = Field(None, max_length=20)
     status: str = Field("active", max_length=20)
     owner_user_id: str | None = Field(None, max_length=64)

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

@@ -6,10 +6,62 @@ from sqlalchemy.orm import Session
 from ..accounts.models import Account
 from ..contacts.models import Contact
 from ..opportunities.models import Opportunity
+from ..pipelines.models import Pipeline, PipelineStage
 from .dto import LeadConvert, LeadCreate, LeadUpdate
 from .models import Lead
 
 
+def _validate_pipeline_stage(
+    db: Session,
+    pipeline_id: int | None,
+    stage_id: int | None,
+    tenant_id: int,
+    company_id: int,
+) -> None:
+    """Valida que embudo/etapa existan en el tenant/company y sean coherentes.
+
+    Evita enlazar la oportunidad convertida a un embudo/etapa de otro tenant
+    (fuga multi-tenant) o a un id inexistente (que produciría un 500).
+    """
+    if pipeline_id is not None:
+        exists = (
+            db.query(Pipeline.id)
+            .filter(
+                Pipeline.id == pipeline_id,
+                Pipeline.tenant_id == tenant_id,
+                Pipeline.company_id == company_id,
+                Pipeline.deleted_at.is_(None),
+            )
+            .first()
+        )
+        if not exists:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="El embudo indicado no existe",
+            )
+    if stage_id is not None:
+        stage = (
+            db.query(PipelineStage)
+            .filter(
+                PipelineStage.id == stage_id,
+                PipelineStage.tenant_id == tenant_id,
+                PipelineStage.company_id == company_id,
+                PipelineStage.deleted_at.is_(None),
+            )
+            .first()
+        )
+        if not stage:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="La etapa indicada no existe",
+            )
+        if pipeline_id is not None and stage.pipeline_id != pipeline_id:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="La etapa no pertenece al embudo indicado",
+            )
+
+
 def get_leads(
     db: Session,
     tenant_id: int,
@@ -102,10 +154,11 @@ def convert_lead(
     db.add(account)
     db.flush()  # necesitamos el id de la cuenta para enlazar contacto/oportunidad
 
-    # 2. Contacto (si el prospecto trae nombre de contacto)
+    # 2. Contacto (si el prospecto trae nombre de contacto no vacío)
     contact = None
-    if lead.contact_name:
+    contact_name = (lead.contact_name or "").strip()
-        parts = lead.contact_name.split(" ", 1)
+    if contact_name:
+        parts = contact_name.split(" ", 1)
         contact = Contact(
             account_id=account.id,
             first_name=parts[0],
@@ -123,6 +176,8 @@ def convert_lead(
     # 3. Oportunidad (opcional)
     opportunity = None
     if payload.create_opportunity:
+        # Valida embudo/etapa en el scope del tenant/company antes de enlazar
+        _validate_pipeline_stage(db, payload.pipeline_id, payload.stage_id, tenant_id, company_id)
         opportunity = Opportunity(
             name=payload.opportunity_name or lead.name,
             account_id=account.id,

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

@@ -10,9 +10,60 @@ from .dto import OpportunityCreate, OpportunityUpdate
 from .models import Opportunity
 
 
+def _get_scoped_stage(
+    db: Session,
+    stage_id: int,
+    tenant_id: int,
+    company_id: int,
+    code: int = status.HTTP_404_NOT_FOUND,
+    detail: str = "Etapa no encontrada",
+) -> PipelineStage:
+    stage = (
+        db.query(PipelineStage)
+        .filter(
+            PipelineStage.id == stage_id,
+            PipelineStage.tenant_id == tenant_id,
+            PipelineStage.company_id == company_id,
+            PipelineStage.deleted_at.is_(None),
+        )
+        .first()
+    )
+    if not stage:
+        raise HTTPException(status_code=code, detail=detail)
+    return stage
+
+
+def _apply_stage_state(opportunity: Opportunity, stage: PipelineStage) -> None:
+    """Deriva estado/probabilidad/cierre desde la etapa destino.
+
+    Fuente única de verdad para create/update/move: la etapa manda. Así una
+    oportunidad nunca queda 'open' en una etapa ganada/perdida (evita conteos
+    inconsistentes en métricas).
+    """
+    opportunity.stage_id = stage.id
+    opportunity.pipeline_id = stage.pipeline_id
+    if stage.is_won:
+        opportunity.status = "won"
+        opportunity.probability = 100
+        opportunity.closed_at = datetime.now(timezone.utc)
+    elif stage.is_lost:
+        opportunity.status = "lost"
+        opportunity.probability = 0
+        opportunity.closed_at = datetime.now(timezone.utc)
+    else:
+        opportunity.status = "open"
+        opportunity.probability = stage.probability
+        opportunity.closed_at = None
+
+
 def _validate_refs(db: Session, data: dict, tenant_id: int, company_id: int) -> None:
-    """Valida que las referencias (cuenta, contacto, embudo, etapa) existan en el tenant/company."""
+    """Valida que cuenta, contacto y embudo existan en el tenant/company.
-    scope = lambda model, _id: (  # noqa: E731
+
+    La etapa se valida por separado (necesitamos el objeto para derivar estado),
+    verificando además que pertenezca al embudo indicado cuando ambos vienen.
+    """
+    def scope(model, _id):
+        return (
             db.query(model.id)
             .filter(
                 model.id == _id,
@@ -22,17 +73,30 @@ def _validate_refs(db: Session, data: dict, tenant_id: int, company_id: int) ->
             )
             .first()
         )
+
     checks = [
         ("account_id", Account, "La cuenta asociada no existe"),
         ("contact_id", Contact, "El contacto asociado no existe"),
         ("pipeline_id", Pipeline, "El embudo asociado no existe"),
-        ("stage_id", PipelineStage, "La etapa asociada no existe"),
     ]
     for field, model, message in checks:
         value = data.get(field)
         if value is not None and not scope(model, value):
             raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message)
 
+    stage_id = data.get("stage_id")
+    if stage_id is not None:
+        stage = _get_scoped_stage(
+            db, stage_id, tenant_id, company_id,
+            code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="La etapa asociada no existe",
+        )
+        pipeline_id = data.get("pipeline_id")
+        if pipeline_id is not None and stage.pipeline_id != pipeline_id:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="La etapa no pertenece al embudo indicado",
+            )
+
 
 def get_opportunities(
     db: Session,
@@ -81,6 +145,10 @@ def create_opportunity(
     data = payload.model_dump()
     _validate_refs(db, data, tenant_id, company_id)
     opportunity = Opportunity(**data, tenant_id=tenant_id, company_id=company_id)
+    # Si se asigna etapa, esta manda sobre estado/probabilidad/cierre
+    if opportunity.stage_id is not None:
+        stage = _get_scoped_stage(db, opportunity.stage_id, tenant_id, company_id)
+        _apply_stage_state(opportunity, stage)
     db.add(opportunity)
     db.commit()
     db.refresh(opportunity)
@@ -95,6 +163,10 @@ def update_opportunity(
     _validate_refs(db, data, tenant_id, company_id)
     for field, value in data.items():
         setattr(opportunity, field, value)
+    # Cambiar de etapa vía PATCH también deriva estado (misma regla que move)
+    if data.get("stage_id") is not None:
+        stage = _get_scoped_stage(db, data["stage_id"], tenant_id, company_id)
+        _apply_stage_state(opportunity, stage)
     db.commit()
     db.refresh(opportunity)
     return opportunity
@@ -105,33 +177,8 @@ def move_opportunity(
 ) -> Opportunity:
     """Mueve la oportunidad a una etapa y deriva estado/probabilidad de la etapa destino."""
     opportunity = get_opportunity(db, opportunity_id, tenant_id, company_id)
-    stage = (
+    stage = _get_scoped_stage(db, stage_id, tenant_id, company_id)
-        db.query(PipelineStage)
+    _apply_stage_state(opportunity, stage)
-        .filter(
-            PipelineStage.id == stage_id,
-            PipelineStage.tenant_id == tenant_id,
-            PipelineStage.company_id == company_id,
-            PipelineStage.deleted_at.is_(None),
-        )
-        .first()
-    )
-    if not stage:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Etapa no encontrada")
-
-    opportunity.stage_id = stage.id
-    opportunity.pipeline_id = stage.pipeline_id
-    if stage.is_won:
-        opportunity.status = "won"
-        opportunity.probability = 100
-        opportunity.closed_at = datetime.now(timezone.utc)
-    elif stage.is_lost:
-        opportunity.status = "lost"
-        opportunity.probability = 0
-        opportunity.closed_at = datetime.now(timezone.utc)
-    else:
-        opportunity.status = "open"
-        opportunity.probability = stage.probability
-        opportunity.closed_at = None
     db.commit()
     db.refresh(opportunity)
     return opportunity

backend/tests/test_leads.py

@@ -1,3 +1,6 @@
+import pytest
+from fastapi import HTTPException
+
 from api.v1.modules.crm.leads import service
 from api.v1.modules.crm.leads.dto import LeadConvert, LeadCreate
 from api.v1.modules.crm.pipelines import service as pipelines_service
@@ -6,6 +9,13 @@ from api.v1.modules.crm.pipelines.dto import PipelineCreate, StageCreate
 T, C = 1, 1
 
 
+def test_convert_rejects_invalid_stage(db):
+    lead = service.create_lead(db, LeadCreate(name="P", company_name="Empresa"), T, C)
+    with pytest.raises(HTTPException) as exc:
+        service.convert_lead(db, lead.id, LeadConvert(create_opportunity=True, stage_id=999), T, C)
+    assert exc.value.status_code == 422
+
+
 def test_create_lead_defaults_to_new(db):
     lead = service.create_lead(db, LeadCreate(name="Prospecto X", company_name="XYZ SA"), T, C)
     assert lead.status == "new"

backend/tests/test_opportunities.py

@@ -2,7 +2,7 @@ import pytest
 from fastapi import HTTPException
 
 from api.v1.modules.crm.opportunities import service
-from api.v1.modules.crm.opportunities.dto import OpportunityCreate
+from api.v1.modules.crm.opportunities.dto import OpportunityCreate, OpportunityUpdate
 from api.v1.modules.crm.pipelines import service as pipelines_service
 from api.v1.modules.crm.pipelines.dto import PipelineCreate, StageCreate
 
@@ -75,3 +75,35 @@ def test_only_one_default_pipeline(db):
     pipelines = pipelines_service.get_pipelines(db, T, C)
     defaults = [p for p in pipelines if p.is_default]
     assert len(defaults) == 1 and defaults[0].name == "P2"
+
+
+def test_create_in_won_stage_derives_state(db):
+    pipeline, _s_open, s_won, _ = _pipeline_with_stages(db)
+    opp = service.create_opportunity(
+        db,
+        OpportunityCreate(name="Directo a ganada", pipeline_id=pipeline.id, stage_id=s_won.id, amount=500),
+        T, C,
+    )
+    assert opp.status == "won"
+    assert opp.probability == 100
+    assert opp.closed_at is not None
+
+
+def test_update_to_lost_stage_derives_state(db):
+    pipeline, s_open, _s_won, s_lost = _pipeline_with_stages(db)
+    opp = service.create_opportunity(db, OpportunityCreate(name="Z", pipeline_id=pipeline.id, stage_id=s_open.id), T, C)
+    updated = service.update_opportunity(db, opp.id, OpportunityUpdate(stage_id=s_lost.id), T, C)
+    assert updated.status == "lost"
+    assert updated.probability == 0
+    assert updated.closed_at is not None
+
+
+def test_create_rejects_stage_from_other_pipeline(db):
+    pipeline, _s_open, _s_won, _s_lost = _pipeline_with_stages(db)
+    other = pipelines_service.create_pipeline(db, PipelineCreate(name="Otro"), T, C)
+    other_stage = pipelines_service.create_stage(db, StageCreate(pipeline_id=other.id, name="X"), T, C)
+    with pytest.raises(HTTPException) as exc:
+        service.create_opportunity(
+            db, OpportunityCreate(name="Y", pipeline_id=pipeline.id, stage_id=other_stage.id), T, C
+        )
+    assert exc.value.status_code == 422