aps-agent/server/aps_domain/explore_boundary.py

# ============================================================
# Explore 与 Runtime 数据边界 v1（moduleId: domain-explore-boundary, 可重生 ✅）
# plan.md §5.1 + §9.7 / 矩阵 55 行：统一通道上下文与存储隔离
#   - readonly_view(world)：主干世界只读视图，任何写操作抛 PermissionError（fail closed）
#   - run_explore(world, fn)：统一 Explore 通道——内部深拷贝沙盒，fn 只拿到沙盒，
#     主干永不外泄写引用；任何 Explore 工具无法取得主干写能力（强制而非约定）
# ============================================================
from __future__ import annotations

import copy
from collections.abc import Callable
from typing import Any, TypeVar

T = TypeVar("T")

# 嵌套写保护深度（足够覆盖 world 顶层 + 常用列表/字典表；防御性限制避免全递归开销）
_MAX_DEPTH = 32


class ReadOnlyWorld:
    """主干世界只读视图：读操作透传，写操作（set/del/嵌套变异）抛 PermissionError。

    用于把主干世界交给只读 Explore 工具；任何尝试写主干的行为立即失败关闭。
    """

    def __init__(self, data: dict[str, Any], depth: int = 0) -> None:
        object.__setattr__(self, "_data", data)
        object.__setattr__(self, "_depth", depth)

    # ---- 读协议 ----
    def __getitem__(self, key: str) -> Any:
        return _wrap(self._data[key], self._depth)

    def get(self, key: str, default: Any = None) -> Any:
        try:
            return self[key]
        except (KeyError, TypeError):
            return default

    def __contains__(self, key: object) -> bool:
        return key in self._data

    def keys(self):
        return self._data.keys()

    def values(self):
        return (_wrap(v, self._depth) for v in self._data.values())

    def items(self):
        return ((k, _wrap(v, self._depth)) for k, v in self._data.items())

    def __iter__(self):
        return iter(self._data)

    def __len__(self) -> int:
        return len(self._data)

    def __repr__(self) -> str:
        return f"ReadOnlyWorld({len(self._data)} keys)"

    # ---- 写协议：fail closed ----
    def __setitem__(self, key: str, value: Any) -> None:
        raise PermissionError(f"Explore 工具尝试写主干世界: key={key!r}")

    def __delitem__(self, key: str) -> None:
        raise PermissionError(f"Explore 工具尝试删除主干世界: key={key!r}")

    def setdefault(self, key: str, default: Any = None) -> Any:
        raise PermissionError(f"Explore 工具尝试 setdefault 主干世界: key={key!r}")

    def update(self, *args, **kwargs) -> None:
        raise PermissionError("Explore 工具尝试 update 主干世界")

    def clear(self) -> None:
        raise PermissionError("Explore 工具尝试 clear 主干世界")

    def pop(self, *args) -> Any:
        raise PermissionError("Explore 工具尝试 pop 主干世界")

    def popitem(self):
        raise PermissionError("Explore 工具尝试 popitem 主干世界")


class _ReadOnlyList:
    """只读列表视图：读透传，写抛 PermissionError。"""

    def __init__(self, data: list[Any], depth: int) -> None:
        self._data = data
        self._depth = depth

    def __getitem__(self, index):
        return _wrap(self._data[index], self._depth)

    def __iter__(self):
        return (_wrap(v, self._depth) for v in self._data)

    def __len__(self) -> int:
        return len(self._data)

    def __contains__(self, item: object) -> bool:
        return item in self._data

    def __repr__(self) -> str:
        return f"ReadOnlyList({len(self._data)} items)"

    def __setitem__(self, index, value) -> None:
        raise PermissionError("Explore 工具尝试写主干列表")

    def __delitem__(self, index) -> None:
        raise PermissionError("Explore 工具尝试删除主干列表")

    def append(self, *args) -> None:
        raise PermissionError("Explore 工具尝试 append 主干列表")

    def extend(self, *args) -> None:
        raise PermissionError("Explore 工具尝试 extend 主干列表")

    def insert(self, *args) -> None:
        raise PermissionError("Explore 工具尝试 insert 主干列表")

    def remove(self, *args) -> None:
        raise PermissionError("Explore 工具尝试 remove 主干列表")

    def pop(self, *args) -> Any:
        raise PermissionError("Explore 工具尝试 pop 主干列表")

    def clear(self) -> None:
        raise PermissionError("Explore 工具尝试 clear 主干列表")

    def sort(self, *args, **kwargs) -> None:
        raise PermissionError("Explore 工具尝试 sort 主干列表")


def _wrap(value: Any, depth: int) -> Any:
    """按深度把 dict/list 包成只读视图（标量原样透传）。"""
    if depth >= _MAX_DEPTH:
        return value
    if isinstance(value, dict):
        return ReadOnlyWorld(value, depth + 1)
    if isinstance(value, list):
        return _ReadOnlyList(value, depth + 1)
    return value


def readonly_view(world: dict[str, Any]) -> ReadOnlyWorld:
    """主干世界只读视图入口（Explore 工具只应拿到此视图）。"""
    return ReadOnlyWorld(world)


def run_explore(world: dict[str, Any], fn: Callable[[dict[str, Any]], T]) -> T:
    """统一 Explore 通道：深拷贝沙盒 → 执行 → 返回结果。

    主干永远以原始 dict 形式留在外部；fn 只接收沙盒 dict（写沙盒安全），
    主干写能力在通道边界处被切断（矩阵 55 行：任何 Explore 工具无法取得主干写能力）。
    """
    sandbox = copy.deepcopy(world)                     # 沙盒：深拷贝隔离（§5.1 铁律）
    return fn(sandbox)