scheduler and context management improvements

Author: Symbiont OSS Sync

Cargo.lock

@@ -102,11 +102,16 @@ symbi/
 ### ✅ Community Features (OSS)
 - **DSL Grammar**: Complete Tree-sitter grammar for agent definitions
 - **Agent Runtime**: Task scheduling, resource management, lifecycle control
+- **Real Task Execution**: Actual process spawning with comprehensive monitoring and metrics
+- **Graceful Shutdown**: Coordinated shutdown with resource cleanup and timeout handling
 - **Tier 1 Sandboxing**: Docker containerized isolation for agent operations
 - **MCP Integration**: Model Context Protocol client for external tools
-- **SchemaPin Security**: Basic cryptographic tool verification 
+- **SchemaPin Security**: Basic cryptographic tool verification
 - **RAG Engine**: Retrieval-augmented generation with vector search
-- **Context Management**: Persistent agent memory and knowledge storage
+- **Advanced Context Management**: Sophisticated memory with importance calculation and search modes
+- **Multi-Modal Search**: Keyword, temporal, similarity, and hybrid search capabilities
+- **Access Control Integration**: Policy engine connected context management with agent-scoped access
+- **Context Archiving**: Automatic archiving with retention policies and compressed storage
 - **Vector Database**: Qdrant integration for semantic search
 - **Comprehensive Secrets Management**: HashiCorp Vault integration with multiple auth methods
 - **Encrypted File Backend**: AES-256-GCM encryption with OS keychain integration

README.md

@@ -22,6 +22,7 @@ path = "examples/full_system.rs"
 name = "symbi-mcp"
 path = "src/bin/symbiont_mcp.rs"
 [dependencies]
+symbi-dsl = { path = "../dsl" }
 tokio = { version = "1.0", features = ["full"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
@@ -45,9 +46,9 @@ bytes = { version = "1.0", features = ["serde"] }
 tempfile = "3.0"
 qdrant-client = "1.14.0"
 flate2 = "1.0"
-candle-core = { version = "0.3", optional = true }
-candle-nn = { version = "0.3", optional = true }
-candle-transformers = { version = "0.3", optional = true }
+candle-core = { version = "0.9.1", optional = true }
+candle-nn = { version = "0.9.1", optional = true }
+candle-transformers = { version = "0.9.1", optional = true }
 tokenizers = { version = "0.15", optional = true }
 hf-hub = { version = "0.3", optional = true }
 regex = "1.0"
@@ -64,6 +65,7 @@ sha2 = "0.10"
 hex = "0.4"
 argon2 = "0.5"
 vaultrs = "0.7"
+sysinfo = "0.30"
 # OS keychain access dependencies
 keyring = { version = "2.0", optional = true }
 security-framework = { version = "2.9", optional = true }
@@ -75,6 +77,7 @@ axum = { version = "0.7", optional = true }
 tower = { version = "0.4", optional = true }
 tower-http = { version = "0.5", features = ["cors", "trace"], optional = true }
 tokio-tungstenite = { version = "0.21", optional = true }
+governor = { version = "0.10", optional = true }
 
 # Enterprise features removed for OSS build
 
@@ -87,7 +90,7 @@ criterion = "0.5"
 default = ["vector-db", "keychain"]
 vector-db = []
 embedding-models = ["candle-core", "candle-nn", "candle-transformers", "tokenizers", "hf-hub"]
-http-api = ["axum", "tower", "tower-http", "tokio-tungstenite"]
+http-api = ["axum", "tower", "tower-http", "tokio-tungstenite", "governor"]
 http-input = ["axum", "tower", "tower-http"]
 keychain = ["keyring", "security-framework", "secret-service", "winapi"]
 enterprise = []  # Enterprise feature for conditional compilation

crates/runtime/Cargo.toml

@@ -37,7 +37,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         ..Default::default()
     };
 
-    let context_manager = Arc::new(StandardContextManager::new(config));
+    let context_manager = Arc::new(StandardContextManager::new(config, "system").await?);
     context_manager.initialize().await?;
     println!("✓ Context manager initialized");
 

crates/runtime/examples/context_example.rs

@@ -23,7 +23,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("\n=== Initializing RAG Engine ===");
 
     let context_manager_config = ContextManagerConfig::default();
-    let context_manager = Arc::new(StandardContextManager::new(context_manager_config));
+    let context_manager = Arc::new(StandardContextManager::new(context_manager_config, "system").await?);
     let rag_engine = StandardRAGEngine::new(context_manager);
 
     // Configure the RAG engine

crates/runtime/examples/rag_example.rs

@@ -6,6 +6,19 @@
 #[cfg(feature = "http-api")]
 use axum::{extract::Request, http::StatusCode, middleware::Next, response::Response};
 
+#[cfg(feature = "http-api")]
+use governor::{
+    clock::DefaultClock,
+    state::{InMemoryState, NotKeyed},
+    Quota, RateLimiter,
+};
+
+#[cfg(feature = "http-api")]
+use std::{net::IpAddr, num::NonZeroU32, sync::{Arc, OnceLock}};
+
+#[cfg(feature = "http-api")]
+use dashmap::DashMap;
+
 /// Authentication middleware for bearer token validation
 #[cfg(feature = "http-api")]
 pub async fn auth_middleware(request: Request, next: Next) -> Result<Response, StatusCode> {
@@ -40,38 +53,193 @@ pub async fn auth_middleware(request: Request, next: Next) -> Result<Response, S
     Ok(next.run(request).await)
 }
 
-/// Rate limiting middleware (placeholder)
+/// Global rate limiter store for per-IP rate limiting
+#[cfg(feature = "http-api")]
+static RATE_LIMITERS: OnceLock<DashMap<IpAddr, Arc<RateLimiter<NotKeyed, InMemoryState, DefaultClock>>>> = OnceLock::new();
+
+/// Get or create a rate limiter for a specific IP address
+#[cfg(feature = "http-api")]
+fn get_rate_limiter_for_ip(ip: IpAddr) -> Arc<RateLimiter<NotKeyed, InMemoryState, DefaultClock>> {
+    let limiters = RATE_LIMITERS.get_or_init(DashMap::new);
+    
+    // Check if limiter exists, if not create one
+    if let Some(limiter) = limiters.get(&ip) {
+        Arc::clone(&limiter)
+    } else {
+        // Create a rate limiter: 100 requests per minute (roughly 1.67 requests per second)
+        let quota = Quota::per_minute(NonZeroU32::new(100).unwrap());
+        let limiter = Arc::new(RateLimiter::direct(quota));
+        limiters.insert(ip, Arc::clone(&limiter));
+        limiter
+    }
+}
+
+/// Extract client IP address from request
+#[cfg(feature = "http-api")]
+fn extract_client_ip(request: &Request) -> IpAddr {
+    // Try to get real IP from X-Forwarded-For header first (for proxy setups)
+    if let Some(forwarded_for) = request.headers().get("x-forwarded-for") {
+        if let Ok(forwarded_str) = forwarded_for.to_str() {
+            // X-Forwarded-For can contain multiple IPs, take the first one
+            if let Some(first_ip) = forwarded_str.split(',').next() {
+                if let Ok(ip) = first_ip.trim().parse::<IpAddr>() {
+                    return ip;
+                }
+            }
+        }
+    }
+    
+    // Try X-Real-IP header
+    if let Some(real_ip) = request.headers().get("x-real-ip") {
+        if let Ok(real_ip_str) = real_ip.to_str() {
+            if let Ok(ip) = real_ip_str.parse::<IpAddr>() {
+                return ip;
+            }
+        }
+    }
+    
+    // Fallback to connection info or default
+    // In a real setup, you'd extract this from the connection info
+    // For now, we'll use a default IP as fallback
+    "127.0.0.1".parse().unwrap()
+}
+
+/// Rate limiting middleware using token bucket algorithm
+///
+/// This middleware implements per-IP rate limiting with a token bucket algorithm.
+/// Each IP address gets 100 requests per minute (approximately 1.67 RPS).
+///
+/// Rate limiters are stored in a global concurrent HashMap and are created
+/// on-demand for each unique IP address.
 #[cfg(feature = "http-api")]
 pub async fn rate_limit_middleware(request: Request, next: Next) -> Result<Response, StatusCode> {
-    // TODO: Implement rate limiting logic
-    // For now, just pass through all requests
-    Ok(next.run(request).await)
+    // Extract client IP address
+    let client_ip = extract_client_ip(&request);
+    
+    // Get the rate limiter for this IP
+    let rate_limiter = get_rate_limiter_for_ip(client_ip);
+    
+    // Check if the request is allowed
+    match rate_limiter.check() {
+        Ok(_) => {
+            // Request is allowed, proceed
+            Ok(next.run(request).await)
+        }
+        Err(_) => {
+            // Rate limit exceeded
+            tracing::warn!("Rate limit exceeded for IP: {}", client_ip);
+            Err(StatusCode::TOO_MANY_REQUESTS)
+        }
+    }
 }
 
-/// Request logging middleware (placeholder)
+/// Enhanced request logging middleware with structured logging
+///
+/// Logs comprehensive request details including:
+/// - HTTP method and URI
+/// - Response status code and processing latency
+/// - Client IP address and response body size
+/// - Uses structured logging with tracing spans for request grouping
 #[cfg(feature = "http-api")]
 pub async fn logging_middleware(request: Request, next: Next) -> Result<Response, StatusCode> {
-    // TODO: Implement request logging
-    // For now, just pass through all requests
+    use std::time::Instant;
+    
+    // Extract request details
     let method = request.method().clone();
     let uri = request.uri().clone();
-
-    tracing::debug!("Incoming request: {} {}", method, uri);
-
+    let client_ip = extract_client_ip(&request);
+    
+    // Create a structured span for this request
+    let span = tracing::info_span!(
+        "http_request",
+        method = %method,
+        uri = %uri,
+        client_ip = %client_ip,
+        status_code = tracing::field::Empty,
+        latency_ms = tracing::field::Empty,
+        response_size = tracing::field::Empty,
+    );
+    
+    let _guard = span.enter();
+    
+    // Record start time for latency calculation
+    let start_time = Instant::now();
+    
+    tracing::info!("Processing request");
+    
+    // Process the request
     let response = next.run(request).await;
-
-    tracing::debug!("Response status: {}", response.status());
-
+    
+    // Calculate latency
+    let latency = start_time.elapsed();
+    let latency_ms = latency.as_millis() as u64;
+    
+    // Extract response details
+    let status_code = response.status();
+    
+    // Try to extract response body size from Content-Length header
+    let response_size = response
+        .headers()
+        .get("content-length")
+        .and_then(|h| h.to_str().ok())
+        .and_then(|s| s.parse::<u64>().ok())
+        .unwrap_or(0);
+    
+    // Record additional fields in the span
+    span.record("status_code", status_code.as_u16());
+    span.record("latency_ms", latency_ms);
+    span.record("response_size", response_size);
+    
+    // Log completion with all details
+    tracing::info!(
+        status_code = status_code.as_u16(),
+        latency_ms = latency_ms,
+        response_size = response_size,
+        "Request completed"
+    );
+    
     Ok(response)
 }
 
-/// Security headers middleware (placeholder)
+/// Security headers middleware
+///
+/// Adds essential security headers to all HTTP responses:
+/// - Strict-Transport-Security: Enforces HTTPS connections
+/// - X-Content-Type-Options: Prevents MIME type sniffing
+/// - X-Frame-Options: Prevents clickjacking attacks
+/// - Content-Security-Policy: Restricts resource loading
 #[cfg(feature = "http-api")]
 pub async fn security_headers_middleware(
     request: Request,
     next: Next,
 ) -> Result<Response, StatusCode> {
-    // TODO: Add security headers
-    // For now, just pass through all requests
-    Ok(next.run(request).await)
+    use axum::http::HeaderValue;
+    
+    // Process the request
+    let mut response = next.run(request).await;
+    
+    // Add security headers to the response
+    let headers = response.headers_mut();
+    
+    headers.insert(
+        "strict-transport-security",
+        HeaderValue::from_static("max-age=63072000; includeSubDomains; preload")
+    );
+    
+    headers.insert(
+        "x-content-type-options",
+        HeaderValue::from_static("nosniff")
+    );
+    
+    headers.insert(
+        "x-frame-options",
+        HeaderValue::from_static("DENY")
+    );
+    
+    headers.insert(
+        "content-security-policy",
+        HeaderValue::from_static("default-src 'self'; frame-ancestors 'none'")
+    );
+    
+    Ok(response)
 }

crates/runtime/src/api/middleware.rs

@@ -44,11 +44,9 @@ pub async fn execute_workflow(
 #[cfg(feature = "http-api")]
 pub async fn get_agent_status(
     State(provider): State<Arc<dyn RuntimeApiProvider>>,
-    Path(_agent_id): Path<String>,
+    Path(agent_id): Path<AgentId>,
 ) -> Result<Json<AgentStatusResponse>, (StatusCode, Json<ErrorResponse>)> {
-    let _agent_id = AgentId::new(); // TODO: Parse agent_id from string parameter
-
-    match provider.get_agent_status(_agent_id).await {
+    match provider.get_agent_status(agent_id).await {
         Ok(status) => Ok(Json(status)),
         Err(e) => Err((
             StatusCode::NOT_FOUND,
@@ -123,7 +121,7 @@ pub async fn create_agent(
 #[cfg(feature = "http-api")]
 pub async fn update_agent(
     State(provider): State<Arc<dyn RuntimeApiProvider>>,
-    Path(agent_id): Path<String>,
+    Path(agent_id): Path<AgentId>,
     Json(request): Json<UpdateAgentRequest>,
 ) -> Result<Json<UpdateAgentResponse>, (StatusCode, Json<ErrorResponse>)> {
     match provider.update_agent(agent_id, request).await {
@@ -143,7 +141,7 @@ pub async fn update_agent(
 #[cfg(feature = "http-api")]
 pub async fn delete_agent(
     State(provider): State<Arc<dyn RuntimeApiProvider>>,
-    Path(agent_id): Path<String>,
+    Path(agent_id): Path<AgentId>,
 ) -> Result<Json<DeleteAgentResponse>, (StatusCode, Json<ErrorResponse>)> {
     match provider.delete_agent(agent_id).await {
         Ok(response) => Ok(Json(response)),
@@ -162,7 +160,7 @@ pub async fn delete_agent(
 #[cfg(feature = "http-api")]
 pub async fn execute_agent(
     State(provider): State<Arc<dyn RuntimeApiProvider>>,
-    Path(agent_id): Path<String>,
+    Path(agent_id): Path<AgentId>,
     Json(request): Json<ExecuteAgentRequest>,
 ) -> Result<Json<ExecuteAgentResponse>, (StatusCode, Json<ErrorResponse>)> {
     match provider.execute_agent(agent_id, request).await {
@@ -182,7 +180,7 @@ pub async fn execute_agent(
 #[cfg(feature = "http-api")]
 pub async fn get_agent_history(
     State(provider): State<Arc<dyn RuntimeApiProvider>>,
-    Path(agent_id): Path<String>,
+    Path(agent_id): Path<AgentId>,
 ) -> Result<Json<GetAgentHistoryResponse>, (StatusCode, Json<ErrorResponse>)> {
     match provider.get_agent_history(agent_id).await {
         Ok(response) => Ok(Json(response)),