05. Environment 与退出协议：把 action 变成真实副作用

本系列基于 mini-SWE-agent 源码学习整理；源码版本固定为 Asuhe404/mini-swe-agent@388da74，便于长期核对实现。

配套练习：下载 05_environment_exit_protocol.py。

系列导航

返回路线图 · 上一篇：04. Agent 核心循环：让模型回答变成下一轮上下文 · 下一篇：06. Model 适配与 action 翻译：把模型方言变成统一命令

适合：零 Python 基础、已有 JavaScript 基础。
重点文件：

• src/minisweagent/environments/local.py
• src/minisweagent/exceptions.py
• tests/environments/test_local.py

本节目标：关闭源码后，能够解释 LocalEnvironment.execute() 的输入、输出和三种结束路径；能够说清 Submitted 为什么语法上是异常、语义上却是成功；能够判断一段完成标记是否真的会结束 Agent。

1. 先用 JavaScript 建立心智模型

1.1 它很像同步版 child_process 适配器

先不要把 Environment 想得太抽象。LocalEnvironment 做的事情可以近似理解为：

import { execFileSync } from "node:child_process";

class LocalEnvironment {
  constructor(options = {}) {
    this.config = ConfigSchema.parse(options);
  }

  execute(action, cwd = "", timeout) {
    const command = action.command === undefined ? "" : action.command;
    cwd = cwd || this.config.cwd || process.cwd();

    let output;
    try {
      const result = runShell(command, {
        cwd,
        env: { ...process.env, ...this.config.env },
        timeout: timeout || this.config.timeout,
      });
      output = {
        output: result.stdoutAndStderr,
        returncode: result.status,
        exception_info: "",
      };
    } catch (error) {
      output = {
        output: error.output ?? "",
        returncode: -1,
        exception_info: "An error occurred...",
        extra: {
          exception_type: error.constructor.name,
          exception: String(error),
        },
      };
    }

    this.checkFinished(output); // 这里可能 throw new Submitted(...)
    return output;
  }
}

对应关系：

最大差异是：这里完全同步。命令运行期间，当前 Python 线程会停在 communicate()，没有 Promise、async 或 await。

1.2 本课最反直觉的三种结果

也就是说：

普通命令失败是 observation；符合协议的成功提交才会抛异常。

1.3 为什么要用异常表示成功

一个 Agent message 可能含多个 action：

outputs = [self.env.execute(action) for action in actions]

Environment 在很深的调用层里发现任务完成。如果只返回一个特殊 dict，每一层都要写 if finished 并手动向上传递。raise Submitted(...) 可以立即跳过：

• 当前 execute() 的普通返回；
• 剩余 action；
• observation 格式化；
• 当前 step()。

最外层 DefaultAgent.run() 统一捕获它，把携带的 exit message 加入历史。这类似在 JS 多层调用中用一个带 payload 的自定义 throw 做非局部返回。

2. 5W2H

3. 一张图看懂完整接力

最重要的代码位置关系：

agents/default.py:154
  -> LocalEnvironment.execute()
       -> _run()
       -> _check_finished()
            -> raise Submitted(...)
  -> execute_actions() 立即中断
  -> DefaultAgent.run() 的 except InterruptAgentFlow
  -> messages 最后一条成为 exit
  -> run() 返回 exit message 的 extra

4. 先认清两个数据协议

4.1 action 输入

本地 Environment 真正使用的字段只有：

{"command": "printf 'hello\n'"}

类型写成普通 dict，因此这是一个“隐式协议”：类型检查器无法证明一定存在字符串 command。缺少 key 时源码会使用空字符串。

4.2 普通 output

成功或普通非零退出：

{
    "output": "hello\n",
    "returncode": 0,
    "exception_info": "",
}

Python 执行层异常：

{
    "output": "已经产生的部分输出",
    "returncode": -1,
    "exception_info": "An error occurred while executing the command: ...",
    "extra": {
        "exception_type": "TimeoutExpired",
        "exception": "...",
    },
}

-1 是项目定义的“执行器异常”哨兵，不是 shell 的普通退出码。

4.3 Submitted 携带的 exit message

{
    "role": "exit",
    "content": submission,
    "extra": {
        "exit_status": "Submitted",
        "submission": submission,
    },
}

这个 dict 不从 execute() 返回，而是放在 Submitted.messages 中向外抛。

5. 逐行读 local.py:1-92

第 1-5 行：标准库工具

import os
import platform
import signal
import subprocess
from typing import Any

• os：当前目录、环境变量、进程组。
• platform：操作系统信息，供提示词模板使用。
• signal：超时后发送 SIGKILL。
• subprocess：创建并管理 shell 进程。
• Any：结果 dict 的 value 不止一种类型。

Any 类似 TS 的 any。它方便，但也说明 output 的形状只靠约定，没有 TypedDict 帮忙检查。

第 7-10 行：第三方与项目依赖

from pydantic import BaseModel

from minisweagent.exceptions import Submitted
from minisweagent.utils.serialize import recursive_merge

BaseModel 负责配置默认值和运行时校验；Submitted 负责成功退出；recursive_merge 把多来源模板变量递归合并。

第 13-16 行：配置 schema

class LocalEnvironmentConfig(BaseModel):
    cwd: str = ""
    env: dict[str, str] = {}
    timeout: int = 30

JS/TS 近似写法：

const LocalEnvironmentConfig = z.object({
  cwd: z.string().default(""),
  env: z.record(z.string()).default({}),
  timeout: z.number().int().default(30),
});

• 空 cwd 表示“没有在配置中指定”。
• env 只保存需要新增或覆盖的变量。
• timeout 单位是秒。
• 普通 Python class 里直接写可变默认值 {} 往往危险；这里 Pydantic v2 会复制它，两个 config 实例不会共享同一个 dict。

第 19-22 行：构造 LocalEnvironment

class LocalEnvironment:
    def __init__(self, *, config_class: type = LocalEnvironmentConfig, **kwargs):
        """This class executes bash commands directly on the local machine."""
        self.config = config_class(**kwargs)

• self 对应 JS 的 this。
• 单独的 * 表示后面的参数只能具名传递。
• config_class 默认是配置 class，也允许测试或子类注入另一种配置类型。
• **kwargs 收集剩余具名参数，再用 config_class(**kwargs) 展开。

例如：

LocalEnvironment(cwd="/tmp/work", timeout=5)

近似：

new LocalEnvironment({ cwd: "/tmp/work", timeout: 5 });

注释写的是 bash，但后面只设置 shell=True。POSIX Python 通常调用 /bin/sh，并不保证是 Bash。

第 24 行：execute 的函数合同

def execute(self, action: dict, cwd: str = "", *, timeout: int | None = None) -> dict[str, Any]:

• action：模型产生的 action dict。
• cwd：本次调用可覆盖配置目录。
• 第二个 *：timeout 必须写成 timeout=...。
• int | None 对应 TS 的 number | null；这里 None 表示未提供。
• 返回 dict[str, Any]，即字符串 key、任意类型 value。

第 25-27 行：取命令并决定 cwd

command = action.get("command", "")
cwd = cwd or self.config.cwd or os.getcwd()

dict.get(key, default) 在 key 缺失时给默认值。注意：若 key 存在但值是 None，不会改成空字符串。

or 和 JS 的 || 一样返回第一个 truthy 值，所以目录优先级是：

本次 execute(cwd=...)
  > 构造器 config.cwd
  > Python 进程当前目录 os.getcwd()

因此空字符串无法用来覆盖一个已有的 config cwd；它被理解为“继续回退”。

第 28-30 行：正常执行路径

try:
    result = _run(command, cwd, os.environ | self.config.env, timeout or self.config.timeout)
    output = {"output": result.stdout, "returncode": result.returncode, "exception_info": ""}

os.environ | self.config.env 是 Python 3.9+ 的字典合并，右侧同名 key 获胜：

const childEnv = { ...process.env, ...config.env };

所以子进程既继承宿主环境，又允许配置覆盖某些值。

timeout 的优先级看似是“本次值 > 配置值”，但 or 按 truthy 判断，因此显式传 timeout=0 会回退到配置 timeout。若要区分 0 和未提供，通常应判断 timeout is None。

_run() 没有开启 check=True。所以 exit 7 会正常返回 CompletedProcess，不会进入 except。

第 31-41 行：把 Python 异常标准化

except Exception as e:
    raw_output = getattr(e, "output", None)
    raw_output = (
        raw_output.decode("utf-8", errors="replace") if isinstance(raw_output, bytes) else (raw_output or "")
    )
    output = {
        "output": raw_output,
        "returncode": -1,
        "exception_info": f"An error occurred while executing the command: {e}",
        "extra": {"exception_type": type(e).__name__, "exception": str(e)},
    }

• except Exception as e 类似 JS catch (e)。
• getattr(e, "output", None) 类似安全读取 e.output，没有属性就给 None。
• isinstance(raw_output, bytes) 判断是否需要 UTF-8 解码。
• errors="replace" 遇到非法字节时用替换字符，不让解码再次失败。
• type(e).__name__ 近似 e.constructor.name。
• 异常文本同时进入适合人读的 exception_info 和结构化 extra。

典型会走这里的情况：

• cwd 不存在；
• 创建进程失败；
• communicate() 超时后重新抛出的 TimeoutExpired。

这里捕获很宽，会把 _run() 内部的编程错误也包装成“命令执行失败”。但 command 和 cwd 的计算在 try 之前，所以非 dict action 等错误仍可能直接逃出。

第 42-43 行：所有结果都检查一次完成协议

self._check_finished(output)
return output

关键是 _check_finished() 在上面 try/except 的外面。因此它抛出的 Submitted 不会被第 31 行误包装成 returncode=-1。

第 45-56 行：完成标记与 submission

lines = output.get("output", "").lstrip().splitlines(keepends=True)
if lines and lines[0].strip() == "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT" and output["returncode"] == 0:
    submission = "".join(lines[1:])
    raise Submitted(
        {
            "role": "exit",
            "content": submission,
            "extra": {"exit_status": "Submitted", "submission": submission},
        }
    )

逐步翻译：

1. output.get("output", "")：读取合并输出。
2. lstrip()：删除整段输出开头的空白。
3. splitlines(keepends=True)：拆成行，同时保留每行换行符。
4. lines and ...：先确保至少有一行，避免访问 lines[0] 越界。
5. lines[0].strip() == marker：第一条有效文本行去掉两端空白后必须精确等于 marker。
6. output["returncode"] == 0：整条 shell 命令最终必须成功。
7. "".join(lines[1:])：marker 后所有文本原样拼成 submission。
8. raise Submitted(message)：立即把 exit message 送回 Agent 外层循环。

预测四段输出：

这里的“输出”实际是 stdout 与 stderr 的合并流。若 stderr 的 warning 出现在 marker 前，也会阻止提交；若 marker 首先由 stderr 打印且最终成功，也可能触发提交。

第 58-59 行：给 Jinja 提供环境信息

def get_template_vars(self, **kwargs) -> dict[str, Any]:
    return recursive_merge(self.config.model_dump(), platform.uname()._asdict(), os.environ, kwargs)

从左到右合并，后面的同名字段优先：

config
  < platform.uname()
  < os.environ
  < 本次 kwargs

这些值最终可被 Agent 的 Jinja 模板读取。config.env 仍是 config 内的嵌套字段，而当前进程的环境变量还会作为顶层模板变量加入。

第 61-69 行：序列化 Environment 配置

def serialize(self) -> dict:
    return {
        "info": {
            "config": {
                "environment": self.config.model_dump(mode="json"),
                "environment_type": f"{self.__class__.__module__}.{self.__class__.__name__}",
            }
        }
    }

• model_dump(mode="json") 生成可写入 JSON 的普通数据。
• f-string 对应 JS template literal。
• class 的模块名与类名一起记录，trajectory 才知道实际用了哪个 Environment。
• 配置中的 env 会被保存；若里面放密钥，trajectory 也可能包含它。

第 72-84 行：创建 shell 进程

def _run(command: str, cwd: str, env: dict[str, str], timeout: int) -> subprocess.CompletedProcess[str]:
    process = subprocess.Popen(
        command,
        shell=True,
        text=True,
        cwd=cwd,
        env=env,
        encoding="utf-8",
        errors="replace",
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        start_new_session=os.name == "posix",
    )

• 前导下划线 _run 只是“模块内部使用”的约定，不是真 private。
• shell=True 允许管道、重定向、$() 等 shell 语法。
• text=True 配合 encoding，让输出直接是 Python str。
• stdout=PIPE 让父进程收集输出。
• stderr=STDOUT 把 stderr 合并到同一条输出流。
• start_new_session=True 在 POSIX 上创建新 session，使 shell 成为新进程组组长。

这里的安全含义非常直接：command 不是参数数组，而是交给 shell 解析的完整字符串。若 command 来自模型，它拥有当前用户在宿主机上的 shell 权限。

第 86-92 行：等待、超时、杀进程组

try:
    stdout, _ = process.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
    os.killpg(process.pid, signal.SIGKILL) if os.name == "posix" else process.kill()
    stdout, _ = process.communicate()
    raise subprocess.TimeoutExpired(command, timeout, output=stdout)
return subprocess.CompletedProcess(command, process.returncode, stdout=stdout)

• communicate() 等待进程结束，同时读尽 pipe，避免常见的管道阻塞。
• tuple 解包中的 _ 表示第二个值不使用；stderr 已合并，所以它通常是 None。
• 超时时，POSIX 用 killpg 杀整个进程组，不只杀外层 shell。
• Windows 分支只 kill() 当前进程，shell 创建的后代可能残留。
• kill 后再次 communicate() 回收进程并拿到剩余/部分输出。
• 随后手动重抛带 output 的 TimeoutExpired，由 execute() 转成统一 dict。
• 正常时手动构造 CompletedProcess，供 execute() 读取 stdout 和 returncode。

SIGKILL 不给进程清理机会，命令可能留下写到一半的文件。第二次 communicate() 也没有 timeout；若后代逃离进程组并持续持有 pipe，理论上仍可能继续等待。

6. 逐行读 exceptions.py:1-26

第 1-6 行：控制流异常基类

class InterruptAgentFlow(Exception):
    """Raised to interrupt the agent flow and add messages."""

    def __init__(self, *messages: dict):
        self.messages = messages
        super().__init__()

JS 类比：

class InterruptAgentFlow extends Error {
  constructor(...messages) {
    super();
    this.messages = messages;
  }
}

• Exception 是 Python 普通异常基类。
• *messages 是 rest parameter，允许传任意条 message。
• Python 把它们收集成 tuple，所以 signal.messages[0] 取第一条。
• super().__init__() 没传文本，因此 str(signal) 通常是空字符串；真正 payload 在 messages。

第 9-26 行：五种语义子类

Exception
└── InterruptAgentFlow
    ├── Submitted
    ├── LimitsExceeded
    │   └── TimeExceeded
    ├── UserInterruption
    └── FormatError

这些 class 只有 docstring，没有新方法。class 名本身就是“事件类型”：

真正让 DefaultAgent.run() 结束的是异常携带的最后消息 role == "exit"，不是异常 class 名本身。

Python except 从上到下匹配，而且子类也匹配父类。因此：

except TimeExceeded:
    ...
except LimitsExceeded:
    ...

顺序不能随意互换。InteractiveAgent 正是先处理不能通过加预算解除的 TimeExceeded，再处理可提高限制的 LimitsExceeded。

两个“超时”不要混为一谈

它们名字相近，但没有继承关系，也不走同一条控制流。

7. Submitted 最终在哪里被接住

local.py 只负责发信号。退出闭环还要看 agents/default.py：

except InterruptAgentFlow as e:
    self.add_messages(*e.messages)
...
if self.messages[-1].get("role") == "exit":
    break
return self.messages[-1].get("extra", {})

一次提交的精确顺序：

1. Model 的 assistant message 已加入历史。
2. execute_actions() 调 env.execute(action)。
3. _check_finished() 抛 Submitted(exit_message)。
4. list comprehension 立即停止，剩余 action 不再执行。
5. observation formatter 没被调用，所以提交 action 没有 observation。
6. run() 用父类 InterruptAgentFlow 捕获信号。
7. exit message 加入 history。
8. finally 仍保存 trajectory。
9. 最后一条 message 是 exit，循环结束并返回其 extra。

tests/agents/test_default.py:130-150 间接验证了这条完整链：两轮后返回 Submitted，submission 是 "Task completed successfully\n"。

8. 逐段读 tests/environments/test_local.py:1-264

这个文件有 18 个测试函数；参数化测试展开 3 个 case，因此 pytest 报告总计 20 cases。

第 1-12 行：测试工具与被测对象

• pytest 对应 Jest/Vitest 测试运行器。
• 顶层 def test_... 会被自动发现，不需要 describe。
• Python 原生 assert actual == expected 近似 expect(actual).toBe(expected)。
• TemporaryDirectory() 创建离开 with 后自动删除的目录。
• Path 是面向对象的路径 API。
• patch.dict(os.environ, ...) 在 with 内临时改环境变量，退出后恢复，近似 vi.stubEnv 加自动 cleanup。

第 15-21 行：配置默认值

构造空配置并分别断言：

cwd == ""
env == {}
timeout == 30

这是默认合同测试。它没有直接验证两个实例的 env 是否彼此隔离。

第 24-30 行：最小成功命令

echo 'hello world' 应得到 rc=0 且 output 包含文本。测试没有断言成功时 exception_info == ""。

第 33-66 行：环境变量的继承、增加与覆盖

三组测试共同证明：

os.environ | self.config.env

• 配置变量会进入子进程；
• 原有宿主变量仍保留；
• 名字冲突时右侧 config 值获胜。

第 69-97 行：cwd 三层优先级

三个测试分别覆盖：

1. 构造器 cwd 生效；
2. execute(..., cwd=...) 覆盖构造器；
3. 两者为空时回退 os.getcwd()。

with TemporaryDirectory() as temp_dir 类似 JS 的 try/finally 资源清理，但语法更紧凑。

第 100-124 行：失败命令与 stderr

• exit 1 返回 rc=1 和空 output，不是 Python 异常。
• 不存在的命令因为 shell=True，通常由 shell 自己报错并给非零码，而不是 Python 直接抛 FileNotFoundError。
• >&2 把文本写到 stderr；由于 stderr=STDOUT，仍能从 result["output"] 读到。

所以字段名 output 更准确地表示“stdout 与 stderr 的合并结果”。

第 127-135 行：timeout 标准化

sleep 2 超过 1 秒后，测试要求：

returncode == -1
exception_info 包含 "timed out"
extra.exception_type == "TimeoutExpired"

这证明命令超时不会直接终止 Agent，而会作为 observation 反馈给下一轮模型。

第 137-192 行：确认整个 POSIX 进程组被杀

这是本文件最重要的进程测试：

1. 临时生成一个永远循环的 Python 子进程脚本。
2. 子进程先把自己的 pid 写入临时文件。
3. LocalEnvironment 用 shell 启动它并在 1 秒后超时。
4. _read_pid 最多轮询约 5 秒等待 pid 文件出现。
5. _process_exited 用 os.kill(pid, 0) 探测进程是否存在；signal 0 不会杀进程。
6. finally 无论断言成功与否都调用清理 helper。
7. Windows 跳过，因为当前实现的进程组保证只适用于 POSIX。

shlex.quote 类似可靠的 shell 参数转义，避免 Python 路径或临时路径中的空格被 shell 拆错。

第 195-200 行：自定义配置

先构造 LocalEnvironmentConfig(timeout=5)，再用 **config.__dict__ 展开成构造器具名参数，最后断言保存值。它只检查配置，没有真的执行一个 5 秒命令。

第 203-216 行：一个模板生成三个 case

@pytest.mark.parametrize(
    ("command", "expected_returncode"),
    [
        ("echo 'test'", 0),
        ("exit 1", 1),
        ("exit 42", 42),
    ],
)

近似：

test.each([
  ["echo 'test'", 0],
  ["exit 1", 1],
  ["exit 42", 42],
])("...", (command, expectedReturncode) => {});

decorator @... 在函数定义时注册参数表；pytest 会把下方测试运行三次。

第 219-229 行：多行输出

strip().split("\n") 去掉整体首尾空白再拆行，断言三行顺序。这里使用 echo -e，不同 /bin/sh 实现的行为可能有可移植性差异。

第 232-249 行：真实文件副作用

命令在临时 cwd 中用 > 创建文件，再用 cat 读取。测试既检查 shell 输出，也从 Python 用 Path.read_text() 验证文件内容；离开 with 后目录自动清理。

第 252-264 行：证明 shell=True 的能力

• | 把一个命令输出传给下一个命令；
• $(...) 先执行内层命令，再把结果替换进外层命令。

这些能力方便模型工作，也正是 shell injection 与任意副作用风险的来源。

当前测试没有直接覆盖的边界

• marker 成功触发 Submitted；
• marker 前有普通文本；
• 有 marker 但 return code 非零；
• marker 从 stderr 打印；
• 本次调用的 timeout 覆盖配置 timeout；
• 缺失 command；
• get_template_vars() 与 serialize()；
• Windows 下 shell 后代进程是否残留。

Local 的成功提交在 Agent 集成测试中有覆盖，但把 marker 边界直接放到 Environment 单测会更容易定位回归。

9. 这些代码的主要使用位置

hello_world.py 只把 YAML 的 agent 段展开给 DefaultAgent；它直接 LocalEnvironment()，因此 default.yaml 的 environment 配置不会在这条入口生效。run/mini.py 才通过 Environment 工厂读取完整 environment 段。

10. 可维护性、性能与安全审查

安全边界

1. LocalEnvironment 不是沙箱。 shell=True 以当前用户权限执行模型生成的字符串，可读写文件、访问网络、启动进程。
2. 子进程继承完整宿主环境。 os.environ 可能包含 API key、token 或内部地址。
3. 模板变量暴露面较大。 get_template_vars() 把当前环境变量合入 Jinja 上下文；配置 env 还可能进入 trajectory。
4. 完成标记可被输出伪造或误触发。 任意成功程序只要把 marker 作为首条有效输出就能结束 Agent。
5. 学习时不要对不可信任务开启 --yolo。使用固定命令、临时目录，真实任务优先隔离 Environment。

性能与可靠性

1. communicate() 把完整合并输出放进内存；极大输出可能造成内存压力，也会扩大下一轮模型上下文。
2. timeout 后 POSIX 直接 SIGKILL，进程没有清理机会，可能留下中间文件状态。
3. Windows 只杀外层进程，后代进程可能继续运行。
4. timeout 后第二次 communicate() 没有新 timeout；逃离进程组并持有 pipe 的后代是极端阻塞风险。
5. docstring 说 bash，实际通常是 /bin/sh；Bash 专属语法可能失败。

可以改进，但要先定义语义

这里先不修改源码，因为本节任务是理解现有合同。任何“更合理”的改法，都可能改变 Model、Agent、其他 Environment 和 trajectory 共同依赖的行为。

11. 无 API 小练习

练习脚本已作为本文附件提供。

它不调用模型、不联网、不写文件，只让真实 LocalEnvironment 在宿主机执行三条固定命令：

1. 打印一个环境变量；
2. 向 stderr 打印文本并 exit 7；
3. 打印完成标记与 submission。

先不要运行，预测三行输出，再执行：

MSWEA_SILENT_STARTUP=1 PYTHONPATH=src .venv/bin/python \
  exercises/05_environment_exit_protocol.py

已验证输出：

0 hello environment
7 problem True
exit Submitted 'learned locally\n'

逐行理解练习：

• 第一个 execute 正常返回，所以可直接读取 dict。
• 第二个命令 rc=7，但 exception_info 仍为空，因为 shell 正常完成了“以 7 退出”。
• 第三个 execute 不会返回；except Submitted as signal 接住成功控制信号。
• signal.messages[0] 是退出消息；repr(...) 会把换行显示成 \n，便于观察 submission 是否保留换行。

动手改两次：

1. 在 marker 前先 printf 'noise\n'。预测为什么不会进入 except Submitted。
2. marker 与正文之后加 exit 1。预测为什么完成文本存在却仍不会提交。

提醒：这个练习里的三条命令是固定且只打印文本的；这不代表 LocalEnvironment 本身是沙箱。

12. 三道检查题（请先回答，不要查答案）

1. exit 7 为什么返回 returncode=7，而不是进入 except Exception？什么情况才会得到 returncode=-1？
2. Submitted 触发必须同时满足哪两个条件？marker 后面的文本怎样变成 submission？
3. 子进程的 TimeoutExpired 与 Agent 的 TimeExceeded 有什么区别，分别在哪里产生、如何被处理？