Sandbox Pattern: Safe Isolated Execution Environments
In intelligent decision platforms, a faulty rule deployment can have catastrophic consequences — a wrong risk control rule might block all legitimate transactions, a flawed reasoning model might produce absurd decision recommendations. The traditional "develop, test, deploy" workflow is too simplistic:
CoomiaPublished on December 20, 202510 min read
Share this articleTwitter / X
Sandbox Pattern: Safe Isolated Execution Environments
“Series: S10 Design Patterns · Article 6 | Level: Advanced | Reading Time: 18 min
#TL;DR
- The Sandbox Pattern provides safe, isolated runtime environments for code execution, rule testing, and policy validation, ensuring experimental operations cannot impact production systems.
- coomia-dip implements sandboxing at multiple levels: World-level logical isolation, Nessie Branch-level data isolation, container-level compute isolation, and Agent-level behavioral isolation.
- Sandboxes support snapshots, replay, comparison, and promotion, making the "development to testing to production" workflow safe and controlled.
#Introduction: Why Sandboxes Are Needed
In intelligent decision platforms, a faulty rule deployment can have catastrophic consequences — a wrong risk control rule might block all legitimate transactions, a flawed reasoning model might produce absurd decision recommendations. The traditional "develop, test, deploy" workflow is too simplistic:
Code
Problem 1: Test environment data differs too much from production — test results are unreliable
Problem 2: Rule impact scope is hard to assess — changing one rule may affect tens of thousands of decisions
Problem 3: Agent behavior works fine in controlled environments but may produce surprises in complex scenarios
Problem 4: Multiple teams modifying rules simultaneously interfere with each other
The Sandbox Pattern solves these problems: run experimental operations in an environment consistent with but completely isolated from production.
#Part 1: coomia-dip Sandbox Architecture
#1.1 Multi-Layer Sandbox System
The coomia-dip sandbox is not a simple "test environment" but a multi-layer isolation system:
Python
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from typing import Any
class SandboxLevel(Enum):
"""Sandbox isolation levels."""
WORLD = "world" # World level: logical isolation
BRANCH = "branch" # Nessie Branch level: data isolation
CONTAINER = "container" # Container level: compute isolation
FULL = "full" # Full isolation: independent infrastructure
@dataclass
class SandboxConfig:
"""Configuration for creating a sandbox."""
name: str
level: SandboxLevel
source_world_id: str
owner_id: str
tenant_id: str
expires_at: datetime | None = None
resource_limits: dict[str, Any] = field(default_factory=lambda: {
"max_cpu": "2",
"max_memory": "4Gi",
"max_storage": "10Gi",
"max_duration_hours": 24,
})
data_snapshot: str = "latest"
include_patterns: list[str] = field(default_factory=list)
exclude_patterns: list[str] = field(default_factory=list)
@dataclass
class Sandbox:
"""A sandbox instance."""
sandbox_id: str
config: SandboxConfig
status: str = "creating"
world_id: str = ""
branch_name: str = ""
created_at: datetime = field(default_factory=datetime.utcnow)
metadata: dict[str, Any] = field(default_factory=dict)
#1.2 Sandbox Lifecycle Management
Python
class SandboxManager:
"""Manage sandbox lifecycle."""
async def create(self, config: SandboxConfig) -> Sandbox:
"""Create a new sandbox with data snapshot."""
sandbox = Sandbox(sandbox_id=generate_id(), config=config)
# Step 1: Create isolated World
world = await self._world_service.create_world(
name=f"sandbox-{sandbox.sandbox_id}",
parent_world_id=config.source_world_id,
isolation_level=config.level.value,
)
sandbox.world_id = world.world_id
# Step 2: Create Nessie Branch (data isolation)
branch = await self._nessie_client.create_branch(
branch_name=f"sandbox/{sandbox.sandbox_id}",
source_ref=config.data_snapshot,
)
sandbox.branch_name = branch.name
# Step 3: Snapshot data (as needed)
if config.include_patterns:
await self._snapshot_data(
source_world=config.source_world_id,
target_branch=branch.name,
include=config.include_patterns,
exclude=config.exclude_patterns,
)
# Step 4: Apply resource limits
await self._resource_manager.apply_limits(
sandbox.sandbox_id, config.resource_limits
)
# Step 5: Schedule auto-expiry
if config.expires_at:
await self._scheduler.schedule_cleanup(
sandbox.sandbox_id, config.expires_at
)
sandbox.status = "ready"
await self._store.save(sandbox)
return sandbox
async def destroy(self, sandbox_id: str) -> None:
"""Destroy a sandbox and clean up all resources."""
sandbox = await self._store.get(sandbox_id)
await self._nessie_client.delete_branch(sandbox.branch_name)
await self._world_service.delete_world(sandbox.world_id)
await self._resource_manager.release(sandbox_id)
sandbox.status = "destroyed"
await self._store.save(sandbox)
async def promote(
self, sandbox_id: str, target_world_id: str, items: list[str] | None = None
) -> dict:
"""Promote sandbox changes to a target environment."""
sandbox = await self._store.get(sandbox_id)
diff = await self._diff_service.compare(
sandbox.branch_name, target_world_id, items=items
)
result = await self._merge_service.merge(
source_branch=sandbox.branch_name,
target_world=target_world_id,
changes=diff.changes,
)
return {
"promoted_items": len(diff.changes),
"conflicts": diff.conflicts,
"merge_result": result.status,
}
#Part 2: Data Isolation with Nessie Git-Like Branching
#2.1 Data Snapshots and Branching
coomia-dip leverages Nessie's Git-like version control to provide data isolation for sandboxes. Each sandbox is like a Git branch — data can be modified independently without affecting the main branch:
Python
class NessieSandboxProvider:
"""Provide data isolation using Nessie branches."""
async def create_data_sandbox(
self,
sandbox_id: str,
source_ref: str = "main",
snapshot_timestamp: datetime | None = None,
) -> str:
"""Create a Nessie branch for sandbox data isolation."""
branch_name = f"sandbox/{sandbox_id}"
if snapshot_timestamp:
commit_hash = await self._nessie.get_commit_at(
source_ref, snapshot_timestamp
)
await self._nessie.create_branch(branch_name, commit_hash)
else:
await self._nessie.create_branch(branch_name, source_ref)
return branch_name
async def get_sandbox_diff(
self, sandbox_branch: str, target_ref: str = "main"
) -> list[dict]:
"""Get differences between sandbox and target reference."""
return await self._nessie.diff(sandbox_branch, target_ref)
async def merge_to_target(
self, sandbox_branch: str, target_ref: str = "main"
) -> dict:
"""Merge sandbox changes to target (promote)."""
try:
result = await self._nessie.merge(
from_branch=sandbox_branch,
to_branch=target_ref,
merge_behavior="NORMAL",
)
return {"status": "merged", "commit": result.commit_hash}
except ConflictError as e:
return {"status": "conflict", "conflicts": e.conflicts}
#2.2 Selective Data Snapshots
Not all data needs to be copied to a sandbox. coomia-dip supports selective snapshots by ObjectType:
Python
class SelectiveSnapshot:
"""Create selective data snapshots for sandboxes."""
async def snapshot(
self,
source_world: str,
target_branch: str,
include_types: list[str],
sample_ratio: float = 1.0,
max_rows_per_type: int | None = None,
) -> dict:
"""Create a selective snapshot with optional sampling."""
stats = {}
for obj_type in include_types:
source_table = await self._catalog.get_table(source_world, obj_type)
if sample_ratio < 1.0 or max_rows_per_type:
rows = await self._sample_data(
source_table, sample_ratio, max_rows_per_type
)
await self._write_to_branch(target_branch, obj_type, rows)
stats[obj_type] = len(rows)
else:
await self._iceberg.create_branch(source_table, target_branch)
stats[obj_type] = "full_snapshot"
return stats
#Part 3: Rule Sandbox — Safe Decision Logic Testing
#3.1 Rule Impact Assessment
Before deploying a new rule to production, assess its impact in a sandbox:
Python
class RuleSandboxEvaluator:
"""Evaluate rule impact in a sandbox environment."""
async def evaluate_rule_impact(
self,
sandbox_id: str,
rule_definition: dict,
test_data_source: str = "production_snapshot",
) -> dict:
"""Evaluate the impact of a rule change using sandbox data."""
sandbox = await self._sandbox_manager.get(sandbox_id)
await self._rule_engine.deploy_in_sandbox(
sandbox.world_id, rule_definition
)
test_data = await self._get_test_data(
sandbox.branch_name, test_data_source
)
results = {
"total_records": len(test_data),
"affected_records": 0,
"decisions": {},
"comparison_with_current": {},
}
current_results = []
new_results = []
for record in test_data:
current = await self._rule_engine.evaluate(
sandbox.config.source_world_id, record
)
current_results.append(current)
new = await self._rule_engine.evaluate(sandbox.world_id, record)
new_results.append(new)
if current.decision != new.decision:
results["affected_records"] += 1
results["impact_rate"] = (
results["affected_records"] / results["total_records"]
if results["total_records"] > 0 else 0
)
results["decision_distribution"] = self._calculate_distribution(new_results)
results["changed_decisions"] = self._find_changes(current_results, new_results)
return results
def _calculate_distribution(self, results: list) -> dict:
distribution: dict[str, int] = {}
for r in results:
distribution[r.decision] = distribution.get(r.decision, 0) + 1
return distribution
def _find_changes(self, current: list, new: list) -> list:
changes = []
for c, n in zip(current, new):
if c.decision != n.decision:
changes.append({
"record_id": c.record_id,
"old_decision": c.decision,
"new_decision": n.decision,
"old_score": c.score,
"new_score": n.score,
})
return changes
#3.2 A/B Test Sandboxes
coomia-dip supports creating multiple sandboxes for A/B testing, comparing the effects of different rule versions:
Python
class ABTestSandbox:
"""Create A/B test sandboxes for comparing rule versions."""
async def create_ab_test(
self,
base_world_id: str,
variants: list[dict],
test_data_config: dict,
) -> dict:
"""Create sandboxes for A/B testing rule variants."""
sandboxes = []
for variant in variants:
sandbox = await self._sandbox_manager.create(
SandboxConfig(
name=f"ab-test-{variant['name']}",
level=SandboxLevel.BRANCH,
source_world_id=base_world_id,
owner_id=variant.get("owner", "system"),
tenant_id=variant["tenant_id"],
)
)
await self._rule_engine.deploy_in_sandbox(
sandbox.world_id, variant["rules"]
)
sandboxes.append({
"variant": variant["name"],
"sandbox_id": sandbox.sandbox_id,
})
results = {}
for sb in sandboxes:
result = await self._evaluator.evaluate_rule_impact(
sb["sandbox_id"], {}, test_data_config
)
results[sb["variant"]] = result
return {
"sandboxes": sandboxes,
"comparison": self._compare_results(results),
}
def _compare_results(self, results: dict) -> dict:
comparison = {}
for variant, result in results.items():
comparison[variant] = {
"impact_rate": result["impact_rate"],
"decision_distribution": result["decision_distribution"],
}
return comparison
#Part 4: Agent Sandbox — Behavioral Isolation and Debugging
#4.1 Agent Behavioral Sandbox
coomia-dip Agents may execute complex multi-step operations. Running Agents in sandboxes allows safe observation and debugging:
Python
class AgentSandbox:
"""Sandbox for safe Agent execution and debugging."""
async def run_agent_in_sandbox(
self,
sandbox_id: str,
agent_config: dict,
input_data: dict,
step_mode: bool = False,
) -> dict:
"""Run an Agent in sandbox with optional step-by-step mode."""
sandbox = await self._sandbox_manager.get(sandbox_id)
agent = await self._agent_factory.create(
agent_config, world_id=sandbox.world_id, sandboxed=True
)
agent.add_interceptor(SandboxInterceptor(
allowed_actions=agent_config.get("allowed_actions", []),
blocked_actions=agent_config.get("blocked_actions", []),
max_steps=agent_config.get("max_steps", 100),
max_cost=agent_config.get("max_cost", 10.0),
))
if step_mode:
return await self._run_step_by_step(agent, input_data)
else:
return await self._run_normal(agent, input_data)
async def _run_step_by_step(self, agent, input_data: dict) -> dict:
"""Execute agent step by step for debugging."""
steps = []
async for step in agent.execute_streaming(input_data):
steps.append({
"step_number": len(steps) + 1,
"action": step.action,
"input": step.input_data,
"output": step.output_data,
"reasoning": step.reasoning,
"timestamp": step.timestamp.isoformat(),
})
return {"steps": steps, "total_steps": len(steps)}
class SandboxInterceptor:
"""Intercept and control Agent actions in sandbox."""
def __init__(
self,
allowed_actions: list[str],
blocked_actions: list[str],
max_steps: int,
max_cost: float,
):
self._allowed = set(allowed_actions)
self._blocked = set(blocked_actions)
self._max_steps = max_steps
self._max_cost = max_cost
self._step_count = 0
self._total_cost = 0.0
async def before_action(self, action: str, data: dict) -> bool:
self._step_count += 1
if self._step_count > self._max_steps:
raise SandboxLimitExceeded(f"Max steps ({self._max_steps}) exceeded")
if action in self._blocked:
return False
if self._allowed and action not in self._allowed:
return False
return True
async def after_action(self, action: str, result: dict, cost: float) -> None:
self._total_cost += cost
if self._total_cost > self._max_cost:
raise SandboxLimitExceeded(
f"Max cost ({self._max_cost}) exceeded: {self._total_cost}"
)
#Part 5: Sandbox Comparison and Promotion
#5.1 Diff Comparison
Before promoting sandbox changes to production, precisely understand what changed:
Python
class SandboxDiffService:
"""Compare sandbox state with production."""
async def diff(self, sandbox_id: str, target_world_id: str) -> dict:
"""Generate comprehensive diff between sandbox and target."""
sandbox = await self._sandbox_manager.get(sandbox_id)
schema_diff = await self._compare_schemas(
sandbox.world_id, target_world_id
)
data_diff = await self._compare_data(
sandbox.branch_name, target_world_id
)
rule_diff = await self._compare_rules(
sandbox.world_id, target_world_id
)
return {
"schema_changes": schema_diff,
"data_changes": data_diff,
"rule_changes": rule_diff,
"total_changes": len(schema_diff) + len(data_diff) + len(rule_diff),
}
#5.2 Safe Promotion Pipeline
Python
class SandboxPromotionPipeline:
"""Safe promotion pipeline from sandbox to production."""
async def promote(
self,
sandbox_id: str,
target_world_id: str,
approval_id: str | None = None,
) -> dict:
"""Execute the promotion pipeline."""
diff = await self._diff_service.diff(sandbox_id, target_world_id)
impact = await self._impact_analyzer.analyze(diff)
if impact["risk_level"] == "high" and not approval_id:
return {
"status": "approval_required",
"impact": impact,
"message": "High-risk changes require explicit approval",
}
compat = await self._compatibility_checker.check(diff, target_world_id)
if not compat["compatible"]:
return {"status": "incompatible", "issues": compat["issues"]}
result = await self._promotion_saga.execute(
sandbox_id, target_world_id, diff
)
return {
"status": result.status,
"promoted_changes": len(diff["schema_changes"]) + len(diff["rule_changes"]),
"saga_id": result.saga_id,
}
#Part 6: Production Practices
#6.1 Sandbox Resource Governance
Sandboxes consume cluster resources and require strict governance:
Python
class SandboxGovernance:
"""Governance policies for sandbox management."""
async def enforce_policies(self) -> list[dict]:
actions = []
expired = await self._store.find_expired()
for sb in expired:
await self._sandbox_manager.destroy(sb.sandbox_id)
actions.append({"action": "destroyed", "reason": "expired", "id": sb.sandbox_id})
idle = await self._store.find_idle(idle_threshold_hours=4)
for sb in idle:
await self._notify_owner(sb, "Your sandbox has been idle for 4+ hours")
actions.append({"action": "notified", "reason": "idle", "id": sb.sandbox_id})
oversize = await self._store.find_over_resource_limits()
for sb in oversize:
await self._sandbox_manager.suspend(sb.sandbox_id)
actions.append({"action": "suspended", "reason": "over_limits", "id": sb.sandbox_id})
return actions
#6.2 Sandbox Auditing
All sandbox operations require audit records:
Python
class SandboxAuditService:
"""Audit trail for sandbox operations."""
async def log_operation(
self, sandbox_id: str, operation: str, actor_id: str, details: dict
) -> None:
await self._event_store.append([
DomainEvent(
event_id=generate_id(),
event_type=f"sandbox.{operation}",
aggregate_id=sandbox_id,
aggregate_type="sandbox",
sequence_number=0,
timestamp=datetime.utcnow(),
payload=details,
metadata=EventMetadata(
actor_id=actor_id,
actor_type="user",
tenant_id=details.get("tenant_id", ""),
world_id=details.get("world_id", ""),
source_plane="platform",
trace_id=generate_trace_id(),
),
)
])
#Key Takeaways
- Multi-Layer Isolation: coomia-dip provides World, Branch, and Container sandbox isolation to meet different security requirements
- Git-Like Data Isolation: Nessie-based branching enables zero-copy data isolation — efficient and safe
- Impact Assessment: Evaluate rule change impact in sandboxes before production deployment to prevent incidents
- Agent Safety: Agents run in sandboxes with behavioral interception and resource limits for safety
- Safe Promotion: The sandbox-to-production promotion workflow includes diff comparison, compatibility checks, and approval processes
- Resource Governance: Strict quota, expiry, and idle detection policies prevent sandbox resource waste
#Next Article
In the next article, we will explore the Federation Pattern — how coomia-dip enables cross-organization, cross-cluster Ontology federated queries and collaboration.
S10-07: Federation Pattern: Cross-Organization Ontology Collaboration
#Tags
#DesignPatterns #Sandbox #Isolation #Nessie #RuleTesting #AgentDebugging #ABTesting #SafePromotion