feat(soak): AppWorld 场景适配器——代码围栏解释器与容器执行器

复刻 dissect 的动作协议当压测负载：两条正则、first_only 策略、五条纠错说明逐字照抄，
提示词模板连同出处一起收进 tools/soak/prompts/。有两处刻意不同，都写在代码注释里——
未闭合围栏那一支不补回三反引号（公共契约要求回填历史的文本不长于模型原文），空围栏
那一支不截断。

执行器在 execute 内部顺带问一次 is_done 填 env_reported_completion，并把环境自己数的
执行次数透出来。那个数字是故障注入的判据来源：验「声明绝不重放的动作没有被执行两次」
必须数环境侧，数库自己的计数器等于我们和我们自己对账。

端到端真跑过一道题（82e2fac_1）：8 步 task_completed，环境判分通过，事件数与步数与
环境执行次数三者相等，耗时 30 秒。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.gitignore

@@ -13,6 +13,13 @@
 /logs/
 /tests/outputs/
 
+# 压测产物。日志与报告里带着模型原文和文档片段，体量也大（一次全量约一百多个运行），
+# 不该进版本库；`.cache/` 是从环境取一次就不变的那些（app 描述清单）。
+/tools/soak/.cache/
+/tools/soak/runs/
+/tools/soak/reports/
+/tools/soak/workspaces/
+
 # Python
 __pycache__/
 *.py[cod]

tools/soak/prompts/appworld/PROVENANCE.md

@@ -0,0 +1,36 @@
+# 这两个提示词文件是什么
+
+`run_prefix.txt` 与 `item_suffix.txt` 合起来是 AppWorld 官方 ReAct baseline 的提示词，
+压测的 AppWorld 场景直接用它们。
+
+原件是 `StonyBrookNLP/appworld` 仓库的
+`experiments/prompts/react_code_agent/instructions.txt`（371 行，Apache License 2.0）。
+这里的两个文件经由 `reference/dissect/config/prompts/appworld/` 转手而来——那边把原件拆成
+两段，并把示例演示里的主管信息换成了固定假值（`Sam Carter / sam.carter@example.com /
+555-0100`）。除拆分与那处替换之外没有其他改动。原始许可与著作权归 AppWorld 作者，
+Apache-2.0 允许再分发。
+
+## 为什么照抄而不自己写一份
+
+压测要拿 dissect 已有的 937 条真实轨迹当对照组——步数分布、停止原因构成、成功率。
+提示词换一份，这三样都会跟着变，对照就失去意义：分不清是内核的行为变了还是提示词变了。
+
+## 改动它们的后果
+
+`observation_template` 与这份提示词是绑死的。提示词里整段示例演示都按
+`Output:\n```\n…\n```\n\n` 的形态写观察，模板改了就会让示例与实况对不上，而那正是这份
+提示词最想避免的事。模板的取值在 `tools/soak/scenarios/appworld.py`，改任何一边都要同时
+改另一边，并且作废与 dissect 基线的对照。
+
+## 变量
+
+`run_prefix.txt` 里只有一个 `{{ app_descriptions }}`，值从环境跑一次
+`print(apis.api_docs.show_app_descriptions())` 拿到，整个压测复用同一份。
+`item_suffix.txt` 里有 `{{ main_user.first_name }}`、`{{ main_user.last_name }}`、
+`{{ main_user.email }}`、`{{ main_user.phone_number }}`、`{{ instruction }}`，
+前四个来自题目的主管信息，最后一个是题面。
+
+渲染器是 `tools/soak/scenarios/appworld.py` 里自己写的极简替换，不是 Jinja——这两个模板
+只用到 `{{ 名字 }}` 与 `{{ 名字.属性 }}` 两种形态，为它装一个模板引擎不值得。
+渲染完会校验文本里不再残留 `{{`：留一个没替换的占位符，模型会看见字面量
+`{{ instruction }}`，而那在轨迹里看起来只是模型表现差。

tools/soak/prompts/appworld/item_suffix.txt

@@ -0,0 +1,5 @@
+USER:
+Using these APIs, now generate code to solve the actual task:
+
+My name is: {{ main_user.first_name }} {{ main_user.last_name }}. My personal email is {{ main_user.email }} and phone number is {{ main_user.phone_number }}.
+Task: {{ instruction }}

tools/soak/prompts/appworld/run_prefix.txt

@@ -0,0 +1,365 @@
+USER:
+I am your supervisor, and you are an AI Assistant whose job is to complete my day-to-day tasks fully autonomously.
+
+To do this, you will need to interact with app(s) (e.g., spotify, venmo etc) using their associated APIs on my behalf. For this you will undertake a *multi-step conversation* using a python REPL environment. That is, you will write the python code, the environment will execute it and show you the result, based on which, you will write python code for the next step and so on, until you've achieved the goal. This environment will let you interact with app(s) using their associated APIs on my behalf.
+
+Here are three key APIs that you need to know to get more information
+
+# To get a list of apps that are available to you.
+
+```python
+print(apis.api_docs.show_app_descriptions())
+```
+
+# To get the list of APIs under any app listed above, e.g. spotify
+
+```python
+print(apis.api_docs.show_api_descriptions(app_name='spotify'))
+```
+
+# To get the specification of a particular api, e.g. spotify app's login api
+
+```python
+print(apis.api_docs.show_api_doc(app_name='spotify', api_name='login'))
+```
+
+Each code execution will produce an output that you can use in subsequent calls. Using these APIs, you can now generate code, that I will execute, to solve the task. Let's start with the task
+
+My name is: Sam Carter. My personal email is sam.carter@example.com and phone number is 555-0100.
+Task: How many playlists do I have in my Spotify playlist library?
+
+ASSISTANT:
+Okay. Let's first find which APIs are available to use in Spotify.
+
+```python
+print(apis.api_docs.show_api_descriptions(app_name='spotify'))
+```
+
+USER:
+Output:
+```
+[
+ ...
+ "login : Login to your account.",
+ "logout : Logout from your account.",
+ ...
+]
+```
+
+ASSISTANT:
+Okay. Looks like I can use the `login` api. Let's find its specifications.
+
+```python
+print(apis.api_docs.show_api_doc(app_name='spotify', api_name='login'))
+```
+
+USER:
+Output:
+```
+{
+ "app_name": "spotify",
+ "api_name": "login",
+ "path": "/spotify/auth/token",
+ "method": "POST",
+ "description": "Login to your account.",
+ "parameters": [
+  {
+   "name": "username",
+   "type": "string",
+   "required": true,
+   "description": "Your account email.",
+   "default": null,
+   "constraints": []
+  },
+  {
+   "name": "password",
+   "type": "string",
+   "required": true,
+   "description": "Your account password.",
+   "default": null,
+   "constraints": []
+  }
+ ],
+ "response_schemas": {
+  "success": {
+   "access_token": "string",
+   "token_type": "string"
+  },
+  "failure": {
+   "message": "string"
+  }
+ }
+}
+```
+
+ASSISTANT:
+I need the supervisor's username and password. Let's see if any app can help me get that.
+
+```python
+print(apis.api_docs.show_app_descriptions())
+```
+
+USER:
+Output:
+```
+{{ app_descriptions }}
+```
+
+ASSISTANT:
+Looks like the supervisor app could help me with that.
+
+```python
+print(apis.api_docs.show_api_descriptions(app_name='supervisor'))
+```
+
+USER:
+Output:
+```
+[
+ ...
+ "show_account_passwords : Show your supervisor's account passwords."
+ ...
+]
+```
+
+ASSISTANT:
+I can use `show_account_passwords` to get the passwords. So I will call that and save it in a variable for future reference.
+
+```python
+passwords=apis.supervisor.show_account_passwords()
+print(passwords)
+```
+
+USER:
+Output:
+```
+[
+ {
+  "account_name": "spotify",
+  "password": "dummy_spotify_pass"
+ },
+ {
+  "account_name": "file_system",
+  "password": "dummy_fs_pass"
+ },
+ ...
+]
+```
+
+ASSISTANT:
+So the Spotify password is an entry in the `passwords` list with the account_name=spotify. I can use it to login now with the provided email address: sam.carter@example.com.
+
+```python
+spotify_password = [account_password["account_name"] == "spotify" for account_password in passwords][0]["password"]
+login_result = apis.spotify.login(username='sam.carter@example.com', password=spotify_password)
+print(login_result)
+```
+
+USER:
+Output:
+```
+{
+ "access_token": "fake_access_token",
+ "token_type": "Bearer"
+}
+```
+
+ASSISTANT:
+First, let's check the available APIs for the Spotify app.
+
+```python
+print(apis.api_docs.show_api_descriptions(app_name='spotify'))
+```
+
+USER:
+Output:
+```
+[
+ ...
+ {
+  "name": "show_playlist_library",
+  "description": "Search or show a list of playlists in your playlist library."
+ },
+ {
+  "name": "show_playlist",
+  "description": "Get detailed information about a specific playlist. You can view your own playlists or others' playlists if they are public."
+ },
+ ...
+]
+```
+
+ASSISTANT:
+The `show_playlist_library` API seems to be the one we need. Let's check its specifications.
+
+```python
+print(apis.api_docs.show_api_doc(app_name='spotify', api_name='show_playlist_library'))
+```
+
+USER:
+Output:
+```
+{
+ "app_name": "spotify",
+ "api_name": "show_playlist_library",
+ "path": "/spotify/library/playlists",
+ "method": "GET",
+ "description": "Search or show a list of playlists in your playlist library.",
+ "parameters": [
+  {
+   "name": "access_token",
+   "type": "string",
+   "required": true,
+   "description": "Access token obtained from spotify app login.",
+   "default": null,
+   "constraints": []
+  },
+  {
+   "name": "query",
+   "type": "string",
+   "required": false,
+   "description": "The search query string.",
+   "default": "",
+   "constraints": []
+  },
+  {
+   "name": "is_public",
+   "type": "boolean",
+   "required": false,
+   "description": "Whether to show public playlists or private playlists.",
+   "default": null,
+   "constraints": []
+  },
+  {
+   "name": "page_index",
+   "type": "integer",
+   "required": false,
+   "description": "The index of the page to return.",
+   "default": 0,
+   "constraints": [
+    "value >= 0"
+   ]
+  },
+  {
+   "name": "page_limit",
+   "type": "integer",
+   "required": false,
+   "description": "The maximum number of results to return per page.",
+   "default": 5,
+   "constraints": [
+    "value >= 1, <= 20"
+   ]
+  },
+  {
+   "name": "sort_by",
+   "type": "string",
+   "required": false,
+   "description": "The attribute to sort the library playlists by prefixed with +/- to reflect ascending/descending. Valid attributes: created_at and title. ...",
+   "default": null,
+   "constraints": []
+  }
+ ],
+ "response_schemas": {
+  "success": [
+   {
+    "playlist_id": 1,
+    "title": "string",
+    "is_public": true,
+    "rating": 0.0,
+    "like_count": 1,
+    "review_count": 1,
+    "owner": {
+     "name": "string",
+     "email": "user@example.com"
+    },
+    "created_at": "2019-01-01T00:00:00",
+    "song_ids": [
+     1
+    ]
+   }
+  ],
+  "failure": {
+   "message": "string"
+  }
+ }
+}
+```
+
+ASSISTANT:
+I need to page through all the playlists to get the list of playlists and save it in `playlists`.
+
+```python
+page_index = 0
+playlists = []
+while page_index < 10:
+  playlist_page = apis.spotify.show_playlist_library(access_token=spotify_access_token, page_index=page_index)
+  if playlist_page:
+    playlists.extend(playlist_page)
+    page_index += 1
+  else:
+    break
+num_playlists = len(playlists)
+print(num_playlists)
+```
+
+USER:
+Output:
+```
+23
+```
+
+ASSISTANT:
+Now that the task is completed, I can call apis.supervisor.complete_task(). Since this task has an answer to be returned, I will pass that as an argument.
+
+```python
+apis.supervisor.complete_task(answer=num_playlists)
+```
+
+USER:
+Output:
+Marked the active task complete.
+
+USER:
+**Key instructions**:
+
+A. General instructions:
+
+- Act fully on your own. You must make all decisions yourself and never ask me or anyone else to confirm or clarify. Your role is to solve the task, not to bounce questions back, or provide me directions to follow.
+- You have full access -- complete permission to operate across my connected accounts and services.
+- Never invent or guess values. For example, if I ask you to play a song, do not assume the ID is 123. Instead, look it up properly through the right API.
+- Never leave placeholders; don't output things like "your_username". Always fill in the real value by retrieving it via APIs (e.g., Supervisor app for credentials).
+- When I omit details, choose any valid value. For example, if I ask you to buy something but don't specify which payment card to use, you may pick any one of my available cards.
+- Avoid collateral damage. Only perform what I explicitly ask for. Example: if I ask you to buy something, do not delete emails, return the order, or perform unrelated account operations.
+
+B. App-specific instructions:
+
+- All my personal information (biographical details, credentials, addresses, cards) is stored in the Supervisor app, accessible via its APIs.
+- Any reference to my friends, family or any other person or relation refers to the people in my phone's contacts list.
+- Always obtain the current date or time, from Python function calls like `datetime.now()`, or from the phone app's get_current_date_and_time API, never from your internal clock.
+- All requests are concerning a single, default (no) time zone.
+- For temporal requests, use proper time boundaries, e.g., when asked about periods like "yesterday", use complete ranges: 00:00:00 to 23:59:59.
+- References to "file system" mean the file system app, not the machine's OS. Do not use OS modules or functions.
+- Paginated APIs: Always process all results, looping through the page_index. Don't stop at the first page.
+
+C. Code-operation instructions
+
+- Make sure to end code blocks with ``` followed by a newline(\n).
+- Remember, you can use the variables in your code in subsequent code blocks.
+- Remember that the email addresses, access tokens and variables (e.g. spotify_password) in the example above are not valid anymore.
+- Always look at API specifications (using apis.api_docs.show_api_doc) before calling an API.
+- Write small chunks of code and only one chunk of code in every step. Make sure everything is working correctly before making any irreversible changes.
+- The Python environment supports the standard library. But system-level operations that may access or affect OS files, processes, etc., are not allowed and will raise an error if called.
+- To interact with apps, only use the provided app APIs, and not the corresponding Python packages, e.g., do NOT use `spotipy` for Spotify.
+- The provided API documentation has both the input arguments and the output JSON format. Use this information when making API calls and parsing their outputs.
+
+D. Task-completion instructions:
+
+You must call the `apis.supervisor.complete_task` API after completing the task.
+- If an answer is needed, e.g., for "How many songs are in the Spotify queue?", call it with the appropriate answer argument value.
+- If no answer is required, e.g., for "Start my Spotify music player.", omit the answer argument (or set it to None/null).
+- The task is doable, but if you cannot find a way, you can call it with status="fail" to exit with failure.
+
+When the answer is given:
+- Keep answers minimal. Return only the entity, number, or direct value requested - not full sentences.
+  E.g., for the song title of the current playing track, return just the title.
+- Numbers must be numeric and not in words.
+  E.g., for the number of songs in the queue, return "10", not "ten".

tools/soak/scenarios/__init__.py

@@ -0,0 +1,10 @@
+"""压测场景：把某一个具体 benchmark 接到 PolyLoop 的公共接缝上。
+
+一个「场景」要交的东西是固定的四样：一个决策解释器（模型输出 → 动作）、一个动作执行器
+（动作 → 观察）、一份上下文（提示词装配成消息序列）、以及一份装好的 `RunRequest`。压测
+的驱动侧只认这四样，换一个 benchmark 就是换一个本包下的模块。
+
+**这一层允许知道 benchmark 的领域细节**，`src/polyloop/` 不允许（CLAUDE.md §1.1）。库里
+一个业务词都不能有，而「AppWorld 的动作是一段 Python 代码」这种话必须写在某个地方，写在
+这里。
+"""

tools/soak/scenarios/appworld.py

@@ -0,0 +1,617 @@
+"""AppWorld 场景：把 `tools/soak/appworld.py` 的环境层包成 PolyLoop 认得的四样东西。
+
+交出去的是一个决策解释器（`AppWorldParser`）、一个动作执行器（`AppWorldExecutor`）、一份
+上下文（`build_context`）和一份装好的运行请求（`build_run_request`）。库本体只认这几个接缝，
+对「AppWorld 的动作是一段 Python 代码」「主管的名字要写进提示词」一无所知，那些全在这里。
+
+**解析协议逐字复刻 dissect 的 `harness/agent/parser.py`，但一行都不 import 它。** PolyLoop
+是被 dissect 依赖的库，反向 import 下游是硬约束（CLAUDE.md §1.2，由 import-linter 断言）。
+那个文件在这里只当协议文档看：两条正则、判定顺序、五条纠错说明的中文原文都照抄，实现是
+独立写的。复刻而不是复用的代价是它会漂移，收益是压测负载与 dissect 的生产解析行为逐位一致，
+于是压出来的解析失败率、步数分布对 dissect 有参考价值。
+
+**提示词模板来自 AppWorld 官方（Apache-2.0），经 dissect 转手拷进 `tools/soak/prompts/`。**
+它用行首独占一行的 `USER:` / `ASSISTANT:` 标记切分消息，不是消息数组——三百多行、十来个
+来回的示例演示当成一篇连续文本读和改，比在结构化配置里写数组可维护得多，而且能与官方模板
+逐行 diff。
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from polyloop.ports import Action, InvalidDecision, ParsedReply
+from polyloop.session import RunRequest
+from polyloop.tools import ToolRegistry
+from polyloop.types import (
+    ActionOutcome,
+    ActionStatus,
+    Budget,
+    Context,
+    Message,
+    ReplayPolicy,
+    Role,
+    SyntheticObservations,
+    TextBlock,
+)
+from tools.soak.appworld import AppWorldError
+
+if TYPE_CHECKING:
+    from polyloop.types import ModelReply
+    from tools.soak.appworld import AppWorldPool, AppWorldSession
+
+
+class AppWorldScenarioError(RuntimeError):
+    """这个场景装配不出来：模板缺变量、格式切不出消息、或环境给的必需信息是空的。
+
+    只有一个错误类型，因为调用方对这几种情形的处置完全一样——压测跑不起来，人得去看一眼。
+    分成三个类只会让每一处 `except` 都要写三个名字。
+    """
+
+
+# ---------------------------------------------------------------------------
+# 一、决策解释：从模型输出里抽出要执行的代码
+# ---------------------------------------------------------------------------
+
+#: 完整的围栏：```python 到配对的 ```，且**闭合围栏必须独占一行**。逐字照抄 dissect 的
+#: `harness/agent/parser.py:44-46`。
+#:
+#: 「独占一行」这个约束是在修一个真 bug。写成非贪婪的 ```` ```python\s*\n(.*?)``` ```` 时，
+#: 匹配遇到代码内部的三反引号就提前收尾——`print("```")` 会被截成 `print("`，一段语法错的
+#: 残码被判成「解析成功」交给环境，同时模型输出被截断到那个伪结尾，后半段真正的代码从历史里
+#: 消失。模型下一轮看到自己被腰斩的输出加一个语法错误，通常重写同一段代码，于是稳定循环到
+#: 步数耗尽。AppWorld 里 agent 写邮件正文、生成 markdown 报告、打印带反引号的 API 文档摘录
+#: 时都会踩到。
+#:
+#: 中间那段 `(?:\r?\n[ \t]*)?` 是可选的，为的是让空块 ```` ```python\n``` ```` 也能被识别成
+#: 「块存在但内容为空」，从而给出对症的纠错说明。每处空白类都带上 `\r`，因为模型的输出可能
+#: 是 Windows 换行——漏掉它的话 ```` ```python\r\n ```` 匹配不上，整条输出退到「未闭合」分支
+#: 去，`\r` 还会留在代码末尾。
+_FULL_BLOCK = re.compile(r"```python[ \t\r]*\n(.*?)(?:\r?\n[ \t]*)?```[ \t\r]*(?=\n|$)", re.DOTALL)
+
+#: 未闭合的围栏：```python 之后一直到文本结束。逐字照抄 `parser.py:52`。
+#:
+#: 它是给「模型配了 stop 序列」那种配置兜底的：AppWorld 官方 CI 配的 stop 是 "```\n"，那样
+#: 模型的输出会正好停在闭合围栏之前，看起来就是缺了结尾。没有这条兜底，那种配置每一步都会
+#: 解析失败。压测这一路不配 stop 序列，所以它基本不会被触发。
+_PARTIAL_BLOCK = re.compile(r"```python[ \t\r]*\n(.*)", re.DOTALL)
+
+#: 追加给纠错说明的一句提示。逐字照抄 `parser.py:59`。
+#:
+#: 多围栏策略取 first_only，模型输出会被截断到第一个围栏结束，它下一轮看到的是自己被腰斩的
+#: 回复。不说明原因的话，它多半会以为输出被网络截断了而原样重发，白烧一步。
+_ONLY_FIRST_BLOCK_HINT = "另外请注意：每一步只写一个代码块，我只会执行第一个。"
+
+#: 未闭合围栏里什么都没有。逐字照抄 `parser.py:114-117`。
+_EMPTY_PARTIAL_BLOCK = (
+    "你的回复里有一个 ```python 代码块，但它是空的。请在代码块里写出这一步要执行的 Python 代码。"
+)
+
+#: 闭合围栏后面还跟着别的东西。逐字照抄 `parser.py:128-131`。
+_FENCE_NOT_ON_ITS_OWN_LINE = (
+    "你的代码块结尾的 ``` 后面还跟了别的内容。请让结尾的 ``` 单独占一行，"
+    "后面直接换行。" + _ONLY_FIRST_BLOCK_HINT
+)
+
+#: 一个围栏都没有。逐字照抄 `parser.py:141-144`。
+_NO_CODE_BLOCK = (
+    "你的回复里没有可执行的代码块。请把这一步要执行的 Python 代码放进一个 "
+    "```python 开头、``` 结尾的代码块里；每一步只写一个代码块。"
+)
+
+#: 第一个完整围栏是空的。逐字照抄 `parser.py:161-165`。
+_EMPTY_FIRST_BLOCK = (
+    "你的回复里第一个 ```python 代码块是空的。"
+    "请在代码块里写出这一步要执行的 Python 代码。" + _ONLY_FIRST_BLOCK_HINT
+)
+
+
+class AppWorldParser:
+    """把模型输出解释成一段要在 AppWorld 里执行的 Python 代码。满足 `polyloop.ports.DecisionParser`。
+
+    **多围栏策略固定为 first_only**，不做成构造参数。dissect 把它做成显式配置项是因为那是
+    它要扫动的实验因子（对照组 smolagents 把所有代码块拼起来执行）；压测只需要一份确定的
+    负载，多留一个开关只会让「这次压的是哪一路」多一个变量。
+
+    **不显式继承那个 Protocol**：结构化子类型不需要继承，继承会让这个模块 import 一个只用来
+    做名义基类的东西。
+    """
+
+    def parameters(self) -> Mapping[str, str]:
+        """上报可复现参数：动作语言与多围栏策略。
+
+        两者都是「模型看得见的东西」的一部分——换一种策略续跑，前几步与后几步对同一份模型
+        输出的解释就不一样了，而两段轨迹在文件里看起来是同一次运行。
+        """
+        return {"kind": "appworld_code_fence", "multi_block_policy": "first_only"}
+
+    def parse(self, reply: ModelReply) -> ParsedReply:
+        """抽出要执行的代码。**同步，且对任何输入都不抛异常。**
+
+        判定顺序照 dissect 的 `parse_code_action`（`parser.py:90-145`）：先找全部完整围栏，
+        没有就退到未闭合围栏，再没有就判失败。三条支路里只有第一条会截断回填历史的那段文本。
+
+        解释不出来时返回「无效决策」而不是抛异常（`design/0007` 决策三）：那一步照常留痕、
+        纠错说明回喂给模型、循环继续。
+        """
+        output = reply.content
+
+        matches = list(_FULL_BLOCK.finditer(output))
+        if matches:
+            first = matches[0]
+            code = first.group(1).strip()
+            if not code:
+                return ParsedReply(
+                    history_text=output,
+                    decision=InvalidDecision(explanation=_EMPTY_FIRST_BLOCK),
+                )
+            # 截断到第一个围栏结束：模型常在代码块后面自行编造「执行结果」，留着的话下一轮
+            # 它会把那段幻想当成真发生过的事。
+            return ParsedReply(
+                history_text=output[: first.end()],
+                decision=Action(text=code, tool_call=None),
+            )
+
+        partial = _PARTIAL_BLOCK.search(output)
+        if partial:
+            code = partial.group(1).strip()
+            if not code:
+                return ParsedReply(
+                    history_text=output,
+                    decision=InvalidDecision(explanation=_EMPTY_PARTIAL_BLOCK),
+                )
+            if "```" in code:
+                # 走到这一支说明：文本里有闭合围栏，但它不满足「独占一行」，例如
+                # ```` ```python\nprint(1)\n``` done ````。这不是「被 stop 序列截断的未闭合
+                # 块」，是格式不规范。若照兜底分支处理，围栏连同后面的散文会一起被当成代码
+                # 交给环境执行，而且判定为解析成功——环境返回一个语法错误，事后统计会把它
+                # 算成「模型写错代码」，而实际是我们解析错了。
+                return ParsedReply(
+                    history_text=output,
+                    decision=InvalidDecision(explanation=_FENCE_NOT_ON_ITS_OWN_LINE),
+                )
+            # **这里与被复刻的 dissect 实现有一处刻意的不同**：dissect 在这一支给历史文本补回
+            # 结尾的三反引号（`parser.py:133-136`），让那条 assistant 消息形态完整；我们不补，
+            # 因为公共契约要求 `len(history_text) <= len(reply.content)`
+            # （`tests/contract/test_decision_parser.py:30`），补一个字符就违约。代价可以接受：
+            # 压测不给模型配 stop 序列，这条路径基本不触发。代码抽取本身照旧兜底。
+            return ParsedReply(
+                history_text=output,
+                decision=Action(text=code, tool_call=None),
+            )
+
+        return ParsedReply(
+            history_text=output,
+            decision=InvalidDecision(explanation=_NO_CODE_BLOCK),
+        )
+
+
+# ---------------------------------------------------------------------------
+# 二、动作执行：把代码交给容器里的 AppWorld 环境
+# ---------------------------------------------------------------------------
+
+
+class AppWorldExecutor:
+    """在一个已经开好的 AppWorld 会话里执行代码。满足 `polyloop.ports.ActionExecutor`。
+
+    一个实例绑定一道题的一次会话，跟 `RunRequest` 一样是每次运行一个。
+    """
+
+    def __init__(self, *, session: AppWorldSession) -> None:
+        """Args:
+        session: 已经实例化好一道题的会话，由 `AppWorldPool.session()` 产出。
+        """
+        self._session = session
+
+    @property
+    def env_executions(self) -> int:
+        """环境自己数的实际执行次数。
+
+        **它必须来自环境，不能是本地计数器。** 故障注入那一步要验的是「库说走了几步」与
+        「环境真的被执行了几次」对不对得上，用本地计数器的话两边都出自我们自己，验的就成了
+        我们和我们自己一致。
+        """
+        return self._session.n_executions
+
+    def parameters(self) -> Mapping[str, str]:
+        """上报可复现参数：动作接口的种类与这次做的是哪道题。
+
+        **刻意不放 `base_url`。** 容器端口是从池里租来的，每次运行、甚至同一次运行的每道题
+        都不同；进了参数快照，续跑时的逐字段比对必然报参数漂移，而那是一次假故障。
+        """
+        return {"kind": "appworld_session", "task_id": self._session.task_id}
+
+    async def execute(self, action: Action) -> ActionOutcome:
+        """执行一段代码，顺带问一次环境「做完了没有」。
+
+        **代码报错、超时、语法错都不是异常**——环境把 traceback 或超时提示原样放在返回文本里，
+        那是给模型看的正常观察，所以这一路是 `EXECUTED` 而不是 `ENV_ERROR`。只有环境本身坏了
+        （连不上、HTTP 非 2xx、返回体不是约定形状）才会抛 `AppWorldError`。这条分界照
+        `polyloop.tools.RegistryExecutor` 的同名分支填。
+
+        **`CancelledError` 不捕获**（CLAUDE.md §1.6）：`AppWorldError` 是 `RuntimeError` 的
+        子类，下面那个 `except` 接不到继承自 `BaseException` 的取消。别的未预料异常同样不接——
+        压测就是要看见它们，吞掉只会让故障变成一条内容古怪的观察（CLAUDE.md §1.7）。
+        """
+        try:
+            observation = await self._session.execute(action.text)
+            # 完成信号只有环境给得出（agent 调没调过 `apis.supervisor.complete_task()`），
+            # 而它每一步都可能翻转，所以每一步都问一次。
+            env_reported_completion = await self._session.is_done()
+        except AppWorldError as exc:
+            return ActionOutcome(
+                status=ActionStatus.ENV_ERROR,
+                observation=str(exc),
+                # 这段文本是环境（或与环境通信的那一层）给的，不是库合成的占位。
+                observation_is_synthetic=False,
+                env_reported_completion=False,
+                observation_truncated_chars=0,
+            )
+
+        return ActionOutcome(
+            status=ActionStatus.EXECUTED,
+            observation=observation,
+            observation_is_synthetic=False,
+            env_reported_completion=env_reported_completion,
+            # 这一层不截断观察。AppWorld 的输出由环境侧自己限长，我们原样转交。
+            observation_truncated_chars=0,
+        )
+
+
+# ---------------------------------------------------------------------------
+# 三、提示词装配：模板文本 → 消息序列
+# ---------------------------------------------------------------------------
+
+#: 提示词模板所在目录。两个文件来自 AppWorld 官方（Apache-2.0），经 dissect 转手，内容不改。
+PROMPTS_DIR = Path(__file__).resolve().parents[1] / "prompts" / "appworld"
+
+#: 行首独占一行的角色标记。逐字照抄 dissect 的 `harness/agent/context.py:31`。
+#: 官方模板只用 USER / ASSISTANT 两种（ReAct 不使用 system 角色），SYSTEM 一并认得。
+_ROLE_MARK = re.compile(r"^(USER|ASSISTANT|SYSTEM):\s*$", re.MULTILINE)
+
+_ROLE_OF = {"USER": Role.USER, "ASSISTANT": Role.ASSISTANT, "SYSTEM": Role.SYSTEM}
+
+#: 模板里的变量占位符：`{{ name }}` 与 `{{ name.attr }}` 两种形态，花括号内两侧允许有空格。
+#:
+#: **刻意不用 jinja2**：当前环境里没有它，而这两个模板一共只用到 `app_descriptions`、
+#: `main_user` 的四个字段和 `instruction`。为一个五行的替换规则拉一个模板引擎进来，压测就多
+#: 了一个装不上就跑不起来的依赖。
+_PLACEHOLDER = re.compile(r"\{\{\s*([A-Za-z_]\w*)(?:\.([A-Za-z_]\w*))?\s*\}\}")
+
+#: 主管信息里提示词要用到的四个字段。
+_SUPERVISOR_FIELDS = ("first_name", "last_name", "email", "phone_number")
+
+
+def split_by_role(text: str, *, where: str) -> list[tuple[Role, str]]:
+    """按行首角色标记把整段模板切成 `(角色, 正文)` 列表。
+
+    Args:
+        text: 模板原文。
+        where: 出错信息里用来指认是哪份模板。
+
+    Raises:
+        AppWorldScenarioError: 没有任何角色标记、第一个标记之前有内容、或某一段是空的。
+    """
+    marks = list(_ROLE_MARK.finditer(text))
+    if not marks:
+        raise AppWorldScenarioError(f"{where} 里没有任何 USER:/ASSISTANT: 行首标记，切不出消息列表")
+    if text[: marks[0].start()].strip():
+        raise AppWorldScenarioError(f"{where} 的第一个角色标记之前有内容，无法归属到某个角色")
+
+    blocks: list[tuple[Role, str]] = []
+    for index, mark in enumerate(marks):
+        end = marks[index + 1].start() if index + 1 < len(marks) else len(text)
+        content = text[mark.end() : end].strip()
+        if not content:
+            raise AppWorldScenarioError(f"{where} 里第 {index + 1} 个消息块是空的")
+        blocks.append((_ROLE_OF[mark.group(1)], content))
+    return blocks
+
+
+def render_template(text: str, variables: Mapping[str, object], *, where: str) -> str:
+    """把 `{{ 变量 }}` 换成取值。
+
+    **变量缺失显式报错，不当空串。** 渲染成空串的表现是提示词里凭空少一段，模型成绩下降，
+    而没有任何地方会告诉你原因。
+
+    **残留的 `{{` 也报错。** 静默留下一个没替换的占位符会让模型看见字面量 `{{ instruction }}`，
+    而那在轨迹里看起来只是模型表现差。这道检查落在**模板**上而不是渲染结果上：合法占位符
+    先从模板里抠掉，剩下的文本里还有 `{{` 就说明写了一个形态不对的占位符（`{{ a-b }}`、
+    少一个右花括号之类）。查渲染结果的话，插值内容里恰好带 `{{` 就会误报一次。
+
+    Raises:
+        AppWorldScenarioError: 有形态不对的占位符、变量没提供、或取值不是字符串。
+    """
+    residual = _PLACEHOLDER.sub("", text)
+    if "{{" in residual:
+        raise AppWorldScenarioError(
+            f"{where} 里有形态不对的占位符：抠掉全部合法的 {{{{ 变量 }}}} 之后仍残留 '{{{{'。"
+            f"渲染器只认 {{{{ name }}}} 与 {{{{ name.attr }}}} 两种形态"
+        )
+
+    def replace(match: re.Match[str]) -> str:
+        return _lookup(match.group(1), match.group(2), variables, where=where)
+
+    return _PLACEHOLDER.sub(replace, text)
+
+
+def render_to_messages(
+    text: str, variables: Mapping[str, object], *, where: str
+) -> tuple[Message, ...]:
+    """先按角色标记切分**模板原文**，再逐段渲染。
+
+    **顺序不能反。** 反过来会开一个「内容能改变结构」的口子：`run_prefix.txt` 里的
+    `{{ app_descriptions }}` 位于一个代码围栏内部，如果先整段渲染再切分，那么只要环境返回的
+    文本里出现独占一行的 `USER:`，前缀就会被切出多余的消息——而且不报错。先切后渲染，插值
+    内容永远只能落在某一条消息**内部**，改不了消息边界。
+
+    Raises:
+        AppWorldScenarioError: 切不出消息、渲染出问题、或某一段渲染后是空的。
+    """
+    messages: list[Message] = []
+    for index, (role, content) in enumerate(split_by_role(text, where=where)):
+        rendered = render_template(content, variables, where=f"{where} 第 {index + 1} 个消息块")
+        rendered = rendered.strip()
+        if not rendered:
+            raise AppWorldScenarioError(f"{where} 里第 {index + 1} 个消息块渲染后是空的")
+        messages.append(Message(role=role, content=(TextBlock(text=rendered),)))
+    return tuple(messages)
+
+
+def build_context(*, session: AppWorldSession, app_descriptions: str) -> Context:
+    """把两份模板渲染成 PolyLoop 的上下文。
+
+    分两段是因为供应商按前缀缓存计费：`run_prefix.txt` 整个压测里逐字节不变，`item_suffix.txt`
+    每道题都不同。把逐题变化的东西排到前面会让缓存静默失效。
+
+    Args:
+        session: 已开好的会话，题面与主管信息从它取。
+        app_descriptions: 全 run 恒定的可用 app 清单，见 `load_app_descriptions`。
+
+    Raises:
+        AppWorldScenarioError: 模板文件缺失、切不出消息、或主管信息有空字段。
+    """
+    run_prefix, item_suffix = _read_templates()
+    supervisor = _verified_supervisor(session)
+    return Context(
+        run_level=render_to_messages(
+            run_prefix,
+            {"app_descriptions": app_descriptions},
+            where="run_prefix.txt",
+        ),
+        goal_level=render_to_messages(
+            item_suffix,
+            {"main_user": supervisor, "instruction": session.instruction},
+            where="item_suffix.txt",
+        ),
+    )
+
+
+def _read_templates() -> tuple[str, str]:
+    """读两份模板原文。缺文件直接报错，不兜底。"""
+    texts: list[str] = []
+    for name in ("run_prefix.txt", "item_suffix.txt"):
+        path = PROMPTS_DIR / name
+        if not path.is_file():
+            raise AppWorldScenarioError(f"提示词模板缺文件：{path}")
+        texts.append(path.read_text(encoding="utf-8"))
+    return texts[0], texts[1]
+
+
+def _verified_supervisor(session: AppWorldSession) -> dict[str, str]:
+    """取主管信息，并确认提示词要用的四个字段都非空。
+
+    单独校验是因为渲染器只拦得住「键不存在」，拦不住「键存在但值是 None 或空串」——后者会
+    被安静地渲染成字面量 `None` 或一段空白，于是提示词变成 `My name is: Jose None.`、
+    `phone number is `，而模型会照着这个去调 API 查一个不存在的人。这类故障不报错，只让
+    成绩变差。
+    """
+    supervisor = dict(session.supervisor)
+    missing = [key for key in _SUPERVISOR_FIELDS if not supervisor.get(key)]
+    if missing:
+        raise AppWorldScenarioError(
+            f"题目 {session.task_id} 的主管信息缺字段 {missing}（值为空或 None）。"
+            f"提示词要用它们介绍任务委托人，缺了会误导模型去查一个不存在的人"
+        )
+    return supervisor
+
+
+def _lookup(
+    name: str, attribute: str | None, variables: Mapping[str, object], *, where: str
+) -> str:
+    """取一个占位符的值。缺任何一环都报错。"""
+    if name not in variables:
+        raise AppWorldScenarioError(
+            f"{where} 用到了变量 {name!r}，但没有提供它；已提供的是 {sorted(variables)}"
+        )
+    value = variables[name]
+
+    if attribute is None:
+        if not isinstance(value, str):
+            raise AppWorldScenarioError(
+                f"{where} 的变量 {name!r} 要直接插进文本，取值必须是字符串，"
+                f"收到 {type(value).__name__}"
+            )
+        return value
+
+    if not isinstance(value, Mapping):
+        raise AppWorldScenarioError(
+            f"{where} 取 {name}.{attribute}，但 {name!r} 不是一份映射，收到 {type(value).__name__}"
+        )
+    if attribute not in value:
+        raise AppWorldScenarioError(
+            f"{where} 取 {name}.{attribute}，但 {name!r} 里没有这个键；实有 {sorted(value)}"
+        )
+    item = value[attribute]
+    if not isinstance(item, str):
+        raise AppWorldScenarioError(
+            f"{where} 的 {name}.{attribute} 取值必须是字符串，收到 {type(item).__name__}"
+        )
+    return item
+
+
+# ---------------------------------------------------------------------------
+# 四、run 级变量：可用 app 清单
+# ---------------------------------------------------------------------------
+
+#: 问环境要 app 清单的那段代码，出自官方提示词自己教的三个 API 之一。
+_APP_DESCRIPTIONS_CODE = "print(apis.api_docs.show_app_descriptions())"
+
+#: 缓存文件的默认位置。它进 `.gitignore`——这份文本是环境的产物，不是源码。
+DEFAULT_APP_DESCRIPTIONS_CACHE = (
+    Path(__file__).resolve().parents[1] / ".cache" / "app_descriptions.txt"
+)
+
+
+async def load_app_descriptions(
+    pool: AppWorldPool,
+    *,
+    task_id: str,
+    cache_path: Path | str = DEFAULT_APP_DESCRIPTIONS_CACHE,
+) -> str:
+    """取可用 app 的清单，优先读磁盘缓存。
+
+    它是 run 级变量：整个压测里所有任务共用同一份，而且它是提示词固定前缀的一部分，逐题重取
+    只会白占一个容器。第一次要开一个会话去问环境，之后直接读缓存。
+
+    Args:
+        pool: 已经 `start()` 过的环境入口。只有缓存不存在时才会用到它。
+        task_id: 借哪道题来开这个会话。问的是全局的 app 文档，跟具体是哪道题无关。
+        cache_path: 缓存文件。
+
+    Raises:
+        AppWorldError: 环境返回了空文本。空的 app 列表会让整批任务全部失败，而表现是
+            「模型不会用 API」——静默用空串的话，这个故障永远查不出来。
+        AppWorldScenarioError: 缓存文件存在但内容是空的。
+    """
+    path = Path(cache_path)
+    if path.exists():
+        cached = path.read_text(encoding="utf-8")
+        if not cached.strip():
+            raise AppWorldScenarioError(
+                f"app 清单的缓存文件 {path} 是空的。删掉它重新取；"
+                f"空清单会让整批任务全部失败，而表现是「模型不会用 API」"
+            )
+        return cached
+
+    async with pool.session(task_id) as session:
+        descriptions = await session.execute(_APP_DESCRIPTIONS_CODE)
+    if not descriptions.strip():
+        raise AppWorldError(
+            f"环境对 {_APP_DESCRIPTIONS_CODE} 返回了空文本。它是提示词里的 app 清单，"
+            f"空的话整批任务都会失败，而表现是「模型不会用 API」"
+        )
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(descriptions, encoding="utf-8")
+    return descriptions
+
+
+# ---------------------------------------------------------------------------
+# 五、装配：预算、合成观察、运行请求
+# ---------------------------------------------------------------------------
+
+#: 环境输出回喂给模型时的包装格式。逐字照抄 dissect 的
+#: `config/agent/appworld.yaml:31`，它本身照抄 AppWorld 官方——提示词里那整段十来个来回的
+#: 示例演示都按这个格式写，改一个字符就会让示例与实况对不上，而那正是这份提示词最想避免的事。
+OBSERVATION_TEMPLATE = "Output:\n```\n{observation}\n```\n\n"
+
+#: 取消进来之后留给库写结束记录的秒数。
+_CANCEL_GRACE_SECONDS = 5.0
+
+
+def build_budget() -> Budget:
+    """AppWorld 这一路的预算。四个数字的来历各不相同，都写在下面。"""
+    return Budget(
+        # 40 而不是 AppWorld 官方当前实验配置的 50，照 dissect 的生产配置
+        # （`config/agent/appworld.yaml:19`）：它对齐的是 ASSAY 公开的裸 ReAct 配置。它必须
+        # 小于等于环境侧的 max_interactions（`AppWorldPool` 的构造参数，默认也是 40），两者
+        # 构成双保险。
+        max_steps=40,
+        # dissect 没有这一维。AppWorld 每步至多一个动作，所以取值等于 max_steps 在行为上无害:
+        # 动作数永远追不上步数，这一维不会先于步数耗尽。
+        max_actions=40,
+        # 3 而不是 1，是因为偶发一次格式失手不该毁掉整道题——纠错说明回喂之后模型通常能自己
+        # 纠正；3 而不是 10，是因为真陷进去之后每多一步都是白烧钱，40 步的预算经不起这种消耗。
+        # 照 `appworld.yaml:36`。
+        max_consecutive_parse_failures=3,
+        # 提示词字符数的硬上限，**不是截断阈值，是安全网**。压测不做上下文截断，但也不能让
+        # 提示词无限增长到撞上模型的上下文窗口。取值宽松，让它极少触发。照 `appworld.yaml:43`。
+        max_prompt_chars=400000,
+    )
+
+
+def build_synthetic_observations() -> SyntheticObservations:
+    """库合成、回填给模型看的那三段观察。
+
+    前两段照 dissect 的 `harness/agent/loop.py:35,38` 逐字。第三段 dissect 没有对应物：它的
+    循环里没有「动作被拒绝」这一档，因为那一档只在工具注册表分发时才可能出现，而 AppWorld
+    的动作是代码不是工具调用。`AppWorldExecutor` 永远不返回 `NOT_EXECUTED`，所以这段文本在
+    这条路上不会被用到；字段是必填的，写一段与另外两段同样形态的话，比起随手填个空串，出现
+    在轨迹里时至少还看得懂。
+    """
+    return SyntheticObservations(
+        action_rejected="[这一步的动作被拒绝，没有交给环境执行]",
+        env_failed="[环境故障，这一步的动作没有被执行]",
+        model_call_failed="[模型调用失败，这一步没有产出]",
+    )
+
+
+def build_run_request(
+    *,
+    run_id: str,
+    session: AppWorldSession,
+    app_descriptions: str,
+    model_binding: Mapping[str, str],
+) -> RunRequest:
+    """把一道题装配成一次运行的请求。
+
+    Args:
+        run_id: 这次运行的标识，同时是日志主键。
+        session: 已经实例化好这道题的会话。
+        app_descriptions: run 级的 app 清单，见 `load_app_descriptions`。
+        model_binding: 项目自己的标识，库不解释、原样透传给每次模型调用。
+
+    Raises:
+        AppWorldScenarioError: 上下文装配不出来。
+    """
+    return RunRequest(
+        run_id=run_id,
+        budget=build_budget(),
+        action_executor=AppWorldExecutor(session=session),
+        # 空注册表：AppWorld 的动作是一段代码，不走工具调用这条路。它与 `action_executor`
+        # 同时存在不是重复——不注册工具的项目就是传一个空注册表加一个环境句柄。
+        tools=ToolRegistry(),
+        context=build_context(session=session, app_descriptions=app_descriptions),
+        # 压测不注入 Skill 条目：注入是另一条正交的路，混进来会让「这次压的是什么」变多一维。
+        injections={},
+        model_binding=model_binding,
+        # **NEVER，不可改成 SAFE。** AppWorld 的代码执行有真实副作用（转账、下单、发消息），
+        # 而恢复时我们并不知道被打断的那次调用有没有真的执行到环境里。状态未知时重放一次
+        # 转账，损坏的是环境状态本身，事后从轨迹里分辨不出来。这条是故障注入那一步要验的
+        # 核心不变量。
+        model_replay_policy=ReplayPolicy.NEVER,
+        observation_template=OBSERVATION_TEMPLATE,
+        cancel_grace_seconds=_CANCEL_GRACE_SECONDS,
+    )
+
+
+__all__ = [
+    "DEFAULT_APP_DESCRIPTIONS_CACHE",
+    "OBSERVATION_TEMPLATE",
+    "PROMPTS_DIR",
+    "AppWorldExecutor",
+    "AppWorldParser",
+    "AppWorldScenarioError",
+    "build_budget",
+    "build_context",
+    "build_run_request",
+    "build_synthetic_observations",
+    "load_app_descriptions",
+    "render_template",
+    "render_to_messages",
+    "split_by_role",
+]

tools/soak/tests/__init__.py

@@ -0,0 +1,14 @@
+"""压测 harness 自己的测试。
+
+**它不进 `make ci`**：`pyproject.toml` 的 `testpaths` 只收 `tests/`，那一套守的是库的公共
+承诺，而这里守的是压测工具。跑法是显式给路径::
+
+    PYTHONUNBUFFERED=1 conda run --live-stream -n PolyLoop \\
+        python -m pytest tools/soak/tests/ -p no:cacheprovider -q
+
+分开是刻意的：压测工具坏了只影响我们自己，不该让库的 CI 变红；反过来，库的 CI 也不该因为
+压测这边加了个依赖就装不上。
+
+这里的测试全都不起容器、不打模型——环境层与场景层的接缝上用的是假对象。真起容器那条路是
+`tools/soak/check_appworld.py`，它单独手动跑。
+"""

tools/soak/tests/test_appworld_scenario.py

@@ -0,0 +1,571 @@
+"""AppWorld 场景适配器的测试。
+
+分三块：解析器、提示词装配、动作执行器。
+
+解析器那块的前五条是 `tests/contract/test_decision_parser.py` 那份公共契约的逐条复刻。契约
+套件本身是给下游接自己的实现用的（在自己的 `conftest.py` 里覆盖 fixture），压测这边不接那
+套装配、只把五条断言照着写一遍——它是任何新适配器的准入标准，压测的适配器也是适配器。
+
+执行器那块用一个假会话，不起容器：这里要验的是「环境返回什么 → 结果对象怎么填」这个映射，
+而那个映射与容器里发生了什么无关。真起容器的那条链路由 `tools/soak/check_appworld.py` 走。
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from polyloop.ports import Action, InvalidDecision
+from polyloop.types import ActionStatus, ModelReply, ReplayPolicy, Role
+from tools.soak.appworld import AppWorldError
+from tools.soak.scenarios.appworld import (
+    OBSERVATION_TEMPLATE,
+    AppWorldExecutor,
+    AppWorldParser,
+    AppWorldScenarioError,
+    build_budget,
+    build_context,
+    build_run_request,
+    build_synthetic_observations,
+    render_to_messages,
+    split_by_role,
+)
+
+
+def _reply(content: str) -> ModelReply:
+    return ModelReply(call_id="call-1", content=content, thinking="")
+
+
+#: 一条正常的、能解析出动作的回复。
+_ACTION_REPLY = _reply(
+    "先看看有哪些 app。\n\n```python\nprint(apis.api_docs.show_app_descriptions())\n```\n"
+)
+
+#: 一条解析不出任何东西的回复。
+_INVALID_REPLY = _reply("我觉得应该先登录 spotify，但我不确定要用哪个接口。")
+
+
+# ---------------------------------------------------------------------------
+# 一、解析器：公共契约的五条
+# ---------------------------------------------------------------------------
+
+
+def test_parse_is_synchronous():
+    """`parse` 是同步的，不是协程（契约一）。"""
+    parsed = AppWorldParser().parse(_ACTION_REPLY)
+
+    assert not hasattr(parsed, "__await__")
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        "```python\nprint(1)\n```\n然后我看到了 42。",  # 成功且要截断
+        "```python\n```",  # 空的完整围栏
+        "```python\nprint(1)",  # 未闭合围栏兜底
+        "```python\nprint(1)\n``` done",  # 闭合围栏后跟了内容
+        "什么代码都没有",  # 一个围栏都没有
+    ],
+)
+def test_history_text_never_grows(content):
+    """`history_text` 不会比模型原文长（契约二）。
+
+    这一条正是我们与被复刻的 dissect 实现分道的地方：dissect 在「未闭合围栏兜底」那一支给
+    历史文本补回结尾的三反引号，补一个字符就会让这条断言失败。
+    """
+    parsed = AppWorldParser().parse(_reply(content))
+
+    assert isinstance(parsed.history_text, str)
+    assert len(parsed.history_text) <= len(content)
+
+
+def test_invalid_decision_carries_a_non_empty_explanation():
+    """无效决策的说明文本非空（契约三）。"""
+    parsed = AppWorldParser().parse(_INVALID_REPLY)
+
+    assert isinstance(parsed.decision, InvalidDecision)
+    assert isinstance(parsed.decision.explanation, str)
+    assert parsed.decision.explanation != ""
+
+
+def test_action_carries_its_trace_form():
+    """动作分支带着这一步在轨迹里长什么样（契约四）。"""
+    parsed = AppWorldParser().parse(_ACTION_REPLY)
+
+    assert isinstance(parsed.decision, Action)
+    assert isinstance(parsed.decision.text, str)
+    assert parsed.decision.text == "print(apis.api_docs.show_app_descriptions())"
+    # AppWorld 的动作是代码，不是工具调用。
+    assert parsed.decision.tool_call is None
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        "",
+        "```",
+        "``````",
+        "```python",
+        "```python\r\n```",
+        "```PYTHON\nprint(1)\n```",
+        "\n\n\n",
+        '```python\nprint("```")\n```',
+        "```python\n```python\n```",
+        "{{ 不是模板 }} ```python``` ```",
+    ],
+)
+def test_any_input_returns_a_decision_rather_than_raising(content):
+    """怎么怪的输入都返回一个决策，不抛异常（契约五）。
+
+    库对解析失败的处置是「留痕 + 回喂纠错说明 + 继续」；抛异常会让整次运行以一个跟模型无关
+    的理由炸掉，而压测里模型什么都吐得出来。
+    """
+    parsed = AppWorldParser().parse(_reply(content))
+
+    assert isinstance(parsed.decision, Action | InvalidDecision)
+    if isinstance(parsed.decision, InvalidDecision):
+        assert parsed.decision.explanation != ""
+
+
+# ---------------------------------------------------------------------------
+# 二、解析器：协议细节
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        "```\nprint(1)\n```",
+        "```bash\nls -la\n```",
+        "```py\nprint(1)\n```",
+    ],
+)
+def test_only_python_fences_count(content):
+    """只认 ```python，裸围栏与别的语言一律判失败。
+
+    放宽会踩到一类静默故障：模型贴一段 bash 或一段 JSON 出来，我们把它当 Python 交给环境，
+    环境回一个语法错，事后统计会记成「模型写错代码」。
+    """
+    parsed = AppWorldParser().parse(_reply(content))
+
+    assert isinstance(parsed.decision, InvalidDecision)
+
+
+def test_first_block_wins_and_the_rest_is_cut_off():
+    """多个围栏取第一个，并把回填历史的文本截断到它结束。
+
+    截断不是可有可无：模型常在代码块后面自行编造「执行结果」，留着的话下一轮它会把那段幻想
+    当成真发生过的事。
+    """
+    content = (
+        "第一步：\n\n```python\nfirst()\n```\n\n"
+        "Output:\n```\n我编的执行结果\n```\n\n"
+        "第二步：\n\n```python\nsecond()\n```\n"
+    )
+
+    parsed = AppWorldParser().parse(_reply(content))
+
+    assert isinstance(parsed.decision, Action)
+    assert parsed.decision.text == "first()"
+    assert parsed.history_text.endswith("```python\nfirst()\n```")
+    assert "second()" not in parsed.history_text
+    assert "我编的执行结果" not in parsed.history_text
+
+
+def test_closing_fence_must_be_alone_on_its_line():
+    """闭合围栏后面跟着别的内容判失败，并给出对症说明。
+
+    照兜底分支处理的话，围栏连同后面的散文会一起被当成代码交给环境，而且判定为解析成功。
+    """
+    parsed = AppWorldParser().parse(_reply("```python\nprint(1)\n``` 好了"))
+
+    assert isinstance(parsed.decision, InvalidDecision)
+    assert "单独占一行" in parsed.decision.explanation
+
+
+def test_code_containing_backticks_survives():
+    """代码内部的三反引号不该把围栏提前切断。
+
+    这是那条「闭合围栏必须独占一行」的正则要修的原 bug：非贪婪匹配会把 `print("` 当成完整
+    代码交出去，剩下半截连同真正的代码从历史里消失。
+    """
+    parsed = AppWorldParser().parse(_reply('```python\nprint("```")\n```\n'))
+
+    assert isinstance(parsed.decision, Action)
+    assert parsed.decision.text == 'print("```")'
+
+
+def test_empty_fence_is_a_failure_with_its_own_explanation():
+    """空围栏判失败，说明与「没有围栏」那条不同。
+
+    五条纠错说明各自对症，压成一句会改掉模型收到的信息，它的纠错行为也就跟着变。
+    """
+    empty = AppWorldParser().parse(_reply("```python\n```"))
+    none_at_all = AppWorldParser().parse(_reply("我不知道该写什么"))
+
+    assert isinstance(empty.decision, InvalidDecision)
+    assert isinstance(none_at_all.decision, InvalidDecision)
+    assert empty.decision.explanation != none_at_all.decision.explanation
+    assert "空的" in empty.decision.explanation
+
+
+def test_unclosed_fence_still_yields_code():
+    """未闭合的围栏仍然抽得出代码，且历史文本不比原文长。
+
+    dissect 在这一支补回结尾的三反引号，我们不补——公共契约要求 history_text 不长于原文。
+    """
+    content = "我来查一下。\n\n```python\nprint(apis.api_docs.show_app_descriptions())"
+
+    parsed = AppWorldParser().parse(_reply(content))
+
+    assert isinstance(parsed.decision, Action)
+    assert parsed.decision.text == "print(apis.api_docs.show_app_descriptions())"
+    assert parsed.history_text == content
+    assert not parsed.history_text.endswith("```")
+
+
+def test_parser_parameters_pin_the_action_language():
+    """参数快照里写明动作语言与多围栏策略。"""
+    assert AppWorldParser().parameters() == {
+        "kind": "appworld_code_fence",
+        "multi_block_policy": "first_only",
+    }
+
+
+# ---------------------------------------------------------------------------
+# 三、提示词装配
+# ---------------------------------------------------------------------------
+
+
+def test_split_by_role_finds_the_boundaries():
+    """行首独占一行的标记才是边界，正文里的 `USER:` 不是。"""
+    text = "USER:\n你好\n\nASSISTANT:\n我在\n\nUSER:\n这里提到 USER: 但不在行首\n"
+
+    blocks = split_by_role(text, where="样例")
+
+    assert [role for role, _ in blocks] == [Role.USER, Role.ASSISTANT, Role.USER]
+    assert blocks[0][1] == "你好"
+    assert blocks[2][1] == "这里提到 USER: 但不在行首"
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        "这份模板一个角色标记都没有",
+        "开头就有内容\nUSER:\n你好\n",
+        "USER:\n\nASSISTANT:\n我在\n",
+    ],
+)
+def test_split_by_role_rejects_malformed_templates(text):
+    """没有标记、标记之前有内容、某一段是空的，三种都报错。"""
+    with pytest.raises(AppWorldScenarioError):
+        split_by_role(text, where="样例")
+
+
+def test_variables_are_rendered_per_block():
+    """`{{ 变量 }}` 与 `{{ 变量.字段 }}` 两种形态都认，花括号内两侧允许有空格。"""
+    text = "USER:\n{{greeting}} {{ user.first_name }} {{  user.last_name  }}\n"
+
+    messages = render_to_messages(
+        text,
+        {"greeting": "你好", "user": {"first_name": "Sam", "last_name": "Carter"}},
+        where="样例",
+    )
+
+    assert len(messages) == 1
+    assert messages[0].content[0].text == "你好 Sam Carter"
+
+
+def test_missing_variable_is_an_error_not_an_empty_string():
+    """变量没提供就报错。渲染成空串的表现是提示词里凭空少一段，模型成绩下降而没人知道原因。"""
+    with pytest.raises(AppWorldScenarioError, match="instruction"):
+        render_to_messages("USER:\n任务：{{ instruction }}\n", {}, where="样例")
+
+
+def test_missing_attribute_is_an_error():
+    """点号取字段时字段不存在同样报错。"""
+    with pytest.raises(AppWorldScenarioError):
+        render_to_messages(
+            "USER:\n{{ user.email }}\n", {"user": {"first_name": "Sam"}}, where="样例"
+        )
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        "USER:\n你好 {{ 中文变量 }}\n",
+        "USER:\n你好 {{ a-b }}\n",
+        "USER:\n你好 {{ name }\n",
+    ],
+)
+def test_leftover_double_braces_are_an_error(text):
+    """形态不对的占位符要当场报错，不能原样留给模型看。
+
+    静默留下一个 `{{ instruction }}` 会让模型看见字面量占位符，而那在轨迹里看起来只是模型
+    表现差。
+    """
+    with pytest.raises(AppWorldScenarioError):
+        render_to_messages(text, {"name": "Sam"}, where="样例")
+
+
+def test_interpolated_content_cannot_move_message_boundaries():
+    """插值内容里出现独占一行的 `USER:` 不会切出多余的消息。
+
+    先切后渲染才有这个性质。反过来的话，`run_prefix.txt` 里那个位于代码围栏内部的
+    `{{ app_descriptions }}` 只要带上一行 `USER:`，前缀就会多切出几条消息——而且不报错。
+    """
+    text = "USER:\n```\n{{ app_descriptions }}\n```\n\nASSISTANT:\n收到\n"
+    payload = "spotify: 音乐\nUSER:\nvenmo: 转账"
+
+    messages = render_to_messages(text, {"app_descriptions": payload}, where="样例")
+
+    assert len(messages) == 2
+    assert messages[0].role is Role.USER
+    assert messages[1].role is Role.ASSISTANT
+    assert "USER:" in messages[0].content[0].text
+
+
+# ---------------------------------------------------------------------------
+# 四、假会话与执行器
+# ---------------------------------------------------------------------------
+
+
+class _FakeSession:
+    """一个假的 `AppWorldSession`：只实现场景层用得到的那几个成员，不碰容器。"""
+
+    def __init__(
+        self,
+        *,
+        output: str = "42\n",
+        done: bool = False,
+        raises: BaseException | None = None,
+        instruction: str = "How many playlists do I have?",
+        supervisor: dict[str, str] | None = None,
+    ) -> None:
+        self.task_id = "82e2fac_1"
+        self.instruction = instruction
+        self.supervisor = (
+            supervisor
+            if supervisor is not None
+            else {
+                "first_name": "Sam",
+                "last_name": "Carter",
+                "email": "sam.carter@example.com",
+                "phone_number": "555-0100",
+            }
+        )
+        self.n_executions = 0
+        self._output = output
+        self._done = done
+        self._raises = raises
+        self.executed: list[str] = []
+
+    async def execute(self, code: str) -> str:
+        self.executed.append(code)
+        if self._raises is not None:
+            raise self._raises
+        self.n_executions += 1
+        return self._output
+
+    async def is_done(self) -> bool:
+        return self._done
+
+
+async def test_normal_execution_is_an_executed_outcome():
+    """代码跑完了就是「已执行」，观察是环境的原样输出。"""
+    session = _FakeSession(output="23\n")
+    executor = AppWorldExecutor(session=session)
+
+    outcome = await executor.execute(Action(text="print(23)", tool_call=None))
+
+    assert outcome.status is ActionStatus.EXECUTED
+    assert outcome.observation == "23\n"
+    assert outcome.observation_is_synthetic is False
+    assert outcome.env_reported_completion is False
+    assert outcome.observation_truncated_chars == 0
+    assert session.executed == ["print(23)"]
+
+
+async def test_code_that_blows_up_is_still_an_executed_outcome():
+    """模型写的代码报错是一条正常观察，不是环境故障。
+
+    压成 `ENV_ERROR` 就等于把「模型写错了」和「环境坏了」混成同一件事，而停止判定对这两者
+    的处置完全不同。
+    """
+    traceback = "Traceback (most recent call last):\nNameError: name 'x' is not defined"
+    executor = AppWorldExecutor(session=_FakeSession(output=traceback))
+
+    outcome = await executor.execute(Action(text="print(x)", tool_call=None))
+
+    assert outcome.status is ActionStatus.EXECUTED
+    assert outcome.observation == traceback
+
+
+async def test_environment_reported_completion_is_forwarded():
+    """环境说做完了，这个事实要原样填进结果——它是「任务完成」这个停止原因的唯一证据来源。"""
+    executor = AppWorldExecutor(session=_FakeSession(done=True))
+
+    outcome = await executor.execute(Action(text="apis.supervisor.complete_task()", tool_call=None))
+
+    assert outcome.env_reported_completion is True
+    assert outcome.status is ActionStatus.EXECUTED
+
+
+async def test_environment_failure_becomes_env_error():
+    """`AppWorldError` 才是环境故障，走 `ENV_ERROR`，观察是那条错误文本本身。"""
+    executor = AppWorldExecutor(
+        session=_FakeSession(raises=AppWorldError("/execute 返回 HTTP 500"))
+    )
+
+    outcome = await executor.execute(Action(text="print(1)", tool_call=None))
+
+    assert outcome.status is ActionStatus.ENV_ERROR
+    assert outcome.observation == "/execute 返回 HTTP 500"
+    assert outcome.observation_is_synthetic is False
+    assert outcome.env_reported_completion is False
+    assert outcome.observation_truncated_chars == 0
+
+
+async def test_cancellation_passes_through():
+    """取消原样穿出去（CLAUDE.md §1.6）。
+
+    吞掉它的后果不是「取消失败」这么直白——是容器租约、连接和临时目录持续泄漏，而且一声
+    不吭。
+    """
+    executor = AppWorldExecutor(session=_FakeSession(raises=asyncio.CancelledError()))
+
+    with pytest.raises(asyncio.CancelledError):
+        await executor.execute(Action(text="print(1)", tool_call=None))
+
+
+async def test_unexpected_errors_are_not_swallowed():
+    """未预料的异常穿出去（CLAUDE.md §1.7）——压测就是要看见它。"""
+    executor = AppWorldExecutor(session=_FakeSession(raises=ZeroDivisionError("boom")))
+
+    with pytest.raises(ZeroDivisionError):
+        await executor.execute(Action(text="print(1)", tool_call=None))
+
+
+async def test_execution_count_comes_from_the_environment():
+    """执行次数转发环境自己的计数，不是本地计数器。
+
+    故障注入那一步要拿它跟库的自述对账，用本地计数器的话对的就是我们和我们自己。
+    """
+    session = _FakeSession()
+    executor = AppWorldExecutor(session=session)
+
+    assert executor.env_executions == 0
+    await executor.execute(Action(text="print(1)", tool_call=None))
+    assert executor.env_executions == 1
+    assert executor.env_executions == session.n_executions
+
+
+def test_executor_parameters_keep_the_port_out():
+    """参数快照里有题号，没有 base_url——端口每次运行都不同，进快照会让续跑必然报参数漂移。"""
+    parameters = AppWorldExecutor(session=_FakeSession()).parameters()
+
+    assert parameters == {"kind": "appworld_session", "task_id": "82e2fac_1"}
+    assert "base_url" not in parameters
+
+
+# ---------------------------------------------------------------------------
+# 五、整体装配
+# ---------------------------------------------------------------------------
+
+
+def test_context_is_built_from_the_vendored_templates():
+    """两份真模板切得出消息，题面与主管信息落在题级那一段。"""
+    session = _FakeSession(instruction="How many playlists do I have?")
+
+    context = build_context(session=session, app_descriptions="spotify: 音乐\nvenmo: 转账")
+
+    assert context.run_level[0].role is Role.USER
+    assert len(context.run_level) > 1
+    run_text = "\n".join(block.text for m in context.run_level for block in m.content)
+    assert "spotify: 音乐" in run_text
+    # 题面属于题级那一段：它逐题变化，混进 run 级会打断供应商的前缀缓存。
+    assert "How many playlists do I have?" not in run_text
+
+    assert len(context.goal_level) == 1
+    goal_text = context.goal_level[0].content[0].text
+    assert "How many playlists do I have?" in goal_text
+    assert "Sam Carter" in goal_text
+    assert "sam.carter@example.com" in goal_text
+    assert "555-0100" in goal_text
+
+
+@pytest.mark.parametrize("blank_field", ["first_name", "last_name", "email", "phone_number"])
+def test_blank_supervisor_fields_are_rejected(blank_field):
+    """主管信息里任何一个字段为空都报错。
+
+    渲染器只拦得住「键不存在」。空值会安静地渲染成一段空白，提示词变成 `phone number is `，
+    模型照着去查一个不存在的人——不报错，只让成绩变差。
+    """
+    supervisor = {
+        "first_name": "Sam",
+        "last_name": "Carter",
+        "email": "sam.carter@example.com",
+        "phone_number": "555-0100",
+    }
+    supervisor[blank_field] = ""
+
+    with pytest.raises(AppWorldScenarioError, match=blank_field):
+        build_context(session=_FakeSession(supervisor=supervisor), app_descriptions="spotify: 音乐")
+
+
+def test_budget_matches_the_production_configuration():
+    """预算四项照 dissect 的生产配置（max_actions 除外，AppWorld 每步至多一个动作）。"""
+    budget = build_budget()
+
+    assert budget.max_steps == 40
+    assert budget.max_actions == 40
+    assert budget.max_consecutive_parse_failures == 3
+    assert budget.max_prompt_chars == 400000
+
+
+def test_synthetic_observations_are_all_non_empty():
+    """三段合成观察都得有内容——它们会被模型看见。"""
+    synthetic = build_synthetic_observations()
+
+    assert synthetic.action_rejected
+    assert synthetic.env_failed
+    assert synthetic.model_call_failed
+
+
+def test_run_request_is_assembled_with_the_pinned_values():
+    """装出来的请求：空注册表、NEVER 重放、逐字照抄的观察格式。"""
+    session = _FakeSession()
+
+    request = build_run_request(
+        run_id="soak-0001",
+        session=session,
+        app_descriptions="spotify: 音乐",
+        model_binding={"scenario": "appworld"},
+    )
+
+    assert request.run_id == "soak-0001"
+    assert isinstance(request.action_executor, AppWorldExecutor)
+    assert request.tools.names() == ()
+    assert request.injections == {}
+    assert request.model_binding == {"scenario": "appworld"}
+    # AppWorld 的代码执行有真实副作用（转账、下单），状态未知时绝不重放。
+    assert request.model_replay_policy is ReplayPolicy.NEVER
+    assert request.observation_template == OBSERVATION_TEMPLATE
+    assert request.observation_template == "Output:\n```\n{observation}\n```\n\n"
+    assert request.cancel_grace_seconds == 5.0
+
+
+def test_run_request_snapshot_has_no_container_port():
+    """参数快照里不该出现容器端口，否则续跑必然报参数漂移。"""
+    request = build_run_request(
+        run_id="soak-0002",
+        session=_FakeSession(),
+        app_descriptions="spotify: 音乐",
+        model_binding={},
+    )
+
+    snapshot = request.parameter_snapshot()
+
+    assert not any("127.0.0.1" in value for value in snapshot.values())
+    assert "82e2fac_1" in snapshot.values()