问题 1：测试环境的数据与生产环境差异太大，测试结果不可信
问题 2：规则的影响范围难以评估——修改一条规则可能影响上万条决策
问题 3：Agent 的行为在受控环境下正常，但在复杂环境下可能出现意外
问题 4：多个团队同时修改规则，彼此干扰

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 级别：逻辑隔离
    BRANCH = "branch"       # Nessie Branch 级别：数据隔离
    CONTAINER = "container" # 容器级别：计算隔离
    FULL = "full"           # 完全隔离：独立基础设施

@dataclass
class SandboxConfig:
    """Configuration for creating a sandbox."""
    name: str
    level: SandboxLevel
    source_world_id: str           # 源 World（用于数据快照）
    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"   # latest | specific_timestamp
    include_patterns: list[str] = field(default_factory=list)  # 包含哪些 ObjectType
    exclude_patterns: list[str] = field(default_factory=list)  # 排除哪些 ObjectType

@dataclass
class Sandbox:
    """A sandbox instance."""
    sandbox_id: str
    config: SandboxConfig
    status: str = "creating"       # creating | ready | running | stopped | expired
    world_id: str = ""             # 沙箱的 World ID
    branch_name: str = ""          # Nessie Branch 名称
    created_at: datetime = field(default_factory=datetime.utcnow)
    metadata: dict[str, Any] = field(default_factory=dict)

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: 创建隔离的 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: 创建 Nessie Branch（数据隔离）
        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: 快照数据（按需）
        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: 设置资源限制
        await self._resource_manager.apply_limits(
            sandbox.sandbox_id, config.resource_limits
        )

        # Step 5: 设置自动过期
        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)

        # 清理 Nessie Branch
        await self._nessie_client.delete_branch(sandbox.branch_name)

        # 清理 World
        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,
        }

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}

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:
            # 获取源数据的 Iceberg 表
            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:
                # 全量快照：利用 Iceberg 的零拷贝分支
                await self._iceberg.create_branch(
                    source_table, target_branch
                )
                stats[obj_type] = "full_snapshot"

        return stats

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:
        """Calculate decision distribution."""
        distribution: dict[str, int] = {}
        for r in results:
            decision = r.decision
            distribution[decision] = distribution.get(decision, 0) + 1
        return distribution

    def _find_changes(self, current: list, new: list) -> list:
        """Find records where decisions changed."""
        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

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:
        """Compare results across variants."""
        comparison = {}
        for variant, result in results.items():
            comparison[variant] = {
                "impact_rate": result["impact_rate"],
                "decision_distribution": result["decision_distribution"],
            }
        return comparison

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 实例（沙箱版本）
        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:
        """Check if action is allowed in sandbox."""
        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:
        """Track action cost."""
        self._total_cost += cost
        if self._total_cost > self._max_cost:
            raise SandboxLimitExceeded(
                f"Max cost ({self._max_cost}) exceeded: {self._total_cost}"
            )

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 差异
        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)
            ),
        }

    async def _compare_schemas(self, source: str, target: str) -> list:
        """Compare Ontology schemas between environments."""
        source_types = await self._ontology.list_object_types(source)
        target_types = await self._ontology.list_object_types(target)

        changes = []
        source_map = {t.name: t for t in source_types}
        target_map = {t.name: t for t in target_types}

        # 新增的类型
        for name in set(source_map) - set(target_map):
            changes.append({"type": "added", "object_type": name})

        # 删除的类型
        for name in set(target_map) - set(source_map):
            changes.append({"type": "removed", "object_type": name})

        # 修改的类型
        for name in set(source_map) & set(target_map):
            if source_map[name].schema != target_map[name].schema:
                changes.append({
                    "type": "modified",
                    "object_type": name,
                    "fields_added": [],
                    "fields_removed": [],
                    "fields_modified": [],
                })

        return changes

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."""
        # Step 1: 生成差异报告
        diff = await self._diff_service.diff(sandbox_id, target_world_id)

        # Step 2: 影响评估
        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",
            }

        # Step 3: 兼容性检查
        compat = await self._compatibility_checker.check(diff, target_world_id)
        if not compat["compatible"]:
            return {
                "status": "incompatible",
                "issues": compat["issues"],
            }

        # Step 4: 执行提升（通过 Saga 保证一致性）
        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,
        }

class SandboxGovernance:
    """Governance policies for sandbox management."""

    async def enforce_policies(self) -> list[dict]:
        """Enforce sandbox governance policies."""
        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

    async def get_quota_usage(self, tenant_id: str) -> dict:
        """Get sandbox quota usage for a tenant."""
        active = await self._store.count_active(tenant_id)
        limits = await self._quota_service.get_limits(tenant_id)
        return {
            "active_sandboxes": active,
            "max_sandboxes": limits["max_sandboxes"],
            "storage_used_gb": await self._resource_manager.get_storage_usage(tenant_id),
            "storage_limit_gb": limits["max_storage_gb"],
        }

class SandboxAuditService:
    """Audit trail for sandbox operations."""

    async def log_operation(
        self,
        sandbox_id: str,
        operation: str,
        actor_id: str,
        details: dict,
    ) -> None:
        """Log a sandbox operation for audit."""
        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(),
                ),
            )
        ])