diff --git a/.markdownlint-cli2.jsonc b/.markdownlint-cli2.jsonc index b56667e..8971204 100644 --- a/.markdownlint-cli2.jsonc +++ b/.markdownlint-cli2.jsonc @@ -25,7 +25,7 @@ // It does NOT catch malformed closing fences with language specifiers (e.g., ```plaintext) // CommonMark spec requires closing fences to be ``` only (no language) // Use separate validation script to check closing fences - "MD040": false, // fenced-code-language (relaxed - flexible language specifiers) + "MD040": true, // fenced-code-language (relaxed - flexible language specifiers) // Formatting - strict whitespace "MD009": true, // no-hard-tabs diff --git a/Cargo.lock b/Cargo.lock index f278cb6..9b5f385 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8060,7 +8060,7 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d8c27177b12a6399ffc08b98f76f7c9a1f4fe9fc967c784c5a071fa8d93cf7e1" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -9249,10 +9249,12 @@ name = "vapora-shared" version = "1.2.0" dependencies = [ "chrono", + "regex", "serde", "serde_json", "surrealdb", "thiserror 2.0.17", + "tokio", "toml", "tracing", "uuid", diff --git a/assets/web/index.html b/assets/web/index.html index 883c411..2083f56 100644 --- a/assets/web/index.html +++ b/assets/web/index.html @@ -1 +1 @@ - Vapora
✅ v1.2.0
Vapora - Development Orchestration

Evaporate complexity

Development Flows

Specialized agentsorchestrate pipelines for design, implementation, testing, documentation and deployment. Agents learn from history and optimize costs automatically.100% self-hosted.

The 4 Problems It Solves

01

Context Switching

Developers jump between tools constantly. Vapora unifies everything in one intelligent system where context flows.

02

Knowledge Fragmentation

Decisions lost in threads, code scattered, docs unmaintained. RAG search and semantic indexing make knowledge discoverable.

03

Manual Coordination

Orchestrating code review, testing, documentation and deployment manually creates bottlenecks. Multi-agent workflows solve this.

04

Dev-Ops Friction

Handoffs between developers and operations lack visibility and context. Vapora maintains unified deployment readiness.

How It Works

🤖

Specialized Agents

Customizable agents for every role: architecture, development, testing, documentation, deployment and more. Agents learn from execution history with recency bias for continuous improvement.

🧠

Intelligent Orchestration

Agents coordinate automatically based on dependencies, context and expertise. Learning-based selection improves over time. Budget enforcement with automatic fallback ensures cost control.

☸️

Cloud-Native & Self-Hosted

Deploy to any Kubernetes cluster (EKS, GKE, AKS, vanilla K8s). Local Docker Compose development. Zero vendor lock-in.

Technology Stack

RustAxumSurrealDBNATS JetStreamLeptos WASMKubernetesPrometheusGrafanaKnowledge Graph

Available Agents

ArchitectSystem design
DeveloperCode implementation
CodeReviewerQuality assurance
TesterTests & benchmarks
DocumenterDocumentation
MarketerMarketing content
PresenterPresentations
DevOpsCI/CD deployment
MonitorHealth & alerting
SecurityAudit & compliance
ProjectManagerRoadmap tracking
DecisionMakerConflict resolution

Ready for intelligent orchestration?

Built with Rust 🦀 | Open Source | Self-Hosted

Explore on GitHub →
+ Vapora
✅ v1.2.0
Vapora - Development Orchestration

Evaporate complexity

Development Flows

Specialized agents orchestrate pipelines for design, implementation, testing, documentation and deployment. Agents learn from history and optimize costs automatically.
100% self-hosted.

The 4 Problems It Solves

01

Context Switching

Developers jump between tools constantly. Vapora unifies everything in one intelligent system where context flows.

02

Knowledge Fragmentation

Decisions lost in threads, code scattered, docs unmaintained. RAG search and semantic indexing make knowledge discoverable.

03

Manual Coordination

Orchestrating code review, testing, documentation and deployment manually creates bottlenecks. Multi-agent workflows solve this.

04

Dev-Ops Friction

Handoffs between developers and operations lack visibility and context. Vapora maintains unified deployment readiness.

How It Works

🤖

Specialized Agents

Customizable agents for every role: architecture, development, testing, documentation, deployment and more. Agents learn from execution history with recency bias for continuous improvement.

🧠

Intelligent Orchestration

Agents coordinate automatically based on dependencies, context and expertise. Learning-based selection improves over time. Budget enforcement with automatic fallback ensures cost control.

☸️

Cloud-Native & Self-Hosted

Deploy to any Kubernetes cluster (EKS, GKE, AKS, vanilla K8s). Local Docker Compose development. Zero vendor lock-in.

Technology Stack

RustAxumSurrealDBNATS JetStreamLeptos WASMKubernetesPrometheusGrafanaKnowledge Graph

Available Agents

ArchitectSystem design
DeveloperCode implementation
CodeReviewerQuality assurance
TesterTests & benchmarks
DocumenterDocumentation
MarketerMarketing content
PresenterPresentations
DevOpsCI/CD deployment
MonitorHealth & alerting
SecurityAudit & compliance
ProjectManagerRoadmap tracking
DecisionMakerConflict resolution

Ready for intelligent orchestration?

Built with Rust 🦀 | Open Source | Self-Hosted

Explore on GitHub →
diff --git a/assets/web/src/index.html b/assets/web/src/index.html index 959a811..0bef7a9 100644 --- a/assets/web/src/index.html +++ b/assets/web/src/index.html @@ -461,7 +461,7 @@

Evaporate complexity

Development Flows

@@ -470,19 +470,19 @@ class="highlight" data-en="Specialized agents" data-es="Agentes especializados" - >Specialized agentsSpecialized agents orchestrate pipelines for design, implementation, testing, documentation and deployment. Agents learn from history and optimize - costs automatically. - +
100% self-hosted. - +

diff --git a/crates/vapora-agents/src/coordinator.rs b/crates/vapora-agents/src/coordinator.rs index 9a16f54..dbd7e99 100644 --- a/crates/vapora-agents/src/coordinator.rs +++ b/crates/vapora-agents/src/coordinator.rs @@ -2,12 +2,14 @@ // Phase 2: Complete implementation with NATS integration use std::collections::HashMap; +use std::path::PathBuf; use std::sync::Arc; use chrono::Utc; use thiserror::Error; use tracing::{debug, info, warn}; use uuid::Uuid; +use vapora_shared::validation::{SchemaRegistry, ValidationPipeline}; use crate::learning_profile::{ExecutionData, LearningProfile, TaskTypeExpertise}; use crate::messages::{AgentMessage, TaskAssignment}; @@ -30,6 +32,9 @@ pub enum CoordinatorError { #[error("Invalid task state: {0}")] InvalidTaskState(String), + + #[error("Validation error: {0}")] + ValidationError(String), } use vapora_llm_router::BudgetManager; @@ -46,6 +51,7 @@ pub struct AgentCoordinator { swarm_coordinator: Option>, learning_profiles: Arc>>, budget_manager: Option>, + validation: Arc, } impl AgentCoordinator { @@ -100,12 +106,25 @@ impl AgentCoordinator { } }); + // Initialize validation pipeline + let schema_dir = + std::env::var("VAPORA_SCHEMA_DIR").unwrap_or_else(|_| "schemas".to_string()); + let schema_path = PathBuf::from(&schema_dir); + let schema_registry = Arc::new(SchemaRegistry::new(schema_path)); + let validation = Arc::new(ValidationPipeline::new(schema_registry)); + + info!( + "Initialized validation pipeline with schema dir: {}", + schema_dir + ); + Ok(Self { registry, nats_client, swarm_coordinator: Some(swarm_coordinator), learning_profiles: Arc::new(std::sync::RwLock::new(HashMap::new())), budget_manager: None, + validation, }) } @@ -118,12 +137,20 @@ impl AgentCoordinator { swarm_coordinator.register_agent(profile.clone()).ok(); } + // Initialize validation pipeline + let schema_dir = + std::env::var("VAPORA_SCHEMA_DIR").unwrap_or_else(|_| "schemas".to_string()); + let schema_path = PathBuf::from(&schema_dir); + let schema_registry = Arc::new(SchemaRegistry::new(schema_path)); + let validation = Arc::new(ValidationPipeline::new(schema_registry)); + Self { registry, nats_client: None, swarm_coordinator: Some(swarm_coordinator), learning_profiles: Arc::new(std::sync::RwLock::new(HashMap::new())), budget_manager: None, + validation, } } @@ -148,6 +175,42 @@ impl AgentCoordinator { context: String, priority: u32, ) -> Result { + // Validate inputs against schema + let input = serde_json::json!({ + "role": role, + "title": &title, + "description": &description, + "context": &context, + "priority": priority, + }); + + let validation_result = self + .validation + .validate("agents/task_assignment", &input) + .await + .map_err(|e| CoordinatorError::ValidationError(e.to_string()))?; + + if !validation_result.valid { + let error_messages: Vec = validation_result + .errors + .iter() + .map(|e| e.to_string()) + .collect(); + + warn!( + "Task assignment validation failed for role {}: {}", + role, + error_messages.join(", ") + ); + + return Err(CoordinatorError::ValidationError(format!( + "Validation failed: {}", + error_messages.join(", ") + ))); + } + + info!("Task assignment validated successfully for role: {}", role); + // Get all available candidates for role let all_agents = self.registry.get_agents_by_role(role); let candidates: Vec<_> = all_agents @@ -563,6 +626,9 @@ mod tests { #[tokio::test] async fn test_coordinator_creation() { + // Set schema directory for tests (relative to workspace root) + std::env::set_var("VAPORA_SCHEMA_DIR", "../../schemas"); + let registry = Arc::new(AgentRegistry::new(5)); let coordinator = AgentCoordinator::with_registry(registry); @@ -571,6 +637,9 @@ mod tests { #[tokio::test] async fn test_task_assignment() { + // Set schema directory for tests (relative to workspace root) + std::env::set_var("VAPORA_SCHEMA_DIR", "../../schemas"); + let registry = Arc::new(AgentRegistry::new(5)); // Register an agent @@ -595,11 +664,18 @@ mod tests { ) .await; + match &task_id { + Ok(_) => {} + Err(e) => eprintln!("Task assignment failed: {}", e), + } assert!(task_id.is_ok()); } #[tokio::test] async fn test_no_available_agent() { + // Set schema directory for tests (relative to workspace root) + std::env::set_var("VAPORA_SCHEMA_DIR", "../../schemas"); + let registry = Arc::new(AgentRegistry::new(5)); let coordinator = AgentCoordinator::with_registry(registry); diff --git a/crates/vapora-backend/src/api/error.rs b/crates/vapora-backend/src/api/error.rs index 692ec77..213292c 100644 --- a/crates/vapora-backend/src/api/error.rs +++ b/crates/vapora-backend/src/api/error.rs @@ -54,6 +54,10 @@ pub fn error_response(error: VaporaError) -> Response { format!("TOML error: {}", msg), ), VaporaError::InternalError(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg), + VaporaError::ValidationError(msg) => ( + StatusCode::BAD_REQUEST, + format!("Validation error: {}", msg), + ), }; let body = Json(json!({ diff --git a/crates/vapora-knowledge-graph/src/reasoning.rs b/crates/vapora-knowledge-graph/src/reasoning.rs index 77b3e38..13d003e 100644 --- a/crates/vapora-knowledge-graph/src/reasoning.rs +++ b/crates/vapora-knowledge-graph/src/reasoning.rs @@ -181,12 +181,14 @@ impl ReasoningEngine { .iter() .filter(|r| !visited.contains(&r.id) && r.solution.is_some()) { - if let Some(ref root) = other.root_cause { - if root.contains(chain.last().unwrap()) { - chain.push(other.solution.clone().unwrap()); - visited.insert(other.id.clone()); - break; - } + let Some(ref root) = other.root_cause else { + continue; + }; + + if root.contains(chain.last().unwrap()) { + chain.push(other.solution.clone().unwrap()); + visited.insert(other.id.clone()); + break; } } diff --git a/crates/vapora-knowledge-graph/src/temporal_kg.rs b/crates/vapora-knowledge-graph/src/temporal_kg.rs index 914f58a..8c5afdd 100644 --- a/crates/vapora-knowledge-graph/src/temporal_kg.rs +++ b/crates/vapora-knowledge-graph/src/temporal_kg.rs @@ -116,17 +116,18 @@ impl TemporalKG { let record = entry.value(); if record.timestamp > cutoff && record.task_type == task_type { - let similarity = if let Some(ref query_emb) = query_embedding { - // Phase 5.1: Use vector embedding similarity - if let Ok(Some(record_emb)) = self.get_or_embed(&record.description).await { + let similarity = match ( + &query_embedding, + self.get_or_embed(&record.description).await, + ) { + (Some(query_emb), Ok(Some(record_emb))) => { + // Phase 5.1: Use vector embedding similarity Self::compute_vector_similarity(query_emb, &record_emb) - } else { - // Fallback to Jaccard if embedding fails + } + _ => { + // Fallback to Jaccard if embedding fails or no embedding provider calculate_similarity(description, &record.description) } - } else { - // Fallback to Jaccard if no embedding provider - calculate_similarity(description, &record.description) }; if similarity >= threshold { @@ -160,15 +161,13 @@ impl TemporalKG { for task in similar_tasks { if task.success { - let confidence = if let Some(ref query_emb) = query_embedding { - if let Ok(Some(task_emb)) = self.get_or_embed(&task.description).await { - Self::compute_vector_similarity(query_emb, &task_emb) - } else { - calculate_similarity(description, &task.description) - } - } else { - calculate_similarity(description, &task.description) - }; + let confidence = + match (&query_embedding, self.get_or_embed(&task.description).await) { + (Some(query_emb), Ok(Some(task_emb))) => { + Self::compute_vector_similarity(query_emb, &task_emb) + } + _ => calculate_similarity(description, &task.description), + }; recommendations.push(Recommendation { source_record_id: task.id.clone(), @@ -284,28 +283,27 @@ impl TemporalKG { for entry in self.records.iter() { let record = entry.value(); if !record.success { - if let Some(error) = &record.error { - let similarity = if let Some(ref pattern_emb) = pattern_embedding { - if let Ok(Some(error_emb)) = self.get_or_embed(error).await { - Self::compute_vector_similarity(pattern_emb, &error_emb) - } else { - calculate_similarity(cause_pattern, error) - } - } else { - calculate_similarity(cause_pattern, error) - }; + let Some(error) = &record.error else { + continue; + }; - if similarity >= threshold { - relationships.push(CausalRelationship { - cause: error.clone(), - effect: record - .solution - .clone() - .unwrap_or_else(|| "unknown".to_string()), - confidence: similarity, - frequency: 1, - }); + let similarity = match (&pattern_embedding, self.get_or_embed(error).await) { + (Some(pattern_emb), Ok(Some(error_emb))) => { + Self::compute_vector_similarity(pattern_emb, &error_emb) } + _ => calculate_similarity(cause_pattern, error), + }; + + if similarity >= threshold { + relationships.push(CausalRelationship { + cause: error.clone(), + effect: record + .solution + .clone() + .unwrap_or_else(|| "unknown".to_string()), + confidence: similarity, + frequency: 1, + }); } } } diff --git a/crates/vapora-llm-router/src/router.rs b/crates/vapora-llm-router/src/router.rs index 9fad536..c144d6f 100644 --- a/crates/vapora-llm-router/src/router.rs +++ b/crates/vapora-llm-router/src/router.rs @@ -158,32 +158,29 @@ impl LLMRouter { } // Check budget if provided - if let Some(role) = agent_role { - if let Some(budget_mgr) = &self.budget_manager { - match budget_mgr.check_budget(role).await { - Ok(status) => { - if status.exceeded { - // Budget exceeded - use fallback provider - info!( - "Budget exceeded for role {}, using fallback provider: {}", - role, status.fallback_provider - ); - return Ok(status.fallback_provider); - } - - if status.near_threshold { - // Budget near threshold - prefer cost-efficient providers - debug!( - "Budget near threshold for role {}, selecting cost-efficient \ - provider", - role - ); - return self.select_cost_efficient_provider(task_type).await; - } - } - Err(e) => { - warn!("Budget check failed: {}, continuing with normal routing", e); - } + if let (Some(role), Some(budget_mgr)) = (agent_role, &self.budget_manager) { + match budget_mgr.check_budget(role).await { + Ok(status) if status.exceeded => { + // Budget exceeded - use fallback provider + info!( + "Budget exceeded for role {}, using fallback provider: {}", + role, status.fallback_provider + ); + return Ok(status.fallback_provider); + } + Ok(status) if status.near_threshold => { + // Budget near threshold - prefer cost-efficient providers + debug!( + "Budget near threshold for role {}, selecting cost-efficient provider", + role + ); + return self.select_cost_efficient_provider(task_type).await; + } + Ok(_) => { + // Budget ok, continue with normal routing + } + Err(e) => { + warn!("Budget check failed: {}, continuing with normal routing", e); } } } @@ -327,13 +324,7 @@ impl LLMRouter { ); // Record spend with budget manager if available - if let Some(role) = agent_role { - if let Some(budget_mgr) = &self.budget_manager { - if let Err(e) = budget_mgr.record_spend(role, cost as u32).await { - warn!("Failed to record budget spend: {}", e); - } - } - } + self.record_budget_spend(agent_role, cost as u32).await; } Ok(response) @@ -411,6 +402,15 @@ impl LLMRouter { latency_ms: provider.latency_ms(), }) } + + /// Record budget spend for agent role + async fn record_budget_spend(&self, agent_role: Option<&str>, cost: u32) { + if let (Some(role), Some(budget_mgr)) = (agent_role, &self.budget_manager) { + if let Err(e) = budget_mgr.record_spend(role, cost).await { + warn!("Failed to record budget spend: {}", e); + } + } + } } #[derive(Debug, Clone)] diff --git a/crates/vapora-mcp-server/src/main.rs b/crates/vapora-mcp-server/src/main.rs index 9652613..ba64d39 100644 --- a/crates/vapora-mcp-server/src/main.rs +++ b/crates/vapora-mcp-server/src/main.rs @@ -1,10 +1,13 @@ // vapora-mcp-server: Model Context Protocol server for VAPORA v1.0 // Phase 2: Standalone MCP server with HTTP endpoints +// Phase 3: Schema validation pipeline integration use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; use axum::{ - extract::{Json, Path}, + extract::{Json, Path, State}, http::StatusCode, response::IntoResponse, routing::{get, post}, @@ -15,6 +18,7 @@ use serde::{Deserialize, Serialize}; use serde_json::json; use tokio::net::TcpListener; use tracing::{info, warn}; +use vapora_shared::validation::{SchemaRegistry, ValidationPipeline}; #[derive(Parser)] #[command(name = "vapora-mcp-server")] @@ -56,6 +60,15 @@ struct PromptDefinition { description: String, } +// ============================================================================ +// App State +// ============================================================================ + +#[derive(Clone)] +struct AppState { + validation: Arc, +} + // ============================================================================ // Handlers // ============================================================================ @@ -153,9 +166,59 @@ async fn list_tools() -> impl IntoResponse { Json(json!({ "tools": tools })) } -async fn invoke_tool(Json(request): Json) -> impl IntoResponse { +async fn invoke_tool( + State(state): State, + Json(request): Json, +) -> impl IntoResponse { info!("Invoking tool: {}", request.tool); + // Validate parameters against schema + let schema_name = format!("tools/{}", request.tool); + let validation_result = match state + .validation + .validate(&schema_name, &request.parameters) + .await + { + Ok(result) => result, + Err(e) => { + warn!("Schema validation error for {}: {}", request.tool, e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ + "success": false, + "error": format!("Schema validation error: {}", e), + })), + ); + } + }; + + // Check if validation passed + if !validation_result.valid { + let error_messages: Vec = validation_result + .errors + .iter() + .map(|e| e.to_string()) + .collect(); + + warn!( + "Validation failed for {}: {}", + request.tool, + error_messages.join(", ") + ); + + return ( + StatusCode::BAD_REQUEST, + Json(json!({ + "success": false, + "error": "Validation failed", + "validation_errors": error_messages, + })), + ); + } + + // Use validated data (with defaults applied) + let validated_params = validation_result.validated_data.unwrap(); + let result = match request.tool.as_str() { "kanban_create_task" => json!({ "success": true, @@ -167,7 +230,7 @@ async fn invoke_tool(Json(request): Json) -> impl IntoRespons "message": "Task updated successfully" }), "get_project_summary" => json!({ - "project_id": request.parameters.get("project_id").and_then(|v| v.as_str()).unwrap_or("unknown"), + "project_id": validated_params.get("project_id").and_then(|v| v.as_str()).unwrap_or("unknown"), "total_tasks": 42, "completed": 15, "in_progress": 12, @@ -188,7 +251,7 @@ async fn invoke_tool(Json(request): Json) -> impl IntoRespons "success": true }), "get_agent_capabilities" => json!({ - "agent_id": request.parameters.get("agent_id").and_then(|v| v.as_str()).unwrap_or("unknown"), + "agent_id": validated_params.get("agent_id").and_then(|v| v.as_str()).unwrap_or("unknown"), "role": "Developer", "capabilities": ["coding", "debugging", "refactoring"], "llm_provider": "claude", @@ -301,6 +364,17 @@ async fn main() -> anyhow::Result<()> { let args = Args::parse(); + // Initialize validation pipeline + let schema_dir = std::env::var("VAPORA_SCHEMA_DIR").unwrap_or_else(|_| "schemas".to_string()); + let schema_path = PathBuf::from(&schema_dir); + + info!("Loading schemas from: {}", schema_path.display()); + + let registry = Arc::new(SchemaRegistry::new(schema_path)); + let validation = Arc::new(ValidationPipeline::new(registry)); + + let state = AppState { validation }; + // Build router let app = Router::new() .route("/health", get(health)) @@ -308,7 +382,8 @@ async fn main() -> anyhow::Result<()> { .route("/mcp/invoke", post(invoke_tool)) .route("/mcp/resources", get(list_resources)) .route("/mcp/resources/:uri", get(get_resource)) - .route("/mcp/prompts", get(list_prompts)); + .route("/mcp/prompts", get(list_prompts)) + .with_state(state); // Bind address let addr = format!("{}:{}", args.host, args.port).parse::()?; @@ -366,7 +441,15 @@ mod tests { #[tokio::test] async fn test_invoke_tool() { - let app = Router::new().route("/mcp/invoke", post(invoke_tool)); + // Create test state + let schema_path = PathBuf::from("schemas"); + let registry = Arc::new(SchemaRegistry::new(schema_path)); + let validation = Arc::new(ValidationPipeline::new(registry)); + let state = AppState { validation }; + + let app = Router::new() + .route("/mcp/invoke", post(invoke_tool)) + .with_state(state); let server = TestServer::new(app).unwrap(); let request = json!({ diff --git a/crates/vapora-shared/Cargo.toml b/crates/vapora-shared/Cargo.toml index b2957b8..17f5101 100644 --- a/crates/vapora-shared/Cargo.toml +++ b/crates/vapora-shared/Cargo.toml @@ -29,6 +29,12 @@ surrealdb = { workspace = true, optional = true } # Logging tracing = { workspace = true } +# Validation +regex = { workspace = true } + +# Async runtime (for validation pipeline) +tokio = { workspace = true, features = ["process", "io-util"] } + [features] default = ["backend"] backend = ["surrealdb"] diff --git a/crates/vapora-shared/src/error.rs b/crates/vapora-shared/src/error.rs index 911852c..86a65f7 100644 --- a/crates/vapora-shared/src/error.rs +++ b/crates/vapora-shared/src/error.rs @@ -22,6 +22,10 @@ pub enum VaporaError { #[error("Invalid input: {0}")] InvalidInput(String), + /// Schema validation error (Nickel contracts) + #[error("Validation error: {0}")] + ValidationError(String), + /// Authentication or authorization error #[error("Unauthorized: {0}")] Unauthorized(String), diff --git a/crates/vapora-shared/src/lib.rs b/crates/vapora-shared/src/lib.rs index c4b760a..cd85957 100644 --- a/crates/vapora-shared/src/lib.rs +++ b/crates/vapora-shared/src/lib.rs @@ -3,5 +3,6 @@ pub mod error; pub mod models; +pub mod validation; pub use error::{Result, VaporaError}; diff --git a/crates/vapora-shared/src/validation/mod.rs b/crates/vapora-shared/src/validation/mod.rs new file mode 100644 index 0000000..4ce735d --- /dev/null +++ b/crates/vapora-shared/src/validation/mod.rs @@ -0,0 +1,12 @@ +// vapora-shared: Validation module - Schema validation pipeline with Nickel +// contracts Phase: Schema Validation Pipeline implementation + +pub mod nickel_bridge; +pub mod pipeline; +pub mod schema_registry; + +pub use nickel_bridge::NickelCli; +pub use pipeline::{ValidationError, ValidationPipeline, ValidationResult}; +pub use schema_registry::{ + CompiledSchema, Contract, FieldSchema, FieldType, SchemaRegistry, SchemaSource, +}; diff --git a/crates/vapora-shared/src/validation/nickel_bridge.rs b/crates/vapora-shared/src/validation/nickel_bridge.rs new file mode 100644 index 0000000..312e3df --- /dev/null +++ b/crates/vapora-shared/src/validation/nickel_bridge.rs @@ -0,0 +1,254 @@ +// vapora-shared: Nickel CLI bridge - Execute Nickel commands for schema +// operations Provides interface to nickel query, typecheck, and export commands + +use std::path::{Path, PathBuf}; +use std::process::Stdio; + +use tokio::process::Command; +use tracing::debug; + +use crate::error::{Result, VaporaError}; + +/// Nickel CLI bridge for schema operations +#[derive(Debug, Clone)] +pub struct NickelCli { + /// Path to nickel executable + nickel_path: PathBuf, + + /// Timeout for nickel commands (milliseconds) + timeout_ms: u64, +} + +impl NickelCli { + /// Create new Nickel CLI bridge with default path + pub fn new() -> Self { + Self { + nickel_path: PathBuf::from("nickel"), + timeout_ms: 30_000, // 30 seconds + } + } + + /// Create Nickel CLI bridge with custom path + pub fn with_path(nickel_path: PathBuf) -> Self { + Self { + nickel_path, + timeout_ms: 30_000, + } + } + + /// Set timeout for Nickel commands + pub fn with_timeout(mut self, timeout_ms: u64) -> Self { + self.timeout_ms = timeout_ms; + self + } + + /// Execute `nickel query` to extract schema metadata + /// + /// Returns JSON representation of Nickel value + pub async fn query(&self, path: &Path, field: Option<&str>) -> Result { + let mut args = vec!["query", "--format", "json"]; + + if let Some(f) = field { + args.push("--field"); + args.push(f); + } + + let path_str = path.to_str().ok_or_else(|| { + VaporaError::ValidationError(format!("Invalid path: {}", path.display())) + })?; + args.push(path_str); + + debug!("Executing: nickel {}", args.join(" ")); + + let output = self.execute_with_timeout(&args).await?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(VaporaError::ValidationError(format!( + "Nickel query failed: {}", + stderr + ))); + } + + let stdout = String::from_utf8_lossy(&output.stdout); + + serde_json::from_str(&stdout).map_err(|e| { + VaporaError::ValidationError(format!("Failed to parse Nickel output: {}", e)) + }) + } + + /// Execute `nickel typecheck` to validate schema + /// + /// Returns Ok(()) if typecheck passes, Err with details if it fails + pub async fn typecheck(&self, path: &Path) -> Result<()> { + let path_str = path.to_str().ok_or_else(|| { + VaporaError::ValidationError(format!("Invalid path: {}", path.display())) + })?; + + let args = vec!["typecheck", path_str]; + + debug!("Executing: nickel {}", args.join(" ")); + + let output = self.execute_with_timeout(&args).await?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(VaporaError::ValidationError(format!( + "Nickel typecheck failed: {}", + stderr + ))); + } + + Ok(()) + } + + /// Execute `nickel export` to generate JSON output + /// + /// Returns exported JSON value + pub async fn export(&self, path: &Path) -> Result { + let path_str = path.to_str().ok_or_else(|| { + VaporaError::ValidationError(format!("Invalid path: {}", path.display())) + })?; + + let args = vec!["export", "--format", "json", path_str]; + + debug!("Executing: nickel {}", args.join(" ")); + + let output = self.execute_with_timeout(&args).await?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(VaporaError::ValidationError(format!( + "Nickel export failed: {}", + stderr + ))); + } + + let stdout = String::from_utf8_lossy(&output.stdout); + + serde_json::from_str(&stdout).map_err(|e| { + VaporaError::ValidationError(format!("Failed to parse Nickel export: {}", e)) + }) + } + + /// Check if Nickel CLI is available + pub async fn is_available(&self) -> bool { + let result = Command::new(&self.nickel_path) + .arg("--version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .await; + + match result { + Ok(status) => status.success(), + Err(_) => false, + } + } + + /// Get Nickel CLI version + pub async fn version(&self) -> Result { + let output = Command::new(&self.nickel_path) + .arg("--version") + .output() + .await + .map_err(|e| { + VaporaError::ValidationError(format!("Failed to get Nickel version: {}", e)) + })?; + + if !output.status.success() { + return Err(VaporaError::ValidationError( + "Failed to get Nickel version".to_string(), + )); + } + + let version = String::from_utf8_lossy(&output.stdout); + Ok(version.trim().to_string()) + } + + /// Execute Nickel command with timeout + async fn execute_with_timeout(&self, args: &[&str]) -> Result { + let child = Command::new(&self.nickel_path) + .args(args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .map_err(|e| { + VaporaError::ValidationError(format!("Failed to execute Nickel: {}", e)) + })?; + + // Wait with timeout + let timeout = tokio::time::Duration::from_millis(self.timeout_ms); + let result = tokio::time::timeout(timeout, child.wait_with_output()).await; + + match result { + Ok(Ok(output)) => Ok(output), + Ok(Err(e)) => Err(VaporaError::ValidationError(format!( + "Nickel command failed: {}", + e + ))), + Err(_) => { + // Timeout occurred - process is still running but we give up + Err(VaporaError::ValidationError(format!( + "Nickel command timed out after {}ms", + self.timeout_ms + ))) + } + } + } +} + +impl Default for NickelCli { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_nickel_cli_creation() { + let cli = NickelCli::new(); + assert_eq!(cli.nickel_path, PathBuf::from("nickel")); + assert_eq!(cli.timeout_ms, 30_000); + } + + #[tokio::test] + async fn test_nickel_cli_with_path() { + let custom_path = PathBuf::from("/usr/local/bin/nickel"); + let cli = NickelCli::with_path(custom_path.clone()); + assert_eq!(cli.nickel_path, custom_path); + } + + #[tokio::test] + async fn test_nickel_cli_with_timeout() { + let cli = NickelCli::new().with_timeout(5_000); + assert_eq!(cli.timeout_ms, 5_000); + } + + #[tokio::test] + #[ignore] // Requires Nickel CLI installed + async fn test_nickel_is_available() { + let cli = NickelCli::new(); + let available = cli.is_available().await; + // This will pass if nickel is in PATH + if available { + println!("Nickel CLI is available"); + } else { + println!("Nickel CLI is not available (install it to run this test)"); + } + } + + #[tokio::test] + #[ignore] // Requires Nickel CLI installed + async fn test_nickel_version() { + let cli = NickelCli::new(); + if cli.is_available().await { + let version = cli.version().await.unwrap(); + println!("Nickel version: {}", version); + assert!(!version.is_empty()); + } + } +} diff --git a/crates/vapora-shared/src/validation/pipeline.rs b/crates/vapora-shared/src/validation/pipeline.rs new file mode 100644 index 0000000..6540891 --- /dev/null +++ b/crates/vapora-shared/src/validation/pipeline.rs @@ -0,0 +1,636 @@ +// vapora-shared: Validation pipeline - Runtime validation against Nickel +// contracts Validates JSON inputs against compiled schemas with detailed error +// reporting + +use std::sync::Arc; + +use regex::Regex; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tracing::{debug, warn}; + +use super::schema_registry::{CompiledSchema, Contract, FieldType, SchemaRegistry}; +use crate::error::Result; + +/// Validation error with field context +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ValidationError { + /// Field name that failed validation + pub field: String, + + /// Human-readable error message + pub message: String, + + /// Contract that was violated + pub contract: String, + + /// Expected value/type + pub expected: Option, + + /// Actual value received + pub actual: Option, +} + +impl ValidationError { + /// Create a new validation error + pub fn new( + field: impl Into, + message: impl Into, + contract: impl Into, + ) -> Self { + Self { + field: field.into(), + message: message.into(), + contract: contract.into(), + expected: None, + actual: None, + } + } + + /// Add expected value to error + pub fn with_expected(mut self, expected: impl Into) -> Self { + self.expected = Some(expected.into()); + self + } + + /// Add actual value to error + pub fn with_actual(mut self, actual: impl Into) -> Self { + self.actual = Some(actual.into()); + self + } +} + +impl std::fmt::Display for ValidationError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Field '{}': {}", self.field, self.message)?; + + if let Some(expected) = &self.expected { + write!(f, " (expected: {})", expected)?; + } + + if let Some(actual) = &self.actual { + write!(f, " (got: {})", actual)?; + } + + write!(f, " [contract: {}]", self.contract) + } +} + +/// Result of validation pipeline +#[derive(Debug, Serialize, Deserialize)] +pub struct ValidationResult { + /// Is input valid? + pub valid: bool, + + /// Validation errors (if any) + pub errors: Vec, + + /// Validated data with defaults applied + pub validated_data: Option, + + /// Schema name that was used + pub schema_name: String, +} + +impl ValidationResult { + /// Create successful validation result + pub fn ok(data: Value, schema_name: impl Into) -> Self { + Self { + valid: true, + errors: vec![], + validated_data: Some(data), + schema_name: schema_name.into(), + } + } + + /// Create failed validation result + pub fn err(errors: Vec, schema_name: impl Into) -> Self { + Self { + valid: false, + errors, + validated_data: None, + schema_name: schema_name.into(), + } + } + + /// Get all error messages + pub fn error_messages(&self) -> Vec { + self.errors.iter().map(|e| e.to_string()).collect() + } + + /// Check if specific field has error + pub fn has_field_error(&self, field_name: &str) -> bool { + self.errors.iter().any(|e| e.field == field_name) + } +} + +/// Main validation pipeline +pub struct ValidationPipeline { + /// Schema registry + registry: Arc, + + /// Enable strict mode (fail on unknown fields) + strict_mode: bool, +} + +impl ValidationPipeline { + /// Create new validation pipeline + pub fn new(registry: Arc) -> Self { + Self { + registry, + strict_mode: false, + } + } + + /// Enable strict mode (reject unknown fields) + pub fn with_strict_mode(mut self, enabled: bool) -> Self { + self.strict_mode = enabled; + self + } + + /// Validate input against a named schema + pub async fn validate(&self, schema_name: &str, input: &Value) -> Result { + let schema = self.registry.load_schema(schema_name).await?; + self.validate_against_schema(&schema, input) + } + + /// Validate input against compiled schema + pub fn validate_against_schema( + &self, + schema: &CompiledSchema, + input: &Value, + ) -> Result { + let mut errors = vec![]; + let mut validated = serde_json::Map::new(); + + let input_obj = match input.as_object() { + Some(obj) => obj, + None => { + errors.push(ValidationError::new( + "", + "Input must be a JSON object", + "type", + )); + return Ok(ValidationResult::err(errors, &schema.name)); + } + }; + + // Validate all defined fields + for field_schema in &schema.fields { + let field_value = input_obj.get(&field_schema.name); + + // Check required fields + if field_schema.required && field_value.is_none() { + errors.push( + ValidationError::new(&field_schema.name, "Required field missing", "required") + .with_expected(format!("{:?}", field_schema.field_type)), + ); + continue; + } + + // Apply default if missing + let value = match field_value { + Some(v) => v.clone(), + None => match &field_schema.default { + Some(default) => { + debug!( + "Applying default for field {}: {:?}", + field_schema.name, default + ); + default.clone() + } + None => continue, + }, + }; + + // Type validation + if let Err(e) = self.validate_type(&field_schema.name, &value, &field_schema.field_type) + { + errors.push(e); + continue; + } + + // Contract validation + for contract in &field_schema.contracts { + if let Err(e) = self.validate_contract(&field_schema.name, &value, contract) { + errors.push(e); + } + } + + validated.insert(field_schema.name.clone(), value); + } + + // Strict mode: check for unknown fields + if self.strict_mode { + for (key, _value) in input_obj { + if !schema.fields.iter().any(|f| &f.name == key) { + errors.push(ValidationError::new( + key, + "Unknown field (strict mode enabled)", + "strict", + )); + } + } + } + + if errors.is_empty() { + Ok(ValidationResult::ok(Value::Object(validated), &schema.name)) + } else { + Ok(ValidationResult::err(errors, &schema.name)) + } + } + + /// Validate value type + fn validate_type( + &self, + field_name: &str, + value: &Value, + expected: &FieldType, + ) -> std::result::Result<(), ValidationError> { + let valid = match expected { + FieldType::String => value.is_string(), + FieldType::Number => value.is_number(), + FieldType::Bool => value.is_boolean(), + FieldType::Array(_) => value.is_array(), + FieldType::Object => value.is_object(), + FieldType::Enum(variants) => value + .as_str() + .map(|s| variants.contains(&s.to_string())) + .unwrap_or(false), + FieldType::Union(types) => { + // Valid if matches any type in union + types + .iter() + .any(|t| self.validate_type(field_name, value, t).is_ok()) + } + }; + + if valid { + Ok(()) + } else { + Err(ValidationError::new(field_name, "Type mismatch", "type") + .with_expected(format!("{:?}", expected)) + .with_actual(format!("{:?}", value))) + } + } + + /// Validate value against contract + fn validate_contract( + &self, + field_name: &str, + value: &Value, + contract: &Contract, + ) -> std::result::Result<(), ValidationError> { + match contract { + Contract::NonEmpty => { + if let Some(s) = value.as_str() { + if s.is_empty() { + return Err(ValidationError::new( + field_name, + "String cannot be empty", + "std.string.NonEmpty", + )); + } + } + } + + Contract::MinLength(min) => { + if let Some(s) = value.as_str() { + if s.len() < *min { + return Err(ValidationError::new( + field_name, + format!("String too short (minimum {} characters)", min), + format!("std.string.length.min {}", min), + ) + .with_expected(format!("≥ {} chars", min)) + .with_actual(format!("{} chars", s.len()))); + } + } + } + + Contract::MaxLength(max) => { + if let Some(s) = value.as_str() { + if s.len() > *max { + return Err(ValidationError::new( + field_name, + format!("String too long (maximum {} characters)", max), + format!("std.string.length.max {}", max), + ) + .with_expected(format!("≤ {} chars", max)) + .with_actual(format!("{} chars", s.len()))); + } + } + } + + Contract::Pattern(pattern) => { + if let Some(s) = value.as_str() { + let re = Regex::new(pattern).map_err(|e| { + ValidationError::new( + field_name, + format!("Invalid regex pattern: {}", e), + "std.string.match", + ) + })?; + + if !re.is_match(s) { + return Err(ValidationError::new( + field_name, + "Value does not match required pattern", + format!("std.string.match {}", pattern), + ) + .with_expected(format!("pattern: {}", pattern)) + .with_actual(s)); + } + } + } + + Contract::Range { min, max } => { + if let Some(n) = value.as_f64() { + if n < *min || n > *max { + return Err(ValidationError::new( + field_name, + format!("Number must be between {} and {}", min, max), + format!("std.number.between {} {}", min, max), + ) + .with_expected(format!("[{}, {}]", min, max)) + .with_actual(n.to_string())); + } + } + } + + Contract::GreaterThan(min) => { + if let Some(n) = value.as_f64() { + if n <= *min { + return Err(ValidationError::new( + field_name, + format!("Number must be greater than {}", min), + format!("std.number.greater_than {}", min), + ) + .with_expected(format!("> {}", min)) + .with_actual(n.to_string())); + } + } + } + + Contract::LessThan(max) => { + if let Some(n) = value.as_f64() { + if n >= *max { + return Err(ValidationError::new( + field_name, + format!("Number must be less than {}", max), + format!("std.number.less_than {}", max), + ) + .with_expected(format!("< {}", max)) + .with_actual(n.to_string())); + } + } + } + + Contract::Email => { + if let Some(s) = value.as_str() { + // Simple email validation (not RFC-compliant, but good enough) + if !s.contains('@') || !s.contains('.') || s.len() < 5 { + return Err(ValidationError::new( + field_name, + "Invalid email format", + "std.string.Email", + ) + .with_expected("user@example.com") + .with_actual(s)); + } + + // Check for multiple @ + if s.matches('@').count() != 1 { + return Err(ValidationError::new( + field_name, + "Invalid email format (multiple @ symbols)", + "std.string.Email", + )); + } + } + } + + Contract::Url => { + if let Some(s) = value.as_str() { + if !s.starts_with("http://") && !s.starts_with("https://") { + return Err(ValidationError::new( + field_name, + "Invalid URL format (must start with http:// or https://)", + "std.string.Url", + ) + .with_expected("https://example.com") + .with_actual(s)); + } + } + } + + Contract::Uuid => { + if let Some(s) = value.as_str() { + let uuid_pattern = + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"; + let re = Regex::new(uuid_pattern).unwrap(); + + if !re.is_match(s) { + return Err(ValidationError::new( + field_name, + "Invalid UUID format", + "std.string.Uuid", + ) + .with_expected("xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx") + .with_actual(s)); + } + } + } + + Contract::Custom(source) => { + // Custom contracts require Nickel evaluation + // For now, warn and skip + warn!( + "Custom contract validation not yet implemented for field {}: {}", + field_name, source + ); + } + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use serde_json::json; + + use super::*; + use crate::validation::schema_registry::{CompiledSchema, Contract, FieldSchema, FieldType}; + + fn create_test_schema() -> CompiledSchema { + CompiledSchema { + name: "test_schema".to_string(), + fields: vec![ + FieldSchema { + name: "title".to_string(), + field_type: FieldType::String, + required: true, + contracts: vec![Contract::NonEmpty, Contract::MaxLength(100)], + default: None, + doc: Some("Task title".to_string()), + }, + FieldSchema { + name: "priority".to_string(), + field_type: FieldType::Number, + required: true, + contracts: vec![Contract::Range { + min: 0.0, + max: 100.0, + }], + default: None, + doc: Some("Priority score".to_string()), + }, + FieldSchema { + name: "email".to_string(), + field_type: FieldType::String, + required: false, + contracts: vec![Contract::Email], + default: Some(json!("noreply@example.com")), + doc: None, + }, + ], + custom_contracts: vec![], + source_path: PathBuf::from("test.ncl"), + } + } + + #[test] + fn test_validation_error_display() { + let error = ValidationError::new("title", "Field is required", "required") + .with_expected("String") + .with_actual("null"); + + let display = error.to_string(); + assert!(display.contains("title")); + assert!(display.contains("required")); + assert!(display.contains("String")); + } + + #[test] + fn test_validation_result_ok() { + let data = json!({"title": "Test"}); + let result = ValidationResult::ok(data.clone(), "test_schema"); + + assert!(result.valid); + assert!(result.errors.is_empty()); + assert_eq!(result.validated_data, Some(data)); + } + + #[test] + fn test_validation_result_err() { + let errors = vec![ValidationError::new("field1", "Error", "contract1")]; + let result = ValidationResult::err(errors, "test_schema"); + + assert!(!result.valid); + assert_eq!(result.errors.len(), 1); + assert!(result.validated_data.is_none()); + } + + #[tokio::test] + async fn test_validate_valid_input() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("/tmp"))); + let pipeline = ValidationPipeline::new(registry); + let schema = create_test_schema(); + + let input = json!({ + "title": "Test Task", + "priority": 50.0 + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + assert!(result.valid, "Validation should pass for valid input"); + assert!(result.errors.is_empty()); + } + + #[tokio::test] + async fn test_validate_missing_required_field() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("/tmp"))); + let pipeline = ValidationPipeline::new(registry); + let schema = create_test_schema(); + + let input = json!({ + "priority": 50.0 + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + assert!(!result.valid); + assert!(result.has_field_error("title")); + } + + #[tokio::test] + async fn test_validate_contract_violation() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("/tmp"))); + let pipeline = ValidationPipeline::new(registry); + let schema = create_test_schema(); + + let input = json!({ + "title": "", // Empty string violates NonEmpty + "priority": 50.0 + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + assert!(!result.valid); + assert!(result.has_field_error("title")); + } + + #[tokio::test] + async fn test_validate_with_defaults() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("/tmp"))); + let pipeline = ValidationPipeline::new(registry); + let schema = create_test_schema(); + + let input = json!({ + "title": "Test", + "priority": 50.0 + // email not provided, should use default + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + assert!(result.valid); + + let validated = result.validated_data.unwrap(); + assert_eq!(validated["email"], "noreply@example.com"); + } + + #[tokio::test] + async fn test_validate_number_range() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("/tmp"))); + let pipeline = ValidationPipeline::new(registry); + let schema = create_test_schema(); + + // Out of range + let input = json!({ + "title": "Test", + "priority": 150.0 // > 100 + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + assert!(!result.valid); + assert!(result.has_field_error("priority")); + } + + #[tokio::test] + async fn test_validate_email_format() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("/tmp"))); + let pipeline = ValidationPipeline::new(registry); + let schema = create_test_schema(); + + let input = json!({ + "title": "Test", + "priority": 50.0, + "email": "invalid-email" // Missing @ + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + assert!(!result.valid); + assert!(result.has_field_error("email")); + } +} diff --git a/crates/vapora-shared/src/validation/schema_registry.rs b/crates/vapora-shared/src/validation/schema_registry.rs new file mode 100644 index 0000000..2bdc830 --- /dev/null +++ b/crates/vapora-shared/src/validation/schema_registry.rs @@ -0,0 +1,481 @@ +// vapora-shared: Schema registry - Loads and caches Nickel schemas for +// validation Compiles Nickel schemas to internal representation for runtime +// validation + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; +use tracing::{debug, info, warn}; + +use super::nickel_bridge::NickelCli; +use crate::error::{Result, VaporaError}; + +/// Compiled schema representation (from Nickel) +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CompiledSchema { + /// Schema name (e.g., "tools/kanban_create_task") + pub name: String, + + /// Field definitions with types and contracts + pub fields: Vec, + + /// Custom contract predicates (Nickel functions) + pub custom_contracts: Vec, + + /// Source file path + pub source_path: PathBuf, +} + +/// Field schema with type and validation contracts +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FieldSchema { + /// Field name (flattened path for nested fields) + pub name: String, + + /// Field type (Nickel type) + pub field_type: FieldType, + + /// Is this field required? + pub required: bool, + + /// Validation contracts (predicates) + pub contracts: Vec, + + /// Default value (if field is optional) + pub default: Option, + + /// Documentation string + pub doc: Option, +} + +/// Nickel field types +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub enum FieldType { + /// String type + String, + + /// Number type (f64) + Number, + + /// Boolean type + Bool, + + /// Array type with element type + Array(Box), + + /// Object/Record type + Object, + + /// Enum with allowed values + Enum(Vec), + + /// Union of types (Nickel `|`) + Union(Vec), +} + +/// Validation contract (Nickel predicate) +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub enum Contract { + /// Non-empty string (std.string.NonEmpty) + NonEmpty, + + /// Minimum string length + MinLength(usize), + + /// Maximum string length + MaxLength(usize), + + /// Regex pattern match + Pattern(String), + + /// Numeric range validation + Range { min: f64, max: f64 }, + + /// Greater than (exclusive) + GreaterThan(f64), + + /// Less than (exclusive) + LessThan(f64), + + /// Email validation + Email, + + /// URL validation + Url, + + /// UUID validation + Uuid, + + /// Custom Nickel predicate (function source code) + Custom(String), +} + +/// Schema source (file or inline definition) +#[derive(Debug, Clone)] +pub enum SchemaSource { + /// Load from Nickel file + File(PathBuf), + + /// Inline Nickel code + Inline(String), + + /// Pre-compiled schema (for testing) + Compiled(CompiledSchema), +} + +/// Registry that loads and caches Nickel schemas +pub struct SchemaRegistry { + /// Cached compiled schemas + schemas: Arc>>, + + /// Base directory for schema files + schema_dir: PathBuf, + + /// Nickel CLI bridge + nickel_cli: NickelCli, +} + +impl SchemaRegistry { + /// Create a new schema registry + pub fn new(schema_dir: PathBuf) -> Self { + Self { + schemas: Arc::new(RwLock::new(HashMap::new())), + schema_dir, + nickel_cli: NickelCli::new(), + } + } + + /// Create registry with custom Nickel CLI path + pub fn with_nickel_cli(schema_dir: PathBuf, nickel_path: PathBuf) -> Self { + Self { + schemas: Arc::new(RwLock::new(HashMap::new())), + schema_dir, + nickel_cli: NickelCli::with_path(nickel_path), + } + } + + /// Load schema from Nickel file, compile and cache + pub async fn load_schema(&self, schema_name: &str) -> Result { + // Check cache first + { + let cache = self.schemas.read().await; + if let Some(schema) = cache.get(schema_name) { + debug!("Schema {} loaded from cache", schema_name); + return Ok(schema.clone()); + } + } + + // Load from file + let schema_path = self.resolve_schema_path(schema_name); + let compiled = self.compile_schema(&schema_path, schema_name).await?; + + // Cache it + { + let mut cache = self.schemas.write().await; + cache.insert(schema_name.to_string(), compiled.clone()); + } + + info!("Schema {} compiled and cached", schema_name); + Ok(compiled) + } + + /// Load schema from source (file, inline, or pre-compiled) + pub async fn load_from_source( + &self, + source: SchemaSource, + schema_name: &str, + ) -> Result { + match source { + SchemaSource::File(path) => self.compile_schema(&path, schema_name).await, + SchemaSource::Inline(code) => self.compile_inline(&code, schema_name).await, + SchemaSource::Compiled(schema) => Ok(schema), + } + } + + /// Resolve schema name to file path + fn resolve_schema_path(&self, schema_name: &str) -> PathBuf { + let filename = if schema_name.ends_with(".ncl") { + schema_name.to_string() + } else { + format!("{}.ncl", schema_name) + }; + + self.schema_dir.join(filename) + } + + /// Compile Nickel schema from file to internal representation + async fn compile_schema(&self, path: &Path, name: &str) -> Result { + if !path.exists() { + return Err(VaporaError::ValidationError(format!( + "Schema file not found: {}", + path.display() + ))); + } + + // Use nickel query to extract schema metadata + let metadata_json = self.nickel_cli.query(path, None).await?; + + self.parse_nickel_metadata(&metadata_json, name, path) + } + + /// Compile inline Nickel code + async fn compile_inline(&self, code: &str, name: &str) -> Result { + // Write to temp file for Nickel CLI processing + let temp_dir = std::env::temp_dir(); + let temp_path = temp_dir.join(format!("vapora_schema_{}.ncl", name.replace('/', "_"))); + + tokio::fs::write(&temp_path, code).await.map_err(|e| { + VaporaError::ValidationError(format!("Failed to write temp file: {}", e)) + })?; + + let result = self.compile_schema(&temp_path, name).await; + + // Clean up temp file + let _ = tokio::fs::remove_file(&temp_path).await; + + result + } + + /// Parse Nickel metadata JSON into CompiledSchema + fn parse_nickel_metadata( + &self, + json: &serde_json::Value, + name: &str, + source_path: &Path, + ) -> Result { + // Extract schema name from metadata or use provided name + let schema_name = json["tool_name"] + .as_str() + .or_else(|| json["schema_name"].as_str()) + .unwrap_or(name) + .to_string(); + + // Extract parameters/fields object + let params = json["parameters"] + .as_object() + .or_else(|| json["fields"].as_object()); + + let fields = match params { + Some(params_obj) => params_obj + .iter() + .map(|(field_name, field_def)| self.parse_field(field_name, field_def)) + .collect::>>()?, + None => { + warn!("No parameters or fields found in schema {}", name); + vec![] + } + }; + + // Extract custom contracts + let custom_contracts = json["contracts"] + .as_object() + .map(|obj| { + obj.iter() + .map(|(name, _code)| name.clone()) + .collect::>() + }) + .unwrap_or_default(); + + Ok(CompiledSchema { + name: schema_name, + fields, + custom_contracts, + source_path: source_path.to_path_buf(), + }) + } + + /// Parse a single field definition from Nickel metadata + fn parse_field(&self, name: &str, def: &serde_json::Value) -> Result { + let field_type = self.parse_type(def)?; + let contracts = self.extract_contracts(def); + let default = def.get("default").cloned(); + let required = default.is_none(); + let doc = def.get("doc").and_then(|v| v.as_str()).map(String::from); + + Ok(FieldSchema { + name: name.to_string(), + field_type, + required, + contracts, + default, + doc, + }) + } + + /// Parse Nickel type from metadata + fn parse_type(&self, def: &serde_json::Value) -> Result { + // Type information comes from Nickel metadata + let type_str = def["type"].as_str().unwrap_or("String"); + + let field_type = match type_str { + "String" => FieldType::String, + "Number" => FieldType::Number, + "Bool" => FieldType::Bool, + "Object" | "Record" => FieldType::Object, + t if t.starts_with("Array") => { + // Parse array element type + FieldType::Array(Box::new(FieldType::String)) + } + t if t.starts_with("Enum") => { + // Extract enum variants + if let Some(variants) = def["variants"].as_array() { + let values = variants + .iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect(); + FieldType::Enum(values) + } else { + FieldType::String + } + } + _ => FieldType::String, + }; + + Ok(field_type) + } + + /// Extract validation contracts from Nickel metadata + fn extract_contracts(&self, def: &serde_json::Value) -> Vec { + let mut contracts = vec![]; + + // Parse contract annotations from Nickel metadata + if let Some(annotations) = def["annotations"].as_array() { + for ann in annotations { + if let Some(contract) = self.parse_contract_annotation(ann) { + contracts.push(contract); + } + } + } + + contracts + } + + /// Parse a single contract annotation + fn parse_contract_annotation(&self, ann: &serde_json::Value) -> Option { + let name = ann["name"].as_str()?; + + match name { + "std.string.NonEmpty" => Some(Contract::NonEmpty), + "std.string.Email" => Some(Contract::Email), + "std.string.Url" => Some(Contract::Url), + "std.string.Uuid" => Some(Contract::Uuid), + n if n.starts_with("std.string.length.min") => { + let min = ann["args"][0].as_u64()? as usize; + Some(Contract::MinLength(min)) + } + n if n.starts_with("std.string.length.max") => { + let max = ann["args"][0].as_u64()? as usize; + Some(Contract::MaxLength(max)) + } + n if n.starts_with("std.number.between") => { + let min = ann["args"][0].as_f64()?; + let max = ann["args"][1].as_f64()?; + Some(Contract::Range { min, max }) + } + n if n.starts_with("std.number.greater_than") => { + let min = ann["args"][0].as_f64()?; + Some(Contract::GreaterThan(min)) + } + n if n.starts_with("std.number.less_than") => { + let max = ann["args"][0].as_f64()?; + Some(Contract::LessThan(max)) + } + n if n.starts_with("std.string.match") => { + let pattern = ann["args"][0].as_str()?.to_string(); + Some(Contract::Pattern(pattern)) + } + _ => { + // Custom contract - store source code + ann["source"] + .as_str() + .map(|source| Contract::Custom(source.to_string())) + } + } + } + + /// Invalidate cache for a specific schema (for hot-reload) + pub async fn invalidate(&self, schema_name: &str) { + let mut cache = self.schemas.write().await; + cache.remove(schema_name); + debug!("Schema {} invalidated from cache", schema_name); + } + + /// Invalidate all cached schemas + pub async fn invalidate_all(&self) { + let mut cache = self.schemas.write().await; + let count = cache.len(); + cache.clear(); + info!("Invalidated {} schemas from cache", count); + } + + /// Get all cached schema names + pub async fn list_cached(&self) -> Vec { + let cache = self.schemas.read().await; + cache.keys().cloned().collect() + } + + /// Check if schema is cached + pub async fn is_cached(&self, schema_name: &str) -> bool { + let cache = self.schemas.read().await; + cache.contains_key(schema_name) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_parse_field_type() { + let registry = SchemaRegistry::new(PathBuf::from("/tmp")); + + let string_def = serde_json::json!({"type": "String"}); + let field_type = registry.parse_type(&string_def).unwrap(); + assert_eq!(field_type, FieldType::String); + + let number_def = serde_json::json!({"type": "Number"}); + let field_type = registry.parse_type(&number_def).unwrap(); + assert_eq!(field_type, FieldType::Number); + } + + #[tokio::test] + async fn test_parse_contract_annotation() { + let registry = SchemaRegistry::new(PathBuf::from("/tmp")); + + let non_empty = serde_json::json!({"name": "std.string.NonEmpty"}); + let contract = registry.parse_contract_annotation(&non_empty); + assert_eq!(contract, Some(Contract::NonEmpty)); + + let min_length = serde_json::json!({ + "name": "std.string.length.min", + "args": [5] + }); + let contract = registry.parse_contract_annotation(&min_length); + assert_eq!(contract, Some(Contract::MinLength(5))); + + let range = serde_json::json!({ + "name": "std.number.between", + "args": [0.0, 100.0] + }); + let contract = registry.parse_contract_annotation(&range); + assert_eq!( + contract, + Some(Contract::Range { + min: 0.0, + max: 100.0 + }) + ); + } + + #[tokio::test] + async fn test_schema_source() { + let inline_source = SchemaSource::Inline(r#"{ name = "test" }"#.to_string()); + assert!(matches!(inline_source, SchemaSource::Inline(_))); + + let file_source = SchemaSource::File(PathBuf::from("test.ncl")); + assert!(matches!(file_source, SchemaSource::File(_))); + } +} diff --git a/crates/vapora-shared/tests/validation_integration.rs b/crates/vapora-shared/tests/validation_integration.rs new file mode 100644 index 0000000..cced481 --- /dev/null +++ b/crates/vapora-shared/tests/validation_integration.rs @@ -0,0 +1,352 @@ +// Integration tests for ValidationPipeline +// Demonstrates end-to-end schema validation workflow + +use std::path::PathBuf; +use std::sync::Arc; + +use serde_json::json; +use vapora_shared::validation::{ + CompiledSchema, Contract, FieldSchema, FieldType, SchemaRegistry, SchemaSource, + ValidationPipeline, +}; + +/// Create a test schema for MCP tool validation +fn create_mcp_tool_schema() -> CompiledSchema { + CompiledSchema { + name: "tools/kanban_create_task".to_string(), + fields: vec![ + FieldSchema { + name: "project_id".to_string(), + field_type: FieldType::String, + required: true, + contracts: vec![Contract::NonEmpty, Contract::Uuid], + default: None, + doc: Some("Project UUID".to_string()), + }, + FieldSchema { + name: "title".to_string(), + field_type: FieldType::String, + required: true, + contracts: vec![ + Contract::NonEmpty, + Contract::MinLength(3), + Contract::MaxLength(200), + ], + default: None, + doc: Some("Task title (3-200 chars)".to_string()), + }, + FieldSchema { + name: "description".to_string(), + field_type: FieldType::String, + required: false, + contracts: vec![], + default: Some(json!("")), + doc: Some("Task description".to_string()), + }, + FieldSchema { + name: "priority".to_string(), + field_type: FieldType::String, + required: true, + contracts: vec![], + default: None, + doc: Some("Task priority".to_string()), + }, + ], + custom_contracts: vec![], + source_path: PathBuf::from("schemas/tools/kanban_create_task.ncl"), + } +} + +/// Create a test schema for agent task assignment +fn create_agent_assignment_schema() -> CompiledSchema { + CompiledSchema { + name: "agents/task_assignment".to_string(), + fields: vec![ + FieldSchema { + name: "role".to_string(), + field_type: FieldType::Enum(vec![ + "developer".to_string(), + "reviewer".to_string(), + "architect".to_string(), + "tester".to_string(), + ]), + required: true, + contracts: vec![], + default: None, + doc: Some("Agent role".to_string()), + }, + FieldSchema { + name: "title".to_string(), + field_type: FieldType::String, + required: true, + contracts: vec![Contract::NonEmpty, Contract::MaxLength(500)], + default: None, + doc: Some("Task title".to_string()), + }, + FieldSchema { + name: "priority".to_string(), + field_type: FieldType::Number, + required: false, + contracts: vec![Contract::Range { + min: 0.0, + max: 100.0, + }], + default: Some(json!(50)), + doc: Some("Priority score (0-100)".to_string()), + }, + ], + custom_contracts: vec![], + source_path: PathBuf::from("schemas/agents/task_assignment.ncl"), + } +} + +#[tokio::test] +async fn test_mcp_tool_valid_input() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + + // Load pre-compiled schema + let schema = create_mcp_tool_schema(); + let source = SchemaSource::Compiled(schema); + let loaded_schema = registry + .load_from_source(source, "tools/kanban_create_task") + .await + .unwrap(); + + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "project_id": "550e8400-e29b-41d4-a716-446655440000", + "title": "Implement validation pipeline", + "priority": "high" + }); + + let result = pipeline + .validate_against_schema(&loaded_schema, &input) + .unwrap(); + + assert!(result.valid, "Valid input should pass validation"); + assert!(result.errors.is_empty()); + assert!(result.validated_data.is_some()); + + // Check defaults applied + let validated = result.validated_data.unwrap(); + assert_eq!(validated["description"], ""); +} + +#[tokio::test] +async fn test_mcp_tool_missing_required_field() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "title": "Test Task", + // Missing project_id (required) + "priority": "high" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + assert!(result.has_field_error("project_id")); + assert!(result.error_messages()[0].contains("Required field missing")); +} + +#[tokio::test] +async fn test_mcp_tool_invalid_uuid() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "project_id": "not-a-valid-uuid", + "title": "Test Task", + "priority": "high" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + assert!(result.has_field_error("project_id")); + + let error_msg = result.error_messages()[0].to_lowercase(); + assert!(error_msg.contains("uuid") || error_msg.contains("format")); +} + +#[tokio::test] +async fn test_mcp_tool_title_too_short() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "project_id": "550e8400-e29b-41d4-a716-446655440000", + "title": "AB", // Only 2 chars, min is 3 + "priority": "high" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + assert!(result.has_field_error("title")); +} + +#[tokio::test] +async fn test_mcp_tool_title_too_long() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry); + + let long_title = "A".repeat(201); // Max is 200 + + let input = json!({ + "project_id": "550e8400-e29b-41d4-a716-446655440000", + "title": long_title, + "priority": "high" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + assert!(result.has_field_error("title")); +} + +#[tokio::test] +async fn test_agent_assignment_valid_input() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_agent_assignment_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "role": "developer", + "title": "Fix authentication bug" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(result.valid); + assert!(result.errors.is_empty()); + + // Check default priority applied + let validated = result.validated_data.unwrap(); + assert_eq!(validated["priority"], 50); +} + +#[tokio::test] +async fn test_agent_assignment_invalid_role() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_agent_assignment_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "role": "invalid_role", // Not in enum + "title": "Test Task" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + assert!(result.has_field_error("role")); +} + +#[tokio::test] +async fn test_agent_assignment_priority_out_of_range() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_agent_assignment_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "role": "developer", + "title": "Test Task", + "priority": 150 // > 100 + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + assert!(result.has_field_error("priority")); +} + +#[tokio::test] +async fn test_strict_mode_rejects_unknown_fields() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry).with_strict_mode(true); + + let input = json!({ + "project_id": "550e8400-e29b-41d4-a716-446655440000", + "title": "Test Task", + "priority": "high", + "unknown_field": "value" // Not in schema + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + assert!(result.has_field_error("unknown_field")); +} + +#[tokio::test] +async fn test_non_strict_mode_allows_unknown_fields() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry).with_strict_mode(false); + + let input = json!({ + "project_id": "550e8400-e29b-41d4-a716-446655440000", + "title": "Test Task", + "priority": "high", + "unknown_field": "value" // Allowed in non-strict mode + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + // Should pass, but unknown_field won't be in validated_data + assert!(result.valid); + let validated = result.validated_data.unwrap(); + assert!(!validated.as_object().unwrap().contains_key("unknown_field")); +} + +#[tokio::test] +async fn test_multiple_validation_errors() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "project_id": "invalid-uuid", + "title": "AB", // Too short + "priority": "high" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + assert!(!result.valid); + // Should have errors for both project_id and title + assert!(result.errors.len() >= 2); + assert!(result.has_field_error("project_id")); + assert!(result.has_field_error("title")); +} + +#[tokio::test] +async fn test_validation_result_serialization() { + let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); + let schema = create_mcp_tool_schema(); + let pipeline = ValidationPipeline::new(registry); + + let input = json!({ + "title": "", // Empty violates NonEmpty + "priority": "high" + }); + + let result = pipeline.validate_against_schema(&schema, &input).unwrap(); + + // Serialize to JSON + let json_result = serde_json::to_string(&result).unwrap(); + assert!(json_result.contains("valid")); + assert!(json_result.contains("errors")); + + // Deserialize back + let _deserialized: vapora_shared::validation::ValidationResult = + serde_json::from_str(&json_result).unwrap(); +} diff --git a/docs/architecture/README.md b/docs/architecture/README.md index c1a5729..f6d49d7 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -10,13 +10,16 @@ Complete system architecture and design documentation for VAPORA. - **[Multi-IA Router](multi-ia-router.md)** — Provider selection, routing rules, and fallback mechanisms - **[Roles, Permissions & Profiles](roles-permissions-profiles.md)** — Cedar policy engine and RBAC implementation - **[Task, Agent & Doc Manager](task-agent-doc-manager.md)** — Task orchestration and documentation lifecycle +- **[Schema Validation Pipeline](schema-validation-pipeline.md)** — Runtime validation with Nickel contracts for MCP tools and agent tasks ## Overview These documents cover: + - Complete system architecture and design decisions - Multi-agent orchestration and coordination patterns - Provider routing and selection strategies - Workflow execution and task management - Security, RBAC, and policy enforcement - Learning-based agent selection and cost optimization +- Runtime schema validation with Nickel contracts diff --git a/docs/architecture/schema-validation-pipeline.md b/docs/architecture/schema-validation-pipeline.md new file mode 100644 index 0000000..ae76859 --- /dev/null +++ b/docs/architecture/schema-validation-pipeline.md @@ -0,0 +1,507 @@ +# Schema Validation Pipeline + +Runtime validation system for MCP tools and agent task assignments using Nickel contracts. + +## Overview + +The Schema Validation Pipeline prevents downstream errors by validating inputs before execution. It uses Nickel schemas with contracts to enforce type safety, business rules, and data constraints at runtime. + +**Problem Solved:** VAPORA previously assumed valid inputs or failed downstream. This caused: + +- Invalid UUIDs reaching database queries +- Empty strings bypassing business logic +- Out-of-range priorities corrupting task queues +- Malformed contexts breaking agent execution + +**Solution:** Validate all inputs against Nickel schemas before execution. + +## Architecture + +```text +┌─────────────────────────────────────────────────────────────────┐ +│ Client Request │ +│ (MCP Tool Invocation / Agent Task Assignment) │ +└────────────────────────────┬────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ ValidationPipeline │ +│ ┌──────────────────────────────────────────────────────────┐ │ +│ │ 1. Load schema from SchemaRegistry (cached) │ │ +│ │ 2. Validate types (String, Number, Array, Object) │ │ +│ │ 3. Check required fields │ │ +│ │ 4. Apply contracts (NonEmpty, UUID, Range, etc.) │ │ +│ │ 5. Apply default values │ │ +│ │ 6. Return ValidationResult (valid + errors + data) │ │ +│ └──────────────────────────────────────────────────────────┘ │ +└────────────────────────────┬────────────────────────────────────┘ + │ + ┌──────────┴──────────┐ + │ │ + ▼ ▼ + ┌──────────────┐ ┌─────────────┐ + │ Valid? │ │ Invalid? │ + │ Execute │ │ Reject │ + │ with data │ │ with errors│ + └──────────────┘ └─────────────┘ +``` + +## Components + +### 1. ValidationPipeline + +Core validation engine in `vapora-shared/src/validation/pipeline.rs`. + +**Key Methods:** + +```rust +pub async fn validate( + &self, + schema_name: &str, + input: &Value +) -> Result +``` + +**Validation Steps:** + +1. Load compiled schema from registry +2. Validate field types (String, Number, Bool, Array, Object) +3. Check required fields (reject if missing) +4. Apply contracts (NonEmpty, UUID, Range, Email, etc.) +5. Apply default values for optional fields +6. Return ValidationResult with errors (if any) + +**Strict Mode:** Rejects unknown fields not in schema. + +### 2. SchemaRegistry + +Schema loading and caching in `vapora-shared/src/validation/schema_registry.rs`. + +**Features:** + +- **Caching:** Compiled schemas cached in memory (Arc + RwLock) +- **Hot Reload:** `invalidate(schema_name)` to reload without restart +- **Schema Sources:** File system, embedded string, or URL (future) + +**Schema Structure:** + +```rust +pub struct CompiledSchema { + pub name: String, + pub fields: HashMap, +} + +pub struct FieldSchema { + pub field_type: FieldType, + pub required: bool, + pub contracts: Vec, + pub default: Option, +} +``` + +### 3. NickelBridge + +CLI integration for Nickel operations in `vapora-shared/src/validation/nickel_bridge.rs`. + +**Operations:** + +- `typecheck(path)` — Validate Nickel syntax +- `export(path)` — Export schema as JSON +- `query(path, field)` — Query specific field +- `is_available()` — Check if Nickel CLI is installed + +**Timeout Protection:** 30s default to prevent DoS from malicious Nickel code. + +## Nickel Schemas + +Located in `schemas/` directory (workspace root). + +### Directory Structure + +```text +schemas/ +├── tools/ # MCP tool parameter validation +│ ├── kanban_create_task.ncl +│ ├── kanban_update_task.ncl +│ ├── assign_task_to_agent.ncl +│ ├── get_project_summary.ncl +│ └── get_agent_capabilities.ncl +└── agents/ # Agent task assignment validation + └── task_assignment.ncl +``` + +### Schema Format + +```nickel +{ + tool_name = "example_tool", + + parameters = { + # Required field with contracts + user_id + | String + | doc "User UUID" + | std.string.NonEmpty + | std.string.match "^[0-9a-f]{8}-[0-9a-f]{4}-...$", + + # Optional field with default + priority + | Number + | doc "Priority score (0-100)" + | std.number.between 0 100 + | default = 50, + }, +} +``` + +### Supported Contracts + +| Contract | Description | Example | +|----------|-------------|---------| +| `std.string.NonEmpty` | String cannot be empty | Required text fields | +| `std.string.length.min N` | Minimum length | `min 3` for titles | +| `std.string.length.max N` | Maximum length | `max 200` for titles | +| `std.string.match PATTERN` | Regex validation | UUID format | +| `std.number.between A B` | Numeric range | `between 0 100` | +| `std.number.greater_than N` | Minimum value (exclusive) | `> -1` | +| `std.number.less_than N` | Maximum value (exclusive) | `< 1000` | +| `std.enum.TaggedUnion` | Enum validation | `[| 'low, 'high |]` | + +## Integration Points + +### MCP Server + +Location: `crates/vapora-mcp-server/src/main.rs` + +```rust +// Initialize validation pipeline +let schema_dir = std::env::var("VAPORA_SCHEMA_DIR") + .unwrap_or_else(|_| "schemas".to_string()); +let registry = Arc::new(SchemaRegistry::new(PathBuf::from(&schema_dir))); +let validation = Arc::new(ValidationPipeline::new(registry)); + +// Add to AppState +#[derive(Clone)] +struct AppState { + validation: Arc, +} + +// Validate in handler +async fn invoke_tool( + State(state): State, + Json(request): Json, +) -> impl IntoResponse { + let schema_name = format!("tools/{}", request.tool); + let validation_result = state + .validation + .validate(&schema_name, &request.parameters) + .await?; + + if !validation_result.valid { + return (StatusCode::BAD_REQUEST, Json(validation_errors)); + } + + // Execute with validated data + let validated_params = validation_result.validated_data.unwrap(); + // ... +} +``` + +### Agent Coordinator + +Location: `crates/vapora-agents/src/coordinator.rs` + +```rust +pub struct AgentCoordinator { + validation: Arc, + // ... +} + +impl AgentCoordinator { + pub async fn assign_task( + &self, + role: &str, + title: String, + description: String, + context: String, + priority: u32, + ) -> Result { + // Validate inputs + let input = serde_json::json!({ + "role": role, + "title": &title, + "description": &description, + "context": &context, + "priority": priority, + }); + + let validation_result = self + .validation + .validate("agents/task_assignment", &input) + .await?; + + if !validation_result.valid { + return Err(CoordinatorError::ValidationError( + validation_result.errors.join(", ") + )); + } + + // Continue with validated inputs + // ... + } +} +``` + +## Usage Patterns + +### 1. Validating MCP Tool Inputs + +```rust +// In MCP server handler +let validation_result = state + .validation + .validate("tools/kanban_create_task", &input) + .await?; + +if !validation_result.valid { + let errors: Vec = validation_result + .errors + .iter() + .map(|e| e.to_string()) + .collect(); + + return (StatusCode::BAD_REQUEST, Json(json!({ + "success": false, + "validation_errors": errors, + }))); +} + +// Use validated data with defaults applied +let validated_data = validation_result.validated_data.unwrap(); +``` + +### 2. Validating Agent Task Assignments + +```rust +// In AgentCoordinator +let input = serde_json::json!({ + "role": role, + "title": title, + "description": description, + "context": context, + "priority": priority, +}); + +let validation_result = self + .validation + .validate("agents/task_assignment", &input) + .await?; + +if !validation_result.valid { + warn!("Validation failed: {:?}", validation_result.errors); + return Err(CoordinatorError::ValidationError( + format!("Invalid input: {}", validation_result.errors.join(", ")) + )); +} +``` + +### 3. Hot Reloading Schemas + +```rust +// Invalidate single schema +registry.invalidate("tools/kanban_create_task").await; + +// Invalidate all schemas (useful for config reload) +registry.invalidate_all().await; +``` + +## Testing Schemas + +### Validate Syntax + +```bash +nickel typecheck schemas/tools/kanban_create_task.ncl +``` + +### Export as JSON + +```bash +nickel export schemas/tools/kanban_create_task.ncl +``` + +### Query Specific Field + +```bash +nickel query --field parameters.title schemas/tools/kanban_create_task.ncl +``` + +## Adding New Schemas + +1. Create `.ncl` file in appropriate directory (`tools/` or `agents/`) +2. Define `tool_name` or `schema_name` +3. Define `parameters` or `fields` with types and contracts +4. Add `doc` annotations for documentation +5. Test with `nickel typecheck` +6. Restart services or use hot-reload + +**Example:** + +```nickel +# schemas/tools/my_new_tool.ncl +{ + tool_name = "my_new_tool", + + parameters = { + name + | String + | doc "User name (3-50 chars)" + | std.string.NonEmpty + | std.string.length.min 3 + | std.string.length.max 50, + + age + | Number + | doc "User age (0-120)" + | std.number.between 0 120, + + email + | String + | doc "User email address" + | std.string.Email, + + active + | Bool + | doc "Account status" + | default = true, + }, +} +``` + +## Configuration + +### Environment Variables + +- `VAPORA_SCHEMA_DIR` — Schema directory path (default: `"schemas"`) + +**In tests:** + +```rust +std::env::set_var("VAPORA_SCHEMA_DIR", "../../schemas"); +``` + +**In production:** + +```bash +export VAPORA_SCHEMA_DIR=/app/schemas +``` + +## Performance Characteristics + +- **Schema Loading:** ~5-10ms (first load, then cached) +- **Validation:** ~0.1-0.5ms per request (in-memory) +- **Hot Reload:** ~10-20ms (invalidates cache, reloads from disk) + +**Optimization:** SchemaRegistry uses `Arc>` for concurrent reads. + +## Security Considerations + +### Timeout Protection + +NickelBridge enforces 30s timeout on all CLI operations to prevent: + +- Infinite loops in malicious Nickel code +- DoS attacks via crafted schemas +- Resource exhaustion + +### Input Sanitization + +Contracts prevent: + +- SQL injection (via UUID/Email validation) +- XSS attacks (via length limits on text fields) +- Buffer overflows (via max length constraints) +- Type confusion (via strict type checking) + +### Schema Validation + +All schemas must pass `nickel typecheck` before deployment. + +## Error Handling + +### ValidationResult + +```rust +pub struct ValidationResult { + pub valid: bool, + pub errors: Vec, + pub validated_data: Option, +} + +pub enum ValidationError { + MissingField(String), + TypeMismatch { field: String, expected: String, got: String }, + ContractViolation { field: String, contract: String, value: String }, + InvalidSchema(String), +} +``` + +### Error Response Format + +```json +{ + "success": false, + "error": "Validation failed", + "validation_errors": [ + "Field 'project_id' must match UUID pattern", + "Field 'title' must be at least 3 characters", + "Field 'priority' must be between 0 and 100" + ] +} +``` + +## Troubleshooting + +### Schema Not Found + +**Error:** `Schema file not found: schemas/tools/my_tool.ncl` + +**Solution:** Check `VAPORA_SCHEMA_DIR` environment variable and ensure schema file exists. + +### Nickel CLI Not Available + +**Error:** `Nickel CLI not found in PATH` + +**Solution:** Install Nickel CLI: + +```bash +cargo install nickel-lang-cli +``` + +### Validation Always Fails + +**Error:** All requests rejected with validation errors + +**Solution:** Check schema syntax with `nickel typecheck`, verify field names match exactly. + +## Future Enhancements + +- [ ] Remote schema loading (HTTP/S3) +- [ ] Schema versioning and migration +- [ ] Custom contract plugins +- [ ] GraphQL schema generation from Nickel +- [ ] OpenAPI spec generation + +## Related Documentation + +- [Nickel Language Documentation](https://nickel-lang.org/) +- [VAPORA Architecture Overview](vapora-architecture.md) +- [Agent Coordination](agent-registry-coordination.md) +- [MCP Protocol Integration](../integrations/mcp-server.md) + +## References + +- **Implementation:** `crates/vapora-shared/src/validation/` +- **Schemas:** `schemas/` +- **Tests:** `crates/vapora-shared/tests/validation_integration.rs` +- **MCP Integration:** `crates/vapora-mcp-server/src/main.rs` +- **Agent Integration:** `crates/vapora-agents/src/coordinator.rs` diff --git a/schemas/README.md b/schemas/README.md new file mode 100644 index 0000000..8fb35a5 --- /dev/null +++ b/schemas/README.md @@ -0,0 +1,128 @@ +# VAPORA Validation Schemas + +Nickel schemas for runtime validation of MCP tools and agent tasks. + +## Directory Structure + +``` +schemas/ +├── tools/ # MCP tool parameter validation +│ ├── kanban_create_task.ncl +│ ├── kanban_update_task.ncl +│ ├── assign_task_to_agent.ncl +│ ├── get_project_summary.ncl +│ └── get_agent_capabilities.ncl +│ +└── agents/ # Agent task assignment validation + └── task_assignment.ncl +``` + +## Schema Format + +Nickel schemas define: + +- **Field types** (String, Number, Bool, Array, Object) +- **Required/optional fields** (via `default` values) +- **Validation contracts** (NonEmpty, Email, UUID, Range, Pattern, etc.) +- **Documentation** (via `doc` annotations) + +### Example + +```nickel +{ + tool_name = "example_tool", + + parameters = { + # Required field with contracts + user_id + | String + | doc "User UUID" + | std.string.NonEmpty + | std.string.match "^[0-9a-f]{8}-[0-9a-f]{4}-...$", + + # Optional field with default + priority + | Number + | doc "Priority score (0-100)" + | std.number.between 0 100 + | default = 50, + }, +} +``` + +## Supported Contracts + +| Contract | Description | Example | +|----------|-------------|---------| +| `std.string.NonEmpty` | String cannot be empty | Required text fields | +| `std.string.length.min N` | Minimum length | `min 3` for titles | +| `std.string.length.max N` | Maximum length | `max 200` for titles | +| `std.string.match PATTERN` | Regex validation | UUID format | +| `std.number.between A B` | Numeric range | `between 0 100` | +| `std.number.greater_than N` | Minimum value (exclusive) | `> -1` | +| `std.number.less_than N` | Maximum value (exclusive) | `< 1000` | +| `std.string.Email` | Email format | user@example.com | +| `std.string.Url` | URL format | https://... | +| `std.string.Uuid` | UUID format | xxxxxxxx-xxxx-... | +| `std.enum.TaggedUnion` | Enum validation | `[| 'low, 'high |]` | + +## Usage + +Schemas are loaded by `ValidationPipeline` at runtime: + +```rust +use vapora_shared::validation::{SchemaRegistry, ValidationPipeline}; + +let registry = Arc::new(SchemaRegistry::new(PathBuf::from("schemas"))); +let pipeline = ValidationPipeline::new(registry); + +// Validate MCP tool input +let result = pipeline.validate("tools/kanban_create_task", &input).await?; + +if !result.valid { + return (StatusCode::BAD_REQUEST, Json(result.errors)); +} +``` + +## Testing Schemas + +Validate schema syntax: + +```bash +nickel typecheck schemas/tools/kanban_create_task.ncl +``` + +Export schema as JSON: + +```bash +nickel export schemas/tools/kanban_create_task.ncl +``` + +Query specific field: + +```bash +nickel query --field parameters.title schemas/tools/kanban_create_task.ncl +``` + +## Adding New Schemas + +1. Create `.ncl` file in appropriate directory +2. Define `tool_name` or `schema_name` +3. Define `parameters` or `fields` with types and contracts +4. Add `doc` annotations for documentation +5. Test with `nickel typecheck` +6. Restart services to reload schemas (or use hot-reload) + +## Hot Reload + +Invalidate cached schema without restart: + +```rust +registry.invalidate("tools/kanban_create_task").await; +``` + +Invalidate all schemas: + +```rust +registry.invalidate_all().await; +``` diff --git a/schemas/agents/task_assignment.ncl b/schemas/agents/task_assignment.ncl new file mode 100644 index 0000000..3f337c6 --- /dev/null +++ b/schemas/agents/task_assignment.ncl @@ -0,0 +1,71 @@ +# Schema for agent task assignment +# Used by AgentCoordinator for internal task assignment validation +{ + schema_name = "task_assignment", + + description = "Validates agent task assignment requests from AgentCoordinator", + + fields = { + # Agent role - must match registered roles + role + | String + | doc "Agent role for task execution" + | std.string.NonEmpty + | std.enum.TaggedUnion + | [| 'developer, 'reviewer, 'architect, 'tester, 'documenter, 'devops, 'monitor, 'security |], + + # Task title + title + | String + | doc "Task title (3-500 chars)" + | std.string.NonEmpty + | std.string.length.min 3 + | std.string.length.max 500, + + # Task description + description + | String + | doc "Detailed task description" + | default = "", + + # Task context (JSON string) + context + | String + | doc "JSON context for task execution" + | default = "{}", + + # Priority score (0-100) + priority + | Number + | doc "Task priority score (0=lowest, 100=highest)" + | std.number.between 0 100 + | default = 50, + + # Optional deadline (ISO 8601 datetime string) + deadline + | String + | doc "Task deadline (ISO 8601 format, optional)" + | default = "", + }, + + # Custom validation contracts + contracts = { + # Title must not be only whitespace + title_not_whitespace = fun fields => + std.string.trim fields.title != "", + + # Context must be valid JSON + context_valid_json = fun fields => + if fields.context == "{}" then + true + else + std.string.is_match "^[\\{\\[]" fields.context, + + # If deadline is provided, must match ISO 8601 pattern + deadline_iso8601 = fun fields => + if fields.deadline == "" then + true + else + std.string.is_match "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}" fields.deadline, + }, +} diff --git a/schemas/tools/assign_task_to_agent.ncl b/schemas/tools/assign_task_to_agent.ncl new file mode 100644 index 0000000..d26bd12 --- /dev/null +++ b/schemas/tools/assign_task_to_agent.ncl @@ -0,0 +1,56 @@ +# Schema for assign_task_to_agent MCP tool +# Validates agent task assignment requests +{ + tool_name = "assign_task_to_agent", + + description = "Assign a task to an agent by role", + + parameters = { + # Agent role - must be valid role + role + | String + | doc "Agent role" + | std.string.NonEmpty + | std.enum.TaggedUnion + | [| 'developer, 'reviewer, 'architect, 'tester, 'documenter, 'devops, 'monitor, 'security |], + + # Task title + task_title + | String + | doc "Task title (3-500 chars)" + | std.string.NonEmpty + | std.string.length.min 3 + | std.string.length.max 500, + + # Task description + task_description + | String + | doc "Task description (detailed requirements)" + | std.string.NonEmpty + | std.string.length.min 10, + + # Optional priority (0-100) + priority + | Number + | doc "Task priority (0-100, default: 50)" + | std.number.between 0 100 + | default = 50, + + # Optional context (JSON string) + context + | String + | doc "Additional context as JSON string (optional)" + | default = "{}", + }, + + # Validation contracts + contracts = { + # Context must be valid JSON if provided + context_is_json = fun params => + if params.context == "{}" then + true + else + # Simple JSON validation - starts with { or [ + std.string.is_match "^[\\{\\[]" params.context, + }, +} diff --git a/schemas/tools/get_agent_capabilities.ncl b/schemas/tools/get_agent_capabilities.ncl new file mode 100644 index 0000000..e3716e4 --- /dev/null +++ b/schemas/tools/get_agent_capabilities.ncl @@ -0,0 +1,16 @@ +# Schema for get_agent_capabilities MCP tool +# Validates agent capability requests +{ + tool_name = "get_agent_capabilities", + + description = "Get detailed agent capabilities and status", + + parameters = { + # Agent UUID or ID + agent_id + | String + | doc "Agent UUID or unique identifier" + | std.string.NonEmpty + | std.string.length.min 3, + }, +} diff --git a/schemas/tools/get_project_summary.ncl b/schemas/tools/get_project_summary.ncl new file mode 100644 index 0000000..4677a43 --- /dev/null +++ b/schemas/tools/get_project_summary.ncl @@ -0,0 +1,16 @@ +# Schema for get_project_summary MCP tool +# Validates project summary requests +{ + tool_name = "get_project_summary", + + description = "Get project summary and statistics", + + parameters = { + # Project UUID + project_id + | String + | doc "UUID of the project" + | std.string.NonEmpty + | std.string.match "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", + }, +} diff --git a/schemas/tools/kanban_create_task.ncl b/schemas/tools/kanban_create_task.ncl new file mode 100644 index 0000000..decaf96 --- /dev/null +++ b/schemas/tools/kanban_create_task.ncl @@ -0,0 +1,45 @@ +# Schema for kanban_create_task MCP tool +# Validates task creation requests with contracts +{ + tool_name = "kanban_create_task", + + description = "Create task in Kanban board with validation", + + parameters = { + # Project UUID - must be valid UUID format + project_id + | String + | doc "UUID of the target project" + | std.string.NonEmpty + | std.string.match "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", + + # Task title - 3 to 200 characters + title + | String + | doc "Task title (3-200 chars)" + | std.string.NonEmpty + | std.string.length.min 3 + | std.string.length.max 200, + + # Optional description + description + | String + | doc "Task description (optional)" + | default = "", + + # Priority level - enum validation + priority + | String + | doc "Task priority level" + | std.string.NonEmpty + | std.enum.TaggedUnion + | [| 'low, 'medium, 'high, 'critical |], + }, + + # Validation contracts (custom predicates) + contracts = { + # Title must not be only whitespace + title_not_whitespace = fun params => + std.string.trim params.title != "", + }, +} diff --git a/schemas/tools/kanban_update_task.ncl b/schemas/tools/kanban_update_task.ncl new file mode 100644 index 0000000..862c910 --- /dev/null +++ b/schemas/tools/kanban_update_task.ncl @@ -0,0 +1,31 @@ +# Schema for kanban_update_task MCP tool +# Validates task status updates and reordering +{ + tool_name = "kanban_update_task", + + description = "Update task status and order in Kanban board", + + parameters = { + # Task UUID + task_id + | String + | doc "UUID of the task to update" + | std.string.NonEmpty + | std.string.match "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", + + # New status - enum validation + status + | String + | doc "New task status" + | std.string.NonEmpty + | std.enum.TaggedUnion + | [| 'todo, 'doing, 'review, 'done |], + + # Optional order within column + order + | Number + | doc "Order within column (0-based index, optional)" + | std.number.greater_than (-1) + | default = 0, + }, +}