响应解析

将模型文本整理为结构化回传

stage 上下文: 该片段位于 _handle_llm_interaction 的响应整理区段，职责是把 llm_response.content 交给 self._parser，再将解析结果改写为本方法的统一返回元组。它不在此处修改实例级寄存状态；除调试日志外，主要产出是命令列表、完成标记、反馈文本，以及解析器给出的附带字段。

这段代码在干什么

代码首先读取 llm_response.content 并调用 self._parser.parse_response(...) 获得 result。随后基于 result.error 与 result.warning 组装单个 feedback 字符串；若存在 warning，还会写一条 self.logger.debug(...)。最后，它把 result.commands 中的每个项转换为 Command，并将 duration 截断到 60 秒，连同 result.is_task_complete、解析器提供的 analysis / plan 以及原始 llm_response 一并返回。

接口 · 参数 / IO

(self, llm_response) -> tuple

• 参数: self: Terminus2 — 提供解析器 self._parser 与日志器 self.logger 的实例对象; llm_response: ? — 输入响应对象；本段代码读取其 .content 字段作为解析输入，并将该对象原样作为返回元组第六项返回
• 读状态: self._parser, self.logger
• 返回: 返回六元组：(commands, result.is_task_complete, feedback, result.analysis, result.plan, llm_response)；其中 commands 为新建的 Command 列表，analysis 与 plan 为解析器提供的不透明值，llm_response 未被改写。
• 副作用: 当 result.warning 为真时调用 self.logger.debug(...) 记录解析警告

执行流

1. 调用 self._parser.parse_response(llm_response.content)，以响应文本生成包含 commands、is_task_complete、error、warning、analysis、plan 等字段的 result。
2. 以空字符串初始化 feedback，然后按分支读取 result.error 与 result.warning：若有 error，先写入 ERROR: ...，并在同一分支下仅当也有 warning 时追加换行后的 WARNINGS: ...；若无 error 但有 warning，则仅写入 WARNINGS: ...。
3. 若 result.warning 存在，额外调用 self.logger.debug(f"Parser warnings: {result.warning}") 记录调试日志。
4. 遍历 result.commands，对每个 parsed_cmd 新建一个 Command；其中 keystrokes 直接沿用 parsed_cmd.keystrokes，duration_sec 使用 min(parsed_cmd.duration, 60) 做上限截断。
5. 返回六个值：构造好的 commands 列表、result.is_task_complete、feedback、result.analysis、result.plan，以及原始 llm_response。

源码

        result = self._parser.parse_response(llm_response.content)

        feedback = ""
        if result.error:
            feedback += f"ERROR: {result.error}"
            if result.warning:
                feedback += f"\nWARNINGS: {result.warning}"
        elif result.warning:
            feedback += f"WARNINGS: {result.warning}"

        if result.warning:
            self.logger.debug(f"Parser warnings: {result.warning}")

        commands = []
        for parsed_cmd in result.commands:
            commands.append(
                Command(
                    keystrokes=parsed_cmd.keystrokes,
                    duration_sec=min(parsed_cmd.duration, 60),
                )
            )

        return (
            commands,
            result.is_task_complete,
            feedback,
            result.analysis,
            result.plan,
            llm_response,
        )

Non-obvious 设计决策

• 反馈文本被折叠为单一字符串 feedback，而不是分别返回 error 与 warning 两个字段；从代码可见，这里采用固定前缀 ERROR: / WARNINGS: 做内嵌标记，以保留两类信息但维持单一返回槽位。
• 在 if result.error 分支中，格式化顺序是先错误、后警告，且二者之间显式插入换行；这说明当两者并存时，代码固定采用该输出顺序，而不是让 warning 独立占用其他位置。
• result.warning 既会进入 feedback，又会单独写入 self.logger.debug(...)；这是把返回给调用方的文本信息与调试日志并行保留，而不是二选一。
• 命令转换时对 parsed_cmd.duration 施加 min(..., 60) 上限，属于显式边界约束；代码只限制最大值，不在此处做最小值校正或其他合法性检查。

上下游关系

• 调用方: _handle_llm_interaction` 的其余未展示代码（当前片段所属函数）
• 核心被调用: self._parser.parse_response; self.logger.debug; Command
• 配置/状态来源: self._parser（本段读取的实例字段）; self.logger（本段读取的实例字段）; llm_response.content（本段读取的输入字段）; result.error / result.warning / result.commands / result.is_task_complete / result.analysis / result.plan（本段读取的解析结果字段）
• 结果去向: 本方法的返回元组第 1 项 commands; 本方法的返回元组第 2 项 result.is_task_complete; 本方法的返回元组第 3 项 feedback; 本方法的返回元组第 4、5 项 result.analysis 与 result.plan; 本方法的返回元组第 6 项原始 llm_response

响应文本容错解析入口

stage 上下文: 该函数是本解析器对外的统一入口：先做一次常规解析，再在首轮解析报错时按顺序尝试自动修复。相较于同 stage 中负责消费解析结果的兄弟函数，这里只负责把原始字符串变成 ParseResult，不处理后续封装或反馈拼接。它本身不维护跨调用状态，职责集中在“尽量解析成功，但保留首个原始失败结果”。

这段代码在干什么

函数接收一个响应字符串 response，先调用 self._try_parse_response(response) 得到首个 ParseResult。若该结果含有 error，它会依次尝试 self._get_auto_fixes() 提供的修复方案；每个修复函数都基于原始 response 与原始 result.error 生成候选文本，并仅在重解析后的 corrected_result.error == "" 时接受该修复。若全部修复都未得到无错误结果，则返回第一次解析得到的原始 result 对象不变；唯一可能的对象修改是成功修复路径上对 corrected_result.warning 的赋值。

接口 · 参数 / IO

(self, response: str) -> ParseResult

• 参数: response: str — 待解析的完整响应字符串，也是所有自动修复尝试的原始输入
• 读状态: self._try_parse_response, self._get_auto_fixes, self._combine_warnings
• 返回: 返回一个 ParseResult：若首轮解析成功则直接返回该结果；若某个自动修复后的重解析得到 error == ""，则返回该修复结果；若所有修复都失败，则返回第一次 _try_parse_response(response) 得到的原始 result 对象且不做修改。
• 副作用: 无 self 状态写入; 成功自动修复时会给局部变量 corrected_result.warning 重新赋值

执行流

1. 先对原始 response 调用 self._try_parse_response(response)，得到首轮解析结果 result。
2. 若 result.error 为真值，则进入自动修复流程；否则跳过修复逻辑并在末尾直接返回该 result。
3. 遍历 self._get_auto_fixes() 返回的每个 (fix_name, fix_function)；对每个候选修复，都以原始 response 和原始 result.error 调用 fix_function(response, result.error)，得到 corrected_response, was_fixed。
4. 只有当 was_fixed 为真时，才对该 corrected_response 再次调用 self._try_parse_response(...) 得到 corrected_result；若这次仍有错误，则不覆盖原始 result，而是继续尝试下一个修复方案。
5. 一旦某个 corrected_result.error == ""，就构造 AUTO-CORRECTED: {fix_name} - please fix this in future responses 警告文本，并通过 self._combine_warnings(auto_warning, corrected_result.warning) 合并进 corrected_result.warning。
6. 在首个成功修复处立即返回该 corrected_result；若循环结束仍无成功修复，则返回最开始的原始 result。

源码

    def parse_response(self, response: str) -> ParseResult:
        """
        Parse a terminus JSON plain response and extract commands.

        Args:
            response: The full LLM response string

        Returns:
            ParseResult with commands, completion status, errors and warnings
        """

        # Try normal parsing first
        result = self._try_parse_response(response)

        if result.error:
            # Try auto-fixes in order until one works
            for fix_name, fix_function in self._get_auto_fixes():
                corrected_response, was_fixed = fix_function(response, result.error)
                if was_fixed:
                    corrected_result = self._try_parse_response(corrected_response)

                    if corrected_result.error == "":
                        # Success! Add auto-correction warning
                        auto_warning = (
                            f"AUTO-CORRECTED: {fix_name} - "
                            "please fix this in future responses"
                        )
                        corrected_result.warning = self._combine_warnings(
                            auto_warning, corrected_result.warning
                        )
                        return corrected_result

        # Return original result if no fix worked
        return result

Non-obvious 设计决策

• 自动修复只在首轮解析出现 result.error 时启动，避免对本已可接受的解析结果做额外变换；这体现了“常规解析优先、容错补救后置”的取舍。
• 每个修复函数都接收原始 response 与原始 result.error，而不是串联使用上一个修复的输出或后续错误信息；这使各修复方案彼此独立，避免连续修补造成来源混杂。
• 函数仅在 corrected_result.error == "" 时接受修复结果，说明其判定标准是“完全消除解析错误”，而不是接受部分改善或不同错误形态。
• 当修复成功时，不静默吞掉这一步，而是通过 AUTO-CORRECTED: ... 警告并与既有 warning 合并；这保留了结果可用性，同时显式暴露输入格式曾有问题。
• 若所有修复都未成功，函数明确回退到第一次解析得到的 result，而不是返回最后一次失败的重解析结果；这保留了最初失败上下文，也避免修复尝试污染原始诊断。

上下游关系

• 调用方: 解析器使用方对本解析器实例的统一响应解析调用; 任何需要把原始响应字符串转换为 ParseResult 的上层流程
• 核心被调用: TerminusJSONPlainParser._try_parse_response; TerminusJSONPlainParser._get_auto_fixes; 自动修复函数 fix_function(response, result.error); TerminusJSONPlainParser._combine_warnings
• 配置/状态来源: self._get_auto_fixes() 提供修复策略顺序; self._try_parse_response 的错误判定结果驱动是否进入修复分支
• 结果去向: 返回给本解析器调用方的 ParseResult; 成功修复路径中带有合并后 warning 的 corrected_result; 总失败路径中原始首轮解析 result 对象
• 同类 sibling: 与 Terminus2._handle_llm_interaction 相比，本函数只负责生成 ParseResult；后者才继续读取其中字段并做后续组织。

JSON 响应的首轮容错解析入口

stage 上下文: 该函数承担 JSON plain 响应的单次解析工作：从原始字符串中抽取 JSON、完成解码与结构核验，再整理出命令、完成状态、分析与计划。它是 parse_response 内部的底层尝试函数；与兄弟函数相比，这里不做多轮修复，只给出当前文本的一次解析结果。

这段代码在干什么

函数接收 response: str，返回一个 ParseResult，其中包含命令列表、任务完成标志、错误字符串、警告字符串以及 analysis/plan。它先通过 _extract_json_content 抽出 JSON 片段并累计格式警告，再依次进行 JSON 解码、结构校验与命令解析。所有失败路径都会返回明确的默认值：命令为空；早期失败时完成状态为 False；在未成功读到结构化字段前，analysis 与 plan 为空字符串。函数本身不写入实例状态，唯一可变数据是局部 warnings 列表，且该列表会被作为参数传入辅助函数以便间接追加告警。

接口 · 参数 / IO

(self, response: str) -> ParseResult

• 参数: response: str — 待解析的完整 LLM 响应文本
• 返回: 返回 ParseResult(commands, is_task_complete, error, warning, analysis, plan)。若 json_content 为空、JSON 解码失败或结构校验失败，则返回 commands=[]、is_task_complete=False，并把 analysis 与 plan 置为 ""。若命令解析失败但 task_complete 为真，则返回 commands=[]、is_task_complete=True、error=""，同时保留已提取的 analysis/plan 并把命令问题降级进 warning；若命令解析失败且未完成，则返回 commands=[]、is_task_complete=False，但同样保留已提取的 analysis/plan。所有返回分支中的 warning 都使用同一格式模式：有告警时为 "- " + "\n- ".join(warnings)，无告警时为空字符串。
• 副作用: 无实例状态写入; 局部 warnings 列表会被 _validate_json_structure(parsed_data, json_content, warnings) 与 _parse_commands(commands_data, warnings) 以可变参数方式间接追加内容

执行流

1. 初始化局部 warnings = []，随后调用 _extract_json_content(response)；该调用同时返回抽取出的 json_content 与 extra_text_warnings，后者立即经 warnings.extend(...) 纳入本次解析的告警累积器。
2. 若 json_content 为空，则直接返回失败结果：commands=[]、is_task_complete=False、error='No valid JSON found in response'、analysis=''、plan=''，并把当前 warnings 统一格式化为 warning 字符串。
3. 若存在 JSON 文本，则尝试 json.loads(json_content)；一旦触发 json.JSONDecodeError，构造 error_msg = 'Invalid JSON: ...'。这里会按 len(json_content) < 200 分支附加完整内容或前 100 字符预览，然后返回同样以空命令、未完成、空 analysis/plan 为默认值的失败结果。
4. JSON 解码成功后，调用 _validate_json_structure(parsed_data, json_content, warnings) 做结构核验；该辅助函数除返回 validation_error 外，还可能通过传入的 warnings 列表间接追加警告。若返回非空错误，则立即以空命令、未完成、空 analysis/plan 返回。
5. 结构通过后，从 parsed_data 读取 task_complete、analysis、plan 与 commands；其中 task_complete 缺省为 False，且若其值是字符串，会按小写后是否属于 ('true', '1', 'yes') 强制归一为布尔值。
6. 接着调用 _parse_commands(commands_data, warnings) 将命令载荷转成命令对象；该调用同样可能通过传入的 warnings 列表补充告警，并返回 (commands, parse_error)。
7. 若 parse_error 非空，则根据 is_complete 分流：任务已完成时，把该错误追加进 warnings，返回 commands=[]、is_task_complete=True、error=''，同时保留先前抽取的 analysis 与 plan；任务未完成时，则把该问题作为 error 返回，is_task_complete=False，也同样保留 analysis 与 plan。
8. 若命令解析无误，则返回成功结果：commands 为解析所得命令列表，is_task_complete 为归一化后的完成状态，error=''，warning 仍采用统一的项目符号字符串格式，analysis 与 plan 原样透传。

源码

    def _try_parse_response(self, response: str) -> ParseResult:
        """
        Try to parse a terminus JSON plain response.

        Args:
            response: The full LLM response string

        Returns:
            ParseResult with commands, completion status, errors and warnings
        """
        warnings = []

        # Check for extra text before/after JSON
        json_content, extra_text_warnings = self._extract_json_content(response)
        warnings.extend(extra_text_warnings)

        if not json_content:
            return ParseResult(
                [],
                False,
                "No valid JSON found in response",
                "- " + "\n- ".join(warnings) if warnings else "",
                "",
                "",
            )

        # Parse JSON
        try:
            parsed_data = json.loads(json_content)
        except json.JSONDecodeError as e:
            # Add debug info
            error_msg = f"Invalid JSON: {str(e)}"
            if len(json_content) < 200:
                error_msg += f" | Content: {repr(json_content)}"
            else:
                error_msg += f" | Content preview: {repr(json_content[:100])}..."
            return ParseResult(
                [],
                False,
                error_msg,
                "- " + "\n- ".join(warnings) if warnings else "",
                "",
                "",
            )

        # Validate structure
        validation_error = self._validate_json_structure(
            parsed_data, json_content, warnings
        )
        if validation_error:
            return ParseResult(
                [],
                False,
                validation_error,
                "- " + "\n- ".join(warnings) if warnings else "",
                "",
                "",
            )

        # Check if task is complete
        is_complete = parsed_data.get("task_complete", False)
        if isinstance(is_complete, str):
            is_complete = is_complete.lower() in ("true", "1", "yes")

        # Extract analysis and plan for reasoning content
        analysis = parsed_data.get("analysis", "")
        plan = parsed_data.get("plan", "")

        # Parse commands
        commands_data = parsed_data.get("commands", [])
        commands, parse_error = self._parse_commands(commands_data, warnings)
        if parse_error:
            # If task is complete, parse errors are just warnings
            if is_complete:
                warnings.append(parse_error)
                return ParseResult(
                    [],
                    True,
                    "",
                    "- " + "\n- ".join(warnings) if warnings else "",
                    analysis,
                    plan,
                )
            return ParseResult(
                [],
                False,
                parse_error,
                "- " + "\n- ".join(warnings) if warnings else "",
                analysis,
                plan,
            )

        return ParseResult(
            commands,
            is_complete,
            "",
            "- " + "\n- ".join(warnings) if warnings else "",
            analysis,
            plan,
        )

Non-obvious 设计决策

• 该函数把告警与致命错误明确分层：warnings 从开头即独立累积，而 error 只在无法继续可信解析时返回。这一点体现在“额外文本”仅入 warning、结构错误直接终止、以及命令解析错误在 is_complete 为真时被降级为 warning。
• 对 warnings 采用“单个可变列表贯穿全流程”的策略，而不是让各辅助函数各自返回独立告警集合；代码通过将同一个 warnings 传给 _validate_json_structure(..., warnings) 与 _parse_commands(..., warnings)，允许辅助函数在原位追加，从而保证最终 warning 输出覆盖抽取、校验与命令解析三个阶段。
• JSON 解码失败时没有无条件回显完整内容，而是按长度做诊断裁剪：len(json_content) < 200 时附上完整 repr(json_content)，否则只附上前 100 个字符预览。这是在诊断可读性与错误消息体积之间做出的显式取舍。
• 对 task_complete 做字符串布尔归一化是一个容错决策：当上游给出 'true'、'1'、'yes' 这类字符串时，函数仍将其解释为完成，而不是把它视为结构错误。这扩大了解析接受面，但仍将可接受取值限制在代码列出的有限集合中。
• 即使命令解析失败，函数也不会丢弃已经成功读取的 analysis 与 plan；两条 parse_error 分支都把这两个字段继续放入返回值。这表明代码将“推理文本可保留”与“命令是否可执行”区分对待。

上下游关系

• 调用方: TerminusJSONPlainParser.parse_response（兄弟单元已说明其首先调用本函数取得首个 ParseResult）
• 核心被调用: self._extract_json_content; json.loads; self._validate_json_structure; self._parse_commands; ParseResult
• 配置/状态来源: 输入参数 response; 局部累积器 warnings; parsed_data['task_complete'] / parsed_data.get('task_complete', False); parsed_data.get('analysis', ''); parsed_data.get('plan', ''); parsed_data.get('commands', [])
• 结果去向: 返回给当前类的 parse_response 作为本轮原始解析结果; 封装为 ParseResult.warning 的项目符号字符串; 封装为 ParseResult.error 的结构化失败信息; 封装为 ParseResult.commands / is_task_complete / analysis / plan
• 同类 sibling: TerminusJSONPlainParser.parse_response：后者围绕本函数结果决定是否尝试自动修复；本函数只负责单次解析，不执行修复循环。

响应文本中的花括号片段提取器

stage 上下文: 该函数位于 Response Parse 阶段内，职责是把一段原始响应按花括号与引号状态扫描，截出首个平衡的 {...} 候选子串。它只处理当前传入的 response 文本，不读取实例状态，也不做语义级 JSON 校验；在本 stage 中属于更底层的文本边界识别步骤。

这段代码在干什么

函数接收 response: str，逐字符扫描，寻找首个处于非字符串上下文中的、花括号计数从 0 开始并回到 0 的 {...} 片段。成功时返回该子串及告警列表；告警仅用于指出该片段前后是否还存在非空额外文本，数量可能为 0、1 或 2 条。若未同时找到起止边界，则返回空字符串和 ['No valid JSON object found']。函数本身无任何对象状态写入或外部副作用。

接口 · 参数 / IO

(self, response: str) -> tuple[str, List[str]]

• 参数: self: ? — 实例方法接收者；本函数实际未读取其属性; response: str — 待扫描的原始响应文本来源
• 返回: 返回二元组 (extracted_text, warnings)：成功时 extracted_text 为首个平衡花括号包围的候选 {...} 子串，warnings 为前后额外文本告警列表（0 到 2 条）；失败时返回 ('', ['No valid JSON object found'])。

执行流

1. 初始化局部变量：warnings 为空列表，边界记录 json_start/json_end 设为 -1，并建立 brace_count、in_string、escape_next 三个扫描状态。
2. 按 enumerate(response) 逐字符遍历；若上一字符触发了转义（escape_next 为真），则当前字符只用于清除该标记，随后直接跳过，不参与任何结构判断。
3. 遇到反斜杠 \ 时，将 escape_next 置为真并继续；遇到双引号 " 且当前不处于“转义后跳过”分支时，翻转 in_string，用于区分字符串内部与外部。
4. 仅当 in_string 为假时才处理花括号：遇到 { 时，若当前 brace_count == 0 就把当前位置记为 json_start，随后递增计数；遇到 } 时先递减计数，若此时计数回到 0 且 json_start != -1，则把 json_end 记为 i + 1 并立即结束扫描。
5. 扫描完成后，若 json_start == -1 或 json_end == -1，说明没有找到同时具备起点和终点的平衡片段，直接返回空字符串和固定错误告警 ['No valid JSON object found']。
6. 若成功得到边界，则分别对 response[:json_start] 与 response[json_end:] 做 strip()；前段非空时加入 Extra text detected before JSON object，后段非空时加入 Extra text detected after JSON object。
7. 最后返回 response[json_start:json_end] 这一候选子串，以及累计得到的 warnings。

源码

    def _extract_json_content(self, response: str) -> tuple[str, List[str]]:
        """Extract JSON content from response, handling extra text."""
        warnings = []

        # Try to find JSON object boundaries
        json_start = -1
        json_end = -1
        brace_count = 0
        in_string = False
        escape_next = False

        for i, char in enumerate(response):
            if escape_next:
                escape_next = False
                continue

            if char == "\\":
                escape_next = True
                continue

            if char == '"' and not escape_next:
                in_string = not in_string
                continue

            if not in_string:
                if char == "{":
                    if brace_count == 0:
                        json_start = i
                    brace_count += 1
                elif char == "}":
                    brace_count -= 1
                    if brace_count == 0 and json_start != -1:
                        json_end = i + 1
                        break

        if json_start == -1 or json_end == -1:
            return "", ["No valid JSON object found"]

        # Check for extra text
        before_text = response[:json_start].strip()
        after_text = response[json_end:].strip()

        if before_text:
            warnings.append("Extra text detected before JSON object")
        if after_text:
            warnings.append("Extra text detected after JSON object")

        return response[json_start:json_end], warnings

Non-obvious 设计决策

• 该实现只做基于 brace_count、in_string、escape_next 的边界提取，而不验证提取结果是否为语义上正确的 JSON；因此它保证的是“首个平衡花括号包围的候选 {...} 子串”，不是已验证合法的 JSON 对象。
• 代码显式区分字符串内外状态，并在 escape_next 分支中直接 continue，使被转义字符完全跳过结构处理；这样可避免字符串内容中的 "、{、} 干扰片段边界识别。
• 函数要求同时找到 json_start 与 json_end 才算成功，任何只有起点没有闭合、或根本未进入有效平衡区间的情况，都统一落入 ('', ['No valid JSON object found'])，从返回值层面保持失败信号明确。
• 额外文本不阻断提取，而是通过 before_text、after_text 的 strip() 结果转化为告警；这是一种“保留主片段、附带报告噪声”的策略。
• 一个不太显眼但确实存在的取舍是：在字符串外遇到 } 时，代码会无条件执行 brace_count -= 1，即使此前尚未看到任何 {；因此 brace_count` 可能变成负数，函数并未对这种情况做保护或纠正。

上下游关系

• 调用方: 未知；提供源码中未出现直接调用点
• 核心被调用: enumerate(response); response[:json_start].strip(); response[json_end:].strip()
• 配置/状态来源: response 参数文本; 局部扫描状态 brace_count; 局部扫描状态 in_string; 局部扫描状态 escape_next
• 结果去向: 返回给其直接调用者的提取子串; 返回给其直接调用者的告警列表

JSON结构与字段约束校验点

stage 上下文: 该函数是一个纯校验辅助例程，职责限于检查传入 data 与 json_content 是否满足本解析器期望的 JSON 结构。它在本体内不构造结果对象，也不操作实例级运行状态；可见输出只有返回的错误字符串与对 warnings 列表的追加。

这段代码在干什么

函数接收 data、原始 json_content 与可变 warnings 列表，先做对象形态与必填字段检查，再对若干字段类型做硬错误或软告警判定。结构性硬失败时返回非空错误字符串；通过硬校验时返回空字符串。其副作用仅限于向 warnings 追加本函数直接生成的三类字段类型告警，以及调用 _check_field_order(...) 委托顺序相关检查继续追加告警。

接口 · 参数 / IO

(self, data: dict, json_content: str, warnings: List[str]) -> str

• 参数: data: dict — 待校验的已解析 JSON 数据；虽有 dict 注解，但函数运行时仍显式拒绝非字典值; json_content: str — 原始 JSON 文本，传给 _check_field_order(...) 作为顺序检查输入; warnings: List[str] — 可变告警列表；本函数会直接追加字段类型告警，并传递给 _check_field_order(...) 继续填充
• 读状态: self.required_fields
• 返回: 返回结构校验结果字符串：硬校验失败时为具体错误信息，全部硬校验通过时返回空字符串 ""。
• 副作用: 当 data.get("analysis", "") 不是 str 时，向 warnings 追加 "Field 'analysis' should be a string"; 当 data.get("plan", "") 不是 str 时，向 warnings 追加 "Field 'plan' should be a string"; 当 data.get("task_complete") 的值不是 None 且也不是 bool 或 str 时，向 warnings 追加 "Field 'task_complete' should be a boolean or string"; 调用 self._check_field_order(data, json_content, warnings)，把顺序相关告警的写入委托给该辅助函数

执行流

1. 先做最外层形态防御：若 data 运行时不是 dict，立即返回 "Response must be a JSON object"，后续所有字段检查都不再执行。
2. 遍历 self.required_fields 收集缺失键；若 missing_fields 非空，返回 "Missing required fields: ..."，这是第二类硬失败出口。
3. 对 analysis 与 plan 做软类型检查：分别通过 data.get("analysis", "")、data.get("plan", "") 取值；值不是字符串时，仅向 warnings 追加告警，不中断校验。
4. 读取 commands = data.get("commands", [])；若其不是列表，则返回 "Field 'commands' must be an array"，这是第三类硬失败出口。
5. 在 commands 通过类型检查后，调用 self._check_field_order(data, json_content, warnings)，把字段顺序相关的检查委托给该函数处理。
6. 读取 task_complete = data.get("task_complete")；仅当该值不是 None 且其类型既非 bool 也非 str 时，追加一条软告警。若值为 None（包括 JSON 中显式 null 导致的情况），该分支不会报错也不会告警。
7. 若以上硬校验均未触发返回，函数最终返回空字符串 ""，表示结构层面的强约束已通过。

源码

    def _validate_json_structure(
        self, data: dict, json_content: str, warnings: List[str]
    ) -> str:
        """Validate the JSON structure has required fields."""
        if not isinstance(data, dict):
            return "Response must be a JSON object"

        # Check for required fields
        missing_fields = []
        for field in self.required_fields:
            if field not in data:
                missing_fields.append(field)

        if missing_fields:
            return f"Missing required fields: {', '.join(missing_fields)}"

        # Validate field types
        if not isinstance(data.get("analysis", ""), str):
            warnings.append("Field 'analysis' should be a string")

        if not isinstance(data.get("plan", ""), str):
            warnings.append("Field 'plan' should be a string")

        commands = data.get("commands", [])
        if not isinstance(commands, list):
            return "Field 'commands' must be an array"

        # Check for correct order of fields (analysis, plan, commands)
        self._check_field_order(data, json_content, warnings)

        # Validate task_complete if present
        task_complete = data.get("task_complete")
        if task_complete is not None and not isinstance(task_complete, (bool, str)):
            warnings.append("Field 'task_complete' should be a boolean or string")

        return ""

Non-obvious 设计决策

• 尽管形参注解写的是 data: dict，函数仍首先用 isinstance(data, dict) 做运行时防御；这表明注解不被视为充分保障，非对象载荷会被显式拒绝而不是交给后续 get(...) 调用隐式出错。
• 该函数把问题分成硬失败与软告警两档：顶层不是对象、缺失必填字段、commands 不是数组都会直接返回错误；而 analysis、plan、task_complete 的类型不匹配只会记录到 warnings。这是函数体内直接可见的校验强度分层。
• task_complete 的判定使用 task_complete is not None 作为前置门槛，而不是简单按键是否存在判断；因此当该字段取到 None 时不会产生任何告警。换言之，显式 null 会被静默接受，这是此处条件写法带来的非显然后果。
• 字段顺序检查没有内联在本函数，而是统一委托给 _check_field_order(...)；本函数只负责在调用点保证前置硬校验已通过，并共享同一个 warnings 容器承接其输出。

上下游关系

• 调用方: 未知；当前源码片段未显示谁调用该函数
• 核心被调用: self._check_field_order
• 配置/状态来源: self.required_fields
• 结果去向: 返回给直接调用者的错误字符串; 由调用者持有并继续使用的 warnings 列表
• 同类 sibling: 与 _extract_json_content 相比，本函数不负责抽取 JSON 片段，只消费已提供的 data/json_content 做结构约束检查。; 与 parse_response、_try_parse_response 这类更高层入口相比，本函数职责更窄，只产生字符串错误与列表告警，不组装更高层解析结果。

响应文本字段顺序软校验器

stage 上下文: 该函数是 JSON 结构校验链路中的一个辅助检查点，职责仅限于观察原始 response 文本里若干字段名的出现顺序。它不决定解析是否成功，只可能向传入的 warnings 列表追加一条顺序告警。就同组兄弟函数分工看，_validate_json_structure(...) 负责调用它，而它专注于顺序这一项软性格式问题。

这段代码在干什么

函数围绕局部常量 expected_order = ["analysis", "plan", "commands"] 检查这三个字段在原始字符串 response 中的文本出现顺序。它对每个字段执行一次 re.search，匹配模式是带双引号的字段名、后接可选空白和冒号；若至少找到两个字段，就把实际顺序与该字段子集的期望顺序比较。函数返回 None，真正产出是必要时向可变参数 warnings 追加一条人类可读的告警；参数 data 在本实现中未被使用。

接口 · 参数 / IO

(self, data: dict, response: str, warnings: List[str]) -> None

• 参数: self: ? — 解析器实例；本函数未读取其实例属性; data: dict — 传入的已解析数据对象；在本函数体内未被使用; response: str — 原始响应文本；字段位置完全从这里通过 re.search(...) 的首个匹配位置取得; warnings: List[str] — 可变告警列表；若顺序不符则在末尾追加一条描述字符串
• 返回: 返回 None；真实输出是可能发生的 warnings.append(...)。
• 副作用: 可能向传入的 warnings 列表追加一条字段顺序告警

执行流

1. 先定义目标顺序 expected_order = ["analysis", "plan", "commands"]，作为后续比较基准。
2. 遍历 expected_order 中的每个字段，为每个字段构造模式 "({field})"\s*:，并对 response 调用一次 re.search(...)；若有匹配，则把该字段记录到 positions[field] = match.start()。这里使用的是每个字段在全文中的首个匹配位置，后续重复出现不会再被考虑。
3. 若 positions 中找到的字段少于两个，则直接 return，因为此时没有足够信息进行顺序比较。
4. 按 expected_order 过滤出实际出现过的字段，组成 present_fields，然后按记录的起始位置排序，得到文本中的实际出现顺序 actual_order。
5. 再按 expected_order 的原始次序抽取当前已出现字段，形成仅针对已出现字段子集的期望顺序 expected_present。
6. 若 actual_order != expected_present，则把两边分别用 → 连接成字符串，并向 warnings 追加 Fields appear in wrong order... 告警；若相同则不产生任何输出。

源码

    def _check_field_order(
        self, data: dict, response: str, warnings: List[str]
    ) -> None:
        """Check if fields appear in the correct order: analysis, plan, commands."""
        # Expected order for required fields
        expected_order = ["analysis", "plan", "commands"]

        # Find positions of each field in the original response
        positions = {}
        for field in expected_order:
            # Look for the field name in quotes
            pattern = f'"({field})"\\s*:'
            match = re.search(pattern, response)
            if match:
                positions[field] = match.start()

        # Check if we have at least 2 fields to compare order
        if len(positions) < 2:
            return

        # Get fields that are present, in the order they appear
        present_fields = []
        for field in expected_order:
            if field in positions:
                present_fields.append((field, positions[field]))

        # Sort by position to get actual order
        actual_order = [
            field for field, pos in sorted(present_fields, key=lambda x: x[1])
        ]

        # Get expected order for present fields only
        expected_present = [f for f in expected_order if f in positions]

        # Compare orders
        if actual_order != expected_present:
            actual_str = " → ".join(actual_order)
            expected_str = " → ".join(expected_present)
            warnings.append(
                f"Fields appear in wrong order. Found: {actual_str}, "
                f"expected: {expected_str}"
            )

Non-obvious 设计决策

• 实现以 response 中的文本位置而非 data 的键顺序为判据；这使检查对象明确落在原始响应呈现顺序上，也解释了为何虽然签名接收了 data，函数体却完全未使用它。
• 每个字段只做一次 re.search(...)，因此只采信首个匹配位置；如果同名字段在后文再次出现，或文本中有重复提及，本实现不会尝试综合多个位置。
• 当找到的字段不足两个时直接返回，表明该检查被设计为“可比较时才发声”的软校验，而不是在信息不足时制造额外告警。
• 比较时使用 expected_present 而不是完整 expected_order，意味着缺失字段本身不在这里被报告；本函数只处理“已出现字段之间的相对顺序”这一件事。

上下游关系

• 调用方: TerminusJSONPlainParser._validate_json_structure
• 核心被调用: re.search; warnings.append; sorted; " → ".join
• 配置/状态来源: 局部常量 expected_order 提供唯一顺序规则; 输入参数 response 提供字段文本位置来源; 输入参数 warnings 提供告警输出容器; 输入参数 data 被传入但在本实现中未使用
• 结果去向: 传入的 warnings 列表; 调用方 TerminusJSONPlainParser._validate_json_structure 的后续处理; 本函数的隐式完成信号：未追加告警即表示顺序检查未发现问题; 返回值固定为 None，不经由返回通道传递结果
• 同类 sibling: TerminusJSONPlainParser._validate_json_structure：负责更广义的结构校验，并把顺序检查委托给本函数。; TerminusJSONPlainParser._try_parse_response：在更上层收集 warnings，本函数只贡献其中可能的一条。; TerminusJSONPlainParser._extract_json_content：同样生成告警，但关注点是 JSON 片段提取而非字段顺序。

命令数组校验与对象化入口

stage 上下文: 该函数位于 stage-4.4 的响应解析链路中，负责把已通过 JSON 解码与结构校验的 commands 数组进一步细化为 ParsedCommand 列表。触发条件是上游 _try_parse_response 已成功取出 commands_data，并把同一个可变 warnings 列表传入本函数。与同阶段兄弟函数相比，_extract_json_content 处理文本包裹问题，_validate_json_structure 处理顶层形态与字段约束，而本函数专门处理单条命令对象的字段级合法性与默认化。

这段代码在干什么

函数接收 commands_data: List[dict] 与调用方提供的 warnings 列表，逐项验证每个命令对象是否可转换为 ParsedCommand。它对命令对象形态、必填字段 keystrokes 及其类型执行硬校验；一旦失败，立即返回空命令列表和错误字符串。对 duration 缺失或类型不合法、出现未知字段、以及多条命令间缺少结尾换行等情况，仅追加告警并继续产出命令对象，因此其副作用主要是原位扩充 warnings。

接口 · 参数 / IO

(self, commands_data: List[dict], warnings: List[str]) -> tuple[List[ParsedCommand], str]

• 参数: self: ? — 实例方法接收者；本函数实现中未读取实例属性; commands_data: List[dict] — 来自上游解析结果中的 commands 数组候选值，待逐条校验并转换; warnings: List[str] — 由调用方传入的可变告警累积器，本函数会向其中追加非致命问题说明
• 返回: 返回二元组 (commands, error_message)：成功时 commands 为 ParsedCommand 列表且 error_message 为空字符串；发生硬校验失败时返回空列表和对应错误文本。
• 副作用: 向传入的 warnings 列表追加关于默认 duration、未知字段、换行建议的告警信息

执行流

1. 初始化局部列表 commands = []，作为成功解析出的 ParsedCommand 累积容器。
2. 按顺序遍历 commands_data 中的每一项 cmd_data，并用 i + 1 生成人类可读的命令序号；若某项不是 dict，立刻返回 ([], "Command N must be an object")。
3. 检查每条命令是否含有必填字段 keystrokes，且其值必须是 str；缺失或类型不符都属于硬错误，立即中止整个命令数组解析并返回空列表与错误字符串。
4. 处理可选字段 duration：若字段存在且值是 int 或 float，直接采用；若存在但类型非法，则向 warnings 追加“Invalid duration value, using default 1.0”并回退到 1.0；若字段缺失，也追加“Missing duration field, using default 1.0”并同样使用 1.0。
5. 计算 unknown_fields = set(cmd_data.keys()) - {"keystrokes", "duration"}；若存在未知字段，则仅记录一条告警，不阻止该命令继续被接受。
6. 若当前命令后面还有下一条命令，但 keystrokes 不以换行符 \n 结尾，则追加一条关于两条命令可能被拼接到同一行的告警；这仍然不视为解析失败。
7. 把当前命令转换为 ParsedCommand(keystrokes=keystrokes, duration=float(duration)) 加入 commands；全部遍历完成后返回 (commands, "")。

源码

    def _parse_commands(
        self, commands_data: List[dict], warnings: List[str]
    ) -> tuple[List[ParsedCommand], str]:
        """Parse commands array into ParsedCommand objects."""
        commands = []

        for i, cmd_data in enumerate(commands_data):
            if not isinstance(cmd_data, dict):
                return [], f"Command {i + 1} must be an object"

            # Check for required keystrokes field
            if "keystrokes" not in cmd_data:
                return [], f"Command {i + 1} missing required 'keystrokes' field"

            keystrokes = cmd_data["keystrokes"]
            if not isinstance(keystrokes, str):
                return [], f"Command {i + 1} 'keystrokes' must be a string"

            # Parse optional fields with defaults
            if "duration" in cmd_data:
                duration = cmd_data["duration"]
                if not isinstance(duration, (int, float)):
                    warnings.append(
                        f"Command {i + 1}: Invalid duration value, using default 1.0"
                    )
                    duration = 1.0
            else:
                warnings.append(
                    f"Command {i + 1}: Missing duration field, using default 1.0"
                )
                duration = 1.0

            # Check for unknown fields
            known_fields = {"keystrokes", "duration"}
            unknown_fields = set(cmd_data.keys()) - known_fields
            if unknown_fields:
                warnings.append(
                    f"Command {i + 1}: Unknown fields: {', '.join(unknown_fields)}"
                )

            # Check for newline at end of keystrokes if followed by another command
            if i < len(commands_data) - 1 and not keystrokes.endswith("\n"):
                warnings.append(
                    f"Command {i + 1} should end with newline when followed "
                    "by another command. Otherwise the two commands will be "
                    "concatenated together on the same line."
                )

            commands.append(
                ParsedCommand(keystrokes=keystrokes, duration=float(duration))
            )

        return commands, ""

Non-obvious 设计决策

• 该函数把 keystrokes 视为唯一真正不可恢复的核心字段：缺失或类型错误即整体失败返回，而不是跳过坏命令继续保留其余命令。这种取舍保证下游执行层拿到的命令序列不存在“没有实际输入内容”的空壳项。
• duration 采取宽容降级而非硬失败：无论是缺失还是类型不合法，都统一回退为 1.0 并留下告警。代码显式区分“字段缺失”和“值非法”两类文案，说明这里优先追求命令可执行性，而非要求模型输出完全规范。
• 未知字段只告警、不报错，体现出对模型输出冗余信息的兼容策略；这与上游 _validate_json_structure 对顶层关键字段的硬性约束形成分层：顶层结构严格，命令对象局部附加信息宽容。
• 对“除最后一条外的命令最好以换行结束”的检查仅给出 advisory warning，因为缺少换行带来的风险是 shell 拼接语义歧义，而不是对象本身无法构造。代码通过 i < len(commands_data) - 1 明确排除了最后一条命令，避免把单条命令的无换行误报为问题。
• 在构造 ParsedCommand 时统一执行 float(duration)，即便前面允许 int 或 float 输入，也将输出规格收敛为单一浮点类型，降低后续调用方处理多种数值类型的复杂度。

上下游关系

• 调用方: TerminusJSONPlainParser._try_parse_response; 经由 TerminusJSONPlainParser.parse_response 的首次解析路径; 经由 TerminusJSONPlainParser.parse_response 的自动修复后重解析路径
• 核心被调用: ParsedCommand(...) 构造函数; warnings.append(...); float(duration); cmd_data.keys()
• 配置/状态来源: 参数 commands_data：来自模型响应中 commands 字段的已解码数组; 参数 warnings：由上游 _try_parse_response 创建并跨子步骤共享的告警容器; 局部常量 known_fields = {"keystrokes", "duration"}：定义命令对象允许字段集合
• 结果去向: 返回给 TerminusJSONPlainParser._try_parse_response，参与组装最终 ParseResult.commands 与 ParseResult.error; 经由 TerminusJSONPlainParser.parse_response 进入可能的自动修复接受/拒绝判定; 经由 Terminus2._handle_llm_interaction 被包装为 Command(keystrokes, duration_sec=min(duration, 60)); 追加到 warnings 的文本最终可能被 _handle_llm_interaction 汇总进 feedback 字符串
• 同类 sibling: TerminusJSONPlainParser._extract_json_content：先抽取 JSON 片段并产生包裹文本告警，再由本函数处理命令项细节; TerminusJSONPlainParser._validate_json_structure：先验证顶层对象与字段类型，本函数继续验证 commands 内部每个元素; TerminusJSONPlainParser._try_parse_response：负责串联 JSON 提取、解码、结构校验与本函数的命令解析

混合内容中的 JSON 候选提取器

stage 上下文: 该函数是一个局部容错辅助项，职责是在 response 文本中寻找可能的 JSON 子串，并返回首个可被 json.loads(...) 接受的候选。就这段源码可见信息而言，它本身不参与状态管理，也不直接组装 ParseResult；与同 stage 的其他解析函数相比，这里只处理“混合内容中的候选抽取与验证”这一窄职责。

这段代码在干什么

函数接收 response: str 与 error: str，但实际只使用 response 作为搜索源，在其中提取若干由花括号包围的候选片段。它按 re.findall(...) 产出的顺序逐个尝试 json.loads(match)，一旦某个候选可成功解码，就立即返回该候选字符串和 True。若所有候选都触发 json.JSONDecodeError，则返回原始 response 与 False。函数不读取或写入任何实例状态，也没有除解析尝试外的外部副作用。

接口 · 参数 / IO

(self, response: str, error: str) -> tuple[str, bool]

• 参数: self: ? — 实例方法接收者；本函数体内未读取实例属性; response: str — 待扫描的完整响应文本，既是候选提取源，也是失败时的原样返回值; error: str — 调用方传入的错误信息；本函数体内未被使用
• 返回: 返回二元组 (text, success)：成功时，text 是 re.findall(json_pattern, response, re.DOTALL) 结果中第一个能被 json.loads(...) 成功解析的候选字符串，success 为 True；若没有任何候选通过该校验，则返回原始 response 与 False。

执行流

1. 定义局部正则 json_pattern = r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}"，把搜索目标限定为基于花括号形态的候选片段，而不是直接把整段 response 当作 JSON 解析。
2. 调用 re.findall(json_pattern, response, re.DOTALL) 收集全部匹配到的候选子串，并存入局部变量 matches。
3. 按 matches 的迭代顺序逐个处理 match，对每个候选执行 json.loads(match) 作为有效性验证。
4. 若某个 match 未抛出异常，则立即返回该 match 与 True，因此成功结果总是“找到顺序上的第一个可解析候选”。
5. 若 json.loads(match) 抛出 json.JSONDecodeError，则通过 except ...: continue 忽略该候选并继续尝试下一个。
6. 当所有候选都未通过验证时，返回原始 response 与 False。

源码

    def _fix_mixed_content(self, response: str, error: str) -> tuple[str, bool]:
        """Extract JSON from response with mixed content."""
        # Look for JSON-like patterns
        json_pattern = r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}"
        matches = re.findall(json_pattern, response, re.DOTALL)

        for match in matches:
            try:
                json.loads(match)
                return match, True
            except json.JSONDecodeError:
                continue

        return response, False

Non-obvious 设计决策

• 候选提取与合法性判定被拆成两步：先用 json_pattern 做花括号启发式筛选，再用 json.loads(...) 做真正的 JSON 校验。这说明正则的职责只是缩小搜索范围，而不是完整表达 JSON 文法；否则单靠该模式无法可靠判断语法有效性。
• 成功路径采用“首个通过即返回”的早停策略，代码上体现为 for match in matches: 内部的 return match, True。这避免了继续扫描后续候选，也明确把返回值绑定为 re.findall(...) 顺序中的第一个合法项。
• 异常容忍被刻意收窄到 json.JSONDecodeError：只有 JSON 解码失败会被吞掉并继续尝试，其他异常类型在本函数中不会被捕获。这一点由 except json.JSONDecodeError: 的精确异常类型可直接看出。
• 形参 error 虽出现在签名中，但函数体完全未引用它，说明当前实现不根据上游错误内容调整提取策略；行为只由 response 决定。

上下游关系

• 调用方: 此片段未展示直接调用者
• 核心被调用: re.findall; json.loads
• 配置/状态来源: response 形参; error 形参（未使用）; 局部常量 json_pattern
• 结果去向: 返回给直接调用方的 (text, success) 二元组; 成功时提供首个可解析候选字符串; 失败时回传原始 response; 布尔标志用于区分是否找到可解析候选

XML 响应解析与自动修复入口

stage 上下文: 该函数是 XML plain parser 对外的统一解析入口：先对原始 response 做一次标准解析，再在首轮出现 result.error 时尝试内建修复。它与同类辅助函数形成分工：真正的解析由 self._try_parse_response(...) 完成，修复候选由 self._get_auto_fixes() 提供，告警拼接由 self._combine_warnings(...) 完成。

这段代码在干什么

函数接收原始响应字符串 response，先调用 self._try_parse_response(response) 得到首个 ParseResult。若该结果带有错误，则按顺序尝试自动修复；只有当某个修复后的文本再次解析得到空错误字符串时，才接受该修复结果。成功接受修复时，函数会在返回前把自动修复说明并入 corrected_result.warning；除此之外不写入实例状态。

接口 · 参数 / IO

(self, response: str) -> ParseResult

• 参数: response: str — 待解析的完整 LLM 响应字符串
• 读状态: self._try_parse_response, self._get_auto_fixes, self._combine_warnings
• 返回: 一个 ParseResult：要么是对原始 response 的首次解析结果，要么是某个成功自动修复后重新解析得到的结果。
• 副作用: 在自动修复成功分支中，返回前会修改 corrected_result.warning

执行流

1. 先对输入 response 执行一次 self._try_parse_response(response)，保存为 result，作为默认返回候选。
2. 若 result.error 为真，进入自动修复流程，并按 self._get_auto_fixes() 给出的顺序遍历每个 (fix_name, fix_function)。
3. 对每个修复函数，调用 fix_function(response, result.error)；这里传入的始终是原始 response 和首轮解析得到的错误信息，而不是前一个修复产物。
4. 只有在 was_fixed 为真时，才会对 corrected_response 再次调用 self._try_parse_response(...) 进行重解析。
5. 若重解析结果满足 corrected_result.error == ""，则构造 AUTO-CORRECTED: {fix_name} - please fix this in future responses 告警，并通过 self._combine_warnings(...) 与已有告警合并后写回 corrected_result.warning，随后立即返回该结果。
6. 若所有修复都未产出无错误结果，则放弃修复路径，返回最开始得到的 result。

源码

    def parse_response(self, response: str) -> ParseResult:
        """
        Parse a terminus XML plain response and extract commands.

        Args:
            response: The full LLM response string

        Returns:
            ParseResult with commands, completion status, errors and warnings
        """
        # Try normal parsing first
        result = self._try_parse_response(response)

        if result.error:
            # Try auto-fixes in order until one works
            for fix_name, fix_function in self._get_auto_fixes():
                corrected_response, was_fixed = fix_function(response, result.error)
                if was_fixed:
                    corrected_result = self._try_parse_response(corrected_response)

                    if corrected_result.error == "":
                        # Success! Add auto-correction warning
                        auto_warning = (
                            f"AUTO-CORRECTED: {fix_name} - "
                            "please fix this in future responses"
                        )
                        corrected_result.warning = self._combine_warnings(
                            auto_warning, corrected_result.warning
                        )
                        return corrected_result

        # Return original result if no fix worked
        return result

Non-obvious 设计决策

• 自动修复只在首轮解析失败时触发，因为分支条件明确是 if result.error:；这避免了对已成功解析的响应做额外改写。
• 每个修复尝试都基于原始 response 与原始 result.error 调用，而不是串联前一个修复结果；这一选择使各修复方案彼此独立，避免级联修补引入新的偏差。
• 修复结果的接受条件是严格的：必须重解析后满足 corrected_result.error == "" 才算成功；仅仅 was_fixed 为真并不足以被采纳。
• 成功修复时不会静默掩盖问题，而是显式向 corrected_result.warning 追加 AUTO-CORRECTED: ... 提示，使返回结果保留“解析成功但原始响应有缺陷”的信息。

上下游关系

• 调用方: 外部通过 XML plain parser 实例调用的响应解析入口; 任何需要把原始响应字符串转换为 ParseResult 的上层解析调用点
• 核心被调用: TerminusXMLPlainParser._try_parse_response; TerminusXMLPlainParser._get_auto_fixes; 各自动修复函数 fix_function(response, result.error); TerminusXMLPlainParser._combine_warnings
• 配置/状态来源: self._get_auto_fixes() 提供的修复策略顺序; self._try_parse_response 的错误判定结果 result.error
• 结果去向: 调用方获得返回的 ParseResult; 成功修复分支返回带合并告警的 corrected_result; 失败时返回首次解析得到的 result
• 同类 sibling: 与 TerminusJSONPlainParser.parse_response 属于同名入口职责：先常规解析，再在失败时尝试自动修复；本条目仅描述当前 XML 实现中该函数体明确体现的行为。

XML 回复分支判定与结果组装

stage 上下文: 该函数负责把一段 XML 形式的回复文本整理为 ParseResult。它本身不直接解析命令语义细节，而是串联若干辅助方法，并依据 response_content、is_complete、sections、parse_error 的不同取值决定返回哪一种结果。与同类入口函数相比，这里呈现的是一次“尝试解析”的具体分支逻辑。

这段代码在干什么

函数接收 response: str，尝试从中提取 <response> 包裹内容、完成标记、分析/计划文本以及命令区段，并返回一个 ParseResult。它显式处理几类情况：缺少 <response>、<commands> 区段缺失、<commands> 存在但为空、命令解析报错、以及正常成功。函数体内不直接写入实例状态；其可见产出仅为返回的 ParseResult，其中错误与警告字段按分支分别填充。

接口 · 参数 / IO

(self, response: str) -> ParseResult

• 参数: response: str — 待解析的完整回复字符串
• 读状态: 本函数体没有直接读取 self 上的数据属性；行为依赖于对 self._check_extra_text、self._extract_response_content、self._check_task_complete、self._extract_sections、self._parse_xml_commands 的调用结果
• 返回: 返回一个 ParseResult。主要分支包括：response_content 为假值时返回 commands=[]、is_task_complete=False、error='No <response> tag found'；commands_content 为假值且 'commands' in sections 时返回空命令、空错误，并保留 is_complete；commands 区段完全缺失且 is_complete=False 时返回 error='Missing <commands> section'；_parse_xml_commands(...) 得到 parse_error 时，若 is_complete=True 则将该错误降级为警告并返回空错误，否则将其作为 error 返回；成功时返回解析出的 commands、is_complete、空错误，以及提取出的 analysis/plan。
• 副作用: 无直接实例状态写入; 原位追加局部列表 warnings，并将其格式化后写入返回值的 warning 字段

执行流

1. 初始化局部 warnings = []，随后调用 self._check_extra_text(response, warnings)；该调用的结果不直接决定控制流，但后续会把累计告警拼接进返回的 warning 字段。
2. 调用 response_content = self._extract_response_content(response)；若 if not response_content 成立，则立即返回 ParseResult([], False, 'No <response> tag found', 格式化告警, '', '')。
3. 对已提取的 response_content 调用 self._check_task_complete(response_content) 得到 is_complete，再调用 sections = self._extract_sections(response_content, warnings)；随后用 sections.get('analysis', '') 与 sections.get('plan', '') 提取 analysis、plan。
4. 读取 commands_content = sections.get('commands', '')；若 if not commands_content 成立且 'commands' in sections，表示命令区段存在但为空。此时仅在 not is_complete 时追加一条关于空命令区段的特定警告，并返回空命令、空错误、当前完成标志以及 analysis/plan。
5. 若 if not commands_content 成立且 'commands' not in sections，表示命令区段完全缺失。此时若 is_complete 为真，则返回空命令、True、空错误；否则返回空命令、False、错误 'Missing <commands> section'。
6. 在存在非空 commands_content 时，调用 commands, parse_error = self._parse_xml_commands(commands_content, warnings)；若 if parse_error 成立，则按 is_complete 分流：完成态下把 parse_error 追加到 warnings 并以空错误返回，未完成态下把 parse_error 直接作为 error 返回。
7. 若以上分支都未提前返回，则以 commands、is_complete、空错误、格式化后的警告串以及前面提取的 analysis、plan 组装最终 ParseResult。

源码

    def _try_parse_response(self, response: str) -> ParseResult:
        """
        Try to parse a terminus XML plain response.

        Args:
            response: The full LLM response string

        Returns:
            ParseResult with commands, completion status, errors and warnings
        """
        warnings = []

        # Check for extra text before/after <response> tags
        self._check_extra_text(response, warnings)

        # Extract <response> content
        response_content = self._extract_response_content(response)
        if not response_content:
            return ParseResult(
                [],
                False,
                "No <response> tag found",
                "- " + "\n- ".join(warnings) if warnings else "",
                "",
                "",
            )

        # Check if task is complete first
        is_complete = self._check_task_complete(response_content)

        # Check for required sections and extract content
        sections = self._extract_sections(response_content, warnings)

        # Extract analysis and plan for reasoning content
        analysis = sections.get("analysis", "")
        plan = sections.get("plan", "")

        # Extract commands section
        commands_content = sections.get("commands", "")
        if not commands_content:
            if "commands" in sections:
                # Commands section exists but is empty
                if not is_complete:
                    warnings.append(
                        "Commands section is empty; not taking any action. "
                        "If you want to wait a specific amount of time please use "
                        "`sleep`, but if you're waiting for a command to finish then "
                        "continue to wait."
                    )
                return ParseResult(
                    [],
                    is_complete,
                    "",
                    "- " + "\n- ".join(warnings) if warnings else "",
                    analysis,
                    plan,
                )
            else:
                # Commands section is missing entirely
                if is_complete:
                    return ParseResult(
                        [],
                        True,
                        "",
                        "- " + "\n- ".join(warnings) if warnings else "",
                        analysis,
                        plan,
                    )
                return ParseResult(
                    [],
                    False,
                    "Missing <commands> section",
                    "- " + "\n- ".join(warnings) if warnings else "",
                    analysis,
                    plan,
                )

        # Parse commands directly from XML (no code blocks in plain format)
        commands, parse_error = self._parse_xml_commands(commands_content, warnings)
        if parse_error:
            # If task is complete, parse errors are just warnings
            if is_complete:
                warnings.append(parse_error)
                return ParseResult(
                    [],
                    True,
                    "",
                    "- " + "\n- ".join(warnings) if warnings else "",
                    analysis,
                    plan,
                )
            return ParseResult(
                [],
                False,
                parse_error,
                "- " + "\n- ".join(warnings) if warnings else "",
                analysis,
                plan,
            )

        return ParseResult(
            commands,
            is_complete,
            "",
            "- " + "\n- ".join(warnings) if warnings else "",
            analysis,
            plan,
        )

Non-obvious 设计决策

• 函数把“命令相关问题”区分为三类：区段存在但为空（if 'commands' in sections 且 if not commands_content）、区段完全缺失（else 分支）、以及区段内容解析失败（if parse_error）。这种分流使不同问题得到不同严重度的返回结果，而不是统一报成同一种解析错误。
• is_complete 会影响错误降级策略：当任务已完成时，缺失命令区段不会报错，命令解析错误也会被转成警告；只有在未完成时，这些情况才会阻断为错误返回。这个取舍直接体现在 if is_complete 的两个分支上。
• 空命令区段的长提示语只在“区段存在、内容为空、且任务未完成”时追加，而不是所有空命令情况都追加。这说明该提示被用于约束未完成任务时的无动作回复，而已完成路径刻意避免附加这条提醒。
• 函数始终把 analysis 与 plan 的提取独立于命令成败：即便后续进入空命令、缺失命令或解析错误分支，只要 sections 中已有这两个字段，对应文本仍会被带入返回值。这是通过在命令判断前先执行 sections.get(...) 实现的。

上下游关系

• 调用方: TerminusXMLPlainParser.parse_response
• 核心被调用: self._check_extra_text; self._extract_response_content; self._check_task_complete; self._extract_sections; self._parse_xml_commands; ParseResult
• 配置/状态来源: 输入参数 response; self._extract_response_content(response) 的返回值 response_content; self._check_task_complete(response_content) 的返回值 is_complete; self._extract_sections(response_content, warnings) 的返回值 sections; self._parse_xml_commands(commands_content, warnings) 的返回值 commands, parse_error
• 结果去向: 返回给本函数调用方的 ParseResult; ParseResult.commands 字段; ParseResult.is_task_complete 字段; ParseResult.error 字段; ParseResult.warning 字段; ParseResult.analysis 字段; ParseResult.plan 字段
• 同类 sibling: TerminusXMLPlainParser.parse_response：该函数的直接上层入口，会先调用本函数获取首个解析结果; TerminusJSONPlainParser._try_parse_response：同名职责对应的 JSON 版本，但此处不描述其内部策略

XML响应块外溢文本告警检查器

stage 上下文: 该函数位于 XML 响应解析阶段，职责很窄：只检查原始 response 中主 <response> 块前后是否夹带额外文本，并把发现结果写入传入的 warnings。与同 stage 的 _try_parse_response 相比，它不负责提取字段、解析命令或构造 ParseResult，而是提供一段结构性告警补充。

这段代码在干什么

函数接收原始响应字符串 response 与可变列表 warnings，围绕首个 <response> 与首个 </response> 的位置检查块外非空白文本。若 <response> 不存在，则直接返回且不追加任何告警；否则最多追加三条具体告警：<response> 前存在额外文本、</response> 后存在额外文本，以及在已发现尾随文本且 response.count("<response>") > 1 时追加一条强调“只应输出一个 <response> 块”的重要告警。返回值为 None，真正产出是对 warnings 的原位修改。

接口 · 参数 / IO

(self, response: str, warnings: List[str]) -> None

• 参数: self: ? — 实例引用；本函数体内未读取任何实例属性; response: str — 待检查的原始模型响应文本，作为 <response>/</response> 与块外文本的搜索源; warnings: List[str] — 由调用方提供的可变告警列表；命中条件时向其中追加人类可读告警
• 返回: 返回 None；实际产出是按条件向 warnings 原位追加最多三条具体字符串告警。
• 副作用: 可能向 warnings 追加 "Extra text detected before <response> tag"; 可能向 warnings 追加 "Extra text detected after </response> tag"; 可能向 warnings 追加 "IMPORTANT: Only issue one <response> block at a time. You issued {total_response_count} and only the first was executed."

执行流

1. 先用 response.find("<response>") 与 response.find("</response>") 定位首个起始标签和首个结束标签，分别保存为 start_pos 与 end_pos。
2. 若 start_pos == -1，函数立即 return；这一分支不写入任何告警。
3. 若找到了 <response>，取 response[:start_pos].strip() 作为 before_text；只要该值非空，就向 warnings 追加 "Extra text detected before <response> tag"。
4. 只有在 end_pos != -1 时，才继续检查结束标签之后的文本：取 response[end_pos + len("</response>") :].strip() 为 after_text；若其非空，就向 warnings 追加 "Extra text detected after </response> tag"。
5. 仍然只在上述“存在尾随文本”的分支里，计算 response.count("<response>") 得到 total_response_count；若其大于 1，再追加一条 IMPORTANT 告警，说明本次输出了多个 <response> 块且只会执行第一个。

源码

    def _check_extra_text(self, response: str, warnings: List[str]) -> None:
        """Check for extra text before/after <response> tags."""
        # Find response tag positions
        start_pos = response.find("<response>")
        end_pos = response.find("</response>")

        if start_pos == -1:
            return  # Will be handled as error later

        # Check text before <response>
        before_text = response[:start_pos].strip()
        if before_text:
            warnings.append("Extra text detected before <response> tag")

        # Check text after </response> if closing tag exists
        if end_pos != -1:
            after_text = response[end_pos + len("</response>") :].strip()
            if after_text:
                warnings.append("Extra text detected after </response> tag")

                # Count total <response> tags
                total_response_count = response.count("<response>")
                if total_response_count > 1:
                    warnings.append(
                        f"IMPORTANT: Only issue one <response> block at a time. "
                        f"You issued {total_response_count} and only the first "
                        f"was executed."
                    )

Non-obvious 设计决策

• 函数把“缺少 <response>”排除在自身职责之外：代码在 start_pos == -1 时直接返回，而不是写告警或报错。这体现了一个边界划分——本函数只处理已出现起始标签时的块外文本问题，不在这里处理起始标签缺失。
• 对 </response> 后文本的检查采用显式门控：只有 end_pos != -1 才会切片并判断 after_text。因此，缺少关闭标签时不会产生“结束标签后有多余文本”的告警，避免在结束边界不存在时构造不可靠结论。
• 重复 <response> 的强调告警不是独立全局检查，而是嵌套在“</response> 后确有额外文本”分支中。也就是说，只有已经检测到关闭标签后仍有内容外溢时，response.count("<response>") > 1 才会触发更强提示；单纯多个 <response> 字样但没有该尾随文本条件，不会在这里告警。
• 文本存在性判断统一经过 .strip()，这意味着纯空白前缀或后缀被视为可接受格式，不会污染 warnings。函数只把非空白可见内容认定为“extra text”。

上下游关系

• 调用方: 未知；提供的源码片段未展示直接调用点; XML 解析路径中的某个上层检查步骤（从参数形态与函数命名可见）
• 核心被调用: str.find; str.strip; list.append; str.count
• 配置/状态来源: 参数 response：唯一的检查输入源; 参数 warnings：唯一的可变输出载体; 局部变量 start_pos：由 response.find("<response>") 得出; 局部变量 end_pos：由 response.find("</response>") 得出
• 结果去向: 传入的 warnings 列表内容; 调用方后续对 warnings 的消费逻辑（本片段未展示具体位置）; 本函数返回点本身不携带结果，所有有效结果均通过副作用输出; 与同 stage 的 _try_parse_response 相比，本函数只产生告警素材，不解析命令或字段
• 同类 sibling: TerminusXMLPlainParser._try_parse_response：负责 XML 主体提取与命令区段解析；本函数仅补充块外文本告警; TerminusXMLPlainParser.parse_response：负责首轮解析与修复尝试；本函数自身不做修复

XML响应区段提取与结构告警器

stage 上下文: 该函数位于 XML 响应解析阶段，职责是把原始 content 中的已知区段整理成字典，并同步补充结构性告警。它由同类解析流程中的 _try_parse_response 调用，和兄弟函数 _check_extra_text、_check_section_order 分别覆盖块外文本、区段顺序等不同维度。相较同 stage 的 JSON 分支，这里处理的是类 XML 标签集合，而不是字段化 JSON 对象。

这段代码在干什么

函数接收响应正文 content 与可变告警列表 warnings，尝试提取 analysis、plan、commands、task_complete 四类标签对应的文本内容。返回值是仅包含“成功识别到的区段”的字典；缺失区段不会出现在返回字典中。副作用仅限于向 warnings 追加缺失必需区段、未知顶层标签、重复标签以及顺序检查相关的告警。

接口 · 参数 / IO

(self, content: str, warnings: List[str]) -> dict

• 参数: content: str — 待扫描的 XML 风格响应文本; warnings: List[str] — 由调用方传入并原位追加诊断信息的告警列表
• 读状态: self.required_sections, self._find_top_level_tags, self._check_section_order
• 返回: 返回 dict：键为已找到的区段名，值为对应内容的去首尾空白字符串；自闭合或显式空标签记为 ""；未找到的区段不会写入返回字典。
• 副作用: 原位修改 warnings，追加缺失必需区段告警; 原位修改 warnings，追加未知顶层标签告警; 原位修改 warnings，追加重复标签告警; 调用 self._check_section_order(content, warnings)，使其继续向 warnings 追加顺序相关告警

执行流

1. 初始化 sections = {} 和 found_sections = set()，并在局部常量 section_patterns` 中为四个已知区段分别准备三类匹配模式：完整包裹标签、自闭合标签、显式空标签。
2. 按 section_patterns.items() 逐个处理区段名；每个区段先用 re.search(full_pattern, content, re.DOTALL) 查找完整标签。若匹配成功，则提取 match.group(1).strip() 写入 sections[section_name]，并把名称加入 found_sections；由于使用 re.search 且随后 continue，每个区段只提取第一个完整匹配。
3. 若完整标签未匹配，则依次检查自闭合模式与空标签模式；任一命中时，都将该区段写为 "" 并登记到 found_sections，然后进入下一个区段。
4. 用 required = set(self.required_sections) 与 missing = required - found_sections 计算缺失区段；对缺失项逐个追加 Missing <...> section 告警，但显式跳过 task_complete，即使它出现在 required_sections 中也不发缺失告警。
5. 调用 self._find_top_level_tags(content) 获取直接子级标签名集合视图，并与 expected_tags = set(self.required_sections + ["task_complete"]) 比较；凡是不在预期集合中的顶层标签，都追加 Unknown tag found: <...> 告警。
6. 对 self.required_sections + ["task_complete"] 中的每个标签，用 re.findall(f"<{section_name}(?:\\s|>|/>)", content) 统计出现次数；当次数大于 1 时，commands 生成一条带 IMPORTANT 的专门告警，其余标签生成通用的 Multiple <...> sections found 告警。
7. 最后调用 self._check_section_order(content, warnings) 进行区段顺序检查，并返回前面累计得到的 sections 字典。

源码

    def _extract_sections(self, content: str, warnings: List[str]) -> dict:
        """Extract analysis, plan, commands, and task_complete sections."""
        sections = {}
        found_sections = set()

        # Define patterns for each section
        section_patterns = {
            "analysis": (
                r"<analysis>(.*?)</analysis>",
                r"<analysis\s*/>",
                r"<analysis></analysis>",
            ),
            "plan": (r"<plan>(.*?)</plan>", r"<plan\s*/>", r"<plan></plan>"),
            "commands": (
                r"<commands>(.*?)</commands>",
                r"<commands\s*/>",
                r"<commands></commands>",
            ),
            "task_complete": (
                r"<task_complete>(.*?)</task_complete>",
                r"<task_complete\s*/>",
                r"<task_complete></task_complete>",
            ),
        }

        for section_name, patterns in section_patterns.items():
            full_pattern, self_closing_pattern, empty_pattern = patterns

            # Try full pattern first
            match = re.search(full_pattern, content, re.DOTALL)
            if match:
                sections[section_name] = match.group(1).strip()
                found_sections.add(section_name)
                continue

            # Try self-closing pattern
            if re.search(self_closing_pattern, content):
                sections[section_name] = ""  # Self-closing = empty content
                found_sections.add(section_name)
                continue

            # Try empty pattern
            if re.search(empty_pattern, content):
                sections[section_name] = ""  # Empty = empty content
                found_sections.add(section_name)
                continue

        # Check for missing required sections
        required = set(self.required_sections)
        missing = required - found_sections
        for section in missing:
            if section != "task_complete":  # task_complete is optional
                warnings.append(f"Missing <{section}> section")

        # Check for unexpected tags at the direct child level of <response> only
        # Find all top-level tags (not nested inside other tags)
        top_level_tags = self._find_top_level_tags(content)
        expected_tags = set(self.required_sections + ["task_complete"])
        unexpected = set(top_level_tags) - expected_tags
        for tag in unexpected:
            warnings.append(
                f"Unknown tag found: <{tag}>, expected "
                f"analysis/plan/commands/task_complete"
            )

        # Check for multiple instances of same tag
        for section_name in self.required_sections + ["task_complete"]:
            tag_count = len(re.findall(f"<{section_name}(?:\\s|>|/>)", content))
            if tag_count > 1:
                if section_name == "commands":
                    warnings.append(
                        f"IMPORTANT: Only issue one <commands> block at a time. "
                        f"You issued {tag_count} and only the first was executed."
                    )
                else:
                    warnings.append(f"Multiple <{section_name}> sections found")

        # Check for correct order of sections (analysis, plan, commands)
        self._check_section_order(content, warnings)

        return sections

Non-obvious 设计决策

• 代码对每个区段采用“完整标签优先，其次自闭合，再次空标签”的判定顺序；这是由三个 re.search(...) 分支的先后顺序直接体现的选择。该选择使非空内容在存在时优先被提取，而 <tag/> 与 <tag></tag> 都被统一折叠为空字符串。
• 代码使用 re.search(..., re.DOTALL) 而不是 re.findall(...) 提取区段正文，因此只保留每类区段的第一个完整匹配，同时允许正文跨多行。这一取舍与后面的重复标签告警配套：函数会提示“有多个”，但返回结果只保留第一处匹配。
• 代码把缺失区段检查建立在 self.required_sections 之上，却对 task_complete 单独豁免缺失告警；这一点由 if section != "task_complete" 明确写死，说明它在本函数里被当作可缺省区段处理。
• 未知标签检查只基于 self._find_top_level_tags(content) 的结果，而不是对全文所有标签做差集；代码注释也明确限定为 direct child level of <response> only。这避免把嵌套标签纳入“未知区段”告警范围。

上下游关系

• 调用方: TerminusXMLPlainParser._try_parse_response
• 核心被调用: re.search; re.findall; self._find_top_level_tags; self._check_section_order
• 配置/状态来源: self.required_sections; 局部常量 section_patterns
• 结果去向: 返回给调用方的 sections 字典; 调用方持有的 warnings 列表
• 同类 sibling: TerminusXMLPlainParser._check_extra_text：同样向 warnings 追加文本结构问题，但关注 <response> 块外文本，而非内部区段集合。; TerminusXMLPlainParser._try_parse_response：在更高一层组织 <response> 提取、完成标记解析与命令区段处理，本函数只负责内部区段提取与校验。; TerminusJSONPlainParser._extract_json_content：二者都做“从原始响应中抽出结构化主体”的工作，但一个面向 XML 标签，一个面向 JSON 对象边界。

XML区段顺序告警检查器

stage 上下文: 该函数是一个纯局部校验例程，职责仅限于检查 content 中若干 XML 风格区段的出现顺序是否符合约定。它不解析完整结构、不验证区段是否必然存在，也不读取任何实例状态；可见产出只有对传入 warnings 列表的条件性追加。

这段代码在干什么

函数扫描输入字符串 content 中 analysis、plan、commands 三类区段的起始标签首次出现位置，并比较这些“实际出现过的区段”之间的先后次序。若至少有两个区段被识别且顺序不符合期望，就向 warnings 追加一条人类可读的告警。函数自身返回 None，没有其他输出或状态写入。

接口 · 参数 / IO

(self, content: str, warnings: List[str]) -> None

• 参数: content: str — 被扫描的文本；函数在其中查找 <analysis、<plan、<commands 形式的起始标签; warnings: List[str] — 可变告警列表；当区段顺序异常时在末尾追加一条说明字符串
• 返回: 返回 None；真正产出是必要时对参数 warnings 的原位追加。
• 副作用: 可能调用 warnings.append(...) 追加一条“Sections appear in wrong order...”告警

执行流

1. 初始化局部字典 positions = {}，准备记录各区段在 content 中的起始位置。
2. 依次处理固定列表 ['analysis', 'plan', 'commands']；对每个区段执行一次 re.search(f"<{section}(?:\\s|>|/>)", content)，只取该区段首个匹配的起始位置 match.start()，若未匹配则该区段不进入 positions。
3. 若 positions 中少于两个区段（len(positions) < 2），函数立即返回，因为不足以比较先后顺序。
4. 定义期望顺序 expected_order = ['analysis', 'plan', 'commands']，然后按该期望顺序筛出实际出现过的区段，形成 present_sections，其中每项为 (section, position)。
5. 按记录的位置值对 present_sections 排序，得到实际出现顺序 actual_order；同时按 expected_order 过滤出仅包含已出现区段的期望子序列 expected_present。
6. 比较 actual_order 与 expected_present；若两者不同，则把二者分别用 → 连接成字符串，并向 warnings 追加一条顺序错误告警。

源码

    def _check_section_order(self, content: str, warnings: List[str]) -> None:
        """Check if sections appear in the correct order: analysis, plan, commands."""
        # Find positions of each section
        positions = {}
        for section in ["analysis", "plan", "commands"]:
            # Look for opening tags
            match = re.search(f"<{section}(?:\\s|>|/>)", content)
            if match:
                positions[section] = match.start()

        # Check if we have at least 2 sections to compare order
        if len(positions) < 2:
            return

        # Expected order
        expected_order = ["analysis", "plan", "commands"]

        # Get sections that are present, in the order they appear
        present_sections = []
        for section in expected_order:
            if section in positions:
                present_sections.append((section, positions[section]))

        # Sort by position to get actual order
        actual_order = [
            section for section, pos in sorted(present_sections, key=lambda x: x[1])
        ]

        # Get expected order for present sections only
        expected_present = [s for s in expected_order if s in positions]

        # Compare orders
        if actual_order != expected_present:
            actual_str = " → ".join(actual_order)
            expected_str = " → ".join(expected_present)
            warnings.append(
                f"Sections appear in wrong order. Found: {actual_str}, "
                f"expected: {expected_str}"
            )

Non-obvious 设计决策

• 该实现只用一次 re.search(...) 获取每类区段的首个起始标签位置，因此重复出现的同名区段不会被完整纳入顺序判断；这把检查目标限定为“首次出现的宏观顺序”，而非完整文档结构一致性。
• 顺序比较只针对 positions 中实际找到的区段子集进行：expected_present = [s for s in expected_order if s in positions]。这意味着缺失区段本身不会在这里触发告警；本函数专注于顺序问题，避免把“存在性”与“排序性”混在同一条规则里。
• 匹配模式使用 <{section}(?:\s|>|/>)，显式接受标签名后接空白、普通闭合尖括号或自闭合写法，而不是要求完整 XML 解析；这是一个宽松的文本级检测策略，代价是它仅凭起始标签形态判断，不验证嵌套或配对正确性。

上下游关系

• 调用方: 未知；源码片段未展示直接调用方; 任何需要对 content 执行区段顺序校验并收集告警的同类内部例程
• 核心被调用: re.search; match.start; sorted; warnings.append
• 配置/状态来源: 固定局部常量 ['analysis', 'plan', 'commands']; 固定局部模式模板 f"<{section}(?:\\s|>|/>)"; 局部期望顺序 expected_order = ['analysis', 'plan', 'commands']
• 结果去向: 传入的 warnings 列表; 调用方后续对 warnings 的消费逻辑
• 同类 sibling: 与 TerminusJSONPlainParser._check_field_order 形态相近：二者都只检查预定义字段/区段的文本出现顺序，并在异常时追加一条告警。; 相较于 TerminusXMLPlainParser._check_extra_text，本函数不关心块外文本，只处理三个特定区段之间的相对位置。

判定任务完成标签的布尔检查器

stage 上下文: 该函数是 XML 纯文本解析器内部的一个细粒度判定步骤，职责仅限于检查传入字符串中是否出现满足特定格式的 <task_complete> 标签。与同类兄弟函数相比，它不负责区段提取、顺序检查或告警累积，只产出一个布尔值。其实现完全自足：只读取参数 response_content，不读取实例状态。

这段代码在干什么

函数 _check_task_complete 接收字符串参数 response_content，检查其中是否存在大小写不敏感的 <task_complete>...</task_complete> 标签，且标签内部文本匹配 \s*true\s*。若找到这样的匹配则返回 True，否则返回 False。函数不修改 warnings、不构造区段字典，也不写入任何 self 状态或外部状态。

接口 · 参数 / IO

(self, response_content: str) -> bool

• 参数: response_content: str — 唯一输入文本；作为正则搜索目标，用于检查是否含有大小写不敏感且内容匹配 \s*true\s* 的 <task_complete> 标签
• 返回: 返回布尔值：仅当 response_content 中存在匹配模式 <task_complete>\s*true\s*</task_complete>（re.IGNORECASE）的片段时返回 True；其中 true 两侧允许出现空白。其余情况一律返回 False。

执行流

1. 调用 re.search(...) 在参数 response_content 上执行一次正则搜索，模式固定为 r"<task_complete>\s*true\s*</task_complete>"，并启用 re.IGNORECASE。
2. 将搜索结果保存到局部变量 true_match。
3. 若 true_match 为真值，则立即返回 True。
4. 若未匹配到上述模式，则落入函数尾部并返回 False。

源码

    def _check_task_complete(self, response_content: str) -> bool:
        """Check if the response indicates the task is complete."""
        # Check for <task_complete>true</task_complete>
        true_match = re.search(
            r"<task_complete>\s*true\s*</task_complete>",
            response_content,
            re.IGNORECASE,
        )
        if true_match:
            return True

        # All other cases (false, empty, self-closing, missing) = not complete
        return False

Non-obvious 设计决策

• 完成判定被收敛为“是否匹配一个唯一正向模式”：代码只接受正则 r"<task_complete>\s*true\s*</task_complete>"，因此判定标准明确且实现极简，但不会区分“显式 false”“空标签”“自闭合标签”“缺失标签”等不同未完成形态。
• 匹配使用 re.IGNORECASE，因此标签名与内部 true 的大小写变体都会被接受；这放宽了输入格式约束，但仍保持结构上的严格性，因为开始标签、结束标签及中间文本模式都必须同时满足。
• 内部文本使用 \s*true\s* 而非字面连续 true，可容忍 true 前后的空白字符；可观察效果是对常见格式化差异更宽容，而无需额外的裁剪步骤。

上下游关系

• 调用方: TerminusXMLPlainParser 内部其他解析步骤会把待检查的响应文本作为 response_content 传入本函数
• 核心被调用: re.search
• 配置/状态来源: 无外部配置或实例字段输入；唯一数据源是显式参数 response_content; 匹配规则直接内嵌在正则字面量 r"<task_complete>\s*true\s*</task_complete>" 中; 大小写处理策略直接来自 re.IGNORECASE 标志; 返回分支仅依赖局部变量 true_match 的真值
• 结果去向: 返回给直接调用方一个布尔结果 True 或 False; 该结果不伴随告警列表修改; 该结果不伴随错误字符串产出; 该结果不写入任何实例寄存状态
• 同类 sibling: 相较于 TerminusXMLPlainParser._extract_sections，本函数不提取多类区段，也不向 warnings 追加缺失/重复/顺序类告警。; 相较于 TerminusXMLPlainParser._check_extra_text 与 _check_section_order，本函数没有原位修改传入列表的副作用，输出形式更单一。; 相较于 TerminusXMLPlainParser._try_parse_response，本函数不构造 ParseResult，仅负责一个局部布尔判定。

XML 命令区段的手工提取器

stage 上下文: 该函数位于 Response Parse 阶段，职责是把已拿到的 xml_content 中 <keystrokes>...</keystrokes> 片段转换为命令对象，同时汇集解析期告警。它不负责整段响应的区段识别，而是聚焦命令明细，和同类兄弟函数 _extract_sections、_check_task_complete 分工互补。相较于 JSON 侧的 _parse_commands，这里处理的是 XML/plain 形态下的命令正文与标签属性问题。

这段代码在干什么

函数接收 xml_content: str 与可变告警列表 warnings，从中提取所有 <keystrokes> 命令块，解析每条命令的 duration 属性，并生成 ParsedCommand 列表。对缺失或非法的时长、非末条命令缺少结尾换行、XML 实体字面量以及字符串中出现的 "\\r\\n"，它会直接向 warnings 追加告警。返回值始终是 (commands, "") 这一二元组，其中错误字符串在本函数内固定为空。

接口 · 参数 / IO

(self, xml_content: str, warnings: List[str]) -> tuple[List[ParsedCommand], str]

• 参数: self: ? — 解析器实例；仅用于调用 self._check_attribute_issues(...); xml_content: str — 待扫描的 XML 风格命令文本；既作为 <keystrokes> 块提取源，也在循环后被整体扫描以检查 XML 实体字面量与 "\\r\\n"; warnings: List[str] — 调用方提供的可变告警容器；本函数直接 append(...) 多类告警，并传给 _check_attribute_issues(...) 供其进一步补充
• 返回: 返回 (commands, "")：前者是按出现顺序构造的 ParsedCommand 列表，后者为固定空字符串；真正的附加产出是对 warnings 的原位修改。
• 副作用: 直接向 warnings 追加时长缺失、时长非法、非末条命令缺少结尾换行、XML 实体字面量、以及 "\\r\\n" 相关告警; 通过调用 self._check_attribute_issues(attributes_str, i + 1, warnings)，可能间接向 warnings 追加属性问题告警

执行流

1. 先编译 keystrokes_pattern = re.compile(r"<keystrokes([^>]*)>(.*?)</keystrokes>", re.DOTALL)，并对 xml_content 执行 findall(...)，得到每个命令的属性串 attributes_str 与正文 keystrokes_content。
2. 按匹配顺序遍历各命令；每轮首先调用 self._check_attribute_issues(attributes_str, i + 1, warnings) 检查当前 <keystrokes ...> 的属性问题。
3. 对每条命令把 duration 先初始化为 1.0，再用 re.search(r'duration\s*=\s*["\']([^"\']*)["\']', attributes_str) 提取时长属性；若存在且可转为 float 则采用该值，若转换抛出 ValueError 或属性缺失，则保留默认值并分别追加对应告警。
4. 仅当当前命令不是最后一条（i < len(matches) - 1）时，检查 keystrokes_content.endswith("\n")；若不以换行结束，则追加“后续命令会拼接到同一行”的告警。
5. 将当前条目构造成 ParsedCommand(keystrokes=keystrokes_content, duration=duration) 并加入 commands 列表。
6. 在命令循环结束后，基于预定义 entities 映射对整个 xml_content 做一次全文扫描；每发现一个 <、>、&、"、' 字面量，就追加一条“本函数不会转换该实体”的告警。
7. 最后再次针对整个 xml_content 检查是否包含字符串 "\\r\\n"；若包含，则追加建议改用 "\\n" 的告警，并返回 (commands, "")。

源码

    def _parse_xml_commands(
        self, xml_content: str, warnings: List[str]
    ) -> tuple[List[ParsedCommand], str]:
        """Parse XML content and extract command objects manually."""

        # Find all keystrokes elements manually for better error reporting
        commands = []
        keystrokes_pattern = re.compile(
            r"<keystrokes([^>]*)>(.*?)</keystrokes>", re.DOTALL
        )

        matches = keystrokes_pattern.findall(xml_content)
        for i, (attributes_str, keystrokes_content) in enumerate(matches):
            # Check for attribute issues
            self._check_attribute_issues(attributes_str, i + 1, warnings)

            # Parse attributes
            duration = 1.0

            # Parse duration attribute
            duration_match = re.search(
                r'duration\s*=\s*["\']([^"\']*)["\']', attributes_str
            )
            if duration_match:
                try:
                    duration = float(duration_match.group(1))
                except ValueError:
                    warnings.append(
                        f"Command {i + 1}: Invalid duration value "
                        f"'{duration_match.group(1)}', using default 1.0"
                    )
            else:
                warnings.append(
                    f"Command {i + 1}: Missing duration attribute, using default 1.0"
                )

            # Check for newline at end of keystrokes
            if i < len(matches) - 1 and not keystrokes_content.endswith("\n"):
                warnings.append(
                    f"Command {i + 1} should end with newline when followed "
                    f"by another command. Otherwise the two commands will be "
                    f"concatenated together on the same line."
                )

            commands.append(
                ParsedCommand(keystrokes=keystrokes_content, duration=duration)
            )

        # Check for XML entities and warn
        entities = {
            "&lt;": "<",
            "&gt;": ">",
            "&amp;": "&",
            "&quot;": '"',
            "&apos;": "'",
        }
        for entity, char in entities.items():
            if entity in xml_content:
                warnings.append(
                    f"Warning: {entity} is read verbatim and not converted to {char}. "
                    f"NEVER USE {entity}, unless you want these exact characters to "
                    f"appear directly in the output."
                )

        # Check for \r\n line endings and warn
        if "\\r\\n" in xml_content:
            warnings.append(
                "Warning: \\r\\n line endings are not necessary - use \\n "
                "instead for simpler output"
            )

        return commands, ""

Non-obvious 设计决策

• 函数采用针对 <keystrokes> 的正则扫描与局部属性解析，而不是在此处引入完整 XML 解析流程；从代码可见，其目标是围绕 attributes_str、keystrokes_content 和命令序号 i + 1 产出更细粒度、定位明确的告警。
• duration 的处理采取“默认接受、异常降级”策略：先设为 1.0，再尝试覆盖；即使属性缺失或 float(...) 失败，也只记录告警而不拒绝该命令。这一取舍直接体现在默认值初始化与 except ValueError 分支上。
• 换行检查被限定在“后面仍有下一条命令”的场景下才触发，而非对最后一条命令一律要求结尾换行；这是代码中 i < len(matches) - 1 条件表达的有意识约束，用于仅拦截可能导致相邻命令文本并行拼接的情况。
• 对 XML 实体与 "\\r\\n" 的处理仅做全文扫描和告警，不做任何替换或解码；这一点由循环中的 warnings.append(...) 与函数末尾直接返回原始提取结果可见。

上下游关系

• 调用方: TerminusXMLPlainParser._try_parse_response（在获得命令区段文本后调用本函数）
• 核心被调用: self._check_attribute_issues; re.compile; keystrokes_pattern.findall; re.search; float; ParsedCommand
• 配置/状态来源: 参数 xml_content：本函数全部解析与全文扫描的输入来源; 参数 warnings：本函数告警输出的承载容器; 局部常量 entities：实体字面量检查规则来源; 局部正则 keystrokes_pattern 与 duration_match 模式：命令块与时长属性的提取规则来源
• 结果去向: 返回的 commands 列表进入调用方后续的 ParseResult.commands 组装路径; 返回的空错误字符串作为调用方判定“本函数未报告硬错误”的信号; 被原位扩充的 warnings 交回调用方，用于后续汇总解析告警
• 同类 sibling: TerminusXMLPlainParser._extract_sections：负责顶层区段提取，本函数只处理已给定的命令内容; TerminusXMLPlainParser._check_extra_text：检查 <response> 块外文本，本函数不关注响应外层结构; TerminusJSONPlainParser._parse_commands：同为命令对象生成，但输入形态是 JSON 命令数组而非 XML 风格文本

XML 命令属性问题告警检查器

stage 上下文: 该函数是 XML 解析器内部的一个局部校验步骤，专门检查单条命令标签的属性字符串 attributes_str。它不负责拒绝输入或构造命令对象，只把发现的问题追加到调用方提供的 warnings。与同类兄弟函数相比，它聚焦于“属性书写形式与属性名”这一窄域，而 _parse_xml_commands 负责整体命令提取与时长解析。

这段代码在干什么

函数接收属性子串 attributes_str、用于生成诊断文案的命令编号 command_num，以及可变列表 warnings。它分三轮独立扫描：未加引号的属性值、使用单引号的属性值、以及不在允许集合内的属性名；每发现一项就向 warnings 追加一条人类可读的告警。函数返回 None，真正产出是对 warnings 的原位累积修改；由于三轮检查互不排斥，同一个属性可累计得到多条告警。

接口 · 参数 / IO

(self, attributes_str: str, command_num: int, warnings: List[str]) -> None

• 参数: self: ? — 实例方法接收者；本函数体内未读取任何实例属性; attributes_str: str — 待检查的原始属性文本，作为三个正则扫描的输入; command_num: int — 写入告警文案中的命令序号标签; warnings: List[str] — 由调用方提供的可变告警列表；本函数向其中追加零到多条消息
• 返回: 返回 None；实际输出是对 warnings 的原位追加。告警是跨三轮独立检查累积的，同一属性可能产生多条。
• 副作用: 每次调用都会在函数内部编译两个正则 unquoted_pattern 与 single_quote_pattern; 按匹配结果向传入的 warnings 列表追加字符串告警

执行流

1. 先在本次调用内部编译 unquoted_pattern = re.compile(r'(\w+)\s*=\s*([^"\'\s>]+)')，用它在 attributes_str 中查找未用引号包裹的 name=value 形式；对 findall(...) 返回的每个 (attr_name, attr_value)，追加一条建议改写为双引号形式的告警，并在文案中带上 command_num。
2. 再编译 single_quote_pattern = re.compile(r"(\w+)\s*=\s*'([^']*)'")，扫描 attributes_str 中使用单引号的属性；对每个匹配追加一条“应使用双引号”的告警，同样带上命令编号与建议格式。
3. 随后在函数内定义允许属性集合 known_attributes = {"duration"}，这是一个局部硬编码常量。
4. 接着执行 all_attributes = re.findall(r"(\w+)\s*=", attributes_str)，从 attributes_str 中提取每一个出现过的 name=；这一轮是逐出现位置检查，不做去重，也不受前两轮是否已命中影响。
5. 最后遍历 all_attributes 中的每个 attr_name，凡是不在 known_attributes 内的都追加“Unknown attribute”告警，并在文案中列出当前已知属性名。

源码

    def _check_attribute_issues(
        self, attributes_str: str, command_num: int, warnings: List[str]
    ) -> None:
        """Check for attribute-related issues."""
        # Check for missing quotes
        unquoted_pattern = re.compile(r'(\w+)\s*=\s*([^"\'\s>]+)')
        unquoted_matches = unquoted_pattern.findall(attributes_str)
        for attr_name, attr_value in unquoted_matches:
            warnings.append(
                f"Command {command_num}: Attribute '{attr_name}' value should be "
                f'quoted: {attr_name}="{attr_value}"'
            )

        # Check for single quotes (should use double quotes for consistency)
        single_quote_pattern = re.compile(r"(\w+)\s*=\s*'([^']*)'")
        single_quote_matches = single_quote_pattern.findall(attributes_str)
        for attr_name, attr_value in single_quote_matches:
            warnings.append(
                f"Command {command_num}: Use double quotes for attribute "
                f"'{attr_name}': {attr_name}=\"{attr_value}\""
            )

        # Check for unknown attribute names
        known_attributes = {"duration"}
        all_attributes = re.findall(r"(\w+)\s*=", attributes_str)
        for attr_name in all_attributes:
            if attr_name not in known_attributes:
                warnings.append(
                    f"Command {command_num}: Unknown attribute '{attr_name}' - "
                    f"known attributes are: {', '.join(sorted(known_attributes))}"
                )

Non-obvious 设计决策

• 函数采用“三轮独立告警”而非单轮统一判定：未加引号、单引号风格、未知属性名分别扫描并分别追加消息。这样会让同一属性同时触发多条告警，例如既因引号形式不合要求，又因名称不在允许集合中被再次报告；代码中没有做互斥或去重。
• 未知属性检查只基于属性名是否出现在局部常量 known_attributes = {"duration"} 中，不参考属性值是否可解析，也不读取任何 self 配置。这使规则来源非常明确，但也意味着允许属性集并非外部可配置状态。
• 函数选择仅追加 warnings 而不返回错误、也不提前中断，表明这里的定位是提示式校验而不是拒绝式验证。代码里三段循环都没有 raise、return 提前结束或错误汇总对象。
• 属性名收集使用 re.findall(r"(\w+)\s*=", attributes_str) 对所有 name= 出现位置逐个处理，不做去重。其结果是重复书写的同名属性会被重复告警，保留了输入文本中的逐处诊断粒度。

上下游关系

• 调用方: TerminusXMLPlainParser._parse_xml_commands（在解析单条 XML 命令时传入属性串与 warnings）
• 核心被调用: re.compile; Pattern.findall; re.findall; warnings.append
• 配置/状态来源: 参数 attributes_str：提供被扫描的原始属性文本; 参数 command_num：提供告警文案中的命令编号; 局部常量 known_attributes = {"duration"}：定义允许的属性名集合; 正则字面量 r'(\w+)\s*=\s*([^"\'\s>]+)'、r"(\w+)\s*=\s*'([^']*)'"、r"(\w+)\s*="：定义三类检查规则
• 结果去向: 调用方持有的 warnings 列表; TerminusXMLPlainParser._parse_xml_commands（通过被原位修改的 warnings 接收结果）
• 同类 sibling: TerminusXMLPlainParser._parse_xml_commands：负责提取 <keystrokes> 与 duration，本函数只补充属性层面的告警; TerminusXMLPlainParser._extract_sections：同样通过可变 warnings 累积解析问题，但其关注点是顶层区段而非命令属性; TerminusXMLPlainParser._check_extra_text：也是仅通过原位追加告警输出结果的辅助检查函数