feat(agent): OscalControlsProvider — pull + snapshot breakpilot OSCAL catalog

Implements ControlsProvider by fetching GET /api/compliance/v1/oscal/catalog from
breakpilot-compliance, snapshotting the exact bytes to disk, and mapping the
catalog into the corpus controls the mapping engine consumes. Deterministic /
offline-capable: on fetch failure it falls back to the last snapshot so scans
still run (matters for on-prem/werkbank).

- controls/oscal_provider.rs: HTTP fetch (bearer-optional) + atomic snapshot +
  offline fallback + naive context ranking; ControlsProvider impl ("breakpilot-oscal")
- controls/mod.rs + lib.rs: register module
- 3 lib tests: url/path building, context ranking, snapshot round-trip + fallback

Not yet wired into startup — no ControlsProvider consumer exists until the
assessment-plan router lands. clippy -D warnings + fmt --all --check clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -47,6 +47,11 @@ PLC_RUNTIME_MAX_LIFETIME_SECS=180
 PLC_RUNTIME_OPENPLC_USER=openplc
 PLC_RUNTIME_OPENPLC_PASSWORD=openplc
 
+# Werkbank runner API (/api/v1/werkbank/jobs/*, /api/v1/werkbank/artifacts/*).
+# When set, mounts the runner-facing queue + artifact endpoints behind this
+# bearer token; runners present the same token. Unset = endpoints not mounted.
+WERKBANK_RUNNER_TOKEN=
+
 # Dashboard
 DASHBOARD_PORT=8080
 AGENT_API_URL=http://localhost:3001

compliance-agent/src/api/handlers/werkbank_jobs.rs

@@ -10,18 +10,20 @@
 //! so a job run by a remote runner lands the same findings an in-process run
 //! would (WB-05, the control-plane cut-over).
 
-use axum::extract::{Extension, Request};
+use axum::extract::{Extension, Path, Request};
 use axum::http::{header, StatusCode};
 use axum::middleware::Next;
 use axum::response::{IntoResponse, Response};
 use axum::Json;
-use mongodb::bson::doc;
+use mongodb::bson::{doc, oid::ObjectId};
 use secrecy::ExposeSecret;
+use serde::{Deserialize, Serialize};
 use std::time::Duration;
 
 use compliance_core::models::werkbank::{
-    CompleteRequest, CompleteResponse, HeartbeatRequest, JobResult, LeaseRequest,
+    CompleteRequest, CompleteResponse, HeartbeatRequest, InputRef, Job, JobResult, LeaseRequest,
 };
+use compliance_core::models::ArtifactKind;
 
 use super::dto::AgentExt;
 use crate::database::Database;
@@ -122,6 +124,100 @@ pub async fn complete(
     Ok(Json(CompleteResponse { recorded }))
 }
 
+/// `GET /api/v1/werkbank/artifacts/{hash}` — serve a content-addressed blob (the
+/// program a runner needs to load). The hash is validated against traversal by
+/// [`crate::ingest::blob::read_blob`]; a runner fetches this for a job's `blob`
+/// input.
+#[tracing::instrument(skip_all, fields(hash = %hash))]
+pub async fn serve_artifact(
+    Extension(agent): AgentExt,
+    Path(hash): Path<String>,
+) -> Result<Response, StatusCode> {
+    let base = std::path::Path::new(&agent.config.artifact_store_base_path);
+    match crate::ingest::blob::read_blob(base, &hash) {
+        Ok(bytes) => {
+            Ok(([(header::CONTENT_TYPE, "application/octet-stream")], bytes).into_response())
+        }
+        Err(_) => Err(StatusCode::NOT_FOUND),
+    }
+}
+
+/// Enqueue a `plc-provision` job for a target: extract its control-logic program,
+/// stash it as a content-addressed blob (which the runner fetches via
+/// [`serve_artifact`]), and queue the job. This is the control-plane "enqueue"
+/// half of the loop — a runner then leases it, provisions, and posts results.
+#[derive(Debug, Deserialize)]
+pub struct EnqueueRequest {
+    /// The tenant whose queue to enqueue into.
+    pub tenant: String,
+    /// The onboarded target to test.
+    pub target_id: String,
+}
+
+/// The enqueued job's id.
+#[derive(Debug, Serialize)]
+pub struct EnqueueResponse {
+    /// The new job id.
+    pub job_id: String,
+    /// Whether this call inserted it (false = already queued).
+    pub enqueued: bool,
+}
+
+#[tracing::instrument(skip_all, fields(tenant = %req.tenant, target = %req.target_id))]
+pub async fn enqueue(
+    Extension(agent): AgentExt,
+    Json(req): Json<EnqueueRequest>,
+) -> Result<Json<EnqueueResponse>, StatusCode> {
+    let db = tenant_db(&agent, &req.tenant).await?;
+    let oid = ObjectId::parse_str(&req.target_id).map_err(|_| StatusCode::BAD_REQUEST)?;
+    let target = db
+        .onboarded_targets()
+        .find_one(doc! { "_id": oid })
+        .await
+        .map_err(internal)?
+        .ok_or(StatusCode::NOT_FOUND)?;
+
+    // Extract the control-logic program from the target's PLC-source artifacts
+    // (same selection as the in-process PLC scan).
+    let ctx = crate::ingest::IngestContext::from_config(&agent.config, &req.target_id);
+    let ingest_set = crate::ingest::ingest_all(&target, &ctx).map_err(internal)?;
+    let program = target
+        .artifacts
+        .iter()
+        .filter(|a| {
+            matches!(
+                a.kind,
+                ArtifactKind::PlcProject | ArtifactKind::GitRepo | ArtifactKind::SourceArchive
+            )
+        })
+        .find_map(|a| {
+            let path = ingest_set
+                .get(&a.id)
+                .and_then(|ia| ia.working_path.clone())?;
+            werkbank_exec::plc::extract_program(&path)
+        })
+        .ok_or(StatusCode::UNPROCESSABLE_ENTITY)?;
+
+    // Stash the program source so the runner can fetch it by hash.
+    let base = std::path::Path::new(&agent.config.artifact_store_base_path);
+    let hash =
+        crate::ingest::blob::store_bytes(base, program.source.as_bytes()).map_err(internal)?;
+
+    let job_id = format!("job_{}", uuid::Uuid::new_v4().simple());
+    let job = Job::plc_provision(
+        &job_id,
+        &req.tenant,
+        &req.target_id,
+        InputRef::blob(hash),
+        agent.config.plc_runtime.max_lifetime_secs,
+    );
+    let enqueued = JobQueue::new(&db)
+        .enqueue(job, chrono::Utc::now())
+        .await
+        .map_err(internal)?;
+    Ok(Json(EnqueueResponse { job_id, enqueued }))
+}
+
 /// Persist a job result's findings against its target: general findings
 /// (dedup'd by fingerprint) and DAST findings. Best-effort — a persistence hiccup
 /// is logged, not surfaced to the runner (its result is already recorded).

compliance-agent/src/api/server.rs

@@ -91,6 +91,14 @@ pub async fn start_api_server(agent: ComplianceAgent, port: u16) -> Result<(), A
                 "/api/v1/werkbank/jobs/complete",
                 post(handlers::werkbank_jobs::complete),
             )
+            .route(
+                "/api/v1/werkbank/jobs/enqueue",
+                post(handlers::werkbank_jobs::enqueue),
+            )
+            .route(
+                "/api/v1/werkbank/artifacts/{hash}",
+                get(handlers::werkbank_jobs::serve_artifact),
+            )
             .layer(middleware::from_fn(
                 handlers::werkbank_jobs::require_runner_token,
             ))

compliance-agent/src/controls/mod.rs

@@ -0,0 +1,10 @@
+//! Controls corpus providers.
+//!
+//! Implementations of [`compliance_core::traits::ControlsProvider`] that supply
+//! the control corpus the mapping engine assesses findings against. Currently:
+//! [`OscalControlsProvider`], which pulls breakpilot-compliance's OSCAL catalog
+//! and snapshots it locally.
+
+mod oscal_provider;
+
+pub use oscal_provider::OscalControlsProvider;

compliance-agent/src/controls/oscal_provider.rs

@@ -0,0 +1,233 @@
+//! Pull + snapshot [`ControlsProvider`] backed by breakpilot-compliance's OSCAL
+//! catalog export.
+//!
+//! Fetches `GET {base}/api/compliance/v1/oscal/catalog?framework=<fw>`, snapshots
+//! the exact bytes to disk (so scans are deterministic and keep working offline /
+//! on-prem), and maps the catalog into the corpus controls the mapping engine
+//! consumes. The producer owns the catalog; we own the assessment — this is the
+//! ingest half of the loop.
+
+use std::path::PathBuf;
+
+use secrecy::{ExposeSecret, SecretString};
+
+use compliance_core::error::CoreError;
+use compliance_core::models::onboarding::ComplianceFramework;
+use compliance_core::models::oscal::OscalDocument;
+use compliance_core::traits::{Control, ControlQuery, ControlsProvider};
+
+/// A [`ControlsProvider`] that pulls the OSCAL catalog from breakpilot-compliance
+/// and snapshots it locally for deterministic / offline reuse.
+pub struct OscalControlsProvider {
+    http: reqwest::Client,
+    base_url: String,
+    token: Option<SecretString>,
+    snapshot_dir: PathBuf,
+}
+
+impl OscalControlsProvider {
+    /// Create a provider. `base_url` is the breakpilot-compliance root (e.g.
+    /// `http://backend-compliance:8002`); `snapshot_dir` is where catalog
+    /// snapshots are written so a later scan can reuse them without the network.
+    pub fn new(
+        http: reqwest::Client,
+        base_url: impl Into<String>,
+        token: Option<SecretString>,
+        snapshot_dir: impl Into<PathBuf>,
+    ) -> Self {
+        Self {
+            http,
+            base_url: base_url.into(),
+            token,
+            snapshot_dir: snapshot_dir.into(),
+        }
+    }
+
+    fn catalog_url(&self, framework: ComplianceFramework) -> String {
+        format!(
+            "{}/api/compliance/v1/oscal/catalog?framework={framework}",
+            self.base_url.trim_end_matches('/')
+        )
+    }
+
+    fn snapshot_path(&self, framework: ComplianceFramework) -> PathBuf {
+        self.snapshot_dir
+            .join(format!("oscal-catalog-{framework}.json"))
+    }
+
+    /// Fetch the raw catalog bytes for a framework over HTTP.
+    async fn fetch_raw(&self, framework: ComplianceFramework) -> Result<Vec<u8>, CoreError> {
+        let mut req = self.http.get(self.catalog_url(framework));
+        if let Some(token) = &self.token {
+            req = req.bearer_auth(token.expose_secret());
+        }
+        let resp = req
+            .send()
+            .await
+            .map_err(|e| CoreError::Http(e.to_string()))?;
+        if !resp.status().is_success() {
+            return Err(CoreError::Http(format!(
+                "catalog fetch for {framework} returned HTTP {}",
+                resp.status()
+            )));
+        }
+        resp.bytes()
+            .await
+            .map(|b| b.to_vec())
+            .map_err(|e| CoreError::Http(e.to_string()))
+    }
+
+    /// Write a catalog snapshot atomically (temp file + rename).
+    async fn write_snapshot(
+        &self,
+        framework: ComplianceFramework,
+        raw: &[u8],
+    ) -> Result<(), CoreError> {
+        tokio::fs::create_dir_all(&self.snapshot_dir).await?;
+        let path = self.snapshot_path(framework);
+        let tmp = path.with_extension("json.tmp");
+        tokio::fs::write(&tmp, raw).await?;
+        tokio::fs::rename(&tmp, &path).await?;
+        Ok(())
+    }
+
+    /// Read a previously written snapshot, if one exists.
+    async fn read_snapshot(
+        &self,
+        framework: ComplianceFramework,
+    ) -> Result<Option<OscalDocument>, CoreError> {
+        match tokio::fs::read(self.snapshot_path(framework)).await {
+            Ok(raw) => Ok(Some(serde_json::from_slice(&raw)?)),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(e) => Err(e.into()),
+        }
+    }
+
+    /// Load the catalog for a framework: fetch fresh + snapshot the exact bytes;
+    /// on network failure, fall back to the last snapshot so scans still run.
+    pub async fn load(&self, framework: ComplianceFramework) -> Result<OscalDocument, CoreError> {
+        match self.fetch_raw(framework).await {
+            Ok(raw) => {
+                let doc: OscalDocument = serde_json::from_slice(&raw)?;
+                if let Err(e) = self.write_snapshot(framework, &raw).await {
+                    tracing::warn!(%framework, error = %e, "failed to write OSCAL snapshot");
+                }
+                Ok(doc)
+            }
+            Err(fetch_err) => match self.read_snapshot(framework).await? {
+                Some(doc) => {
+                    tracing::warn!(
+                        %framework, error = %fetch_err,
+                        "OSCAL catalog fetch failed; falling back to snapshot"
+                    );
+                    Ok(doc)
+                }
+                None => Err(fetch_err),
+            },
+        }
+    }
+}
+
+/// Order controls whose title/text mention the query context first (stable), then
+/// truncate to the requested limit. Naive relevance — refined when the assessment
+/// layer lands.
+fn rank_and_truncate(mut controls: Vec<Control>, context: &str, limit: usize) -> Vec<Control> {
+    if !context.is_empty() {
+        let needle = context.to_lowercase();
+        controls.sort_by_key(|c| {
+            let hit =
+                c.title.to_lowercase().contains(&needle) || c.text.to_lowercase().contains(&needle);
+            u8::from(!hit)
+        });
+    }
+    controls.truncate(limit);
+    controls
+}
+
+impl ControlsProvider for OscalControlsProvider {
+    fn name(&self) -> &str {
+        "breakpilot-oscal"
+    }
+
+    async fn controls(&self, query: &ControlQuery<'_>) -> Result<Vec<Control>, CoreError> {
+        let mut out: Vec<Control> = Vec::new();
+        for &framework in query.frameworks {
+            match self.load(framework).await {
+                Ok(doc) => out.extend(doc.to_controls()),
+                Err(e) => {
+                    tracing::warn!(%framework, error = %e, "skipping framework: catalog unavailable")
+                }
+            }
+        }
+        Ok(rank_and_truncate(out, query.context, query.limit))
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use super::*;
+
+    const MINI_CATALOG: &str = r#"{"catalog":{"uuid":"u","metadata":{"title":"T",
+      "version":"1.0.0","oscal-version":"1.1.2","props":[{"name":"framework","value":"cra"}]},
+      "groups":[{"id":"g","title":"G","controls":[{"id":"cra-ai-1","title":"MFA",
+      "props":[],"parts":[{"name":"statement","prose":"require mfa"}]}]}]}}"#;
+
+    fn provider(dir: &std::path::Path) -> OscalControlsProvider {
+        OscalControlsProvider::new(reqwest::Client::new(), "http://unused/", None, dir)
+    }
+
+    #[test]
+    fn builds_catalog_url_and_snapshot_path() {
+        let p = provider(std::path::Path::new("/snap"));
+        assert_eq!(
+            p.catalog_url(ComplianceFramework::Cra),
+            "http://unused/api/compliance/v1/oscal/catalog?framework=cra"
+        );
+        assert_eq!(
+            p.snapshot_path(ComplianceFramework::Cra),
+            std::path::Path::new("/snap/oscal-catalog-cra.json")
+        );
+    }
+
+    #[test]
+    fn ranks_context_hits_first_then_truncates() {
+        let mk = |id: &str, title: &str| Control {
+            id: id.into(),
+            framework: ComplianceFramework::Cra,
+            title: title.into(),
+            text: String::new(),
+            source: None,
+        };
+        let controls = vec![
+            mk("a", "logging policy"),
+            mk("b", "multi-factor auth"),
+            mk("c", "backup"),
+        ];
+        let ranked = rank_and_truncate(controls, "auth", 2);
+        assert_eq!(ranked.len(), 2);
+        assert_eq!(ranked[0].id, "b"); // the "auth" hit floats to the top
+    }
+
+    #[tokio::test]
+    async fn snapshot_round_trip_and_offline_fallback() {
+        let dir = std::env::temp_dir().join(format!("oscal-test-{}", uuid::Uuid::new_v4()));
+        let p = provider(&dir);
+        assert!(p
+            .read_snapshot(ComplianceFramework::Cra)
+            .await
+            .unwrap()
+            .is_none());
+        p.write_snapshot(ComplianceFramework::Cra, MINI_CATALOG.as_bytes())
+            .await
+            .unwrap();
+        let doc = p
+            .read_snapshot(ComplianceFramework::Cra)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(doc.to_controls().len(), 1);
+        assert_eq!(doc.framework(), Some(ComplianceFramework::Cra));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}

compliance-agent/src/ingest/blob.rs

@@ -32,6 +32,30 @@ pub fn hash_file(path: &Path) -> Result<(String, u64), AgentError> {
     Ok((hex::encode(hasher.finalize()), total))
 }
 
+/// Store raw bytes in the content-addressed blob store under `base`, returning
+/// the SHA-256 digest. Used to stash a small derived artifact (e.g. the extracted
+/// PLC program source) so a Werkbank runner can fetch it by hash. Idempotent.
+pub fn store_bytes(base: &Path, bytes: &[u8]) -> Result<String, AgentError> {
+    let sha = hex::encode(Sha256::digest(bytes));
+    let dir = base.join("blobs").join(&sha[0..2]);
+    fs::create_dir_all(&dir)?;
+    let dest = dir.join(&sha);
+    if !dest.exists() {
+        fs::write(&dest, bytes)?;
+    }
+    Ok(sha)
+}
+
+/// Read a blob's bytes by its SHA-256 digest. Rejects a non-hex/wrong-length hash
+/// so a request can't traverse outside the blob store.
+pub fn read_blob(base: &Path, sha: &str) -> Result<Vec<u8>, AgentError> {
+    if sha.len() != 64 || !sha.bytes().all(|b| b.is_ascii_hexdigit()) {
+        return Err(AgentError::Other(format!("invalid content hash '{sha}'")));
+    }
+    let path = base.join("blobs").join(&sha[0..2]).join(sha);
+    Ok(fs::read(path)?)
+}
+
 /// Copy `src` into the content-addressed blob store under `base`, returning the
 /// stored path. Idempotent: an already-present blob is not rewritten.
 pub fn store_file(base: &Path, src: &Path, sha: &str) -> Result<PathBuf, AgentError> {

compliance-agent/src/ingest/mod.rs

@@ -6,7 +6,7 @@
 //! is also the reconciliation key against sibling products (a firmware sha256
 //! matches tramiton's `Artifact.sha256`).
 
-mod blob;
+pub(crate) mod blob;
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};

compliance-agent/src/lib.rs

@@ -4,6 +4,7 @@ pub mod agent;
 pub mod api;
 pub mod classify;
 pub mod config;
+pub mod controls;
 pub mod database;
 pub mod error;
 pub mod ingest;

compliance-agent/tests/werkbank_api.rs

@@ -11,7 +11,7 @@ mod common;
 
 use std::sync::Arc;
 
-use axum::routing::post;
+use axum::routing::{get, post};
 use axum::{middleware, Extension, Router};
 
 use compliance_agent::agent::ComplianceAgent;
@@ -19,7 +19,9 @@ use compliance_agent::api::handlers::werkbank_jobs;
 use compliance_agent::database::DatabasePool;
 use compliance_agent::werkbank::JobQueue;
 use compliance_core::models::werkbank::{InputRef, Job, JobResult, JobStatus, LeasedJob};
-use compliance_core::models::{Finding, ScanType, Severity};
+use compliance_core::models::{
+    Artifact, Finding, OnboardedTarget, PlcFormat, ScanType, Severity, TargetType,
+};
 
 use common::{dev_config, TEST_RUNNER_TOKEN};
 
@@ -58,6 +60,14 @@ async fn start() -> Option<Harness> {
             "/api/v1/werkbank/jobs/complete",
             post(werkbank_jobs::complete),
         )
+        .route(
+            "/api/v1/werkbank/jobs/enqueue",
+            post(werkbank_jobs::enqueue),
+        )
+        .route(
+            "/api/v1/werkbank/artifacts/{hash}",
+            get(werkbank_jobs::serve_artifact),
+        )
         .layer(middleware::from_fn(werkbank_jobs::require_runner_token))
         .layer(Extension(Arc::new(agent)));
 
@@ -177,6 +187,65 @@ async fn lease_complete_persists_findings_against_the_target() {
     h.cleanup().await;
 }
 
+#[tokio::test]
+async fn enqueue_extracts_program_stores_a_blob_and_serves_it() {
+    let Some(h) = start().await else { return };
+    let db = h.pool.for_tenant_id(TENANT).await.unwrap();
+
+    // A PlcSps target with a single complete ST program uploaded.
+    let dir = std::env::temp_dir().join(format!("wbq-prog-{}", uuid::Uuid::new_v4()));
+    std::fs::create_dir_all(&dir).unwrap();
+    let st = dir.join("main.st");
+    std::fs::write(
+        &st,
+        "PROGRAM Main\nEND_PROGRAM\nCONFIGURATION C\n  RESOURCE R\nEND_CONFIGURATION\n",
+    )
+    .unwrap();
+    let mut target = OnboardedTarget::new("plc".into(), TargetType::PlcSps);
+    let mut art = Artifact::plc_project("main.st", PlcFormat::StructuredText);
+    art.stored_path = Some(st.to_string_lossy().to_string());
+    target.artifacts.push(art);
+    let ins = db.onboarded_targets().insert_one(&target).await.unwrap();
+    let target_id = ins.inserted_id.as_object_id().unwrap().to_hex();
+
+    // Enqueue → a plc-provision job whose program is a content-addressed blob.
+    let resp = h
+        .post(
+            "/api/v1/werkbank/jobs/enqueue",
+            Some(TEST_RUNNER_TOKEN),
+            serde_json::json!({ "tenant": TENANT, "target_id": target_id }),
+        )
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200, "enqueue should succeed");
+    let body: serde_json::Value = resp.json().await.unwrap();
+    let job_id = body["job_id"].as_str().unwrap().to_string();
+
+    let rec = JobQueue::new(&db).get(&job_id).await.unwrap().unwrap();
+    let hash = rec
+        .job
+        .inputs
+        .get("program")
+        .and_then(|i| i.blob.clone())
+        .expect("program blob");
+
+    // Serve the blob back and confirm it's the program source (what the runner
+    // would fetch).
+    let served = h
+        .client
+        .get(format!("{}/api/v1/werkbank/artifacts/{hash}", h.base_url))
+        .bearer_auth(TEST_RUNNER_TOKEN)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(served.status(), 200);
+    assert!(served.text().await.unwrap().contains("CONFIGURATION"));
+
+    h.cleanup().await;
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
 #[tokio::test]
 async fn empty_queue_leases_nothing() {
     let Some(h) = start().await else { return };

compliance-core/src/models/mod.rs

@@ -10,6 +10,7 @@ pub mod mcp;
 pub mod mcp_token;
 pub mod notification;
 pub mod onboarding;
+pub mod oscal;
 pub mod pentest;
 pub mod repository;
 pub mod sbom;
@@ -39,6 +40,7 @@ pub use onboarding::{
     GitArtifactConfig, IssueTrackerConfig, OnboardedTarget, PlcArtifactConfig, PlcFormat,
     TargetScanConfig, TargetType, TargetTypeCandidate, WebArtifactConfig,
 };
+pub use oscal::OscalDocument;
 pub use pentest::{
     AttackChainNode, AttackNodeStatus, AuthMode, CodeContextHint, Environment, IdentityProvider,
     PentestAuthConfig, PentestConfig, PentestEvent, PentestMessage, PentestSession, PentestStats,

compliance-core/src/models/oscal.rs

@@ -0,0 +1,250 @@
+//! OSCAL 1.1 catalog types + mapping into the controls corpus.
+//!
+//! Deserialises the OSCAL catalog served by breakpilot-compliance
+//! (`GET /api/compliance/v1/oscal/catalog`) and maps its controls into the
+//! framework-agnostic [`crate::traits::Control`] that the mapping engine consumes.
+//! Only the fields we use are modelled; unknown OSCAL fields are ignored so the
+//! producer can add detail without breaking us.
+//!
+//! Scope boundary: this is the *catalog* (domain content). Assessment objectives
+//! and scanner routing live in our assessment layer, not here — see
+//! [`crate::traits::ControlsProvider`].
+
+use serde::Deserialize;
+
+use crate::models::onboarding::ComplianceFramework;
+use crate::traits::Control as CorpusControl;
+
+/// A parsed OSCAL catalog document (`{"catalog": {...}}`).
+#[derive(Debug, Clone, Deserialize)]
+pub struct OscalDocument {
+    pub catalog: Catalog,
+}
+
+/// An OSCAL catalog: metadata + a tree of control groups.
+#[derive(Debug, Clone, Deserialize)]
+pub struct Catalog {
+    pub uuid: String,
+    pub metadata: Metadata,
+    #[serde(default)]
+    pub groups: Vec<Group>,
+    #[serde(rename = "back-matter", default)]
+    pub back_matter: Option<BackMatter>,
+}
+
+/// Catalog metadata (title/version + provenance props).
+#[derive(Debug, Clone, Deserialize)]
+pub struct Metadata {
+    pub title: String,
+    pub version: String,
+    #[serde(rename = "oscal-version")]
+    pub oscal_version: String,
+    #[serde(default)]
+    pub props: Vec<Prop>,
+}
+
+/// A name/value property, optionally namespaced.
+#[derive(Debug, Clone, Deserialize)]
+pub struct Prop {
+    pub name: String,
+    pub value: String,
+    #[serde(default)]
+    pub ns: Option<String>,
+}
+
+/// A control group (may nest sub-groups and controls).
+#[derive(Debug, Clone, Deserialize)]
+pub struct Group {
+    #[serde(default)]
+    pub id: String,
+    #[serde(default)]
+    pub title: String,
+    #[serde(default)]
+    pub controls: Vec<Control>,
+    #[serde(default)]
+    pub groups: Vec<Group>,
+}
+
+/// An OSCAL control (may nest enhancement controls).
+#[derive(Debug, Clone, Deserialize)]
+pub struct Control {
+    pub id: String,
+    #[serde(default)]
+    pub title: String,
+    #[serde(default)]
+    pub props: Vec<Prop>,
+    #[serde(default)]
+    pub parts: Vec<Part>,
+    #[serde(default)]
+    pub links: Vec<Link>,
+    #[serde(default)]
+    pub controls: Vec<Control>,
+}
+
+/// A control part (e.g. the `statement`), may nest sub-parts.
+#[derive(Debug, Clone, Deserialize)]
+pub struct Part {
+    #[serde(default)]
+    pub name: String,
+    #[serde(default)]
+    pub prose: Option<String>,
+    #[serde(default)]
+    pub parts: Vec<Part>,
+}
+
+/// A link, e.g. a `reference` to a back-matter resource.
+#[derive(Debug, Clone, Deserialize)]
+pub struct Link {
+    pub href: String,
+    #[serde(default)]
+    pub rel: Option<String>,
+}
+
+/// Back-matter holding referenced resources (e.g. the CRA measures).
+#[derive(Debug, Clone, Deserialize)]
+pub struct BackMatter {
+    #[serde(default)]
+    pub resources: Vec<Resource>,
+}
+
+/// A back-matter resource referenced by control links.
+#[derive(Debug, Clone, Deserialize)]
+pub struct Resource {
+    pub uuid: String,
+    #[serde(default)]
+    pub title: Option<String>,
+    #[serde(default)]
+    pub description: Option<String>,
+}
+
+impl Metadata {
+    /// First prop value with the given name.
+    pub fn prop(&self, name: &str) -> Option<&str> {
+        self.props
+            .iter()
+            .find(|p| p.name == name)
+            .map(|p| p.value.as_str())
+    }
+}
+
+impl Control {
+    /// First prop value with the given name.
+    pub fn prop(&self, name: &str) -> Option<&str> {
+        self.props
+            .iter()
+            .find(|p| p.name == name)
+            .map(|p| p.value.as_str())
+    }
+
+    /// The control's `statement` prose, if present.
+    pub fn statement(&self) -> Option<&str> {
+        self.parts
+            .iter()
+            .find(|p| p.name == "statement")
+            .and_then(|p| p.prose.as_deref())
+    }
+}
+
+impl OscalDocument {
+    /// The framework this catalog declares (`metadata.props[name="framework"]`).
+    pub fn framework(&self) -> Option<ComplianceFramework> {
+        framework_from_str(self.catalog.metadata.prop("framework")?)
+    }
+
+    /// The catalog `content-hash` prop — consumers pin this to snapshot/detect drift.
+    pub fn content_hash(&self) -> Option<&str> {
+        self.catalog.metadata.prop("content-hash")
+    }
+
+    /// Flatten the catalog into the corpus controls the mapping engine consumes.
+    pub fn to_controls(&self) -> Vec<CorpusControl> {
+        let framework = self.framework().unwrap_or(ComplianceFramework::Cra);
+        let source_label = self.catalog.metadata.title.as_str();
+        let mut out = Vec::new();
+        for group in &self.catalog.groups {
+            collect_group(group, framework, source_label, &mut out);
+        }
+        out
+    }
+}
+
+/// Map an OSCAL framework token (e.g. `"cra"`) to [`ComplianceFramework`] via its
+/// serde snake_case representation.
+fn framework_from_str(raw: &str) -> Option<ComplianceFramework> {
+    serde_json::from_value(serde_json::Value::String(raw.to_string())).ok()
+}
+
+fn collect_group(
+    group: &Group,
+    framework: ComplianceFramework,
+    source_label: &str,
+    out: &mut Vec<CorpusControl>,
+) {
+    for control in &group.controls {
+        collect_control(control, framework, source_label, out);
+    }
+    for sub in &group.groups {
+        collect_group(sub, framework, source_label, out);
+    }
+}
+
+fn collect_control(
+    control: &Control,
+    framework: ComplianceFramework,
+    source_label: &str,
+    out: &mut Vec<CorpusControl>,
+) {
+    let source = match control.prop("annex-anchor") {
+        Some(anchor) => Some(format!("{source_label} · {anchor}")),
+        None => Some(source_label.to_string()),
+    };
+    out.push(CorpusControl {
+        id: control.id.clone(),
+        framework,
+        title: control.title.clone(),
+        text: control.statement().unwrap_or_default().to_string(),
+        source,
+    });
+    for enhancement in &control.controls {
+        collect_control(enhancement, framework, source_label, out);
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use super::*;
+
+    const CATALOG: &str = include_str!("../../tests/data/cra_catalog.json");
+
+    fn parse() -> OscalDocument {
+        serde_json::from_str(CATALOG).unwrap()
+    }
+
+    #[test]
+    fn parses_full_catalog() {
+        let doc = parse();
+        assert_eq!(doc.catalog.metadata.oscal_version, "1.1.2");
+        assert!(!doc.catalog.groups.is_empty());
+        assert!(doc.catalog.back_matter.is_some());
+    }
+
+    #[test]
+    fn maps_all_controls_to_corpus() {
+        let doc = parse();
+        let controls = doc.to_controls();
+        assert_eq!(controls.len(), 40);
+        assert_eq!(doc.framework(), Some(ComplianceFramework::Cra));
+
+        let c8 = controls.iter().find(|c| c.id == "cra-ai-8").unwrap();
+        assert_eq!(c8.framework, ComplianceFramework::Cra);
+        assert!(!c8.title.is_empty());
+        assert!(!c8.text.is_empty(), "statement prose should map into text");
+        assert!(c8.source.as_deref().unwrap_or_default().contains("Annex I"));
+    }
+
+    #[test]
+    fn exposes_content_hash_for_snapshotting() {
+        assert_eq!(parse().content_hash().map(str::len), Some(64));
+    }
+}