fix: expand bgp pipeline and stabilize backend tests

backend/app/api/v1/bgp.py

@@ -8,7 +8,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.security import get_current_user
 from app.db.session import get_db
 from app.models.bgp_anomaly import BGPAnomaly
-from app.models.collected_data import CollectedData
+from app.models.bgp_incident import BGPIncident
+from app.models.bgp_observation import BGPObservation
 from app.models.user import User
 
 router = APIRouter()
@@ -48,12 +49,12 @@ async def list_bgp_events(
     db: AsyncSession = Depends(get_db),
 ):
     stmt = (
-        select(CollectedData)
-        .where(CollectedData.source.in_(BGP_SOURCES))
-        .order_by(CollectedData.reference_date.desc().nullslast(), CollectedData.id.desc())
+        select(BGPObservation)
+        .where(BGPObservation.source.in_(BGP_SOURCES))
+        .order_by(BGPObservation.observed_at.desc(), BGPObservation.id.desc())
     )
     if source:
-        stmt = stmt.where(CollectedData.source == source)
+        stmt = stmt.where(BGPObservation.source == source)
 
     result = await db.execute(stmt)
     records = result.scalars().all()
@@ -62,18 +63,17 @@ async def list_bgp_events(
 
     filtered = []
     for record in records:
-        metadata = record.extra_data or {}
-        if prefix and metadata.get("prefix") != prefix:
+        if prefix and record.prefix != prefix:
             continue
-        if origin_asn is not None and metadata.get("origin_asn") != origin_asn:
+        if origin_asn is not None and record.origin_asn != origin_asn:
             continue
-        if peer_asn is not None and metadata.get("peer_asn") != peer_asn:
+        if peer_asn is not None and record.peer_asn != peer_asn:
             continue
-        if collector and metadata.get("collector") != collector:
+        if collector and record.collector != collector:
             continue
-        if event_type and metadata.get("event_type") != event_type:
+        if event_type and record.event_type != event_type:
             continue
-        if (dt_from or dt_to) and not _matches_time(record.reference_date, dt_from, dt_to):
+        if (dt_from or dt_to) and not _matches_time(record.observed_at, dt_from, dt_to):
             continue
         filtered.append(record)
 
@@ -92,7 +92,7 @@ async def get_bgp_event(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
-    record = await db.get(CollectedData, event_id)
+    record = await db.get(BGPObservation, event_id)
     if not record or record.source not in BGP_SOURCES:
         raise HTTPException(status_code=404, detail="BGP event not found")
     return record.to_dict()
@@ -180,3 +180,44 @@ async def get_bgp_anomaly(
     if not record:
         raise HTTPException(status_code=404, detail="BGP anomaly not found")
     return record.to_dict()
+
+
+@router.get("/incidents")
+async def list_bgp_incidents(
+    severity: Optional[str] = Query(None),
+    incident_type: Optional[str] = Query(None),
+    status: Optional[str] = Query(None),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200),
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    stmt = select(BGPIncident).order_by(BGPIncident.created_at.desc(), BGPIncident.id.desc())
+    if severity:
+        stmt = stmt.where(BGPIncident.severity == severity)
+    if incident_type:
+        stmt = stmt.where(BGPIncident.incident_type == incident_type)
+    if status:
+        stmt = stmt.where(BGPIncident.status == status)
+
+    result = await db.execute(stmt)
+    records = result.scalars().all()
+    offset = (page - 1) * page_size
+    return {
+        "total": len(records),
+        "page": page,
+        "page_size": page_size,
+        "data": [record.to_dict() for record in records[offset : offset + page_size]],
+    }
+
+
+@router.get("/incidents/{incident_id}")
+async def get_bgp_incident(
+    incident_id: int,
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    record = await db.get(BGPIncident, incident_id)
+    if not record:
+        raise HTTPException(status_code=404, detail="BGP incident not found")
+    return record.to_dict()

backend/app/api/v1/visualization.py

@@ -202,6 +202,44 @@ def convert_satellite_to_geojson(records: List[CollectedData]) -> Dict[str, Any]
     return {"type": "FeatureCollection", "features": features}
 
 
+def dedupe_satellite_records(records: List[CollectedData]) -> List[CollectedData]:
+    """Keep only the newest record for each satellite identity."""
+    latest_by_key: Dict[str, CollectedData] = {}
+
+    for record in records:
+        metadata = record.extra_data or {}
+        norad_id = metadata.get("norad_cat_id")
+        dedupe_key = (
+            str(norad_id)
+            if norad_id not in (None, "")
+            else str(record.source_id or record.entity_key or record.name or record.id)
+        )
+
+        existing = latest_by_key.get(dedupe_key)
+        if existing is None or (record.id or 0) > (existing.id or 0):
+            latest_by_key[dedupe_key] = record
+
+    return sorted(latest_by_key.values(), key=lambda item: item.id or 0, reverse=True)
+
+
+def dedupe_collected_records(records: List[CollectedData]) -> List[CollectedData]:
+    """Keep only the newest record for each collected entity."""
+    latest_by_key: Dict[str, CollectedData] = {}
+
+    for record in records:
+        dedupe_key = str(
+            record.source_id
+            or record.entity_key
+            or record.name
+            or record.id
+        )
+        existing = latest_by_key.get(dedupe_key)
+        if existing is None or (record.id or 0) > (existing.id or 0):
+            latest_by_key[dedupe_key] = record
+
+    return sorted(latest_by_key.values(), key=lambda item: item.id or 0, reverse=True)
+
+
 def convert_supercomputer_to_geojson(records: List[CollectedData]) -> Dict[str, Any]:
     """Convert TOP500 supercomputer records to GeoJSON"""
     features = []
@@ -410,7 +448,7 @@ async def get_cables_geojson(db: AsyncSession = Depends(get_db)):
     try:
         stmt = select(CollectedData).where(CollectedData.source == "arcgis_cables")
         result = await db.execute(stmt)
-        records = result.scalars().all()
+        records = dedupe_collected_records(list(result.scalars().all()))
 
         if not records:
             raise HTTPException(
@@ -430,15 +468,15 @@ async def get_landing_points_geojson(db: AsyncSession = Depends(get_db)):
     try:
         landing_stmt = select(CollectedData).where(CollectedData.source == "arcgis_landing_points")
         landing_result = await db.execute(landing_stmt)
-        records = landing_result.scalars().all()
+        records = dedupe_collected_records(list(landing_result.scalars().all()))
         
         relation_stmt = select(CollectedData).where(CollectedData.source == "arcgis_cable_landing_relation")
         relation_result = await db.execute(relation_stmt)
-        relation_records = relation_result.scalars().all()
+        relation_records = dedupe_collected_records(list(relation_result.scalars().all()))
         
         cable_stmt = select(CollectedData).where(CollectedData.source == "arcgis_cables")
         cable_result = await db.execute(cable_stmt)
-        cable_records = cable_result.scalars().all()
+        cable_records = dedupe_collected_records(list(cable_result.scalars().all()))
         
         city_to_cable_ids_map = {}
         for rel in relation_records:
@@ -476,15 +514,15 @@ async def get_landing_points_geojson(db: AsyncSession = Depends(get_db)):
 async def get_all_geojson(db: AsyncSession = Depends(get_db)):
     cables_stmt = select(CollectedData).where(CollectedData.source == "arcgis_cables")
     cables_result = await db.execute(cables_stmt)
-    cables_records = cables_result.scalars().all()
+    cables_records = dedupe_collected_records(list(cables_result.scalars().all()))
 
     points_stmt = select(CollectedData).where(CollectedData.source == "arcgis_landing_points")
     points_result = await db.execute(points_stmt)
-    points_records = points_result.scalars().all()
+    points_records = dedupe_collected_records(list(points_result.scalars().all()))
     
     relation_stmt = select(CollectedData).where(CollectedData.source == "arcgis_cable_landing_relation")
     relation_result = await db.execute(relation_stmt)
-    relation_records = relation_result.scalars().all()
+    relation_records = dedupe_collected_records(list(relation_result.scalars().all()))
     
     city_to_cable_ids_map = {}
     for rel in relation_records:
@@ -542,10 +580,11 @@ async def get_satellites_geojson(
         .where(CollectedData.name != "Unknown")
         .order_by(CollectedData.id.desc())
     )
-    if limit is not None:
-        stmt = stmt.limit(limit)
     result = await db.execute(stmt)
-    records = result.scalars().all()
+    records = dedupe_satellite_records(list(result.scalars().all()))
+
+    if limit is not None:
+        records = records[:limit]
 
     if not records:
         return {"type": "FeatureCollection", "features": [], "count": 0}
@@ -567,10 +606,11 @@ async def get_supercomputers_geojson(
         select(CollectedData)
         .where(CollectedData.source == "top500")
         .where(CollectedData.name != "Unknown")
-        .limit(limit)
+        .order_by(CollectedData.id.desc())
     )
     result = await db.execute(stmt)
-    records = result.scalars().all()
+    records = dedupe_collected_records(list(result.scalars().all()))
+    records = records[:limit]
 
     if not records:
         return {"type": "FeatureCollection", "features": [], "count": 0}
@@ -592,10 +632,11 @@ async def get_gpu_clusters_geojson(
         select(CollectedData)
         .where(CollectedData.source == "epoch_ai_gpu")
         .where(CollectedData.name != "Unknown")
-        .limit(limit)
+        .order_by(CollectedData.id.desc())
     )
     result = await db.execute(stmt)
-    records = result.scalars().all()
+    records = dedupe_collected_records(list(result.scalars().all()))
+    records = records[:limit]
 
     if not records:
         return {"type": "FeatureCollection", "features": [], "count": 0}
@@ -645,11 +686,11 @@ async def get_all_visualization_data(db: AsyncSession = Depends(get_db)):
     """
     cables_stmt = select(CollectedData).where(CollectedData.source == "arcgis_cables")
     cables_result = await db.execute(cables_stmt)
-    cables_records = list(cables_result.scalars().all())
+    cables_records = dedupe_collected_records(list(cables_result.scalars().all()))
 
     points_stmt = select(CollectedData).where(CollectedData.source == "arcgis_landing_points")
     points_result = await db.execute(points_stmt)
-    points_records = list(points_result.scalars().all())
+    points_records = dedupe_collected_records(list(points_result.scalars().all()))
 
     satellites_stmt = (
         select(CollectedData)
@@ -657,7 +698,7 @@ async def get_all_visualization_data(db: AsyncSession = Depends(get_db)):
         .where(CollectedData.name != "Unknown")
     )
     satellites_result = await db.execute(satellites_stmt)
-    satellites_records = list(satellites_result.scalars().all())
+    satellites_records = dedupe_satellite_records(list(satellites_result.scalars().all()))
 
     supercomputers_stmt = (
         select(CollectedData)
@@ -665,7 +706,7 @@ async def get_all_visualization_data(db: AsyncSession = Depends(get_db)):
         .where(CollectedData.name != "Unknown")
     )
     supercomputers_result = await db.execute(supercomputers_stmt)
-    supercomputers_records = list(supercomputers_result.scalars().all())
+    supercomputers_records = dedupe_collected_records(list(supercomputers_result.scalars().all()))
 
     gpu_stmt = (
         select(CollectedData)
@@ -673,7 +714,7 @@ async def get_all_visualization_data(db: AsyncSession = Depends(get_db)):
         .where(CollectedData.name != "Unknown")
     )
     gpu_result = await db.execute(gpu_stmt)
-    gpu_records = list(gpu_result.scalars().all())
+    gpu_records = dedupe_collected_records(list(gpu_result.scalars().all()))
 
     cables = (
         convert_cable_to_geojson(cables_records)

backend/app/db/session.py

@@ -91,6 +91,8 @@ async def init_db():
     import app.models.datasource_config  # noqa: F401
     import app.models.alert  # noqa: F401
     import app.models.bgp_anomaly  # noqa: F401
+    import app.models.bgp_incident  # noqa: F401
+    import app.models.bgp_observation  # noqa: F401
     import app.models.collected_data  # noqa: F401
     import app.models.system_setting  # noqa: F401
 

backend/app/models/__init__.py

@@ -6,6 +6,8 @@ from app.models.datasource import DataSource
 from app.models.datasource_config import DataSourceConfig
 from app.models.alert import Alert, AlertSeverity, AlertStatus
 from app.models.bgp_anomaly import BGPAnomaly
+from app.models.bgp_incident import BGPIncident
+from app.models.bgp_observation import BGPObservation
 from app.models.system_setting import SystemSetting
 
 __all__ = [
@@ -20,4 +22,6 @@ __all__ = [
     "AlertSeverity",
     "AlertStatus",
     "BGPAnomaly",
+    "BGPIncident",
+    "BGPObservation",
 ]

backend/app/models/bgp_incident.py

@@ -0,0 +1,64 @@
+"""BGP incident model for aggregated routing events."""
+
+from datetime import datetime
+
+from sqlalchemy import Column, DateTime, Float, ForeignKey, Index, Integer, JSON, String, Text
+
+from app.core.time import to_iso8601_utc
+from app.db.session import Base
+
+
+class BGPIncident(Base):
+    __tablename__ = "bgp_incidents"
+
+    id = Column(Integer, primary_key=True, index=True)
+    snapshot_id = Column(Integer, ForeignKey("data_snapshots.id"), nullable=True, index=True)
+    task_id = Column(Integer, ForeignKey("collection_tasks.id"), nullable=True, index=True)
+    source = Column(String(100), nullable=False, index=True)
+    incident_key = Column(String(255), nullable=False, index=True)
+    incident_type = Column(String(50), nullable=False, index=True)
+    title = Column(String(255), nullable=False)
+    summary = Column(Text, nullable=False)
+    severity = Column(String(20), nullable=False, index=True)
+    status = Column(String(20), nullable=False, default="active", index=True)
+    confidence = Column(Float, nullable=False, default=0.5)
+    started_at = Column(DateTime(timezone=True), nullable=False, default=datetime.utcnow, index=True)
+    ended_at = Column(DateTime(timezone=True), nullable=True)
+    affected_prefixes = Column(JSON, default=list)
+    affected_asns = Column(JSON, default=list)
+    affected_collectors = Column(JSON, default=list)
+    affected_regions = Column(JSON, default=list)
+    related_cables = Column(JSON, default=list)
+    related_ixps = Column(JSON, default=list)
+    evidence_refs = Column(JSON, default=list)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=datetime.utcnow, index=True)
+
+    __table_args__ = (
+        Index("idx_bgp_incidents_source_created", "source", "created_at"),
+        Index("idx_bgp_incidents_type_status", "incident_type", "status"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "snapshot_id": self.snapshot_id,
+            "task_id": self.task_id,
+            "source": self.source,
+            "incident_key": self.incident_key,
+            "incident_type": self.incident_type,
+            "title": self.title,
+            "summary": self.summary,
+            "severity": self.severity,
+            "status": self.status,
+            "confidence": self.confidence,
+            "started_at": to_iso8601_utc(self.started_at),
+            "ended_at": to_iso8601_utc(self.ended_at),
+            "affected_prefixes": self.affected_prefixes or [],
+            "affected_asns": self.affected_asns or [],
+            "affected_collectors": self.affected_collectors or [],
+            "affected_regions": self.affected_regions or [],
+            "related_cables": self.related_cables or [],
+            "related_ixps": self.related_ixps or [],
+            "evidence_refs": self.evidence_refs or [],
+            "created_at": to_iso8601_utc(self.created_at),
+        }

backend/app/models/bgp_observation.py

@@ -0,0 +1,62 @@
+"""BGP raw observation model for routing event ingestion."""
+
+from sqlalchemy import Column, DateTime, ForeignKey, Index, Integer, JSON, String, Text
+from sqlalchemy.sql import func
+
+from app.core.time import to_iso8601_utc
+from app.db.session import Base
+
+
+class BGPObservation(Base):
+    __tablename__ = "bgp_observations"
+
+    id = Column(Integer, primary_key=True, index=True)
+    snapshot_id = Column(Integer, ForeignKey("data_snapshots.id"), nullable=True, index=True)
+    task_id = Column(Integer, ForeignKey("collection_tasks.id"), nullable=True, index=True)
+    source = Column(String(100), nullable=False, index=True)
+    ingest_batch_id = Column(String(100), nullable=True, index=True)
+    source_event_id = Column(String(100), nullable=True, index=True)
+    collector = Column(String(100), nullable=True, index=True)
+    peer_asn = Column(Integer, nullable=True, index=True)
+    peer_ip = Column(String(100), nullable=True)
+    prefix = Column(String(64), nullable=True, index=True)
+    event_type = Column(String(32), nullable=False, index=True)
+    as_path = Column(JSON, default=list)
+    origin_asn = Column(Integer, nullable=True, index=True)
+    next_hop = Column(String(100), nullable=True)
+    communities = Column(JSON, default=list)
+    observed_at = Column(DateTime(timezone=True), nullable=False, index=True)
+    collector_geo = Column(JSON, default=dict)
+    raw_payload = Column(JSON, default=dict)
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), index=True)
+    note = Column(Text, nullable=True)
+
+    __table_args__ = (
+        Index("idx_bgp_obs_source_observed", "source", "observed_at"),
+        Index("idx_bgp_obs_collector_prefix", "collector", "prefix"),
+        Index("idx_bgp_obs_task_source_event", "task_id", "source_event_id"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "snapshot_id": self.snapshot_id,
+            "task_id": self.task_id,
+            "source": self.source,
+            "ingest_batch_id": self.ingest_batch_id,
+            "source_event_id": self.source_event_id,
+            "collector": self.collector,
+            "peer_asn": self.peer_asn,
+            "peer_ip": self.peer_ip,
+            "prefix": self.prefix,
+            "event_type": self.event_type,
+            "as_path": self.as_path or [],
+            "origin_asn": self.origin_asn,
+            "next_hop": self.next_hop,
+            "communities": self.communities or [],
+            "observed_at": to_iso8601_utc(self.observed_at),
+            "collector_geo": self.collector_geo or {},
+            "raw_payload": self.raw_payload or {},
+            "created_at": to_iso8601_utc(self.created_at),
+            "note": self.note,
+        }

backend/app/services/bgp_detectors.py

@@ -0,0 +1,193 @@
+"""Detector helpers for BGP anomaly generation."""
+
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+from datetime import UTC, datetime
+from typing import Any
+
+from app.models.bgp_anomaly import BGPAnomaly
+
+
+def detect_origin_change_anomalies(
+    *,
+    source: str,
+    snapshot_id: int | None,
+    task_id: int | None,
+    events: list[dict[str, Any]],
+    previous_origin_map: dict[str, set[int]],
+) -> list[BGPAnomaly]:
+    prefix_to_origins: defaultdict[str, set[int]] = defaultdict(set)
+    for event in events:
+        metadata = event.get("metadata") or {}
+        prefix = metadata.get("prefix")
+        origin_asn = metadata.get("origin_asn")
+        if prefix and origin_asn is not None:
+            prefix_to_origins[str(prefix)].add(int(origin_asn))
+
+    anomalies: list[BGPAnomaly] = []
+    for prefix, origins in prefix_to_origins.items():
+        historic = previous_origin_map.get(prefix, set())
+        new_origins = sorted(origin for origin in origins if origin not in historic)
+        if not historic or not new_origins:
+            continue
+
+        for new_origin in new_origins:
+            sample_event = next(
+                (
+                    event
+                    for event in events
+                    if (event.get("metadata") or {}).get("prefix") == prefix
+                    and int((event.get("metadata") or {}).get("origin_asn") or -1) == new_origin
+                ),
+                {},
+            )
+            sample_metadata = sample_event.get("metadata") or {}
+            sample_enrichment = sample_metadata.get("enrichment") or {}
+            anomalies.append(
+                BGPAnomaly(
+                    snapshot_id=snapshot_id,
+                    task_id=task_id,
+                    source=source,
+                    anomaly_type="origin_change",
+                    severity="critical",
+                    status="active",
+                    entity_key=f"origin_change:{prefix}:{new_origin}",
+                    prefix=prefix,
+                    origin_asn=sorted(historic)[0],
+                    new_origin_asn=new_origin,
+                    peer_scope=[],
+                    started_at=datetime.now(UTC),
+                    confidence=0.86,
+                    summary=f"Prefix {prefix} is now originated by AS{new_origin}, outside the current baseline.",
+                    evidence={
+                        "previous_origins": sorted(historic),
+                        "current_origins": sorted(origins),
+                        "events": [sample_metadata] if sample_metadata else [],
+                        "origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
+                        "new_origin_asn_profile": sample_enrichment.get("new_origin_asn_profile"),
+                        "rpki_validation": sample_enrichment.get("rpki_validation"),
+                        "prefix_scope": sample_enrichment.get("prefix_scope"),
+                        "impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
+                    },
+                )
+            )
+
+    return anomalies
+
+
+def detect_more_specific_burst_anomalies(
+    *,
+    source: str,
+    snapshot_id: int | None,
+    task_id: int | None,
+    events: list[dict[str, Any]],
+) -> list[BGPAnomaly]:
+    prefix_to_more_specifics: defaultdict[str, list[dict[str, Any]]] = defaultdict(list)
+    for event in events:
+        metadata = event.get("metadata") or {}
+        prefix = metadata.get("prefix")
+        enrichment = metadata.get("enrichment") or {}
+        if prefix and enrichment.get("is_more_specific"):
+            prefix_to_more_specifics[str(prefix).split("/")[0]].append(event)
+
+    anomalies: list[BGPAnomaly] = []
+    for root_prefix, more_specifics in prefix_to_more_specifics.items():
+        if len(more_specifics) < 2:
+            continue
+
+        sample = more_specifics[0].get("metadata") or {}
+        sample_enrichment = sample.get("enrichment") or {}
+        anomalies.append(
+            BGPAnomaly(
+                snapshot_id=snapshot_id,
+                task_id=task_id,
+                source=source,
+                anomaly_type="more_specific_burst",
+                severity="high",
+                status="active",
+                entity_key=f"more_specific_burst:{root_prefix}:{len(more_specifics)}",
+                prefix=sample.get("prefix"),
+                origin_asn=sample.get("origin_asn"),
+                new_origin_asn=None,
+                peer_scope=sorted(
+                    {
+                        str(item.get("metadata", {}).get("collector") or "")
+                        for item in more_specifics
+                        if item.get("metadata", {}).get("collector")
+                    }
+                ),
+                started_at=datetime.now(UTC),
+                confidence=0.72,
+                summary=f"{len(more_specifics)} more-specific announcements clustered around {root_prefix}.",
+                evidence={
+                    "events": [item.get("metadata") for item in more_specifics[:10]],
+                    "rpki_validation": sample_enrichment.get("rpki_validation"),
+                    "origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
+                    "prefix_scope": sample_enrichment.get("prefix_scope"),
+                    "impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
+                },
+            )
+        )
+
+    return anomalies
+
+
+def detect_mass_withdrawal_anomalies(
+    *,
+    source: str,
+    snapshot_id: int | None,
+    task_id: int | None,
+    events: list[dict[str, Any]],
+) -> list[BGPAnomaly]:
+    withdrawal_counter: Counter[tuple[str, int | None]] = Counter()
+    for event in events:
+        metadata = event.get("metadata") or {}
+        prefix = metadata.get("prefix")
+        if prefix and metadata.get("event_type") == "withdrawal":
+            withdrawal_counter[(str(prefix), metadata.get("origin_asn"))] += 1
+
+    anomalies: list[BGPAnomaly] = []
+    for (prefix, origin_asn), count in withdrawal_counter.items():
+        if count < 3:
+            continue
+        sample_event = next(
+            (
+                event
+                for event in events
+                if (event.get("metadata") or {}).get("prefix") == prefix
+                and (event.get("metadata") or {}).get("event_type") == "withdrawal"
+            ),
+            {},
+        )
+        sample_metadata = sample_event.get("metadata") or {}
+        sample_enrichment = sample_metadata.get("enrichment") or {}
+
+        anomalies.append(
+            BGPAnomaly(
+                snapshot_id=snapshot_id,
+                task_id=task_id,
+                source=source,
+                anomaly_type="mass_withdrawal",
+                severity="high" if count < 8 else "critical",
+                status="active",
+                entity_key=f"mass_withdrawal:{prefix}:{origin_asn}:{count}",
+                prefix=prefix,
+                origin_asn=origin_asn,
+                new_origin_asn=None,
+                peer_scope=[],
+                started_at=datetime.now(UTC),
+                confidence=min(0.55 + (count * 0.05), 0.95),
+                summary=f"{count} withdrawal events observed for {prefix} in the current ingest window.",
+                evidence={
+                    "withdrawal_count": count,
+                    "events": [sample_metadata] if sample_metadata else [],
+                    "origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
+                    "rpki_validation": sample_enrichment.get("rpki_validation"),
+                    "prefix_scope": sample_enrichment.get("prefix_scope"),
+                    "impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
+                },
+            )
+        )
+
+    return anomalies

backend/app/services/bgp_enrichment.py

@@ -0,0 +1,280 @@
+"""Enrichment helpers for BGP observation and anomaly pipelines."""
+
+from __future__ import annotations
+
+import ipaddress
+from collections import defaultdict
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.bgp_observation import BGPObservation
+from app.models.collected_data import CollectedData
+
+
+def _safe_int(value: Any) -> int | None:
+    try:
+        if value in (None, ""):
+            return None
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _parse_timestamp(value: Any) -> datetime:
+    if isinstance(value, datetime):
+        return value.astimezone(UTC) if value.tzinfo else value.replace(tzinfo=UTC)
+
+    if isinstance(value, (int, float)):
+        return datetime.fromtimestamp(value, tz=UTC)
+
+    if isinstance(value, str) and value:
+        normalized = value.replace("Z", "+00:00")
+        parsed = datetime.fromisoformat(normalized)
+        return parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC)
+
+    return datetime.now(UTC)
+
+
+def _dedupe_as_path(as_path: list[int]) -> list[int]:
+    deduped: list[int] = []
+    for asn in as_path:
+        if not deduped or deduped[-1] != asn:
+            deduped.append(asn)
+    return deduped
+
+
+def _compact_locations(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    results: list[dict[str, Any]] = []
+    seen: set[tuple[Any, ...]] = set()
+    for item in items:
+        key = (
+            item.get("country"),
+            item.get("city"),
+            item.get("latitude"),
+            item.get("longitude"),
+        )
+        if key in seen:
+            continue
+        seen.add(key)
+        results.append(item)
+    return results
+
+
+def extract_bgp_network_fields(prefix: str) -> dict[str, Any]:
+    if not prefix:
+        return {
+            "prefix_family": None,
+            "prefix_length": None,
+            "prefix_supernet": None,
+            "is_more_specific": False,
+        }
+
+    try:
+        network = ipaddress.ip_network(prefix, strict=False)
+    except ValueError:
+        return {
+            "prefix_family": None,
+            "prefix_length": None,
+            "prefix_supernet": None,
+            "is_more_specific": False,
+        }
+
+    supernet_prefix = 16 if network.version == 4 else 32
+    if network.prefixlen > supernet_prefix:
+        prefix_supernet = str(network.supernet(new_prefix=supernet_prefix))
+    else:
+        prefix_supernet = str(network)
+
+    return {
+        "prefix_family": f"ipv{network.version}",
+        "prefix_length": int(network.prefixlen),
+        "prefix_supernet": prefix_supernet,
+        "is_more_specific": network.prefixlen > (24 if network.version == 4 else 48),
+    }
+
+
+async def enrich_bgp_events_for_batch(
+    db: AsyncSession,
+    *,
+    source: str,
+    events: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    if not events:
+        return []
+
+    prefixes = {
+        str((event.get("metadata") or {}).get("prefix") or "").strip()
+        for event in events
+        if (event.get("metadata") or {}).get("prefix")
+    }
+    prefix_values = sorted(prefix for prefix in prefixes if prefix)
+    origin_asns = sorted(
+        {
+            asn
+            for event in events
+            for asn in [
+                _safe_int((event.get("metadata") or {}).get("origin_asn")),
+                _safe_int((event.get("metadata") or {}).get("new_origin_asn")),
+            ]
+            if asn is not None
+        }
+    )
+
+    historical_prefix_baseline: dict[str, dict[str, Any]] = {}
+    if prefix_values:
+        previous_result = await db.execute(
+            select(BGPObservation).where(
+                BGPObservation.source == source,
+                BGPObservation.prefix.in_(prefix_values),
+            )
+        )
+        by_prefix: defaultdict[str, list[BGPObservation]] = defaultdict(list)
+        for observation in previous_result.scalars().all():
+            if observation.prefix:
+                by_prefix[observation.prefix].append(observation)
+
+        for prefix, observations in by_prefix.items():
+            unique_origins = sorted(
+                {
+                    observation.origin_asn
+                    for observation in observations
+                    if observation.origin_asn is not None
+                }
+            )
+            unique_collectors = sorted(
+                {
+                    observation.collector
+                    for observation in observations
+                    if observation.collector
+                }
+            )
+            historical_prefix_baseline[prefix] = {
+                "historical_origin_asns": unique_origins,
+                "historical_collectors": unique_collectors,
+                "historical_observation_count": len(observations),
+                "historical_regions": _compact_locations(
+                    [
+                        observation.collector_geo or {}
+                        for observation in observations
+                        if observation.collector_geo
+                    ]
+                ),
+            }
+
+    asn_profiles: dict[int, dict[str, Any]] = {}
+    if origin_asns:
+        peeringdb_result = await db.execute(
+            select(CollectedData).where(CollectedData.source == "peeringdb_network")
+        )
+        for record in peeringdb_result.scalars().all():
+            metadata = record.extra_data or {}
+            asn = _safe_int(metadata.get("asn"))
+            if asn is None or asn not in origin_asns:
+                continue
+            current = asn_profiles.get(asn)
+            if current and (current.get("id") or 0) > (record.id or 0):
+                continue
+            asn_profiles[asn] = {
+                "id": record.id,
+                "asn": asn,
+                "name": record.name,
+                "country": metadata.get("country"),
+                "city": metadata.get("city"),
+                "source": "peeringdb_network",
+                "info_type": metadata.get("info_type"),
+                "info_traffic": metadata.get("info_traffic"),
+                "info_ratio": metadata.get("info_ratio"),
+                "ix_count": metadata.get("ix_count"),
+                "url": metadata.get("url"),
+            }
+
+    collector_counts: defaultdict[str, int] = defaultdict(int)
+    for event in events:
+        collector = (event.get("metadata") or {}).get("collector")
+        if collector:
+            collector_counts[str(collector)] += 1
+
+    enriched: list[dict[str, Any]] = []
+    for event in events:
+        metadata = dict(event.get("metadata") or {})
+        prefix = str(metadata.get("prefix") or "").strip()
+        as_path = metadata.get("as_path") or []
+        normalized_as_path = [asn for asn in (_safe_int(item) for item in as_path) if asn is not None]
+        deduped_as_path = _dedupe_as_path(normalized_as_path)
+        collector = str(metadata.get("collector") or "").strip()
+        collector_location = metadata.get("collector_location") or {}
+        baseline = historical_prefix_baseline.get(prefix, {})
+        observed_at = _parse_timestamp(metadata.get("timestamp") or event.get("reference_date"))
+        origin_asn = _safe_int(metadata.get("origin_asn"))
+        new_origin_asn = _safe_int(metadata.get("new_origin_asn"))
+        observed_regions = _compact_locations(
+            [
+                {
+                    "country": collector_location.get("country"),
+                    "city": collector_location.get("city"),
+                    "latitude": collector_location.get("latitude"),
+                    "longitude": collector_location.get("longitude"),
+                }
+            ]
+        )
+        baseline_regions = baseline.get("historical_regions", [])
+        prefix_scope_regions = _compact_locations([*observed_regions, *baseline_regions])
+
+        enrichment = {
+            **extract_bgp_network_fields(prefix),
+            "observed_at": observed_at.isoformat(),
+            "normalized_as_path": normalized_as_path,
+            "deduped_as_path": deduped_as_path,
+            "deduped_as_path_length": len(deduped_as_path),
+            "path_prepending": len(normalized_as_path) > len(deduped_as_path),
+            "collector_region": {
+                "city": collector_location.get("city"),
+                "country": collector_location.get("country"),
+            },
+            "collector_observation_count_in_batch": collector_counts.get(collector, 0),
+            "batch_visibility_collectors": sorted(collector_counts.keys()),
+            "prefix_baseline": baseline,
+            "is_new_origin_for_prefix": (
+                origin_asn is not None
+                and origin_asn
+                not in set(baseline.get("historical_origin_asns", []))
+            ),
+            "rpki_validation": {
+                "status": "unknown",
+                "reason": "no_rpki_roa_dataset_configured",
+            },
+            "origin_asn_profile": asn_profiles.get(origin_asn),
+            "new_origin_asn_profile": asn_profiles.get(new_origin_asn),
+            "prefix_scope": {
+                "countries": sorted(
+                    {
+                        item.get("country")
+                        for item in prefix_scope_regions
+                        if item.get("country")
+                    }
+                ),
+                "cities": sorted(
+                    {
+                        item.get("city")
+                        for item in prefix_scope_regions
+                        if item.get("city")
+                    }
+                ),
+                "regions": prefix_scope_regions,
+            },
+        }
+
+        enriched.append(
+            {
+                **event,
+                "metadata": {
+                    **metadata,
+                    "enrichment": enrichment,
+                },
+            }
+        )
+
+    return enriched

backend/app/services/bgp_incidents.py

@@ -0,0 +1,151 @@
+"""Incident aggregation helpers for BGP anomalies."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.bgp_anomaly import BGPAnomaly
+from app.models.bgp_incident import BGPIncident
+
+
+def _severity_rank(value: str | None) -> int:
+    mapping = {"critical": 4, "high": 3, "medium": 2, "low": 1, "info": 0}
+    return mapping.get(str(value or "").lower(), 0)
+
+
+def _pick_severity(values: list[str]) -> str:
+    ordered = sorted(values, key=_severity_rank, reverse=True)
+    return ordered[0] if ordered else "medium"
+
+
+def _collector_regions_from_anomaly(anomaly: BGPAnomaly) -> list[dict]:
+    evidence = anomaly.evidence or {}
+    regions = evidence.get("impacted_regions") or []
+    if regions:
+        return regions
+
+    collected = []
+    for item in evidence.get("events") or []:
+        collector = item.get("collector")
+        location = item.get("collector_location") or {}
+        if collector or location:
+            collected.append(
+                {
+                    "collector": collector,
+                    "country": location.get("country"),
+                    "city": location.get("city"),
+                    "latitude": location.get("latitude"),
+                    "longitude": location.get("longitude"),
+                }
+            )
+    return collected
+
+
+async def create_bgp_incidents_for_anomalies(
+    db: AsyncSession,
+    *,
+    source: str,
+    snapshot_id: int | None,
+    task_id: int | None,
+    anomalies: list[BGPAnomaly],
+) -> int:
+    if not anomalies:
+        return 0
+
+    grouped: dict[str, list[BGPAnomaly]] = {}
+    for anomaly in anomalies:
+        incident_key = f"{anomaly.anomaly_type}:{anomaly.prefix or 'unknown'}:{anomaly.new_origin_asn or anomaly.origin_asn or 'na'}"
+        grouped.setdefault(incident_key, []).append(anomaly)
+
+    existing_result = await db.execute(
+        select(BGPIncident.incident_key).where(BGPIncident.incident_key.in_(sorted(grouped.keys())))
+    )
+    existing_keys = {row[0] for row in existing_result.fetchall()}
+
+    created = 0
+    for incident_key, items in grouped.items():
+        if incident_key in existing_keys:
+            continue
+
+        items = sorted(items, key=lambda item: item.created_at or item.started_at or datetime.now(UTC))
+        primary = items[0]
+        prefixes = sorted({item.prefix for item in items if item.prefix})
+        asns = sorted(
+            {
+                asn
+                for item in items
+                for asn in [item.origin_asn, item.new_origin_asn]
+                if asn is not None
+            }
+        )
+        collectors = sorted(
+            {
+                collector
+                for item in items
+                for collector in (item.peer_scope or [])
+                if collector
+            }
+        )
+        regions: list[dict] = []
+        seen_regions: set[tuple] = set()
+        for item in items:
+            for region in _collector_regions_from_anomaly(item):
+                region_key = (
+                    region.get("collector"),
+                    region.get("country"),
+                    region.get("city"),
+                )
+                if region_key in seen_regions:
+                    continue
+                seen_regions.add(region_key)
+                regions.append(region)
+
+        if not collectors:
+            collectors = sorted(
+                {
+                    region.get("collector")
+                    for region in regions
+                    if region.get("collector")
+                }
+            )
+
+        evidence_refs = [item.entity_key for item in items if item.entity_key]
+        severity = _pick_severity([item.severity for item in items])
+        confidence = max((item.confidence or 0.0) for item in items)
+        title = f"{primary.anomaly_type.replace('_', ' ').title()} incident on {primary.prefix or 'unknown prefix'}"
+        summary = (
+            f"{len(items)} anomaly signal(s) grouped into one {primary.anomaly_type} incident, "
+            f"affecting {len(prefixes) or 1} prefix scope(s) across {len(collectors)} collector(s)."
+        )
+
+        db.add(
+            BGPIncident(
+                snapshot_id=snapshot_id,
+                task_id=task_id,
+                source=source,
+                incident_key=incident_key,
+                incident_type=primary.anomaly_type,
+                title=title,
+                summary=summary,
+                severity=severity,
+                status="active",
+                confidence=confidence,
+                started_at=primary.started_at or datetime.now(UTC),
+                affected_prefixes=prefixes,
+                affected_asns=asns,
+                affected_collectors=collectors,
+                affected_regions=regions,
+                related_cables=[],
+                related_ixps=[],
+                evidence_refs=evidence_refs,
+            )
+        )
+        created += 1
+
+    if created:
+        await db.commit()
+
+    return created

backend/app/services/collectors/bgp_common.py

@@ -3,8 +3,7 @@
 from __future__ import annotations
 
 import hashlib
-import ipaddress
-from collections import Counter, defaultdict
+from collections import defaultdict
 from datetime import UTC, datetime
 from typing import Any
 
@@ -12,7 +11,15 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.bgp_anomaly import BGPAnomaly
+from app.models.bgp_observation import BGPObservation
 from app.models.collected_data import CollectedData
+from app.services.bgp_incidents import create_bgp_incidents_for_anomalies
+from app.services.bgp_detectors import (
+    detect_mass_withdrawal_anomalies,
+    detect_more_specific_burst_anomalies,
+    detect_origin_change_anomalies,
+)
+from app.services.bgp_enrichment import enrich_bgp_events_for_batch, extract_bgp_network_fields
 
 
 RIPE_RIS_COLLECTOR_COORDS: dict[str, dict[str, Any]] = {
@@ -105,6 +112,10 @@ def normalize_bgp_event(payload: dict[str, Any], *, project: str) -> dict[str, A
     timestamp = _parse_timestamp(payload.get("timestamp") or payload.get("time") or payload.get("ts"))
     collector = str(payload.get("collector") or payload.get("host") or payload.get("router") or "unknown")
     peer_asn = _safe_int(payload.get("peer_asn") or payload.get("peer"))
+    peer_ip = payload.get("peer_ip") or payload.get("peer_address")
+    if peer_ip in (None, ""):
+        peer_candidate = payload.get("peer")
+        peer_ip = str(peer_candidate) if isinstance(peer_candidate, str) and ":" in peer_candidate else peer_candidate
     origin_asn = _safe_int(payload.get("origin_asn")) or (as_path[-1] if as_path else None)
     source_material = "|".join(
         [
@@ -118,37 +129,33 @@ def normalize_bgp_event(payload: dict[str, Any], *, project: str) -> dict[str, A
     )
     source_id = hashlib.sha1(source_material.encode("utf-8")).hexdigest()[:24]
 
-    prefix_length = None
-    is_more_specific = False
-    if prefix:
-        try:
-            network = ipaddress.ip_network(prefix, strict=False)
-            prefix_length = int(network.prefixlen)
-            is_more_specific = prefix_length > (24 if network.version == 4 else 48)
-        except ValueError:
-            prefix_length = None
-
     collector_location = RIPE_RIS_COLLECTOR_COORDS.get(collector, {})
+    network_fields = extract_bgp_network_fields(prefix)
     metadata = {
         "project": project,
         "collector": collector,
         "peer_asn": peer_asn,
-        "peer_ip": payload.get("peer_ip") or payload.get("peer_address"),
+        "peer_ip": peer_ip,
         "event_type": event_type,
         "prefix": prefix,
         "origin_asn": origin_asn,
         "as_path": as_path,
-        "communities": payload.get("communities") or payload.get("attrs", {}).get("communities") or [],
+        "communities": payload.get("communities")
+        or payload.get("community")
+        or payload.get("attrs", {}).get("communities")
+        or [],
         "next_hop": payload.get("next_hop") or payload.get("attrs", {}).get("next_hop"),
         "med": payload.get("med") or payload.get("attrs", {}).get("med"),
         "local_pref": payload.get("local_pref") or payload.get("attrs", {}).get("local_pref"),
         "timestamp": timestamp.isoformat(),
         "as_path_length": len(as_path),
-        "prefix_length": prefix_length,
-        "is_more_specific": is_more_specific,
         "visibility_weight": 1,
         "collector_location": collector_location,
         "raw_message": raw_message,
+        "prefix_family": network_fields.get("prefix_family"),
+        "prefix_length": network_fields.get("prefix_length"),
+        "prefix_supernet": network_fields.get("prefix_supernet"),
+        "is_more_specific": network_fields.get("is_more_specific", False),
     }
 
     return {
@@ -165,6 +172,57 @@ def normalize_bgp_event(payload: dict[str, Any], *, project: str) -> dict[str, A
     }
 
 
+async def save_bgp_observations_for_batch(
+    db: AsyncSession,
+    *,
+    source: str,
+    snapshot_id: int | None,
+    task_id: int | None,
+    events: list[dict[str, Any]],
+) -> int:
+    if not events:
+        return 0
+
+    ingest_batch_id = f"{source}:{task_id or 'adhoc'}:{snapshot_id or 'nosnapshot'}"
+    created = 0
+
+    for event in events:
+        metadata = event.get("metadata", {}) or {}
+        collector_location = metadata.get("collector_location") or {}
+        observed_at = _parse_timestamp(
+            metadata.get("timestamp") or event.get("reference_date")
+        )
+
+        db.add(
+            BGPObservation(
+                snapshot_id=snapshot_id,
+                task_id=task_id,
+                source=source,
+                ingest_batch_id=ingest_batch_id,
+                source_event_id=event.get("source_id"),
+                collector=metadata.get("collector"),
+                peer_asn=_safe_int(metadata.get("peer_asn")),
+                peer_ip=metadata.get("peer_ip"),
+                prefix=metadata.get("prefix"),
+                event_type=str(metadata.get("event_type") or "announcement"),
+                as_path=metadata.get("as_path") or [],
+                origin_asn=_safe_int(metadata.get("origin_asn")),
+                next_hop=metadata.get("next_hop"),
+                communities=metadata.get("communities") or [],
+                observed_at=observed_at,
+                collector_geo=collector_location,
+                raw_payload=metadata.get("raw_message") or {},
+                note=event.get("description"),
+            )
+        )
+        created += 1
+
+    if created:
+        await db.commit()
+
+    return created
+
+
 async def create_bgp_anomalies_for_batch(
     db: AsyncSession,
     *,
@@ -176,12 +234,17 @@ async def create_bgp_anomalies_for_batch(
     if not events:
         return 0
 
-    pending_anomalies: list[BGPAnomaly] = []
-    prefix_to_origins: defaultdict[str, set[int]] = defaultdict(set)
-    prefix_to_more_specifics: defaultdict[str, list[dict[str, Any]]] = defaultdict(list)
-    withdrawal_counter: Counter[tuple[str, int | None]] = Counter()
+    enriched_events = await enrich_bgp_events_for_batch(
+        db,
+        source=source,
+        events=events,
+    )
 
-    prefixes = {event["metadata"].get("prefix") for event in events if event.get("metadata", {}).get("prefix")}
+    prefixes = {
+        event["metadata"].get("prefix")
+        for event in enriched_events
+        if event.get("metadata", {}).get("prefix")
+    }
     previous_origin_map: dict[str, set[int]] = defaultdict(set)
 
     if prefixes:
@@ -199,97 +262,27 @@ async def create_bgp_anomalies_for_batch(
             if prefix and origin is not None:
                 previous_origin_map[prefix].add(origin)
 
-    for event in events:
-        metadata = event.get("metadata", {})
-        prefix = metadata.get("prefix")
-        origin_asn = _safe_int(metadata.get("origin_asn"))
-        if not prefix:
-            continue
-
-        if origin_asn is not None:
-            prefix_to_origins[prefix].add(origin_asn)
-
-        if metadata.get("is_more_specific"):
-            prefix_to_more_specifics[prefix.split("/")[0]].append(event)
-
-        if metadata.get("event_type") == "withdrawal":
-            withdrawal_counter[(prefix, origin_asn)] += 1
-
-    for prefix, origins in prefix_to_origins.items():
-        historic = previous_origin_map.get(prefix, set())
-        new_origins = sorted(origin for origin in origins if origin not in historic)
-        if historic and new_origins:
-            for new_origin in new_origins:
-                pending_anomalies.append(
-                    BGPAnomaly(
-                    snapshot_id=snapshot_id,
-                    task_id=task_id,
-                    source=source,
-                    anomaly_type="origin_change",
-                    severity="critical",
-                    status="active",
-                    entity_key=f"origin_change:{prefix}:{new_origin}",
-                    prefix=prefix,
-                    origin_asn=sorted(historic)[0],
-                    new_origin_asn=new_origin,
-                    peer_scope=[],
-                    started_at=datetime.now(UTC),
-                    confidence=0.86,
-                    summary=f"Prefix {prefix} is now originated by AS{new_origin}, outside the current baseline.",
-                    evidence={"previous_origins": sorted(historic), "current_origins": sorted(origins)},
-                )
-                )
-
-    for root_prefix, more_specifics in prefix_to_more_specifics.items():
-        if len(more_specifics) >= 2:
-            sample = more_specifics[0]["metadata"]
-            pending_anomalies.append(
-                BGPAnomaly(
-                snapshot_id=snapshot_id,
-                task_id=task_id,
-                source=source,
-                anomaly_type="more_specific_burst",
-                severity="high",
-                status="active",
-                entity_key=f"more_specific_burst:{root_prefix}:{len(more_specifics)}",
-                prefix=sample.get("prefix"),
-                origin_asn=_safe_int(sample.get("origin_asn")),
-                new_origin_asn=None,
-                peer_scope=sorted(
-                    {
-                        str(item.get("metadata", {}).get("collector") or "")
-                        for item in more_specifics
-                        if item.get("metadata", {}).get("collector")
-                    }
-                ),
-                started_at=datetime.now(UTC),
-                confidence=0.72,
-                summary=f"{len(more_specifics)} more-specific announcements clustered around {root_prefix}.",
-                evidence={"events": [item.get("metadata") for item in more_specifics[:10]]},
-            )
-            )
-
-    for (prefix, origin_asn), count in withdrawal_counter.items():
-        if count >= 3:
-            pending_anomalies.append(
-                BGPAnomaly(
-                snapshot_id=snapshot_id,
-                task_id=task_id,
-                source=source,
-                anomaly_type="mass_withdrawal",
-                severity="high" if count < 8 else "critical",
-                status="active",
-                entity_key=f"mass_withdrawal:{prefix}:{origin_asn}:{count}",
-                prefix=prefix,
-                origin_asn=origin_asn,
-                new_origin_asn=None,
-                peer_scope=[],
-                started_at=datetime.now(UTC),
-                confidence=min(0.55 + (count * 0.05), 0.95),
-                summary=f"{count} withdrawal events observed for {prefix} in the current ingest window.",
-                evidence={"withdrawal_count": count},
-            )
-            )
+    pending_anomalies = [
+        *detect_origin_change_anomalies(
+            source=source,
+            snapshot_id=snapshot_id,
+            task_id=task_id,
+            events=enriched_events,
+            previous_origin_map=previous_origin_map,
+        ),
+        *detect_more_specific_burst_anomalies(
+            source=source,
+            snapshot_id=snapshot_id,
+            task_id=task_id,
+            events=enriched_events,
+        ),
+        *detect_mass_withdrawal_anomalies(
+            source=source,
+            snapshot_id=snapshot_id,
+            task_id=task_id,
+            events=enriched_events,
+        ),
+    ]
 
     if not pending_anomalies:
         return 0
@@ -302,12 +295,21 @@ async def create_bgp_anomalies_for_batch(
     existing_keys = {row[0] for row in existing_result.fetchall()}
 
     created = 0
+    created_anomalies: list[BGPAnomaly] = []
     for anomaly in pending_anomalies:
         if anomaly.entity_key in existing_keys:
             continue
         db.add(anomaly)
+        created_anomalies.append(anomaly)
         created += 1
 
     if created:
         await db.commit()
+        await create_bgp_incidents_for_anomalies(
+            db,
+            source=source,
+            snapshot_id=snapshot_id,
+            task_id=task_id,
+            anomalies=created_anomalies,
+        )
     return created

backend/app/services/collectors/bgpstream.py

@@ -10,7 +10,11 @@ import urllib.request
 from typing import Any
 
 from app.services.collectors.base import BaseCollector
-from app.services.collectors.bgp_common import create_bgp_anomalies_for_batch, normalize_bgp_event
+from app.services.collectors.bgp_common import (
+    create_bgp_anomalies_for_batch,
+    normalize_bgp_event,
+    save_bgp_observations_for_batch,
+)
 
 
 class BGPStreamBackfillCollector(BaseCollector):
@@ -98,6 +102,13 @@ class BGPStreamBackfillCollector(BaseCollector):
             return result
 
         snapshot_id = await self._resolve_snapshot_id(db, result.get("task_id"))
+        observation_count = await save_bgp_observations_for_batch(
+            db,
+            source=self.name,
+            snapshot_id=snapshot_id,
+            task_id=result.get("task_id"),
+            events=getattr(self, "_latest_transformed_batch", []),
+        )
         anomaly_count = await create_bgp_anomalies_for_batch(
             db,
             source=self.name,
@@ -105,6 +116,7 @@ class BGPStreamBackfillCollector(BaseCollector):
             task_id=result.get("task_id"),
             events=getattr(self, "_latest_transformed_batch", []),
         )
+        result["observations_created"] = observation_count
         result["anomalies_created"] = anomaly_count
         return result
 

backend/app/services/collectors/ris_live.py

@@ -8,7 +8,11 @@ import urllib.request
 from typing import Any
 
 from app.services.collectors.base import BaseCollector
-from app.services.collectors.bgp_common import create_bgp_anomalies_for_batch, normalize_bgp_event
+from app.services.collectors.bgp_common import (
+    create_bgp_anomalies_for_batch,
+    normalize_bgp_event,
+    save_bgp_observations_for_batch,
+)
 
 
 class RISLiveCollector(BaseCollector):
@@ -109,6 +113,13 @@ class RISLiveCollector(BaseCollector):
             return result
 
         snapshot_id = await self._resolve_snapshot_id(db, result.get("task_id"))
+        observation_count = await save_bgp_observations_for_batch(
+            db,
+            source=self.name,
+            snapshot_id=snapshot_id,
+            task_id=result.get("task_id"),
+            events=getattr(self, "_latest_transformed_batch", []),
+        )
         anomaly_count = await create_bgp_anomalies_for_batch(
             db,
             source=self.name,
@@ -116,6 +127,7 @@ class RISLiveCollector(BaseCollector):
             task_id=result.get("task_id"),
             events=getattr(self, "_latest_transformed_batch", []),
         )
+        result["observations_created"] = observation_count
         result["anomalies_created"] = anomaly_count
         return result