158 lines
5.6 KiB
Python
158 lines
5.6 KiB
Python
# ============================================================
|
||
# 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)
|