feat(werkbank): make the loop runnable — enqueue + artifact serve/fetch (WB-05b) (#209)

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,
             ))