命令执行 → 观察 → 轨迹步骤

执行结果封装为观测与轨迹步

stage 上下文: 该片段位于 stage-4.9 的主体后半段，承接前面已解析出的命令与完成信号，在命令执行后把终端输出整理成下一轮发送给模型的 observation。它同时把本轮代理消息、伪工具调用、观测结果与 token/cost 统计打包为一条 agent 轨迹步。作为本 stage 的首个已翻译单元，它对应“执行命令之后，形成可记录的单步试验事实”这一职责。

这段代码在干什么

本段先调用 self._execute_commands(commands, self._session) 取得 terminal_output，再依据 is_task_complete、self._pending_completion 与 feedback 生成发送回模型的 observation。随后它用 chat 当前累计值减去调用前基线，计算本步消耗，并把解析出的命令与任务完成声明转写为 ToolCall 记录；若开启 self._save_raw_content_in_trajectory，则只记录观测、不生成工具调用。最终产出不是返回值，而是对 self._pending_completion、可选的 self._last_response_model_name 的写入，以及向 self._trajectory_steps 追加一条 Step。

接口 · 参数 / IO

(self) -> None

• 参数: commands: list[?] — 本轮已解析出的命令序列；既传给 _execute_commands(...)，也决定是否生成 bash_command 型 ToolCall; is_task_complete: bool — 代理是否声称任务完成；控制完成确认分支、self._pending_completion 写入以及是否生成 mark_task_complete 调用; feedback: ? — 上一轮反馈文本；仅当其真值成立且包含 "WARNINGS:" 时被前缀进 observation; episode: ? — 用于构造 tool_call_id，如 call_{episode}_{i+1} 与 call_{episode}_task_complete; message_content: ? — 写入轨迹 Step.message 的代理消息正文; chat.total_cache_tokens: number — 与 tokens_before_cache 做差，得到本步缓存命中 token 数; chat.total_cost: number — 与 cost_before 做差，得到本步成本; chat.total_input_tokens: number — 与 tokens_before_input 做差，得到本步 prompt token 数; chat.total_output_tokens: number — 与 tokens_before_output 做差，得到本步 completion token 数; tokens_before_cache: number — 本次 LLM 调用前的缓存 token 基线; cost_before: number — 本次 LLM 调用前的成本基线; tokens_before_input: number — 本次 LLM 调用前的输入 token 基线; tokens_before_output: number — 本次 LLM 调用前的输出 token 基线; llm_response.model_name: str | None — 若存在则更新 self._last_response_model_name，并优先写入 Step.model_name; llm_response.reasoning_content: ? — 写入轨迹步的 reasoning_content; llm_response.prompt_token_ids: ? — 写入 Metrics.prompt_token_ids; llm_response.completion_token_ids: ? — 写入 Metrics.completion_token_ids; llm_response.logprobs: ? — 写入 Metrics.logprobs
• 读状态: self._session, self._pending_completion, self._save_raw_content_in_trajectory, self._trajectory_steps, self._model_name
• 返回: 无返回值；对外可见产出是派生出的 observation / tool_calls / metrics 被封装进一条新 Step 追加到 self._trajectory_steps，并伴随对完成确认状态与最近模型名的状态写入。
• 副作用: 调用 await self._execute_commands(commands, self._session) 执行命令并取得终端输出; 把 self._pending_completion 维持原值、设为 True 或设为 False; 把当前 self._pending_completion 快照到局部变量 was_pending_completion（本片段内未再使用）; 当 llm_response.model_name 为真值时写入 self._last_response_model_name; 向 self._trajectory_steps 追加一条 Step

执行流

1. 先以 commands 和 self._session 调用 self._execute_commands(...)，得到 timeout_occurred, terminal_output；随后立即把当前 self._pending_completion 复制到局部变量 was_pending_completion。
2. 构造 observation：若 is_task_complete 为真且 self._pending_completion 已经为真，则直接使用原始 terminal_output；若 is_task_complete 为真但此前未 pending，则把 self._pending_completion 设为 True，并改用 self._get_completion_confirmation_message(terminal_output)；若 is_task_complete 为假，则把 self._pending_completion 设为 False，再根据 feedback 是否含 "WARNINGS:" 选择在 self._limit_output_length(terminal_output) 前附加警告前缀或仅保留截断后的输出。
3. 从 chat 的累计计数减去 tokens_before_cache、cost_before，得到 cache_tokens_used 与 step_cost，供稍后写入 Metrics。
4. 初始化 tool_calls=None 与空的 observation_results；若 self._save_raw_content_in_trajectory 为假，则在非 raw 模式下为每个 commands 元素生成一个 function_name="bash_command" 的 ToolCall，参数只保留 cmd.keystrokes 与 cmd.duration_sec。
5. 继续在非 raw 模式下按精确分支追加 ObservationResult(content=observation)：有命令时，在所有命令型 ToolCall 创建完后追加一次；若 is_task_complete 为真，则额外追加一个 mark_task_complete 调用，并且仅在“没有命令”时补加 observation result；若 is_task_complete 为假且也没有命令，则仅追加 observation result。最后把 tool_calls 设为 tool_calls_list or None。
6. 若 self._save_raw_content_in_trajectory 为真，则不生成任何 ToolCall，保持 tool_calls 为 None，但仍然无条件追加一个 ObservationResult(content=observation)。
7. 构建并追加 Step：若 llm_response.model_name 存在则先更新 self._last_response_model_name，然后以当前轨迹长度生成 step_id，填入时间戳、source="agent"、模型名、message_content、llm_response.reasoning_content、前述 tool_calls 与 Observation(results=observation_results)，以及由 chat 基线差分得到的 Metrics；其中 cached_tokens 与 cost_usd 仅在差值大于 0 时写入，否则写 None。

源码

            timeout_occurred, terminal_output = await self._execute_commands(
                commands,
                self._session,
            )

            # Capture the pending completion state before potentially modifying it
            was_pending_completion = self._pending_completion

            # Construct the observation (what gets sent back to the LLM)
            if is_task_complete:
                if self._pending_completion:
                    observation = terminal_output
                else:
                    self._pending_completion = True
                    observation = self._get_completion_confirmation_message(
                        terminal_output
                    )
            else:
                self._pending_completion = False
                if feedback and "WARNINGS:" in feedback:
                    observation = (
                        f"Previous response had warnings:\n{feedback}\n\n"
                        f"{self._limit_output_length(terminal_output)}"
                    )
                else:
                    observation = self._limit_output_length(terminal_output)

            # Record the step in trajectory
            cache_tokens_used = chat.total_cache_tokens - tokens_before_cache
            step_cost = chat.total_cost - cost_before

            # Create tool_calls array from commands and task completion
            # Note: Although Terminus 2 doesn't offer native tool calling (it uses text-based command parsing),
            # we represent parsed commands as tool calls for better trajectory analysis and compatibility with tooling.
            # However, when raw_content mode is enabled, we skip tool_calls generation to preserve the raw LLM response.
            tool_calls: list[ToolCall] | None = None
            observation_results: list[ObservationResult] = []

            if not self._save_raw_content_in_trajectory:
                # Only create tool_calls when NOT in raw_content mode
                tool_calls_list: list[ToolCall] = []

                if commands:
                    for i, cmd in enumerate(commands):
                        tool_call_id = f"call_{episode}_{i + 1}"
                        tool_calls_list.append(
                            ToolCall(
                                tool_call_id=tool_call_id,
                                function_name="bash_command",
                                arguments={
                                    "keystrokes": cmd.keystrokes,
                                    "duration": cmd.duration_sec,
                                },
                            )
                        )

                    # Add observation result after all tool calls are created
                    # Note: All commands share the same terminal output in this architecture,
                    # so we omit source_call_id to indicate the observation applies to the entire step.
                    observation_results.append(
                        ObservationResult(
                            content=observation,
                        )
                    )

                # Add task_complete as a tool call if the agent marked the task complete
                if is_task_complete:
                    task_complete_call_id = f"call_{episode}_task_complete"
                    tool_calls_list.append(
                        ToolCall(
                            tool_call_id=task_complete_call_id,
                            function_name="mark_task_complete",
                            arguments={},
                        )
                    )
                    # If there are no commands, we still need to add an observation result
                    if not commands:
                        observation_results.append(
                            ObservationResult(
                                content=observation,
                            )
                        )
                elif not commands:
                    # No commands and no task completion, just the observation
                    observation_results.append(
                        ObservationResult(
                            content=observation,
                        )
                    )

                tool_calls = tool_calls_list or None
            else:
                # In raw_content mode, just add observation without tool_calls
                observation_results.append(
                    ObservationResult(
                        content=observation,
                    )
                )

            # Build the step object using Pydantic models
            if llm_response.model_name:
                self._last_response_model_name = llm_response.model_name
            self._trajectory_steps.append(
                Step(
                    step_id=len(self._trajectory_steps) + 1,
                    timestamp=datetime.now(timezone.utc).isoformat(),
                    source="agent",
                    model_name=llm_response.model_name or self._model_name,
                    message=message_content,
                    reasoning_content=llm_response.reasoning_content,
                    tool_calls=tool_calls,
                    observation=Observation(results=observation_results),
                    metrics=Metrics(
                        prompt_tokens=chat.total_input_tokens - tokens_before_input,
                        completion_tokens=chat.total_output_tokens
                        - tokens_before_output,
                        cached_tokens=cache_tokens_used
                        if cache_tokens_used > 0
                        else None,
                        cost_usd=step_cost if step_cost > 0 else None,
                        prompt_token_ids=llm_response.prompt_token_ids,
                        completion_token_ids=llm_response.completion_token_ids,
                        logprobs=llm_response.logprobs,
                    ),
                )
            )

Non-obvious 设计决策

• 完成声明采用两段式确认：第一次遇到 is_task_complete 且 self._pending_completion 仍为假时，不直接把原始输出送回，而是切换到 self._get_completion_confirmation_message(...) 并把 pending 置真；只有再次进入完成分支且 pending 已真时，observation 才回落为原始 terminal_output。这体现出对“代理误报完成”的显式缓冲。
• 非完成路径会无条件把 self._pending_completion 设为 False，说明该确认状态不是粘性的；只要代理本轮没有再次声明完成，就立即清除先前的待确认状态。
• 轨迹层把文本解析出的命令重写为 ToolCall(function_name="bash_command", arguments={...})，但在 self._save_raw_content_in_trajectory 为真时完全跳过此转换，仅保留 ObservationResult。这一分支是显式的记录策略取舍：分析友好的结构化轨迹与原始响应保真不能同时兼得。
• ObservationResult 的追加位置不是统一出口，而是随 commands / is_task_complete 组合精细分支：代码刻意避免在有命令时为每条命令分别绑定输出，因为注释已说明该架构下所有命令共享同一份 terminal_output，因此 observation 只按“整步”记录一次。

上下游关系

• 调用方: Terminus2._run_agent_loop 的当前迭代主体（同一方法内前序代码）
• 核心被调用: self._execute_commands; self._get_completion_confirmation_message; self._limit_output_length; ToolCall; ObservationResult; Observation; Metrics; Step
• 配置/状态来源: self._pending_completion; self._save_raw_content_in_trajectory; self._trajectory_steps; self._model_name; self._session
• 结果去向: self._pending_completion; self._last_response_model_name; self._trajectory_steps

批量终端命令执行与超时收口

stage 上下文: 它位于 stage-4.9 的命令执行子步骤，负责把前面已解析好的 Command 列表真正送入活动中的 TmuxSession。触发条件是 _run_agent_loop 已拿到模型响应并完成命令解析，准备进入“执行→观测”环节。与同 stage 的后续兄弟逻辑相比，这里只产出原始执行结果 (timeout_occurred, terminal_output)，不负责 observation、ToolCall 或轨迹写入；那些由 _run_agent_loop 在拿到本函数返回值后继续完成。

这段代码在干什么

本函数顺序执行 commands 中的每条终端命令，对每条命令调用 session.send_keys(...)，并把命令自带的 duration_sec 作为最小等待时长。若某条命令触发 TimeoutError，函数立即停止后续执行，返回 True 与一段格式化后的超时文本；该文本包含超时命令、等待秒数，以及经 self._limit_output_length(...) 截断的当前终端输出。若整批命令都未超时，则返回 False 和最后一次 session.get_incremental_output() 的截断结果。本函数不写入任何实例状态，真正的下游用途是供 _run_agent_loop 继续构造 observation 与轨迹步。

接口 · 参数 / IO

(self, commands: list[Command], session: TmuxSession) -> tuple[bool, str]

• 参数: commands: list[Command] — 已从模型回复解析出的命令批次；每项提供 keystrokes 与 duration_sec; session: TmuxSession — 当前活动终端会话；负责发送按键并读取增量输出
• 读状态: self._timeout_template, self._limit_output_length
• 返回: 返回 (timeout_occurred, terminal_output)：前者表示本批命令是否在执行中遇到 TimeoutError，后者是供上层直接当作执行观测基础的字符串；超时时为模板化消息，正常时为当前终端增量输出的截断版本。
• 副作用: 向 session 发送命令按键序列; 从 session 读取增量终端输出

执行流

1. 按 for command in commands 的顺序遍历整批命令，逐条处理，不做并发调度或重排。
2. 对每条命令调用 await session.send_keys(command.keystrokes, block=False, min_timeout_sec=command.duration_sec)，把命令文本和该命令声明的持续时间直接交给会话层执行。
3. 如果某次 send_keys(...) 抛出 TimeoutError，立即进入 except 分支：先用 await session.get_incremental_output() 读取当前终端状态，再经 self._limit_output_length(...) 截断，并填入 self._timeout_template.format(...) 生成统一的超时反馈文本。
4. 发生超时时立刻 return True, <formatted_message>，因此后续命令不会继续执行，本批次也不会再读取“正常结束”输出。
5. 如果循环完整结束而没有超时，则在末尾调用一次 await session.get_incremental_output() 取得本批执行后的最新增量输出，并经 self._limit_output_length(...) 截断。
6. 最后返回 False, <limited_output>，把“未超时”和终端输出一起交给上层 _run_agent_loop 继续做 observation 构造。

源码

    async def _execute_commands(
        self,
        commands: list[Command],
        session: TmuxSession,
    ) -> tuple[bool, str]:
        """Execute a batch of commands in the terminal.

        Args:
            commands: List of commands to execute
            session: TmuxSession instance

        Returns:
            Tuple of (timeout_occurred, terminal_output)
        """
        for command in commands:
            try:
                await session.send_keys(
                    command.keystrokes,
                    block=False,
                    min_timeout_sec=command.duration_sec,
                )
            except TimeoutError:
                return True, self._timeout_template.format(
                    timeout_sec=command.duration_sec,
                    command=command.keystrokes,
                    terminal_state=self._limit_output_length(
                        await session.get_incremental_output()
                    ),
                )

        return False, self._limit_output_length(await session.get_incremental_output())

Non-obvious 设计决策

• 超时被转换为普通字符串返回，而不是继续向上抛异常：代码在 except TimeoutError 中直接生成模板化文本并返回 (True, ...)。这样上层循环可以把超时当作一种可消费的执行观测，而不是把整轮 agent 控制流打断。
• 一旦某条命令超时就停止整批执行：except 内的立即 return 明确选择“首个超时即收口”。这种取舍避免在终端已进入不确定状态时继续发送后续命令，降低连锁污染后续观察的风险。
• 无论超时还是正常结束，都统一经过 self._limit_output_length(...) 裁剪终端文本。该做法把输出长度控制前移到执行层，保证下游 _run_agent_loop 收到的是有界文本，避免超长终端内容直接扩散到 observation 与轨迹。
• 命令执行使用 block=False，但仍给出 min_timeout_sec=command.duration_sec：这里隐含的取舍是让会话层按“非阻塞发送 + 至少等待指定时长”的契约工作，从而由 Command 自身携带每条命令的观察窗口，而不是在本函数里自行实现睡眠或轮询逻辑。

上下游关系

• 调用方: Terminus2._run_agent_loop
• 核心被调用: TmuxSession.send_keys; TmuxSession.get_incremental_output; self._limit_output_length; self._timeout_template.format
• 配置/状态来源: commands[].keystrokes; commands[].duration_sec; self._timeout_template; session
• 结果去向: 返回给 Terminus2._run_agent_loop 的 timeout_occurred 分支判断; 返回给 Terminus2._run_agent_loop 的 terminal_output 作为 observation 原料; stage-4.9 后续的完成确认/普通观测分支; 后续轨迹步记录中的终端观测文本
• 同类 sibling: 与同 stage 的 _run_agent_loop 后半段形成前后衔接：本函数只负责拿到执行结果，兄弟逻辑再基于 is_task_complete 与 self._pending_completion 生成最终 observation、ToolCall 和 Step。