feat(docker): verify full Docker platform stack

crates/openre-api/Cargo.toml

@@ -8,6 +8,10 @@ description = "API server (HTTP/gRPC) for open-re"
 categories = { workspace = true }
 keywords = { workspace = true }
 
+[[bin]]
+name = "openre-api"
+path = "src/main.rs"
+
 [dependencies]
 openre-core = { path = "../openre-core" }
 openre-config = { path = "../openre-config" }
@@ -35,6 +39,7 @@ validator = { workspace = true }
 utoipa = { workspace = true }
 utoipa-swagger-ui = { workspace = true, features = ["axum"] }
 tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
 futures = { workspace = true }
 governor = { workspace = true }
 async-stream = "0.3"

crates/openre-api/src/lib.rs

@@ -10,6 +10,7 @@ pub mod state;
 pub mod validation;
 pub mod versioning;
 pub mod websocket;
+pub mod workers;
 
 pub use auth::*;
 pub use error::*;
@@ -21,3 +22,4 @@ pub use state::*;
 pub use validation::*;
 pub use versioning::*;
 pub use websocket::*;
+pub use workers::*;

crates/openre-api/src/main.rs

@@ -0,0 +1,38 @@
+//! API server binary for open-re
+
+use openre_api::{AppState, http::start_server};
+use std::sync::Arc;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Initialize tracing
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::EnvFilter::new(
+            std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into()),
+        ))
+        .with(tracing_subscriber::fmt::layer())
+        .init();
+
+    // Load configuration (uses Figment: config.toml, env vars, etc.)
+    let config = openre_config::Config::load()?;
+
+    // Create application state
+    let state = Arc::new(AppState::new(config).await?);
+
+    // Get server address from environment or use default
+    let host = std::env::var("HOST").unwrap_or_else(|_| "0.0.0.0".into());
+    let port = std::env::var("PORT").unwrap_or_else(|_| "8080".into());
+    let addr = format!("{}:{}", host, port);
+
+    // Run database migrations
+    tracing::info!("Running database migrations...");
+    state.global_store.run_migrations().await?;
+    tracing::info!("Database migrations completed");
+
+    // Start the server
+    tracing::info!("Starting server on {}", addr);
+    start_server(state, &addr).await?;
+
+    Ok(())
+}

crates/openre-api/src/state.rs

@@ -12,7 +12,7 @@ use openre_security_ai::{
     FindingProvider, ScanStorageFindingProvider, SecurityAnalyst, SecurityAnalystImpl,
 };
 use openre_storage::{GlobalStore, ObjectStore};
-use openre_telemetry::{metrics::MetricsRegistry, TelemetryHandle};
+use openre_telemetry::{metrics::MetricsRegistry, TelemetryHandle, init_telemetry, TelemetryGuards};
 use std::num::NonZeroU32;
 use std::sync::Arc;
 use std::time::Duration;
@@ -22,14 +22,26 @@ use tokio::sync::RwLock;
 pub struct Telemetry {
     pub metrics: MetricsRegistry,
     pub _handle: TelemetryHandle,
+    pub _guards: TelemetryGuards,
 }
 
 impl Telemetry {
-    pub fn new(_config: &openre_config::TelemetryConfig) -> ApiResult<Self> {
-        Ok(Self {
-            metrics: MetricsRegistry::new(),
-            _handle: TelemetryHandle,
-        })
+    pub async fn new(config: &openre_config::TelemetryConfig) -> ApiResult<Self> {
+        let full_config = openre_config::Config {
+            server: Default::default(),
+            database: Default::default(),
+            redis: Default::default(),
+            storage: Default::default(),
+            plugins: Default::default(),
+            ai: Default::default(),
+            queue: Default::default(),
+            telemetry: config.clone(),
+            security: Default::default(),
+            auth: Default::default(),
+        };
+        let guards = init_telemetry(&full_config).await?;
+
+        Ok(Self { metrics: MetricsRegistry::new(), _handle: TelemetryHandle, _guards: guards })
     }
 }
 
@@ -62,7 +74,7 @@ impl AppState {
     /// Create new application state
     pub async fn new(config: Config) -> ApiResult<Self> {
         // Initialize telemetry
-        let telemetry = Arc::new(Telemetry::new(&config.telemetry)?);
+        let telemetry = Arc::new(Telemetry::new(&config.telemetry).await?);
 
         // Initialize stores
         let global_store = Arc::new(GlobalStore::new(&config.database).await?);

crates/openre-api/src/workers.rs

@@ -0,0 +1,103 @@
+//! Job handlers for open-re workers
+
+use crate::AppState;
+use openre_core::ids::FileId;
+use openre_core::traits::JobType;
+use openre_queue::{BoxedJobHandler, Job, JobHandler};
+use std::sync::Arc;
+use tokio::io::AsyncReadExt;
+use tracing::{error, info, warn};
+
+/// Analysis job handler
+pub struct AnalysisJobHandler {
+    state: Arc<AppState>,
+}
+
+impl AnalysisJobHandler {
+    pub fn new(state: Arc<AppState>) -> Self {
+        Self { state }
+    }
+}
+
+#[async_trait::async_trait]
+impl JobHandler for AnalysisJobHandler {
+    fn job_type(&self) -> JobType {
+        JobType::Analysis
+    }
+
+    async fn handle(&self, job: Job) -> openre_core::error::OpenreResult<serde_json::Value> {
+        info!(job_id = %job.id, "Starting analysis job");
+
+        // Extract payload
+        let file_id_str = job.payload.get("file_id").and_then(|v| v.as_str());
+        let stages = job.payload.get("stages").and_then(|v| v.as_array());
+        let _config = job.payload.get("config");
+
+        let file_id_str = match file_id_str {
+            Some(id) => id,
+            None => {
+                error!(job_id = %job.id, "No file_id in job payload");
+                return Err(openre_core::Error::InvalidInput("Missing file_id".into()));
+            }
+        };
+
+        // Parse file ID
+        let file_id = file_id_str.parse::<FileId>().map_err(|_| {
+            openre_core::Error::InvalidInput(format!("Invalid file_id: {}", file_id_str))
+        })?;
+
+        // Validate stages - at least one stage must be specified
+        let stage_count = stages.map(|s| s.len()).unwrap_or(0);
+        if stage_count == 0 {
+            warn!(job_id = %job.id, "No analysis stages specified, using default stages");
+        }
+
+        // Verify file exists in object storage (streaming check, no full load)
+        // We just open the stream and read a small amount to verify accessibility
+        let mut file_stream = self.state.object_store.get_object(file_id).await?;
+        let mut verify_buffer = [0u8; 1024];
+        let bytes_read = tokio::io::AsyncReadExt::read(&mut file_stream, &mut verify_buffer).await?;
+        if bytes_read == 0 {
+            error!(job_id = %job.id, file_id = %file_id_str, "File is empty or inaccessible");
+            return Err(openre_core::Error::InvalidInput("File is empty".into()));
+        }
+
+        // Get file size from object store metadata (avoid loading entire file)
+        let file_size = self.state.object_store.get_size(file_id).await.unwrap_or(0);
+
+        // TODO: Implement actual analysis pipeline
+        // This would include:
+        // 1. Binary format identification (ELF, PE, Mach-O, WASM)
+        // 2. Architecture detection
+        // 3. Function discovery and CFG construction
+        // 4. Data flow analysis
+        // 5. Type recovery
+        // 6. Decompilation
+        // 7. AI enrichment (if enabled)
+        // 8. Export results
+
+        let start_time = std::time::Instant::now();
+
+        // Placeholder for actual analysis - in production this runs the full pipeline
+        // For now, we return a structured result indicating the job was received
+        // and what stages would be run
+        let result = serde_json::json!({
+            "file_id": file_id_str,
+            "status": "completed",
+            "stages_requested": stage_count,
+            "stages_completed": 0,  // Will be updated by actual pipeline
+            "functions_found": 0,
+            "analysis_duration_ms": start_time.elapsed().as_millis() as u64,
+            "file_size_bytes": file_size,
+            "note": "Analysis pipeline not yet implemented - this is a scaffold"
+        });
+
+        info!(job_id = %job.id, file_size = file_size, "Analysis job scaffold completed");
+        Ok(result)
+    }
+}
+
+/// Get all job handlers for the worker
+pub fn get_job_handlers(state: Arc<AppState>) -> Vec<BoxedJobHandler> {
+    vec![Arc::new(AnalysisJobHandler::new(state))]
+}

crates/openre-cli/Cargo.toml

@@ -32,6 +32,7 @@ serde_json = { workspace = true }
 uuid = { workspace = true }
 chrono = { workspace = true }
 reqwest = { workspace = true }
+tracing = { workspace = true }
 indicatif = "0.17"
 colored = "2.0"
 tabled = { version = "0.16", features = ["derive"] }

crates/openre-cli/src/commands/file.rs

@@ -138,7 +138,7 @@ impl FileCommands {
 
                 if !response.status().is_success() {
                     let error = response.text().await?;
-                    return Err(CliError::ApiError(error));
+                    return Err(error.into());
                 }
 
                 let file_response: FileResponse = response.json().await?;

crates/openre-cli/src/commands/mod.rs

@@ -13,3 +13,4 @@ pub mod project;
 pub mod report;
 pub mod scan;
 pub mod server;
+pub mod worker;

crates/openre-cli/src/commands/worker.rs

@@ -0,0 +1,94 @@
+//! Worker commands
+
+use crate::{CliError, Context};
+use clap::{Parser, Subcommand};
+use colored::Colorize;
+use openre_api::{AppState, get_job_handlers};
+use openre_config::Config;
+use openre_queue::{WorkerPool, WorkerMetrics as QueueWorkerMetrics};
+use openre_telemetry::metrics::{MetricsRegistry, WorkerMetrics as TelemetryWorkerMetrics};
+use std::sync::Arc;
+use tracing::{error, info};
+
+#[derive(Subcommand)]
+pub enum WorkerCommands {
+    /// Start the worker
+    Start {
+        /// Number of concurrent jobs
+        #[arg(short, long, default_value = "4")]
+        concurrency: usize,
+
+        /// Queue priorities to process (comma-separated)
+        #[arg(short, long, default_value = "high,default,low")]
+        priorities: String,
+
+        /// Enable AI capabilities
+        #[arg(long)]
+        ai_enabled: bool,
+    },
+}
+
+impl WorkerCommands {
+    pub async fn execute(self, mut ctx: Context) -> Result<(), CliError> {
+        match self {
+            WorkerCommands::Start {
+                concurrency,
+                priorities,
+                ai_enabled,
+            } => {
+                println!("{} Starting worker with concurrency: {}", "✓".green(), concurrency);
+                println!("  Priorities: {}", priorities);
+                println!("  AI enabled: {}", ai_enabled);
+
+                // Load configuration (uses Figment: config.toml, env vars, etc.)
+                let config = Config::load()?;
+
+                // Create application state (reusing API state creation, which initializes telemetry)
+                let state = Arc::new(AppState::new(config.clone()).await?);
+
+                // Get job handlers
+                let handlers = get_job_handlers(state.clone());
+
+                // Create worker config from app config
+                let worker_config = openre_config::WorkerConfig {
+                    min_workers: 1,
+                    max_workers: concurrency,
+                    max_concurrent_jobs: concurrency,
+                    max_memory_mb: 4096,
+                    heartbeat_interval_secs: 10,
+                    graceful_shutdown_timeout_secs: 60,
+                    target_queue_depth_per_worker: 10,
+                };
+
+                // Create worker metrics (using the telemetry from AppState which has Prometheus exporter running)
+                let worker_metrics = Arc::new(TelemetryWorkerMetrics::new(&state.telemetry.metrics));
+
+                // Create worker pool
+                let mut worker_pool = WorkerPool::new(
+                    state.queue_manager.clone(),
+                    worker_config,
+                    config.queue.clone(),
+                    worker_metrics,
+                );
+
+                // Start the worker pool
+                info!("Starting worker pool with {} workers", concurrency);
+                worker_pool.start(handlers).await.map_err(|e| CliError::CoreError(e))?;
+
+                // Start scheduler
+                info!("Starting scheduler");
+                state.scheduler.start().await;
+
+                // Wait for shutdown signal
+                tokio::signal::ctrl_c().await?;
+                info!("Shutdown signal received, stopping workers...");
+
+                // Graceful shutdown
+                worker_pool.stop().await.map_err(|e| CliError::CoreError(e))?;
+
+                println!("{} Worker stopped gracefully", "✓".green());
+            }
+        }
+        Ok(())
+    }
+}