运行收尾

运行收尾的上下文汇总与落盘段

stage 上下文: 该片段位于 Terminus2.run 的 stage-5，对外部 context 进行集中回填，并在末尾调用 _dump_trajectory()。代码可见的职责是汇总主 chat 与 subagent 的 rollout/token/cost 信息，重建 context.metadata，并按条件附带完整消息列表。

这段代码在干什么

本段先把 self._chat.rollout_details 与 self._subagent_rollout_details 拼接后写入 context.rollout_details。随后将主 chat 的 token/cost 统计与 self._subagent_metrics 中的对应统计求和，写回 context.n_input_tokens、context.n_output_tokens、context.n_cache_tokens 与 context.cost_usd，其中总 cost 不大于 0 时显式写成 None。接着以新的字典覆盖 context.metadata，写入 n_episodes、api_request_times_msec、summarization_count，并在 self._store_all_messages 为真时额外放入 all_messages。最后调用 _dump_trajectory() 触发轨迹导出。

接口 · 参数 / IO

(self, context) -> None

• 参数: self: Terminus2 — 提供主 chat、subagent 汇总指标、运行计数与轨迹导出能力; context: 具有可写属性 rollout_details、n_input_tokens、n_output_tokens、n_cache_tokens、cost_usd、metadata 的对象 — 接收本次运行的对外汇总结果
• 读状态: self._chat.rollout_details, self._subagent_rollout_details, self._chat.total_input_tokens, self._subagent_metrics.total_prompt_tokens, self._chat.total_output_tokens, self._subagent_metrics.total_completion_tokens, self._chat.total_cache_tokens, self._subagent_metrics.total_cached_tokens, self._chat.total_cost, self._subagent_metrics.total_cost_usd, self._n_episodes, self._api_request_times, self._summarization_count, self._store_all_messages, self._chat.messages
• 返回: 此代码片段中未出现 return 语句。
• 副作用: 覆盖写入 context.rollout_details; 覆盖写入 context.n_input_tokens; 覆盖写入 context.n_output_tokens; 覆盖写入 context.n_cache_tokens; 覆盖写入 context.cost_usd; 以新字典覆盖 context.metadata; 在条件满足时写入 context.metadata["all_messages"]; 调用 self._dump_trajectory() 导出轨迹

执行流

1. 将 self._chat.rollout_details 与 self._subagent_rollout_details 连接后赋给 context.rollout_details，把主对话与 subagent 的 rollout 明细汇总到同一出口字段。
2. 分别把主 chat 的输入、输出、缓存 token 总数与 self._subagent_metrics 中对应总数相加，写入 context.n_input_tokens、context.n_output_tokens、context.n_cache_tokens。
3. 计算 total_cost = self._chat.total_cost + self._subagent_metrics.total_cost_usd，若结果大于 0 则写入 context.cost_usd，否则写入 None。
4. 构造一个新的 context.metadata 字典，包含 "n_episodes": self._n_episodes、"api_request_times_msec": self._api_request_times、"summarization_count": self._summarization_count。
5. 若 self._store_all_messages 为真，则把 self._chat.messages 直接放入 context.metadata["all_messages"]。
6. 调用 self._dump_trajectory()，执行轨迹导出。

源码

            context.rollout_details = (
                self._chat.rollout_details + self._subagent_rollout_details
            )

            # Include subagent metrics in context totals
            context.n_input_tokens = (
                self._chat.total_input_tokens
                + self._subagent_metrics.total_prompt_tokens
            )
            context.n_output_tokens = (
                self._chat.total_output_tokens
                + self._subagent_metrics.total_completion_tokens
            )
            context.n_cache_tokens = (
                self._chat.total_cache_tokens
                + self._subagent_metrics.total_cached_tokens
            )
            total_cost = self._chat.total_cost + self._subagent_metrics.total_cost_usd
            context.cost_usd = total_cost if total_cost > 0 else None
            context.metadata = {
                "n_episodes": self._n_episodes,
                "api_request_times_msec": self._api_request_times,
                "summarization_count": self._summarization_count,
            }
            if self._store_all_messages:
                context.metadata["all_messages"] = self._chat.messages

            # Dump trajectory to JSON
            self._dump_trajectory()

Non-obvious 设计决策

• cost 字段采用 total_cost if total_cost > 0 else None，把零或负值统一折叠为缺省态，而不是保留 0 或负数；这是代码中显式存在的表示决策。
• context.metadata 不是增量更新，而是直接用一个新字典整体替换；这意味着先前若已有其他 metadata 键，按该片段可见行为会被丢弃，只保留这里列出的键及条件性的 all_messages。
• api_request_times_msec 直接引用 self._api_request_times，条件分支中的 all_messages 直接引用 self._chat.messages；片段中未显示复制操作，因此后续若底层对象继续变动，context.metadata 中看到的也可能随之变化。
• 完整消息历史仅在 self._store_all_messages 为真时附加，默认 metadata 保持为较小的固定集合，避免无条件暴露或携带整段消息列表。

上下游关系

• 调用方: Terminus2.run
• 核心被调用: self._dump_trajectory
• 配置/状态来源: self._chat; self._subagent_rollout_details; self._subagent_metrics; self._n_episodes; self._api_request_times; self._summarization_count; self._store_all_messages
• 结果去向: context.rollout_details; context.n_input_tokens; context.n_output_tokens; context.n_cache_tokens; context.cost_usd; context.metadata; 轨迹导出目标（经由 _dump_trajectory()）

轨迹导出的薄封装方法

stage 上下文: 该函数本身只有一层转调：把实例状态中的 self._summarization_count 取出后，传给 _dump_trajectory_with_continuation_index(...)。相较于同 stage 中负责汇总 metadata 与 token 统计的 run() 收尾逻辑，这里不直接组装数据，只负责发起一次带固定参数来源的委托调用。

这段代码在干什么

此函数不接收除 self 外的任何输入，也不直接写入实例状态。它读取 self._summarization_count，并将该值作为唯一实参传给 self._dump_trajectory_with_continuation_index(...)。函数返回 None；若存在 JSON 生成或落盘副作用，也只会发生在被调函数内部而非这里。

接口 · 参数 / IO

(self) -> None

• 参数: self: ? — 提供 _summarization_count 状态与 _dump_trajectory_with_continuation_index 绑定方法的实例
• 读状态: self._summarization_count, self._dump_trajectory_with_continuation_index
• 返回: 返回 None；直接可见的产出仅是一次对 _dump_trajectory_with_continuation_index(...) 的调用
• 副作用: 间接触发 self._dump_trajectory_with_continuation_index(self._summarization_count) 的执行

执行流

1. 读取实例属性 self._summarization_count 作为本次调用的唯一参数来源。
2. 调用 self._dump_trajectory_with_continuation_index(self._summarization_count)，将实际处理委托给下层 helper。
3. 函数体结束后隐式返回 None。

源码

    def _dump_trajectory(self) -> None:
        """Dump trajectory data to JSON file following ATIF format."""
        self._dump_trajectory_with_continuation_index(self._summarization_count)

Non-obvious 设计决策

• 该方法将实参来源硬编码为实例内的 self._summarization_count，而不是把 continuation index 暴露为调用参数；这是源码中唯一可见的封装取舍。
• 函数体没有分支、循环、异常处理或额外状态改写；可见设计是保持为单纯转调包装层，把具体工作留给 _dump_trajectory_with_continuation_index(...)。

上下游关系

• 调用方: 未在给定源码片段中展示；外部若需按当前 self._summarization_count 发起轨迹导出，可调用本方法
• 核心被调用: Terminus2._dump_trajectory_with_continuation_index
• 配置/状态来源: self._summarization_count
• 结果去向: _dump_trajectory_with_continuation_index(...) 的形参位置; 本函数的直接返回值为 None
• 同类 sibling: Terminus2.run：兄弟收尾逻辑先汇总 context 中的 rollout、token 与 metadata，再调用轨迹导出；本函数自身只承担转调，不做这些汇总

轨迹 JSON 落盘与续段元数据组装点

stage 上下文: 该函数位于 run 收尾阶段所依赖的轨迹导出链路中，负责把当前内存中的 trajectory 组装成可落盘 JSON。与同 stage 的兄弟函数 _dump_trajectory 相比，后者只转发一个 continuation index；本函数才实际读取 self._context、拼装 Trajectory/Agent/FinalMetrics，并决定输出文件名。

这段代码在干什么

函数接收 continuation_index，读取 self._context、self._trajectory_steps、self._linear_history、self._summarization_count、agent/model 配置与 self.logs_dir，构造一个 Trajectory 对象并序列化到磁盘。若 self._context 不存在，则仅记录 warning 并直接返回。其真正产出不是返回值，而是写出 trajectory.json 或在 self._linear_history 为真且 continuation_index > 0 时写出 trajectory.cont-<n>.json，同时可能在 JSON 中写入 continuation_index 与 continued_trajectory_ref。

接口 · 参数 / IO

(self, continuation_index: int) -> None

• 参数: continuation_index: int — 当前轨迹段的续写编号；参与 agent.extra 与输出文件名选择，并与 self._summarization_count 一起决定是否生成 continued_trajectory_ref
• 读状态: self._context, self._context.n_input_tokens, self._context.n_output_tokens, self._context.n_cache_tokens, self._context.cost_usd, self._parser_name, self._temperature, self._llm_kwargs, self._linear_history, self._summarization_count, self._session_id, self._trajectory_steps, self._model_name, self.logs_dir, self.logger, self.name(), self.version()
• 返回: 返回 None；真正结果是向 self.logs_dir 下写入轨迹 JSON 文件，或在缺少 self._context / 写文件异常时仅记录日志。
• 副作用: 当 self._context 为空时调用 self.logger.warning(...); 向 self.logs_dir / "trajectory.json" 写文件，或在 self._linear_history 为真且 continuation_index > 0 时向 self.logs_dir / f"trajectory.cont-{continuation_index}.json" 写文件; 成功写入后调用 self.logger.debug(...); 写入失败时调用 self.logger.error(...)

执行流

1. 先检查 self._context；若其为假值，则记录 No context available, skipping trajectory dump 并直接返回，不再尝试构造或写入任何轨迹文件。
2. 基于 self._context.n_input_tokens、self._context.n_output_tokens、self._context.n_cache_tokens or 0、self._context.cost_usd if self._context.cost_usd else None 组装 FinalMetrics。
3. 构造 agent_extra：固定写入 parser=self._parser_name，在 self._temperature is not None 时补入 temperature，在 self._llm_kwargs 为真时补入 llm_kwargs，并且仅当 self._linear_history 为真且 continuation_index > 0 时写入 continuation_index。
4. 计算是否生成续段引用：先令 continued_trajectory_ref = None，随后仅当 self._linear_history 为真且 continuation_index < self._summarization_count 时，将其设为 trajectory.cont-<continuation_index + 1>.json。
5. 以 self._session_id、self.name()、self.version() or "unknown"、self._model_name、上一步生成的 agent_extra、self._trajectory_steps、final_metrics 与 continued_trajectory_ref 构造 Trajectory。
6. 根据文件名规则选择输出路径：若 self._linear_history 为真且 continuation_index > 0，使用 trajectory.cont-<n>.json；否则一律使用 trajectory.json。随后在 try 中调用 format_trajectory_json(trajectory.to_json_dict()) 生成字符串并写入文件；成功时记 debug，异常时捕获并记 error。

源码

    def _dump_trajectory_with_continuation_index(self, continuation_index: int) -> None:
        """Dump trajectory data to JSON file with specified continuation index.

        Args:
            continuation_index: The continuation index to use for filename and metadata.
                               For the initial trajectory, use 0.
                               For the first continuation, use 1, etc.
        """
        if not self._context:
            self.logger.warning("No context available, skipping trajectory dump")
            return

        # Construct the trajectory using Pydantic models for validation
        final_metrics = FinalMetrics(
            total_prompt_tokens=self._context.n_input_tokens,
            total_completion_tokens=self._context.n_output_tokens,
            total_cached_tokens=self._context.n_cache_tokens or 0,
            total_cost_usd=self._context.cost_usd if self._context.cost_usd else None,
        )

        agent_extra: dict[str, Any] = {
            "parser": self._parser_name,
        }
        if self._temperature is not None:
            agent_extra["temperature"] = self._temperature
        if self._llm_kwargs:
            agent_extra["llm_kwargs"] = self._llm_kwargs
        if self._linear_history and continuation_index > 0:
            agent_extra["continuation_index"] = continuation_index

        # Determine if this trajectory will be continued
        # In linear_history mode, when saving during summarization (i.e., continuation_index < _summarization_count),
        # this trajectory will have a continuation
        continued_trajectory_ref = None
        if self._linear_history and continuation_index < self._summarization_count:
            # This trajectory segment will be continued in the next file
            next_continuation_index = continuation_index + 1
            continued_trajectory_ref = f"trajectory.cont-{next_continuation_index}.json"

        trajectory = Trajectory(
            session_id=self._session_id,
            agent=Agent(
                name=self.name(),
                version=self.version() or "unknown",
                model_name=self._model_name,
                extra=agent_extra,
            ),
            steps=self._trajectory_steps,
            final_metrics=final_metrics,
            continued_trajectory_ref=continued_trajectory_ref,
        )

        # Determine trajectory filename based on continuation index
        if self._linear_history and continuation_index > 0:
            trajectory_path = (
                self.logs_dir / f"trajectory.cont-{continuation_index}.json"
            )
        else:
            trajectory_path = self.logs_dir / "trajectory.json"

        try:
            with open(trajectory_path, "w") as f:
                json_str = format_trajectory_json(trajectory.to_json_dict())
                f.write(json_str)
            self.logger.debug(f"Trajectory dumped to {trajectory_path}")
        except Exception as e:
            self.logger.error(f"Failed to dump trajectory: {e}")

Non-obvious 设计决策

• 函数把 self._context 的存在性作为整个导出的前置条件；若缺失则仅告警并返回，而不是抛异常中断调用方，这一容错分支在开头的显式检查中直接体现。
• FinalMetrics.total_cached_tokens 对 self._context.n_cache_tokens 采用 or 0，而 FinalMetrics.total_cost_usd 对 self._context.cost_usd 采用“假值即 None”的写法，因此缓存 token 缺省会落成数值 0，而 cost 的假值会在 JSON 模型里表现为 None。
• 线性历史相关元数据被拆成两层条件：agent_extra["continuation_index"] 与续段文件名都要求 self._linear_history and continuation_index > 0，但 continued_trajectory_ref 还额外要求 continuation_index < self._summarization_count；也就是说，是否写成 continuation 文件与是否声明“后面还有下一段”是两个独立判定。
• 文件写入被包在宽泛的 try/except Exception 中；异常不会向外传播，调用方只能通过 error 日志得知导出失败。

上下游关系

• 调用方: Terminus2._dump_trajectory（已知兄弟源码显示其将一个 continuation index 转发到本函数）
• 核心被调用: FinalMetrics(...); Agent(...); Trajectory(...); trajectory.to_json_dict(); format_trajectory_json(...); open(...); self.logger.warning(...); self.logger.debug(...); self.logger.error(...); self.name(); self.version()
• 配置/状态来源: self._context.n_input_tokens; self._context.n_output_tokens; self._context.n_cache_tokens; self._context.cost_usd; self._parser_name; self._temperature; self._llm_kwargs; self._linear_history; self._summarization_count; self._session_id; self._trajectory_steps; self._model_name; self.logs_dir
• 结果去向: Trajectory.final_metrics; Trajectory.agent.extra; Trajectory.continued_trajectory_ref; self.logs_dir / "trajectory.json"; self.logs_dir / f"trajectory.cont-{continuation_index}.json"; 日志输出（warning/debug/error）
• 同类 sibling: Terminus2._dump_trajectory：兄弟函数只负责把 continuation index 传入本函数，本函数才完成实际组装与落盘

会话收尾中的录屏停止与回收段

stage 上下文: 该段属于 stage-5 的 teardown 收尾路径，在运行循环无论以何种方式结束后执行。它不处理 trial 级统计汇总，而是聚焦 TmuxSession.stop 内部的录屏关闭与 cast 文件回收。相较同 stage 中 Terminus2.run 负责把指标写回 context、_dump_trajectory 负责最终轨迹落盘，这里负责让 asciinema 录制会话有机会正常结束，并把远端产物取回本地。

这段代码在干什么

本段仅在 self._remote_asciinema_recording_path 存在时启动录屏收尾流程：先记录一条 debug 日志，再向会话发送 Ctrl-D 以结束录制，并短暂等待写盘完成。若同时配置了 self._local_asciinema_recording_path，则确保本地目标目录存在，并调用 self.environment.download_file(...) 将远端 cast 文件下载到本地。函数本身不返回业务值，实际产出是远端录制停止以及本地录屏文件可用。

接口 · 参数 / IO

(self) -> None

• 参数: self: ? — TmuxSession 实例；提供录屏路径、发送按键能力、日志器与环境对象
• 读状态: self._remote_asciinema_recording_path, self._logger, self._local_asciinema_recording_path, self.environment
• 返回: 返回 None；真正产出是结束远端 asciinema 录制，并在配置了本地路径时把录制文件下载到本地。
• 副作用: 调用 self._logger.debug(...) 写出停止录制日志; 调用 self.send_keys(keys=['C-d'], min_timeout_sec=0.1) 向会话发送结束录制按键; 等待约 0.5 秒以给远端录制文件写盘留出时间; 在本地创建 self._local_asciinema_recording_path.parent 目录; 调用 self.environment.download_file(...) 从远端下载录制文件到本地

执行流

1. 先检查 self._remote_asciinema_recording_path；只有存在远端录制目标时，才进入后续关闭与回收流程，否则该段无动作结束。
2. 进入流程后，先通过 self._logger.debug("Stopping recording.") 记录调试信息，然后调用 await self.send_keys(keys=['C-d'], min_timeout_sec=0.1) 向录制所在会话发送 Ctrl-D，请求结束录制。
3. 随后局部导入 asyncio 并执行 await asyncio.sleep(0.5)，显式留出一个很短的等待窗口，让远端录制文件完成最终写入。
4. 如果 self._local_asciinema_recording_path 也存在，则先对其父目录执行 mkdir(parents=True, exist_ok=True)，确保本地落点可写。
5. 在本地路径存在的前提下，调用 await self.environment.download_file(source_path=str(self._remote_asciinema_recording_path), target_path=self._local_asciinema_recording_path)，把远端录制文件下载到本地，以便后续使用本地 cast 文件。

源码

        if self._remote_asciinema_recording_path:
            self._logger.debug("Stopping recording.")
            await self.send_keys(
                keys=["C-d"],
                min_timeout_sec=0.1,
            )

            # Wait a moment for the recording to finish writing
            import asyncio

            await asyncio.sleep(0.5)

            if self._local_asciinema_recording_path:
                self._local_asciinema_recording_path.parent.mkdir(
                    parents=True, exist_ok=True
                )
                # Ensure recording exists locally before merging markers
                await self.environment.download_file(
                    source_path=str(self._remote_asciinema_recording_path),
                    target_path=self._local_asciinema_recording_path,
                )

Non-obvious 设计决策

• 该段以 self._remote_asciinema_recording_path 作为总开关，意味着只有确实启用了远端录制时才尝试停止与下载，避免对未录制的会话做无意义的终止动作。
• 结束录制后没有立即下载，而是固定等待 0.5 秒；这是对异步写盘时序的显式容忍，避免在录制进程刚收到 Ctrl-D 时就读取到尚未完全落盘的文件。
• 本地回收被设计成可选：self._local_asciinema_recording_path 不存在时不会报错也不会创建默认路径，而是仅完成远端录制停止。这体现了“停止录制”和“本地保存副本”两个目标的解耦。
• 下载前显式创建父目录并使用 parents=True, exist_ok=True，选择的是幂等式目录准备策略；这样调用方无需预先保证目录结构存在，减少 teardown 末尾因路径缺失导致的失败面。

上下游关系

• 调用方: TmuxSession.stop; stage-5 teardown 路径中的会话关闭流程; 框架在试验收尾时触发的 tmux/asciinema 停止逻辑
• 核心被调用: self._logger.debug; self.send_keys; asyncio.sleep; Path.mkdir; self.environment.download_file
• 配置/状态来源: self._remote_asciinema_recording_path; self._local_asciinema_recording_path; self.environment
• 结果去向: 远端 asciinema 录制进程; 本地录制文件路径 self._local_asciinema_recording_path; 后续需要使用本地 cast 文件的 marker 合并流程; 日志输出通道
• 同类 sibling: Terminus2.run：同属 stage-5 收尾，但其职责是把 rollout/token/cost/metadata 汇总写回 context。; Terminus2._dump_trajectory：同一收尾阶段中的另一条落盘路径，负责轨迹 JSON 导出，而本段负责录屏 cast 文件回收。