compliance-agent/src/pipeline/orchestrator.rs

@@ -259,68 +259,13 @@ impl PipelineOrchestrator {
                 .await?;
         }
 
-        // Persist CVE alerts and create notifications
+        // Persist CVE alerts and create notifications (shared with the PLC path).
-        {
+        let new_notif_count = self
-            use compliance_core::models::notification::{parse_severity, CveNotification};
+            .persist_cve_alerts(&repo_id, &repo.name, &cve_alerts)
-
-            let repo_name = repo.name.clone();
-            let mut new_notif_count = 0u32;
-
-            for alert in &cve_alerts {
-                // Upsert the alert
-                let filter = doc! {
-                    "cve_id": &alert.cve_id,
-                    "repo_id": &alert.repo_id,
-                };
-                let update = mongodb::bson::to_document(alert)
-                    .map(|d| doc! { "$set": d })
-                    .unwrap_or_else(|_| doc! {});
-                self.db
-                    .cve_alerts()
-                    .update_one(filter, update)
-                    .upsert(true)
             .await?;
-
-                // Create notification (dedup by cve_id + repo + package + version)
-                let notif_filter = doc! {
-                    "cve_id": &alert.cve_id,
-                    "repo_id": &alert.repo_id,
-                    "package_name": &alert.affected_package,
-                    "package_version": &alert.affected_version,
-                };
-                let severity = parse_severity(alert.severity.as_deref(), alert.cvss_score);
-                let mut notification = CveNotification::new(
-                    alert.cve_id.clone(),
-                    repo_id.clone(),
-                    repo_name.clone(),
-                    alert.affected_package.clone(),
-                    alert.affected_version.clone(),
-                    severity,
-                );
-                notification.cvss_score = alert.cvss_score;
-                notification.summary = alert.summary.clone();
-                notification.url = Some(format!("https://osv.dev/vulnerability/{}", alert.cve_id));
-
-                let notif_update = doc! {
-                    "$setOnInsert": mongodb::bson::to_bson(&notification).unwrap_or_default()
-                };
-                if let Ok(result) = self
-                    .db
-                    .cve_notifications()
-                    .update_one(notif_filter, notif_update)
-                    .upsert(true)
-                    .await
-                {
-                    if result.upserted_id.is_some() {
-                        new_notif_count += 1;
-                    }
-                }
-            }
-
         if new_notif_count > 0 {
             tracing::info!("[{repo_id}] Created {new_notif_count} CVE notification(s)");
         }
-        }
 
         // Stage 6: Issue Creation
         tracing::info!("[{repo_id}] Stage 6: Issue Creation");
@@ -558,7 +503,10 @@ impl PipelineOrchestrator {
             target_id,
         );
         if !sbom.is_empty() {
-            if let Err(e) = self.persist_control_app_sbom(target_id, sbom).await {
+            if let Err(e) = self
+                .persist_control_app_sbom(target_id, &target.name, sbom)
+                .await
+            {
                 tracing::warn!(target_id, error = %e, "control-app SBOM persist failed");
             }
         }
@@ -573,6 +521,7 @@ impl PipelineOrchestrator {
     async fn persist_control_app_sbom(
         &self,
         target_id: &str,
+        target_name: &str,
         mut entries: Vec<SbomEntry>,
     ) -> Result<(), AgentError> {
         if entries.is_empty() {
@@ -622,25 +571,80 @@ impl PipelineOrchestrator {
                     .await?;
             }
         }
-        for alert in &alerts {
+        let new_notifs = self
-            let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id };
+            .persist_cve_alerts(target_id, target_name, &alerts)
-            if let Ok(d) = mongodb::bson::to_document(alert) {
-                self.db
-                    .cve_alerts()
-                    .update_one(filter, doc! { "$set": d })
-                    .upsert(true)
             .await?;
-            }
-        }
         tracing::info!(
             target_id,
             components = entries.len(),
             alerts = alerts.len(),
+            notifications = new_notifs,
             "control-app SBOM stored"
         );
         Ok(())
     }
 
+    /// Upsert CVE alerts for a target and create dedup'd CVE notifications;
+    /// returns the number of newly-created notifications. Shared by the SAST
+    /// pipeline and the PLC control-app SBOM path, so every SBOM source (source,
+    /// firmware, CODESYS libraries/runtime) raises the same notifications.
+    async fn persist_cve_alerts(
+        &self,
+        repo_id: &str,
+        repo_name: &str,
+        alerts: &[CveAlert],
+    ) -> Result<u32, AgentError> {
+        use compliance_core::models::notification::{parse_severity, CveNotification};
+
+        let mut new_notif = 0u32;
+        for alert in alerts {
+            let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id };
+            let update = mongodb::bson::to_document(alert)
+                .map(|d| doc! { "$set": d })
+                .unwrap_or_else(|_| doc! {});
+            self.db
+                .cve_alerts()
+                .update_one(filter, update)
+                .upsert(true)
+                .await?;
+
+            // Dedup notifications by cve + repo + package + version.
+            let notif_filter = doc! {
+                "cve_id": &alert.cve_id,
+                "repo_id": &alert.repo_id,
+                "package_name": &alert.affected_package,
+                "package_version": &alert.affected_version,
+            };
+            let severity = parse_severity(alert.severity.as_deref(), alert.cvss_score);
+            let mut notification = CveNotification::new(
+                alert.cve_id.clone(),
+                repo_id.to_string(),
+                repo_name.to_string(),
+                alert.affected_package.clone(),
+                alert.affected_version.clone(),
+                severity,
+            );
+            notification.cvss_score = alert.cvss_score;
+            notification.summary = alert.summary.clone();
+            notification.url = Some(format!("https://osv.dev/vulnerability/{}", alert.cve_id));
+            let notif_update = doc! {
+                "$setOnInsert": mongodb::bson::to_bson(&notification).unwrap_or_default()
+            };
+            if let Ok(result) = self
+                .db
+                .cve_notifications()
+                .update_one(notif_filter, notif_update)
+                .upsert(true)
+                .await
+            {
+                if result.upserted_id.is_some() {
+                    new_notif += 1;
+                }
+            }
+        }
+        Ok(new_notif)
+    }
+
     /// Ingest the target's artifacts, classify (tramiton for firmware/RTOS/Yocto,
     /// heuristics otherwise), and store the detected classification on the target.
     /// Best-effort — never fails the scan.