51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""Sandbox 容器管理(DESIGN §7.5)。
|
|
|
|
模块边界:
|
|
- `network.py`:Docker network ensure(`zcbot-sandbox-net`,`--internal` 隔离 outbound + cross-container)
|
|
- `pool.py`:per-user 容器生命周期(ensure / mark_active / reap_idle / shutdown_all)
|
|
- `__init__.py`:module-level singleton(`init_pool` / `get_pool`),给 web lifespan 与
|
|
`agent_builder` 共享同一个池实例。
|
|
|
|
不在本目录:`shell` / `run_python` 工具的 docker exec 调用 ── 那是 `core/executor_docker.py`,
|
|
调用本模块的 `pool.ensure(user_id)` 拿到容器名后再 exec。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from .pool import SandboxPool, container_name, setup_pool
|
|
from .network import NETWORK_NAME, ensure_network
|
|
|
|
|
|
__all__ = [
|
|
"SandboxPool",
|
|
"container_name",
|
|
"setup_pool",
|
|
"NETWORK_NAME",
|
|
"ensure_network",
|
|
"init_pool",
|
|
"get_pool",
|
|
]
|
|
|
|
|
|
# Module-level singleton。web lifespan 启动钩子调 `init_pool(user_root_base)`,
|
|
# `agent_builder` 在构造 DockerExecutor 时 `get_pool()` 拿同一实例。
|
|
# 未初始化 → `get_pool()` 返 None,agent_builder 此时必须不走 docker 分支。
|
|
_pool: Optional[SandboxPool] = None
|
|
|
|
|
|
def init_pool(user_root_base: Path) -> SandboxPool:
|
|
"""幂等初始化 module-level pool。返回 pool 实例。
|
|
|
|
lifespan 调一次;ensure_network 内部也幂等。重复调用返回同一实例(不重新建)。
|
|
"""
|
|
global _pool
|
|
if _pool is None:
|
|
_pool = setup_pool(user_root_base)
|
|
return _pool
|
|
|
|
|
|
def get_pool() -> Optional[SandboxPool]:
|
|
return _pool
|