Refactor: Enforce ExecutionRequest contract and implement safe sandbox execution (ENH-001/Phase 9)

Author: nadeem4

audit/remediation_plan.md

@@ -61,7 +61,7 @@ This document serves as the master backlog for addressing findings from the Arch
 
 ## 🚀 Enhancements (Architecture Upgrade)
 
-- [ ] **ENH-001: Sandboxed Execution Service** (P0 - Critical)
+- [x] **ENH-001: Sandboxed Execution Service** (P0 - Critical)
   - **Value**: Prevents driver crashes from taking down the Agent and mitigates RCE risks.
   - **Action**: Move `ExecutorNode` logic to a separate gRPC service/sidecar with limited privileges.
 

docs/architecture/decisions/ADR-002_circuit_breakers.md

@@ -0,0 +1,126 @@
+# ADR-002: Circuit Breaker Pattern (ENH-002)
+
+## 1. Problem Statement
+
+The NL2SQL Agent relies on multiple unreliable downstream dependencies:
+
+1. **LLM Providers** (OpenAI/Anthropic): Subject to rate limits, latency spikes, and 503 outages.
+2. **Vector Stores** (Chroma/Weaviate): Network I/O, potential lock contention.
+3. **SQL Databases**: Network partitions, connection pool exhaustion, "Hard Down" instances.
+
+### The Risks
+
+* **Cascading Failure**: If the LLM is down, 100 concurrent users will launch 100 requests. If we just retry, we amplify the load (Retry Storms).
+* **Resource Exhaustion**: Threads hanging on `socket.recv` block the entire web server, even for health checks.
+* **Poor UX**: Users wait 60s+ for a timeout instead of getting an immediate "Service Unavailable" message.
+* **Degraded Functionality**: Without isolation, a DB failure kills the entire agent, even if the user only wanted to chat or check schema.
+
+## 2. Options Analysis
+
+### Option A: Retries Only (Current State)
+
+We currently implement Exponential Backoff.
+
+* **Pros**: Simple. Handles transient blips.
+* **Cons**: catastrophic during full outages. If OpenAI is down for 1 hour, every request still tries 3-5 times, wasting CPU and I/O.
+
+### Option B: In-Process Circuit Breaker (`pybreaker`) - **Recommended**
+
+Use a Python library to track failure rates in memory.
+
+* **Mechanism**:
+  * **Closed**: Normal operation.
+  * **Open**: After `fail_max` errors, fail immediately (raise `CircuitBreakerError`). Avoid downstream calls.
+  * **Half-Open**: After `reset_timeout`, let one request through to test the waters.
+* **Pros**: lightweight, easy to integrate via decorators, no extra infrastructure.
+* **Cons**: State is per-process (not distributed). Each pod must rediscover the outage independently.
+
+### Option C: Service Mesh (Istio / Linkerd)
+
+Offload resilience to the network layer (Envoy sidecars).
+
+* **Pros**: Distributed, language-agnostic, central control.
+* **Cons**: Massive operational overkill for the current scale. Depends on k8s deployment.
+
+## 3. Decision
+
+We will implement **Option B** using the `pybreaker` library.
+
+### Rationale
+
+* **Complexity**: Low. It requires only a decorator on service methods.
+* **Granularity**: We can define different breakers for different failure domains (e.g., `LLM_BREAKER`, `DB_BREAKER`).
+* **Dependencies**: Adds one small pure-Python dependency.
+
+## 4. Implementation Details
+
+### Thresholds & Classification
+
+**LLM Breaker (`LLM_BREAKER`)**:
+
+* `fail_max = 5`: Open after 5 consecutive failures.
+* `reset_timeout = 60s`: Wait 1 minute.
+* **Failure Classification**:
+  * **Failures**: Network Errors, Timed Out, 5xx (Server Error).
+  * **Ignored**: 429 (Rate Limit), 400 (Bad Request), 401 (Auth). *Rationale: Rate limits are soft failures and should trigger backoff, not circuit breaking.*
+
+**DB/Vector Breaker (`DB_BREAKER`, `VECTOR_BREAKER`)**:
+
+* `fail_max = 5`
+* `reset_timeout = 30s` (Faster recovery for infra blips).
+* **Failure Classification**: Any unhandled exception or connection error.
+
+### State Transitions (Critical)
+
+* **Closed -> Open**: Triggered by `fail_max` consecutive errors.
+* **Open -> Half-Open**: After `reset_timeout` passes.
+* **Half-Open -> Closed**: If the *single probe request* succeeds.
+* **Half-Open -> Open**: If the *single probe request* fails.
+* **Invariant**: The transition to Half-Open MUST be protected by a lock to prevent a Thundering Herd (only ONE request acts as the probe).
+
+### Tiered Resilience (Degradation Strategy)
+
+We will implement "Graceful Degradation" rather than "Hard Stop" when a breaker opens:
+
+| Breaker Open | Impact | Fallback Behavior |
+| :--- | :--- | :--- |
+| **DB** | Cannot execute SQL | **Accept Query**: Validate intent via LLM. <br> **Response**: "I understood your query, but the database is currently unreachable. Here is the SQL I would have run: ..." |
+| **Vector** | Cannot route/search | **Static Route**: Fallback to direct text search (if available) or warn user. <br> **LLM**: Proceed with "Best Effort" routing without examples. |
+| **LLM** | Cannot understand | **Critical Failure**: Return "Service Unavailable. Please try again later." |
+
+### Observability
+
+Circuit breaker state changes must be exported as metrics for alerting (SLO dashboards):
+
+* `breaker_open_total{type="llm|db|vector"}`
+* `breaker_half_open_total{type="llm|db|vector"}`
+* `breaker_closed_total{type="llm|db|vector"}`
+* `breaker_failure_total{type="...", error="timeout|503"}`
+* `breaker_ignored_failure_total{type="...", error="429|400"}`
+
+## 5. Integration with Sandbox
+
+The Circuit Breaker sits **upstream** of the Sandbox:
+`Agent -> Circuit Breaker -> Sandbox -> SQL Driver`
+
+This ensures that:
+
+1. **Crash Isolation**: Sandbox handles Segfaults.
+2. **Failure Isolation**: Breaker handles Timeouts/Network Partitions.
+
+### Scope
+
+We will wrap the following critical paths:
+
+1. `LLMService.invoke()`: Protects against AI provider outages.
+2. `VectorStore.search()`: Protects against Retrieval outages.
+3. `Adapter.connect()`: Protects against SQL DB outages.
+
+## 5. Consequences
+
+* **Positive**:
+  * **Fail Fast**: Usage during outages will return <10ms errors instead of 30s timeouts.
+  * **Reduced Load**: Downstream services get breathing room to recover.
+  * **Negative**:
+  * **Debugging**: Developers must distinguish between "Service Down" and "Circuit Open" errors.
+  * **Local State**: Process restarts reset the breaker state.

packages/core/src/nl2sql/common/contracts.py

@@ -0,0 +1,45 @@
+"""
+Contract definitions for the Execution Service Interface.
+
+This module defines the Pydantic models used to communicate between the 
+Control Plane (Nodes) and the Data Plane (Execution Sandbox).
+"""
+from typing import Dict, Any, Optional, Literal, List
+from pydantic import BaseModel, Field, ConfigDict
+
+class ExecutionRequest(BaseModel):
+    """Payload for requesting an operation from the properties sandbox."""
+    
+    mode: Literal["execute", "dry_run", "cost_estimate", "fetch_schema"] = Field(
+        ..., description="The type of operation to perform."
+    )
+    datasource_id: str = Field(..., description="The unique ID of the target datasource.")
+    engine_type: str = Field(..., description="The SQL engine type (e.g. postgres, sqlite).")
+    connection_args: Dict[str, Any] = Field(
+        ..., description="Connection arguments (host, user, password, etc.)."
+    )
+    sql: Optional[str] = Field(None, description="The SQL query to execute (required for execute/dry_run).")
+    parameters: Dict[str, Any] = Field(
+        default_factory=dict, description="Query parameters for parameterized execution."
+    )
+    limits: Dict[str, int] = Field(
+        default_factory=dict, 
+        description="Execution limits (e.g., {'row_limit': 1000, 'timeout_ms': 5000})."
+    )
+
+    model_config = ConfigDict(extra="ignore")
+
+
+class ExecutionResult(BaseModel):
+    """Standardized response from the execution sandbox."""
+    
+    success: bool = Field(..., description="Whether the operation completed successfully.")
+    data: Optional[Any] = Field(
+        None, description="The result payload (Rows, CostEstimate, Schema, etc.)."
+    )
+    error: Optional[str] = Field(None, description="Error message if failed.")
+    metrics: Dict[str, float] = Field(
+        default_factory=dict, description="Performance metrics (e.g., execution_time_ms)."
+    )
+
+    model_config = ConfigDict(extra="ignore")

packages/core/src/nl2sql/common/sandbox.py

@@ -90,3 +90,67 @@ def get_indexing_pool() -> ProcessPoolExecutor:
         ProcessPoolExecutor: The shared indexing pool instance.
     """
     return SandboxManager.get_indexing_pool()
+
+from typing import Callable, Any, Dict
+from nl2sql.common.contracts import ExecutionRequest, ExecutionResult
+
+def execute_in_sandbox(
+    pool: ProcessPoolExecutor,
+    func: Callable[[ExecutionRequest], ExecutionResult],
+    request: ExecutionRequest,
+    timeout_sec: int = 30
+) -> ExecutionResult:
+    """Executes a function in the sandbox with centralized error handling.
+
+    This helper centralizes the logic for handling:
+    1. BrokenProcessPool (Worker Crash/Segfault/OOM)
+    2. TimeoutError (Job hung)
+    3. Generic Exception (Serialization error or other infra failure)
+
+    Args:
+        pool (ProcessPoolExecutor): The execution pool to usage.
+        func (Callable): The function to execute (must accept ExecutionRequest and return ExecutionResult).
+        request (ExecutionRequest): The contract payload.
+        timeout_sec (int): Maximum time to wait for the result.
+
+    Returns:
+        ExecutionResult: The result, safely wrapping any infrastructure errors.
+    """
+    try:
+        future = pool.submit(func, request)
+        return future.result(timeout=timeout_sec)
+    
+    except TimeoutError:
+        logger.error(f"Sandbox Timeout ({timeout_sec}s) for {func.__name__}")
+        return ExecutionResult(
+            success=False,
+            error=f"Operation timed out after {timeout_sec} seconds. The worker process may be hung.",
+            metrics={"execution_time_ms": timeout_sec * 1000}
+        )
+        
+    except Exception as e:
+        msg = str(e)
+        logger.error(f"Sandbox Exception: {msg}")
+        
+        is_crash = False
+        try:
+            from concurrent.futures.process import BrokenProcessPool
+            if isinstance(e, BrokenProcessPool):
+                is_crash = True
+        except ImportError:
+            pass
+            
+        if "BrokenProcessPool" in msg or "Terminated" in msg:
+            is_crash = True
+            
+        if is_crash:
+            return ExecutionResult(
+                success=False,
+                error=f"SANDBOX CRASH: The worker process terminated abruptly (Segfault/OOM). ({msg})",
+                metrics={"is_crash": 1.0}
+            )
+            
+        return ExecutionResult(
+            success=False,
+            error=f"Sandbox Execution Failed: {msg}"
+        )