沙箱层:Docker Compose + tmux 会话(最巧的一环)
30 秒导读: 要给"任意 AI agent"打分,先得给它一个干净、隔离、可观测的终端去操作。 Terminal-Bench 的答案出奇地简单:每个任务起一套 Docker Compose 容器,在容器里开一个 tmux 会话,然后规定——agent 与世界的唯一交互方式就是"往这个会话敲键、从这个会话读屏"。 本章讲这层沙箱内部怎么实现:容器生命周期、把文件塞进容器、以及 tmux 会话那几个精巧的假动作。
本章是 02-harness-loop(一次 trial 的主线)的"下一层"。02 讲何时创建会话、
何时调 send_keys;本章只讲会话内部怎么把这些动作变成真实终端行为。任务数据长什么样见
01-task-anatomy,被评测的 agent 怎么用这个会话见 04-agents。
1. 这是什么(零基础也能懂)
一句话定义: 这一层是一个"带遥控器的隔离终端"——用 Docker 把每个任务关进独立容器,用一个 tmux 会话当遥控器,让评测框架能程序化地"敲键盘 + 看屏幕"。
它解决什么问题: 想象你要公平比较十个不同的 AI coding agent。它们内部千差万别(有的是纯 API
循环,有的是装在容器里的 CLI 工具),但它们都得在终端里干活——ls、cat、vim、pytest。
于是就有了一个统一的抽象:
不管你是什么 agent,你能做的只有两件事——(1) 往一个终端敲一串键,(2) 读回那个终端此刻的屏幕。
只 要把 agent 的能力收窄到这两个动作,任何模型都能被同一套 harness 驱动、被同一套测试脚本评判。 这就是"用一个 tmux 会话当作 agent 唯一接口"的核心价值。
一句话直觉/类比: 把 tmux 会话想成一个远程终端的录屏 + 键盘注入。harness 是坐在另一头的人,
它看不到 agent 的"想法",只能看到"屏幕现在长啥样",也只能"替它按键"。屏幕(capture-pane)+
键盘(send-keys)就是全部 API。
用起来什么样: 上层代码几乎就是一个上下文管理器 + 两三个方法调用:
# 示意,非源码:harness 视角下这层沙箱用起来的样子
with spin_up_terminal(...) as terminal: # 起容器 + 进入/退出自动清理
session = terminal.create_session("agent") # 在容器里开一个 tmux 会话
session.send_keys(["ls -la", "Enter"], # 敲键,并阻塞等它跑完
block=True)
screen = session.capture_pane() # 读回当前屏幕文本
三个词就是这一层的全部词汇表:spin_up_terminal(容器)、create_session(会话)、
send_keys / capture_pane(键盘 + 屏幕)。
2. 顶层全景(这层怎么转)
三个类分工明确,是一条"外套内"的包含链:Terminal 管会话字典 → DockerComposeManager 管容器 生命周期 → TmuxSession 管会话内的键盘与屏幕。
2.1 部件职责
| 部件 | 干什么 | 文件 |
|---|---|---|
spin_up_terminal | 上下文管理器:进入时起容器,退出时无条件清理 | terminal/terminal.py:147 |
Terminal | 门面:持有容器句柄 + 一个"会话名→会话"的字典,负责建/ 取会话 | terminal/terminal.py:11 |
DockerComposeManager | 容器生命周期:build/up -d/down,以及把文件塞进容器 | terminal/docker_compose_manager.py:27 |
TmuxSession | 一个 tmux 会话:敲键(阻塞/非阻塞)、读屏、录制、增量输出 | terminal/tmux_session.py:14 |
TerminalCommand | 一条命令的数据模型(命令文本 + 超时 + 是否阻塞 + 是否回车) | terminal/models.py:9 |
2.2 包含关系与数据流
怎么读这张图:从上到下是"谁持有谁";箭头是控制流,命令往下走、屏幕文本往上回。
spin_up_terminal(...) ── 上下文管理器,负责 start()/stop() 收尾
│
▼
┌─────────────────────────────┐
│ Terminal │ 门面
│ _sessions: dict[str, │ ← "agent"、"tests" 两个会话就存这
│ TmuxSession] │
└───────┬──────────────┬───────┘
│ 持有 │ 建/取会话
▼ ▼
┌──────────────────────┐ ┌───────────────────────────────┐
│ DockerComposeManager │ │ TmuxSession ("agent") │
│ build / up -d │ │ send_keys() ──敲键──►┐ │
│ down (清理) │ │ capture_pane() ◄─读屏─┤ │
│ copy_to_container │ │ │ │
└──────────┬───────────┘ └──────────── ────────────┼───────┘
│ 拥有 container 句柄 │
▼ │
╔══════════════════════════════════╗ │
║ Docker 容器(隔离沙箱) ║◄─────────┘
║ └─ tmux server ║ exec_run(["tmux", ...])
║ └─ session "agent" ║
║ └─ session "tests" ║
╚══════════════════════════════════╝
2.3 主线走一遍(高层)
一次 trial 里,这层沙箱经历五步(对应 harness.py:717-782 的调用序列):
spin_up_terminal(...)进入 → 触发DockerComposeManager.start():docker compose build+up -d,拿到容器句柄。terminal.create_session("agent", as_configured_user=True)→ 在容器里tmux new-session, 开始 asciinema 录制。- agent 反复
send_keys(...)敲命令、capture_pane()读屏(这就是 04 章 agent 的整个生命)。 - 若测试要另起壳:
create_session("tests", as_configured_user=False)→ 用 root 开第二个会话跑测试。 spin_up_terminal退出 →finally里terminal.stop()→DockerComposeManager.stop()(docker compose down),无论中途是否抛异常都执行。
关键设计:容器是每任务隔离的,会话是容器内的具名句柄。同一容器里可以有多个会话(agent 一个、 tests 一个),它们共享文件系统但各有独立的屏幕缓冲。
3. 核心机制(逐个拆,由浅入深)
3.1 生命周期:上下文管理器保证"起了必清"
要解决的小问题: 容器是重资源,跑几百个任务时绝不能泄漏——哪怕某个任务 中途崩了,也要把 容器拆掉。
思路: 用 Python 的 @contextmanager。进入 try 里 start(),finally 里 stop()——异常
照样走 finally,容器必被回收。
# terminal/terminal.py:147-179 —— spin_up_terminal(节选骨架)
@contextmanager
def spin_up_terminal(...) -> Generator[Terminal, None, None]:
terminal = Terminal(...)
try:
terminal.start() # → DockerComposeManager.start(): build + up -d
yield terminal # 交给 harness 用
finally:
terminal.stop() # → 关会话 + docker compose down,异常也走这里
Terminal.stop()(terminal/terminal.py:104)做三件事:逐个 session.stop()(停录制)、
_compose_manager.stop()(拆容器)、清空 _sessions 字典。
3.2 容器编排:三条 docker compose 命令撑起整个沙箱
要解决的小问题: 每个任务自带一个 docker-compose.yaml(定义它需要的镜像、服务、卷)。
harness 要能对任意这样的 compose 文件做统一的"起、拿、拆"。
关键技巧——用 -p <项目名> 做命名空间隔离。 所有 compose 命令都带上 -p 容器名,让并发跑的
不同任务各自成一个 compose project,互不干扰:
# terminal/docker_compose_manager.py:83-92 —— get_docker_compose_command
return [
"docker", "compose",
"-p", self._client_container_name, # 项目名 = 命名空间
"-f", str(self._docker_compose_path.resolve().absolute()),
*command, # build / up -d / down ...
]
生命周期就三步,都走 _run_docker_compose_command(docker_compose_manager.py:94,用
subprocess.run(..., check=True, capture_output=True)):
| 阶段 | 命令 | 代码 |
|---|---|---|
| 起 | build(除非 no_rebuild)+ up -d | start() docker_compose_manager.py:124-134 |
| 拿 | client.containers.get(名字) 取容器句柄 | start() docker_compose_manager.py:130 |
| 拆 | down;cleanup 时再 down --rmi all --volumes | stop() docker_compose_manager.py:136-145 |
磁盘不爆的兜底——_cleanup_build_cache。 跑成百上千个任务,构建缓存会撑爆磁盘。cleanup
模式下额外调 docker buildx prune,用 --max-used-space 30GB 把缓存压在上限内;buildx 不存在
(退出码 125)时只警告、不报错:
# terminal/docker_compose_manager.py:151-167 —— _cleanup_build_cache(节选)
subprocess.run(
["docker", "buildx", "prune", "--force", "--max-used-space", "30GB"],
capture_output=True, text=True, check=True,
)
# except CalledProcessError: e.returncode == 125 → 只 warning(buildx 未装)
3.3 把文件塞进容器:tar 打包 + put_archive
要解决的小问题: harness 常要把宿主机上的文件送进容器——测试脚本、参考解、或那个取时间戳的
小脚本(见 3.7)。容器可能没挂载对应卷,docker cp 又要另起子进程。
思路: 用 Docker SDK 的 container.put_archive——它吃一个 tar 字节流,直接解包进容器目录。
于是"复制文件"变成"内存里打个 tar 包再灌进去",全程不落盘、不起子进程。
copy_to_container(paths, container_dir)
│
├─ _create_tar_archive(paths) # 内存 BytesIO,tarfile.open(mode="w")
│ ├─ 文件 → tar.add(arcname=名字)
│ └─ 目录 → rglob("*") 逐个 add
│
├─ container.exec_run("mkdir -p 目标目录")
└─ container.put_archive(目标目录, tar_stream.read()) # 解包落地
真实实现两段:_create_tar_archive(docker_compose_manager.py:171-188,把路径列表打进内存
io.BytesIO)和 copy_to_container(docker_compose_manager.py:190-214,先 mkdir -p 再
put_archive)。TmuxSession.copy_to_container(tmux_session.py:384)只是转发到这个静态方法。
3.4 开会话:tmux new-session + 落日志 + 录屏
要解决的小问题: 会话不仅要能敲键,还要全程留痕——一份纯文本日志(给 livestream 和调试)、 一份 asciinema 录像(给回放和时间戳)。
核心命令 _tmux_start_session。 一条 tmux new-session 命令用 \; 串起三件事:
# terminal/tmux_session.py:90-102 —— _tmux_start_session(属性,返回命令列表)
f"tmux new-session -x 160 -y 40 -d -s {self._session_name} \\; " # 160×40 后台会话
f"set-option -t {self._session_name} history-limit 50000 \\; " # 回滚缓冲 5 万行
f'pipe-pane -t {self._session_name} "cat > {self.logging_path}"' # 屏幕实时管到日志文件
三个设计点:
-x 160 -y 40固定终端尺寸——让不同 agent 看到同样宽高的屏幕,读屏结果可复现。history-limit 50000把回滚缓冲开到 5 万行,配合capture_pane(capture_entire=True)能捞回被刷出屏幕的长输出(见 3.6)。pipe-pane ... cat > 日志把屏幕字节流实时管进/logs/<会话名>.log,livestream 就是 tail 这个文件。
start()(tmux_session.py:128-145)在开完会话后,如果录制没 禁用,就 send_keys 敲
asciinema rec --stdin <会话名>.cast 再 clear,从此这个会话的一切都被录进 .cast 文件。
3.5 阻塞式 send_keys:本章最巧的一环
要解决的小问题: 敲下 sleep 5 && ls 后,harness 怎么知道命令跑完了、可以去读屏了?
tmux send-keys 是"发了就返回"的,它不等命令执行完。轮询屏幕判断"是否出现提示符"既脆弱又不准。
思路(精髓): 借 tmux 自己的信号量。tmux 有一对命令 tmux wait -S <名>(发信号)和
tmux wait <名>(等信号)。于是:在用户命令末尾偷偷追加一句"命令跑完就发 done 信号",然后另起
一条 exec 去"等 done 信号"——等到了,就等于原命令跑完了。
怎么读这张图:上半是注入的键,下半是并行的等待;两者靠 done 这个 tmux 信号汇合。
send_keys(["sleep 5 && ls", "Enter"], block=True)
│
▼ _prepare_keys:先剥掉尾部回车,再接上完成信号
实际敲进会话的键:
"sleep 5 && ls" + "; tmux wait -S done" + "Enter"
└──────── 用户命令 ────────┘ └──完成即发 done 信号──┘
│
│ (几乎同时,另开一条 exec 阻塞等待)
▼
_exec_run(["timeout", "180s", "tmux", "wait", "done"])
│
├─ 收到 done(命令跑完)→ 退出码 0 → 返回,去读屏
└─ 180s 没等到 → timeout 退出码 ≠ 0 → raise TimeoutError
第一步 _prepare_keys:判断该不该阻塞,并拼上信号命令。
# terminal/tmux_session.py:180-205 —— _prepare_keys(节选)
if not block or not keys or not self._is_executing_command(keys[-1]):
return keys, False # 不是"要执行"的键,不阻塞
keys = self._prevent_execution(keys) # 先剥掉尾部的 Enter/换行
keys.extend([self._TMUX_COMPLETION_COMMAND, "Enter"]) # 追加 "; tmux wait -S done" + Enter
return keys, True
_TMUX_COMPLETION_COMMAND 就是常量 "; tmux wait -S done"(tmux_session.py:18)。
注意顺序:必须先把用户原本的回车去掉,否则命令会先被回车执行、把 ; tmux wait -S done
甩在后面单独成一行。这就是 _prevent_execution 的活。
第二步 _prevent_execution:去掉多余回车。 从键列表尾部往前,把"回车键"(Enter/C-m
等,见 _ENTER_KEYS 集合 tmux_session.py:15)或"以换行结尾的字符串"逐个剥掉/去尾,直到最后一个
键不再是"执行动作":
# terminal/tmux_session.py:165-178 —— _prevent_execution(节选)
while keys and self._is_executing_command(keys[-1]):
if self._is_enter_key(keys[-1]):
keys.pop() # 整个是回车键 → 删掉
else:
stripped_key = keys[-1].rstrip(self._NEWLINE_CHARS) # 字符串尾部换行 → 去掉
keys[-1] = stripped_key if stripped_key else ... # 去空则 pop
第三步 _send_blocking_keys:敲键 + 用 timeout 包住等待。
# terminal/tmux_session.py:207-222 —— _send_blocking_keys(节选)
self.container.exec_run(self._tmux_send_keys(keys), user=self._user) # 敲键(含追加的信号命令)
result = self._exec_run(
["timeout", f"{max_timeout_sec}s", "tmux", "wait", "done"] # 阻塞等 done,最多 Ns
)
if result.exit_code != 0:
raise TimeoutError(f"Command timed out after {max_timeout_sec} seconds")
用系统 timeout Ns 包住 tmux wait done:命令按时跑完则 wait 收到信号、退出码 0;超时则
timeout 杀掉 wait、退出码非 0 → 抛 TimeoutError(默认 max_timeout_sec=180.0,见
send_keys 签名 tmux_session.py:249-255)。
非阻塞分支(block=False): 走 _send_non_blocking_keys(tmux_session.py:224-235)——
敲完就返回,只用 min_timeout_sec 硬睡一小会儿让输出稳定,不等命令结束。适合"启动一个交互式程序
然后要读中间态"的场景。