跳到主要内容

数据截至 (上游 commit 5c995c365dfb)

运行时与会话

本章讲"执行"这一层:对外的接口长什么样,本地实现怎么管多个 shell 会话,以及一个 BashSession 从生到死的过程。输出/退出码怎么抠这个最难的点留到 02 章。

1. 接口:一个 Runtime 能做 8 件事​

AbstractRuntime 是所有运行时的契约(runtime/abstract.py:234)。它全部是 async 抽象方法,可以分成三组:

组方法干什么
会话create_session / run_in_session / close_session开/用/关一个长活 REPL(如 bash)
一次性执行execute跑一条子进程命令,像 subprocess.run(),跑完即弃
文件 & 探活read_file / write_file / upload / is_alive / close读写文件、上传、健康检查、关运行时

会话 vs execute 的区别是核心。 这俩很容易混:

  • run_in_session 把命令送进一个已经开着、有状态的 shell。前一条命令 cd /tmp 了,后一条命令就在 /tmp 里——状态延续。这就是"像人开着终端"。
  • execute 起一个全新子进程跑完就结束,没有跨命令状态。对应 subprocess.run()(runtime/local.py:419-453)。

数据模型:请求和响应都是 pydantic​

每个动作的输入输出都是 pydantic 模型,这让它们能直接 JSON 序列化走 HTTP(这正是 02/03 章远程化的基础)。最该认识的两个:

BashAction(你要跑什么,runtime/abstract.py:51)关键字段:

字段含义
command要跑的命令
session在哪个会话跑(默认 "default")
timeout超时,None 表示不限
check退出码策略:"raise" / "silent" / "ignore"(abstract.py:71-76)
is_interactive_command是否是 gdb 这类不退出的交互程序
expect除 PS1 外还要等待哪些输出串

BashObservation(你拿回什么,runtime/abstract.py:109):output(输出)、exit_code(退出码,可能 None)、failure_reason、expect_string(命中了哪个等待串而终止)。

一个巧思:IsAliveResponse 重写了 __bool__(abstract.py:19-20),所以可以直接 if await runtime.is_alive():。

为什么都用判别 union? 像 Action = BashAction | BashInterruptAction,用 action_type 字段判别(abstract.py:105)。这让 FastAPI 收到 JSON 时能自动反序列化成正确的子类型,也为将来加 session_type 留口子。目前 union 里 bash 是唯一成员。

2. LocalRuntime:用一个字典管多个会话​

LocalRuntime 是真正干活的实现(runtime/local.py:365)。它的状态简单到一句话:一个 name -> Session 的字典(runtime/local.py:375)。

# 示意,非源码 —— 演示 LocalRuntime 的会话管理骨架
class LocalRuntime:
    def __init__(self):
        self._sessions: dict[str, Session] = {}   # 会话名 -> 会话对象

    async def create_session(self, request):
        if request.session in self._sessions:      # 重名直接报错
            raise SessionExistsError(...)
        session = BashSession(request)
        self._sessions[request.session] = session
        return await session.start()               # 真正 spawn 一个 bash

    async def run_in_session(self, action):
        if action.session not in self._sessions:    # 没开过就报错
            raise SessionDoesNotExistError(...)
        return await self._sessions[action.session].run(action)

真实实现见 create_session(runtime/local.py:390-401)、run_in_session(:403-408)、close_session(:410-417)。重点看: 多会话并行就是"字典里多个 key";agent 想同时开 bash + ipython + gdb,就 create 三个不同 session 名字的会话。close() 会遍历字典关掉所有会话(runtime/local.py:474-478)。

execute(不进会话的一次性命令)则完全独立——直接 subprocess.run,带 timeout / cwd / env / check,并能把 stderr 合并进 stdout(merge_output_streams,runtime/local.py:419-453)。

3. BashSession:把一个 bash 进程包成可控 REPL​

BashSession 是"一个长活终端"的封装(runtime/local.py:129)。注释直接点明:"它基本就是一个 pexpect.REPLWrapper"(runtime/local.py:134)。

为什么用 pexpect? 因为要跑 ipython / gdb 这种交互程序,你不能用 subprocess.run(它跑完才返回)。pexpect 能伪装成一个终端(pty),持续地往 shell 写命令、读输出,正是"人坐在终端前"的程序化版本。

生命周期:start → run* → close​

Session 抽象基类只要求三个方法(runtime/local.py:118-126):start() / run() / close()。

start():spawn 一个 bash,并驯服它的提示符。 这是全章最关键的设计动作:

# 示意,非源码 —— BashSession.start 的核心思路
self._shell = pexpect.spawn(
    "/usr/bin/env bash",
    echo=False,                      # 关回显,免得命令本身混进输出
    env={**os.environ, "PS1": self._ps1, "PS2": "", "PS0": ""},
)
# 把 PS1 固定成一个我们认得的哨兵串,PS2/PS0 清空
self._shell.sendline("export PS1='SHELLPS1PREFIX' ; export PS2='' ; export PS0=''")
self._shell.expect(self._ps1)        # 等到哨兵出现 = shell 就绪

真实实现见 BashSession.start(runtime/local.py:157-175)。重点看 _ps1 = "SHELLPS1PREFIX"(runtime/local.py:138):整个输出解析的命脉就是"把 PS1 提示符换成一个固定的、不会自然出现的字符串",这样"命令跑完、回到提示符"就变成"在输出流里 expect 到这个串"。详见 02 章。

为什么要 _get_reset_commands 强行重设 PS1/PS2/PS0(runtime/local.py:149-155)?因为用户的 startup_source 启动脚本经常会覆盖 PS1(abstract.py:25-28 的注释自陈),必须在 source 之后再盖回来。

run():按命令类型分流。 run 是个分流器(runtime/local.py:218-243):

            ┌─ BashInterruptAction ──▶ interrupt()   发 Ctrl-C 终止当前命令
 run(action)─┤
            ├─ is_interactive_* ─────▶ _run_interactive()  不取退出码
            └─ 普通命令 ─────────────▶ _run_normal()    切分 + 跑 + 抠退出码

三条分支都在 02 章细讲。这里只记结论:普通命令走 _run_normal,跑完若 check=="raise" 且退出码非 0,就抛 NonZeroExitCodeError(runtime/local.py:238-242)。

interrupt():打断卡住的命令。 不是简单发 Ctrl-C 就完事——它先重试 n_retry 次发中断信号(runtime/local.py:188-201),失败了还有兜底:Ctrl-Z 把任务挂到后台,再 kill -9 %1 干掉(runtime/local.py:202-216)。这是给"命令彻底卡死"准备的最后手段。

close(): 关掉 pexpect 进程并置空(runtime/local.py:353-358)。

4. 巧妙之处​

  • 回显关闭 + 哨兵 PS1 是整个解析的地基。 echo=False 让命令文本不混进输出,固定 PS1 让"命令结束"可被 expect 精确捕捉(runtime/local.py:159-164)。一个看似不起眼的环境变量设置,撑起了后面所有解析。
  • 会话即字典,并行天然免费。 没有线程池、没有锁——多会话就是字典多个 key,每个 key 一个独立 pexpect 进程(runtime/local.py:375)。并发的复杂度被推到了"开多个会话"这个最简单的模型上。
  • IsAliveResponse.__bool__ 让健康检查能直接当布尔用(runtime/abstract.py:19),是 _wait_until_alive 轮询逻辑能写得很干净的原因(见 03 章)。

5. 代码地图​

主题文件路径关键符号
执行接口定义src/swerex/runtime/abstract.pyAbstractRuntime
请求/响应模型src/swerex/runtime/abstract.pyBashAction、BashObservation、IsAliveResponse、Command
本地运行时src/swerex/runtime/local.pyLocalRuntime、LocalRuntime.create_session、LocalRuntime.execute
会话抽象src/swerex/runtime/local.pySession
bash 会话src/swerex/runtime/local.pyBashSession、BashSession.start、BashSession.run、BashSession.interrupt
异常src/swerex/exceptions.pySessionExistsError、SessionDoesNotExistError、NonZeroExitCodeError