Servicio: Motor de Integridad (compute + gate)

Descripción

El servicio Motor de Integridad es la columna vertebral de la F2. Centraliza la lógica que:

Computa el estado en vivo del radar plano↔realidad (compute_integrity()).
Integra veredictos humanos (IntegrityAck) para excluir false-alarms de las métricas.
Alimenta automáticamente la métrica del gate (gate_stats()).

Ubicación: racks/services/integrity.py (módulo, no clase).
Versión: v1.59.0 (motor) + v1.60.0 (veredictos + gate).

Funciones públicas

1. compute_integrity(org, fresh_days=7)

Propósito: Estado completo del radar en vivo, incluyendo el efecto de veredictos previos.

Firma:

def compute_integrity(organization, fresh_days: int = DEFAULT_FRESH_DAYS) -> dict

Parámetros:

organization: FK(Organization) — org a validar.
fresh_days: int = 7 — ventana de “visto recientemente en red” (DeviceProfile.last_seen).

Retorna:

{
    "summary": {
        "devices": 250,
        "matched_alive": 248,
        "matched_stale": 1,
        "ip_conflict": 0,
        "unobservable": 1,
        "muted": 0,
        "fidelity_pct": 99.1,
        "coverage_pct": 99.5,
        "undocumented": 3,
    },
    "racks": [
        {
            "rack_id": 1,
            "rack_name": "Rack-A",
            "fidelity_pct": 100.0,
            "coverage_pct": 100.0,
            "matched_alive": 20,
            "matched_stale": 0,
            "ip_conflict": 0,
            "unobservable": 0,
            "muted": 0,
            "findings": [
                {
                    "device_id": 42,
                    "name": "Switch-A",
                    "u_position": 10,
                    "classification": "matched_stale",
                    "reason": "Plan says present, network cannot confirm (7+ days)",
                    "evidence": {...},
                    "ack": {  # Si hay veredicto vigente
                        "verdict": "plan_updated",
                        "muted_until": "2026-07-20T10:00:00Z",
                        "note": "Ya retirado"
                    }
                },
                ...
            ]
        },
        ...
    ],
    "undocumented": [  # DeviceProfile vivos sin plano
        {
            "profile_id": 100,
            "ip": "10.0.1.50",
            "hostname": "scanner-lab",
            "vendor": "HP",
            "model": "OfficeJet",
            "device_type": "printer",
            "last_seen": "2026-07-17T14:30:00Z",
            "ack": {  # Si hay veredicto vigente
                "verdict": "false_alarm",
                "muted_until": "2026-08-16T10:00:00Z",
                "note": "Es temporal, ignorar"
            }
        },
        ...
    ]
}

Lógica:

Carga veredictos vigentes:

acks = _load_active_acks(organization, now)
# Índice: {("device", id, kind): ack_obj, ...}

Itera devices por rack:
- Para cada Device en la org (no deletado, no template).
- Busca DeviceProfile que le corresponda (match por IP+MAC o mejor heurística).
- Valida que esté “vivo en red” (last_seen < 7 días).
Clasifica cada device:
- matched_alive: Hay match en ambos planos (plan + red) y vivo recientemente.
- matched_stale: Hay match pero no se ve en red (> 7 días sin confirmación).
- ip_conflict: Dos devices afirman la misma IP (después de resolver MACs duales, etc.).
- unobservable: Imposible de validar (sin IPv4, sin MAC, etc.).

Aplica veredictos previos:

ack = acks.get(("device", device.id, CLASS_MATCHED_STALE))
if ack:
    finding["ack"] = _ack_info(ack)
    if ack.verdict == VERDICT_FALSE_ALARM:
        counts["muted"] += 1  # Excluir de fidelity
    else:
        counts[CLASS_MATCHED_STALE] += 1  # Contar (plano aún diverge)
else:
    counts[CLASS_MATCHED_STALE] += 1  # Sin veredicto

Calcula métricas:

observable = total - unobservable - muted
fidelity_pct = 100 * matched_alive / observable
coverage_pct = 100 * observable / total

Itera DeviceProfiles sin Device (undocumented):
- Los que estén “vivos en red” (last_seen < 7 días) y NO tengan device asignado.
- Aplica same veredictos → excluye si false_alarm (cuarentena).

Guarda snapshot diario:

RackIntegritySnapshot.objects.get_or_create(
    organization=org,
    rack=rack,
    date=date.today(),
    defaults={"fidelity_pct": fidelity_pct, "coverage_pct": coverage_pct}
)

2. gate_stats(org)

Propósito: Métrica de calibración del motor para el gate #202.

Firma:

def gate_stats(organization) -> dict

Retorna:

{
    "verdicts": {
        "plan_updated": 50,
        "real_change": 42,
        "false_alarm": 8,
    },
    "signal": 92,  # plan_updated + real_change
    "noise": 8,    # false_alarm
    "total": 100,
    "signal_pct": 92.0,  # Si total > 0, else None
}

Lógica:

rows = IntegrityAck.objects.filter(organization=organization) \
    .values("verdict").annotate(n=Count("id"))
by = {r["verdict"]: r["n"] for r in rows}

signal = by.get(VERDICT_PLAN_UPDATED, 0) + by.get(VERDICT_REAL_CHANGE, 0)
noise = by.get(VERDICT_FALSE_ALARM, 0)
total = signal + noise
signal_pct = 100.0 * signal / total if total else None

Interpretación:

signal_pct > 85% → Go para producción (motor es confiable).
signal_pct < 70% → NoGo (demasiado ruido, tuning requerido).
signal_pct vacío → aún sin veredictos (beta activa, esperando feedback).

Funciones privadas

_load_active_acks(org, now)

Carga veredictos vigentes (no mutados aún).

def _load_active_acks(organization, now) -> dict

Retorna:

{
    ("device", 42, "matched_stale"): ack_obj1,
    ("profile", 100, "undocumented"): ack_obj2,
    ...
}

Lógica:

qs = IntegrityAck.objects.filter(
    organization=organization,
    muted_until__gt=now
).order_by("created_at")

for ack in qs:
    key = ("device", ack.device_id, ack.kind) if ack.device_id else \
          ("profile", ack.profile_id, ack.kind)
    acks[key] = ack  # Si hay duplicados, el más reciente sobrescribe

Nota: Si hay varios acks para el mismo sujeto+kind, gana el más reciente (por orden de creación).

_ack_info(ack)

Serializa un IntegrityAck para respuesta JSON.

def _ack_info(ack) -> dict:
    return {
        "verdict": ack.verdict,
        "muted_until": ack.muted_until.isoformat(),
        "note": ack.note or None,
    }

_evidence(device, mip, profile, target)

Construye la cadena de evidencia de un hallazgo (qué datos llevaron a la conclusión).

def _evidence(device, mip, profile, target) -> dict:
    return {
        "device": {
            "name": device.name,
            "ip": mip.ip if mip else None,
            "mac": mip.mac if mip else None,
        },
        "profile": {
            "ip": profile.ip_address if profile else None,
            "hostname": profile.hostname if profile else None,
            "last_seen": _iso(_profile_last_seen(profile)),
        },
        "match_type": target.match_type if target else None,
    }

Constantes

# Tipos de hallazgo (cls, kind)
CLASS_MATCHED_ALIVE = "matched_alive"
CLASS_MATCHED_STALE = "matched_stale"
CLASS_IP_CONFLICT = "ip_conflict"
CLASS_UNOBSERVABLE = "unobservable"

# Veredictos (F2)
VERDICT_PLAN_UPDATED = "plan_updated"
VERDICT_REAL_CHANGE = "real_change"
VERDICT_FALSE_ALARM = "false_alarm"
VERDICTS = (VERDICT_PLAN_UPDATED, VERDICT_REAL_CHANGE, VERDICT_FALSE_ALARM)
ACK_KINDS = (CLASS_MATCHED_STALE, CLASS_IP_CONFLICT, "undocumented")

# Ventanas de silencio (días)
ACK_MUTE_DAYS = {
    VERDICT_PLAN_UPDATED: 3,
    VERDICT_REAL_CHANGE: 30,
    VERDICT_FALSE_ALARM: 30,
}

# Defaults
DEFAULT_FRESH_DAYS = 7  # DeviceProfile.last_seen < 7 días = vivo

Complejidad y performance

Queries principales

compute_integrity(org):
- Device.objects.filter(rack__organization=org, deleted_at__isnull=True) — O(devices).
- DeviceProfile.objects.filter(organization=org) — O(profiles).
- Match heurística — O(devices * profiles) en worst case (optimizable con índices).
- Load acks — O(acks) con índice org+muted_until.
- Total: O(devices * profiles) + índice opt. → ~100ms para 500 devices + 1000 profiles.
gate_stats(org):
- IntegrityAck.objects.filter(organization=org).values("verdict").annotate(Count) — O(acks) con índice.
- Total: ~5ms.

Almacenamiento

IntegrityAck: 1 fila por veredicto → crece con volumen de cambios (típicamente 50-200/mes por org).
RackIntegritySnapshot: 1 fila/rack/día → crece O(racks * days) (típicamente 30-200 MB/año).

Integración con API y vistas

GET /api/racks/integrity/overview

@router.get("/overview")
def overview(request):
    org = require_org(request)
    result = integrity.compute_integrity(org)
    history = RackIntegritySnapshot.objects.filter(...).order_by("-date")[:30][::-1]
    return {"result": result, "history": history, "gate": integrity.gate_stats(org)}

POST /api/racks/integrity/ack

@router.post("/ack")
def ack_finding(request, data: AckIn):
    org = require_org(request)
    require_perm(request, "racks", "edit")
    ack = IntegrityAck.objects.create(...)
    return {"ok": True, "ack_id": ack.id, "gate": integrity.gate_stats(org)}

Pruebas unitarias

10 tests en tests/racks/test_integrity_api.py:

test_gating_without_module — 403 sin módulo activo.
test_compute_integrity_basic — cómputo sin veredictos.
test_false_alarm_excludes_fidelity — false_alarm → muted, no castiga %.
test_gate_stats_signal_ratio — cálculo correcto de signal_pct.
test_ack_silences_by_verdict — muted_until correcto (3/30/30 días).
test_rls_isolation — usuarios nunca ven acks cross-org.

Véase también

[[feature—racks—integridad-f2-radar-interno]]
[[entity—racks—model—integrity-ack]]
[[entity—racks—endpoint—integrity-overview-ack]]
[[entity—racks—model—rack]]
[[entity—network—model—device-profile]]
[[concept—racks—calibracion-sensor]]
[[decision—20260717—integrity-apagado-por-defecto]]