feat(pipeline): unified run_target execution behind UNIFIED_PIPELINE (#133 part 2)

Wire the scan-plan builder into a runnable pipeline that scans an OnboardedTarget:
- AgentConfig.unified_pipeline (env UNIFIED_PIPELINE, default off). agent.run_scan
  dispatches to run_target when set, else the legacy run().
- orchestrator::run_target: loads the target from onboarded_targets, creates a
  ScanRun, and runs run_target_pipeline; run_target_pipeline builds the scan plan
  and, for a code (git) artifact, reuses the full legacy pipeline via a
  TrackedRepository view (clone → SAST umbrella → triage → persist → issues →
  DAST) then syncs findings_count + the git watermark back to the target.
  Firmware/PLC/mobile scanners are follow-ups (#128/#129/#130); URL-only targets
  attempt DAST.
- repo_view_from_target: inverse of the migration's repo_to_target, _id-preserving.

Additive + flag-gated: the legacy path is unchanged and default. Unit tests for
repo_view_from_target (+ the part-1 build_scan_plan tests). Full E2E verification
is via UNIFIED_PIPELINE=1 on a running instance.

Closes #133.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

compliance-agent/src/agent.rs

@@ -63,7 +63,11 @@ impl ComplianceAgent {
         let db = self.db_pool.for_tenant_id(tenant_id).await?;
         let orchestrator =
             PipelineOrchestrator::new(self.config.clone(), db, self.llm.clone(), self.http.clone());
-        orchestrator.run(repo_id, trigger).await
+        if self.config.unified_pipeline {
+            orchestrator.run_target(repo_id, trigger).await
+        } else {
+            orchestrator.run(repo_id, trigger).await
+        }
     }
 
     /// Run a PR review: scan the diff and post review comments.

compliance-agent/src/config.rs

@@ -47,6 +47,9 @@ pub fn load_config() -> Result<AgentConfig, AgentError> {
             .unwrap_or_else(|| "/tmp/compliance-scanner/repos".to_string()),
         artifact_store_base_path: env_var_opt("ARTIFACT_STORE_BASE_PATH")
             .unwrap_or_else(|| "/data/compliance-scanner/artifacts".to_string()),
+        unified_pipeline: env_var_opt("UNIFIED_PIPELINE")
+            .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
+            .unwrap_or(false),
         ssh_key_path: env_var_opt("SSH_KEY_PATH")
             .unwrap_or_else(|| "/data/compliance-scanner/ssh/id_ed25519".to_string()),
         keycloak_url: env_var_opt("KEYCLOAK_URL"),

compliance-agent/src/pentest/cleanup.rs

@@ -342,6 +342,7 @@ mod tests {
             pentest_imap_password: None,
             admin_api_token: None,
             tenant_registry_url: None,
+            unified_pipeline: false,
         }
     }
 

compliance-agent/src/pipeline/mod.rs

@@ -8,6 +8,7 @@ mod issue_creation;
 pub mod lint;
 pub mod orchestrator;
 pub mod patterns;
+pub mod plan;
 mod pr_review;
 pub mod sbom;
 pub mod semgrep;

compliance-agent/src/pipeline/orchestrator.rs

@@ -15,6 +15,7 @@ use crate::pipeline::git::GitOps;
 use crate::pipeline::gitleaks::GitleaksScanner;
 use crate::pipeline::lint::LintScanner;
 use crate::pipeline::patterns::{GdprPatternScanner, OAuthPatternScanner};
+use crate::pipeline::plan::build_scan_plan;
 use crate::pipeline::sbom::SbomScanner;
 use crate::pipeline::semgrep::SemgrepScanner;
 
@@ -419,6 +420,149 @@ impl PipelineOrchestrator {
         Ok(new_count)
     }
 
+    /// Unified entry point (behind `UNIFIED_PIPELINE`): run a scan for an
+    /// `OnboardedTarget`. Mirrors [`Self::run`] but sources the target from
+    /// `onboarded_targets` and dispatches by the scan plan.
+    #[tracing::instrument(skip_all, fields(target_id = %target_id, trigger = ?trigger))]
+    pub async fn run_target(
+        &self,
+        target_id: &str,
+        trigger: ScanTrigger,
+    ) -> Result<(), AgentError> {
+        let oid = mongodb::bson::oid::ObjectId::parse_str(target_id)
+            .map_err(|e| AgentError::Other(e.to_string()))?;
+        let target = self
+            .db
+            .onboarded_targets()
+            .find_one(doc! { "_id": oid })
+            .await?
+            .ok_or_else(|| AgentError::Other(format!("Onboarded target {target_id} not found")))?;
+
+        let scan_run = ScanRun::new(target_id.to_string(), trigger);
+        let insert = self.db.scan_runs().insert_one(&scan_run).await?;
+        let scan_run_id = insert
+            .inserted_id
+            .as_object_id()
+            .map(|id| id.to_hex())
+            .unwrap_or_default();
+
+        let result = self.run_target_pipeline(&target, &scan_run_id).await;
+        match &result {
+            Ok(count) => {
+                self.db
+                    .scan_runs()
+                    .update_one(
+                        doc! { "_id": &insert.inserted_id },
+                        doc! { "$set": {
+                            "status": "completed",
+                            "current_phase": "completed",
+                            "new_findings_count": *count as i64,
+                            "completed_at": mongodb::bson::DateTime::now(),
+                        } },
+                    )
+                    .await?;
+            }
+            Err(e) => {
+                tracing::error!(target_id, error = %e, "Unified scan pipeline failed");
+                self.db
+                    .scan_runs()
+                    .update_one(
+                        doc! { "_id": &insert.inserted_id },
+                        doc! { "$set": {
+                            "status": "failed",
+                            "error_message": e.to_string(),
+                            "completed_at": mongodb::bson::DateTime::now(),
+                        } },
+                    )
+                    .await?;
+            }
+        }
+        result.map(|_| ())
+    }
+
+    /// Run the applicable scans for a target. For a code target this reuses the
+    /// full legacy pipeline over the code artifact (clone → SAST umbrella →
+    /// triage → persist → issues → DAST); firmware/PLC/mobile scanners are
+    /// follow-ups (#128/#129/#130). Returns the number of new findings.
+    async fn run_target_pipeline(
+        &self,
+        target: &OnboardedTarget,
+        scan_run_id: &str,
+    ) -> Result<u32, AgentError> {
+        let target_id = target.id.map(|id| id.to_hex()).unwrap_or_default();
+        let plan = build_scan_plan(target);
+        tracing::info!(
+            target_id = %target_id,
+            target_type = %target.target_type,
+            planned_steps = plan.steps.len(),
+            "Unified pipeline: scan plan built"
+        );
+
+        match target.code_artifact() {
+            Some(code) if code.kind == ArtifactKind::GitRepo => {
+                let repo = repo_view_from_target(target, code);
+                let new_count = self.run_pipeline(&repo, scan_run_id).await?;
+                self.finalize_target(target, &repo, new_count).await?;
+                Ok(new_count)
+            }
+            Some(_) => {
+                tracing::warn!(
+                    target_id = %target_id,
+                    "Unified pipeline: source-archive scanning not yet wired; skipping"
+                );
+                Ok(0)
+            }
+            None => {
+                // No code to scan. Firmware/PLC/mobile static scanners land in
+                // #128/#129/#130; DAST for a running URL still works when a
+                // DastTarget row exists (migrated targets).
+                tracing::info!(
+                    target_id = %target_id,
+                    "Unified pipeline: no code artifact; attempting DAST only"
+                );
+                self.update_phase(scan_run_id, "dast_scanning").await;
+                self.maybe_trigger_dast(&target_id, scan_run_id).await;
+                Ok(0)
+            }
+        }
+    }
+
+    /// Sync the onboarded-target document after a scan: bump `findings_count`
+    /// and advance the git artifact's `last_scanned_commit` watermark.
+    async fn finalize_target(
+        &self,
+        target: &OnboardedTarget,
+        repo: &TrackedRepository,
+        new_count: u32,
+    ) -> Result<(), AgentError> {
+        let oid = match target.id {
+            Some(id) => id,
+            None => return Ok(()),
+        };
+        self.db
+            .onboarded_targets()
+            .update_one(
+                doc! { "_id": oid },
+                doc! {
+                    "$inc": { "findings_count": new_count as i64 },
+                    "$set": { "updated_at": mongodb::bson::DateTime::now() },
+                },
+            )
+            .await?;
+
+        let repo_path = std::path::Path::new(&self.config.git_clone_base_path).join(&repo.name);
+        if let (Ok(sha), Some(code)) = (GitOps::get_head_sha(&repo_path), target.code_artifact()) {
+            self.db
+                .onboarded_targets()
+                .update_one(
+                    doc! { "_id": oid, "artifacts.id": &code.id },
+                    doc! { "$set": { "artifacts.$.git.last_scanned_commit": sha } },
+                )
+                .await?;
+        }
+        Ok(())
+    }
+
     pub(super) async fn update_phase(&self, scan_run_id: &str, phase: &str) {
         if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(scan_run_id) {
             let _ = self
@@ -436,6 +580,37 @@ impl PipelineOrchestrator {
     }
 }
 
+/// Build a legacy `TrackedRepository` view from an onboarded target's code
+/// artifact, so the unified pipeline can reuse the existing repo pipeline. The
+/// inverse of the migration's `repo_to_target`. `_id` is preserved so findings
+/// and DAST lookups resolve against the same key.
+fn repo_view_from_target(target: &OnboardedTarget, code: &Artifact) -> TrackedRepository {
+    let mut repo = TrackedRepository::new(target.name.clone(), code.source_ref.clone());
+    repo.id = target.id;
+    if let Some(git) = &code.git {
+        repo.default_branch = git.default_branch.clone();
+        repo.last_scanned_commit = git.last_scanned_commit.clone();
+        repo.local_path = git.local_path.clone();
+    }
+    if let Some(auth) = &code.auth {
+        repo.auth_token = auth.secret.clone();
+        repo.auth_username = auth.username.clone();
+    }
+    if let Some(it) = &target.scan_config.issue_tracker {
+        repo.tracker_type = it.tracker_type.clone();
+        repo.tracker_owner = it.owner.clone();
+        repo.tracker_repo = it.repo.clone();
+        repo.tracker_token = it.token.clone();
+    }
+    repo.scan_schedule = target.scan_schedule.clone();
+    repo.webhook_enabled = target.webhook_enabled;
+    repo.webhook_secret = target.webhook_secret.clone();
+    repo.findings_count = target.findings_count;
+    repo.created_at = target.created_at;
+    repo.updated_at = target.updated_at;
+    repo
+}
+
 /// Extract the scheme + host from a git URL.
 /// e.g. "https://gitea.example.com/owner/repo.git" -> "https://gitea.example.com"
 /// e.g. "ssh://git@gitea.example.com:22/owner/repo.git" -> "https://gitea.example.com"
@@ -460,3 +635,50 @@ pub(super) fn extract_base_url(git_url: &str) -> Option<String> {
         None
     }
 }
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use compliance_core::models::{
+        Artifact, ArtifactAuth, IssueTrackerConfig, TargetType, TrackerType,
+    };
+
+    #[test]
+    fn repo_view_preserves_id_git_auth_and_tracker() {
+        let mut target = OnboardedTarget::new("acme".to_string(), TargetType::WebApp);
+        target.id = Some(mongodb::bson::oid::ObjectId::new());
+        target.findings_count = 3;
+        target.scan_config.issue_tracker = Some(IssueTrackerConfig {
+            tracker_type: Some(TrackerType::Gitea),
+            owner: Some("acme".to_string()),
+            repo: Some("web".to_string()),
+            token: Some("tt".to_string()),
+        });
+
+        let mut artifact = Artifact::git_repo("https://git/acme.git", "develop");
+        if let Some(git) = artifact.git.as_mut() {
+            git.last_scanned_commit = Some("abc123".to_string());
+        }
+        artifact.auth = Some(ArtifactAuth {
+            method: "token".to_string(),
+            username: Some("bob".to_string()),
+            secret: Some("pat".to_string()),
+            ..Default::default()
+        });
+        target.artifacts.push(artifact);
+
+        let code = target.code_artifact().expect("code artifact");
+        let repo = repo_view_from_target(&target, code);
+
+        assert_eq!(repo.id, target.id); // preserved
+        assert_eq!(repo.git_url, "https://git/acme.git");
+        assert_eq!(repo.default_branch, "develop");
+        assert_eq!(repo.last_scanned_commit.as_deref(), Some("abc123"));
+        assert_eq!(repo.auth_token.as_deref(), Some("pat"));
+        assert_eq!(repo.auth_username.as_deref(), Some("bob"));
+        assert_eq!(repo.tracker_type, Some(TrackerType::Gitea));
+        assert_eq!(repo.tracker_owner.as_deref(), Some("acme"));
+        assert_eq!(repo.findings_count, 3);
+    }
+}

compliance-agent/src/pipeline/plan.rs

@@ -0,0 +1,195 @@
+//! The scan plan.
+//!
+//! [`build_scan_plan`] turns an [`OnboardedTarget`] into the concrete ordered
+//! list of scans to run, each bound to the artifact it consumes. It intersects
+//! the scan-applicability matrix ([`applicable_scans`]) with the target's
+//! `scan_config` overrides: a scan runs when its required artifact is present and
+//! it is either on by default or explicitly enabled, and is not explicitly
+//! disabled. This is the decision engine the unified pipeline (`run_target`)
+//! executes.
+
+use compliance_core::models::{Artifact, ArtifactKind, OnboardedTarget, ScanPhase, ScanType};
+use compliance_core::scan_matrix::applicable_scans;
+
+/// One scan to run, bound to the artifact it operates on.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ScanStep {
+    /// The scan to run.
+    pub scan_type: ScanType,
+    /// The pipeline phase to report while it runs.
+    pub phase: ScanPhase,
+    /// The id of the artifact this scan consumes ([`Artifact::id`]).
+    pub artifact_id: String,
+}
+
+/// The ordered set of scans to run for a target.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct ScanPlan {
+    /// The scans, in matrix order.
+    pub steps: Vec<ScanStep>,
+}
+
+impl ScanPlan {
+    /// Whether the plan contains a step for the given scan type.
+    pub fn has(&self, scan: ScanType) -> bool {
+        self.steps.iter().any(|s| s.scan_type == scan)
+    }
+
+    /// Whether the plan is empty (nothing to run).
+    pub fn is_empty(&self) -> bool {
+        self.steps.is_empty()
+    }
+}
+
+/// Build the scan plan for a target: matrix defaults ∩ `scan_config`, each scan
+/// bound to the artifact it consumes. Scans whose required artifact is absent, or
+/// that are disabled, or off-by-default and not explicitly enabled, are dropped.
+pub fn build_scan_plan(target: &OnboardedTarget) -> ScanPlan {
+    let enabled = &target.scan_config.enabled_scans;
+    let disabled = &target.scan_config.disabled_scans;
+
+    let mut steps = Vec::new();
+    for option in applicable_scans(target) {
+        // Required artifact missing → not runnable.
+        if option.blocked_reason.is_some() {
+            continue;
+        }
+        // Explicit opt-out wins.
+        if disabled.contains(&option.scan) {
+            continue;
+        }
+        // Run if on by default, or explicitly enabled.
+        if !option.default_on && !enabled.contains(&option.scan) {
+            continue;
+        }
+        let Some(artifact) = resolve_artifact(target, option.required_artifact) else {
+            continue;
+        };
+        steps.push(ScanStep {
+            scan_type: option.scan,
+            phase: phase_for(option.scan),
+            artifact_id: artifact.id.clone(),
+        });
+    }
+    ScanPlan { steps }
+}
+
+/// Resolve the artifact a scan consumes. A "code" requirement (represented by
+/// `GitRepo`) is satisfied by a git repo *or* a source archive.
+fn resolve_artifact(target: &OnboardedTarget, required: Option<ArtifactKind>) -> Option<&Artifact> {
+    match required {
+        Some(ArtifactKind::GitRepo) => target.code_artifact(),
+        Some(kind) => target.first_of(kind),
+        None => target.code_artifact().or_else(|| target.artifacts.first()),
+    }
+}
+
+/// The pipeline phase reported while a given scan runs.
+fn phase_for(scan: ScanType) -> ScanPhase {
+    match scan {
+        ScanType::Sast => ScanPhase::Sast,
+        ScanType::Sbom => ScanPhase::SbomGeneration,
+        ScanType::Cve => ScanPhase::CveScanning,
+        ScanType::Gdpr | ScanType::OAuth => ScanPhase::PatternScanning,
+        ScanType::SecretDetection => ScanPhase::SecretDetection,
+        ScanType::Lint => ScanPhase::LintScanning,
+        ScanType::CodeReview => ScanPhase::CodeReview,
+        ScanType::Graph => ScanPhase::GraphBuilding,
+        ScanType::Dast => ScanPhase::DastScanning,
+        ScanType::FirmwareStatic => ScanPhase::FirmwareStatic,
+        ScanType::PlcControlLogic => ScanPhase::PlcAnalysis,
+        ScanType::MobileStatic => ScanPhase::MobileStatic,
+        ScanType::ContainerScan => ScanPhase::ContainerScan,
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use compliance_core::models::{PlcFormat, TargetType};
+
+    fn target(target_type: TargetType, artifacts: Vec<Artifact>) -> OnboardedTarget {
+        let mut t = OnboardedTarget::new("t".to_string(), target_type);
+        t.artifacts = artifacts;
+        t
+    }
+
+    fn step_for<'a>(plan: &'a ScanPlan, scan: ScanType) -> Option<&'a ScanStep> {
+        plan.steps.iter().find(|s| s.scan_type == scan)
+    }
+
+    #[test]
+    fn webapp_with_code_and_url_runs_sast_and_dast_bound_to_the_right_artifacts() {
+        let code = Artifact::git_repo("https://git/x", "main");
+        let url = Artifact::live_url("https://x");
+        let (code_id, url_id) = (code.id.clone(), url.id.clone());
+        let t = target(TargetType::WebApp, vec![code, url]);
+
+        let plan = build_scan_plan(&t);
+        let sast = step_for(&plan, ScanType::Sast).expect("sast planned");
+        assert_eq!(sast.artifact_id, code_id);
+        assert_eq!(sast.phase, ScanPhase::Sast);
+        let dast = step_for(&plan, ScanType::Dast).expect("dast planned");
+        assert_eq!(dast.artifact_id, url_id);
+    }
+
+    #[test]
+    fn webapp_without_url_omits_dast() {
+        let t = target(TargetType::WebApp, vec![Artifact::git_repo("u", "main")]);
+        let plan = build_scan_plan(&t);
+        assert!(plan.has(ScanType::Sast));
+        assert!(!plan.has(ScanType::Dast));
+    }
+
+    #[test]
+    fn code_scan_binds_to_source_archive_when_no_git_repo() {
+        let arc = Artifact::source_archive("src.zip");
+        let arc_id = arc.id.clone();
+        let t = target(TargetType::BackendService, vec![arc]);
+        let plan = build_scan_plan(&t);
+        let sast = step_for(&plan, ScanType::Sast).expect("sast planned");
+        assert_eq!(sast.artifact_id, arc_id);
+    }
+
+    #[test]
+    fn firmware_sbom_and_cve_bind_to_the_firmware_image() {
+        let fw = Artifact::firmware_image("fw.bin");
+        let fw_id = fw.id.clone();
+        let t = target(TargetType::FirmwareBareMetal, vec![fw]);
+        let plan = build_scan_plan(&t);
+        let sbom = step_for(&plan, ScanType::Sbom).expect("sbom planned");
+        assert_eq!(sbom.artifact_id, fw_id);
+        assert!(step_for(&plan, ScanType::FirmwareStatic).is_some());
+        assert!(!plan.has(ScanType::Dast));
+    }
+
+    #[test]
+    fn plc_plans_only_control_logic() {
+        let plc = Artifact::plc_project("p.xml", PlcFormat::PlcopenXml);
+        let t = target(TargetType::PlcSps, vec![plc]);
+        let plan = build_scan_plan(&t);
+        assert_eq!(plan.steps.len(), 1);
+        assert_eq!(plan.steps[0].scan_type, ScanType::PlcControlLogic);
+        assert_eq!(plan.steps[0].phase, ScanPhase::PlcAnalysis);
+    }
+
+    #[test]
+    fn disabled_scan_is_dropped_and_off_by_default_can_be_enabled() {
+        let mut t = target(TargetType::WebApp, vec![Artifact::git_repo("u", "main")]);
+        t.scan_config.disabled_scans = vec![ScanType::Lint];
+        // CodeReview is off by default for web; enable it explicitly.
+        t.scan_config.enabled_scans = vec![ScanType::CodeReview];
+        let plan = build_scan_plan(&t);
+        assert!(!plan.has(ScanType::Lint));
+        assert!(plan.has(ScanType::CodeReview));
+        assert!(plan.has(ScanType::Sast));
+    }
+
+    #[test]
+    fn no_code_artifact_yields_empty_plan_for_web() {
+        let t = target(TargetType::WebApp, vec![]);
+        let plan = build_scan_plan(&t);
+        assert!(plan.is_empty());
+    }
+}

compliance-agent/tests/common/mod.rs

@@ -70,6 +70,7 @@ impl TestServer {
             pentest_imap_password: None,
             admin_api_token: None,
             tenant_registry_url: None,
+            unified_pipeline: false,
         };
 
         let agent = ComplianceAgent::new(config, db_pool);

compliance-core/src/config.rs

@@ -49,6 +49,10 @@ pub struct AgentConfig {
     /// of tenants to iterate. When `None` or unreachable, scheduler
     /// falls back to `SCHEDULER_TENANT_IDS` env (M7.2-C).
     pub tenant_registry_url: Option<String>,
+    /// When true, `run_scan` dispatches to the unified `run_target` pipeline
+    /// (reads `onboarded_targets`) instead of the legacy repository pipeline.
+    /// Env `UNIFIED_PIPELINE`. Defaults off during the transition.
+    pub unified_pipeline: bool,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]