Run evidence, ready to inspect.

============================= test session starts =============================
platform win32 -- Python 3.13.9, pytest-9.1.1, pluggy-1.6.0
rootdir: [local path]
configfile: pyproject.toml
plugins: anyio-4.14.2, cov-7.1.0
collected 12 items

tests\test_maintenance_case.py FFF..FF.F.F.                              [100%]

================================== FAILURES ===================================
__ test_observation_status[read_file-1 | from errors import NotFound-False] ___

action = 'read_file', observation = '1 | from errors import NotFound'
failed = False

    @pytest.mark.parametrize(
        "action,observation,failed",
        [
            ("read_file", "1 | from errors import NotFound", False),
            ("edit_file", "已替换 1 处。\n2 | raise ValueError('invalid')", False),
            ("list_directory", "errors.py\nmain.py", False),
            ("run_python", "stdout:\nERROR: expected fixture\nreturncode: 0", False),
            ("run_tests", "stdout:\nall tests passed\nreturncode: 0", False),
            ("run_python", "stderr:\nprocess stopped\nreturncode: 2", True),
            ("run_shell", "stderr:\nprocess stopped\nreturncode: 255", True),
            ("read_file", "文件 missing.py 不存在。", True),
            ("read_file", "路径 tests 不是文件。", True),
            ("edit_file", "Tool execution failed: permission denied", True),
            ("edit_file", "[语法检查] 未通过：invalid syntax", True),
            ("third_party_tool", "Error: extension failed", True),
        ],
    )
    def test_observation_status(action, observation, failed):
        # 历史接口没有 action 参数；调用真实旧接口，避免以 TypeError 冒充缺陷复现。
        kwargs = {"action": action} if "action" in inspect.signature(is_failure_observation).parameters else {}
>       assert is_failure_observation(observation, **kwargs) is failed
E       AssertionError: assert True is False
E        +  where True = is_failure_observation('1 | from errors import NotFound', **{})

tests\test_maintenance_case.py:30: AssertionError
_ test_observation_status[edit_file-\u5df2\u66ff\u6362 1 \u5904\u3002\n2 | raise ValueError('invalid')-False] _

action = 'edit_file', observation = "已替换 1 处。\n2 | raise ValueError('invalid')"
failed = False

    @pytest.mark.parametrize(
        "action,observation,failed",
        [
            ("read_file", "1 | from errors import NotFound", False),
            ("edit_file", "已替换 1 处。\n2 | raise ValueError('invalid')", False),
            ("list_directory", "errors.py\nmain.py", False),
            ("run_python", "stdout:\nERROR: expected fixture\nreturncode: 0", False),
            ("run_tests", "stdout:\nall tests passed\nreturncode: 0", False),
            ("run_python", "stderr:\nprocess stopped\nreturncode: 2", True),
            ("run_shell", "stderr:\nprocess stopped\nreturncode: 255", True),
            ("read_file", "文件 missing.py 不存在。", True),
            ("read_file", "路径 tests 不是文件。", True),
            ("edit_file", "Tool execution failed: permission denied", True),
            ("edit_file", "[语法检查] 未通过：invalid syntax", True),
            ("third_party_tool", "Error: extension failed", True),
        ],
    )
    def test_observation_status(action, observation, failed):
        # 历史接口没有 action 参数；调用真实旧接口，避免以 TypeError 冒充缺陷复现。
        kwargs = {"action": action} if "action" in inspect.signature(is_failure_observation).parameters else {}
>       assert is_failure_observation(observation, **kwargs) is failed
E       assert True is False
E        +  where True = is_failure_observation("已替换 1 处。\n2 | raise ValueError('invalid')", **{})

tests\test_maintenance_case.py:30: AssertionError
______ test_observation_status[list_directory-errors.py\nmain.py-False] _______

action = 'list_directory', observation = 'errors.py\nmain.py', failed = False

    @pytest.mark.parametrize(
        "action,observation,failed",
        [
            ("read_file", "1 | from errors import NotFound", False),
            ("edit_file", "已替换 1 处。\n2 | raise ValueError('invalid')", False),
            ("list_directory", "errors.py\nmain.py", False),
            ("run_python", "stdout:\nERROR: expected fixture\nreturncode: 0", False),
            ("run_tests", "stdout:\nall tests passed\nreturncode: 0", False),
            ("run_python", "stderr:\nprocess stopped\nreturncode: 2", True),
            ("run_shell", "stderr:\nprocess stopped\nreturncode: 255", True),
            ("read_file", "文件 missing.py 不存在。", True),
            ("read_file", "路径 tests 不是文件。", True),
            ("edit_file", "Tool execution failed: permission denied", True),
            ("edit_file", "[语法检查] 未通过：invalid syntax", True),
            ("third_party_tool", "Error: extension failed", True),
        ],
    )
    def test_observation_status(ac
[truncated: showing first 4800 and last 3200 of 12691 chars (227 lines total); narrow the command or use read_file with line_start/line_end on specific files to view the omitted range]
"third_party_tool", "Error: extension failed", True),
        ],
    )
    def test_observation_status(action, observation, failed):
        # 历史接口没有 action 参数；调用真实旧接口，避免以 TypeError 冒充缺陷复现。
        kwargs = {"action": action} if "action" in inspect.signature(is_failure_observation).parameters else {}
>       assert is_failure_observation(observation, **kwargs) is failed
E       AssertionError: assert False is True
E        +  where False = is_failure_observation('路径 tests 不是文件。', **{})

tests\test_maintenance_case.py:30: AssertionError
_ test_observation_status[edit_file-[\u8bed\u6cd5\u68c0\u67e5] \u672a\u901a\u8fc7\uff1ainvalid syntax-True] _

action = 'edit_file', observation = '[语法检查] 未通过：invalid syntax', failed = True

    @pytest.mark.parametrize(
        "action,observation,failed",
        [
            ("read_file", "1 | from errors import NotFound", False),
            ("edit_file", "已替换 1 处。\n2 | raise ValueError('invalid')", False),
            ("list_directory", "errors.py\nmain.py", False),
            ("run_python", "stdout:\nERROR: expected fixture\nreturncode: 0", False),
            ("run_tests", "stdout:\nall tests passed\nreturncode: 0", False),
            ("run_python", "stderr:\nprocess stopped\nreturncode: 2", True),
            ("run_shell", "stderr:\nprocess stopped\nreturncode: 255", True),
            ("read_file", "文件 missing.py 不存在。", True),
            ("read_file", "路径 tests 不是文件。", True),
            ("edit_file", "Tool execution failed: permission denied", True),
            ("edit_file", "[语法检查] 未通过：invalid syntax", True),
            ("third_party_tool", "Error: extension failed", True),
        ],
    )
    def test_observation_status(action, observation, failed):
        # 历史接口没有 action 参数；调用真实旧接口，避免以 TypeError 冒充缺陷复现。
        kwargs = {"action": action} if "action" in inspect.signature(is_failure_observation).parameters else {}
>       assert is_failure_observation(observation, **kwargs) is failed
E       AssertionError: assert False is True
E        +  where False = is_failure_observation('[语法检查] 未通过：invalid syntax', **{})

tests\test_maintenance_case.py:30: AssertionError
=========================== short test summary info ===========================
FAILED tests/test_maintenance_case.py::test_observation_status[read_file-1 | from errors import NotFound-False]
FAILED tests/test_maintenance_case.py::test_observation_status[edit_file-\u5df2\u66ff\u6362 1 \u5904\u3002\n2 | raise ValueError('invalid')-False]
FAILED tests/test_maintenance_case.py::test_observation_status[list_directory-errors.py\nmain.py-False]
FAILED tests/test_maintenance_case.py::test_observation_status[run_python-stderr:\nprocess stopped\nreturncode: 2-True]
FAILED tests/test_maintenance_case.py::test_observation_status[run_shell-stderr:\nprocess stopped\nreturncode: 255-True]
FAILED tests/test_maintenance_case.py::test_observation_status[read_file-\u8def\u5f84 tests \u4e0d\u662f\u6587\u4ef6\u3002-True]
FAILED tests/test_maintenance_case.py::test_observation_status[edit_file-[\u8bed\u6cd5\u68c0\u67e5] \u672a\u901a\u8fc7\uff1ainvalid syntax-True]
========================= 7 failed, 5 passed in 2.67s =========================
returncode: 1

"""观察结果的判定与边界处理。

``is_failure_observation`` 从 ``ReactAgent._is_failure_observation`` 提出来，让不住在
内核里的能力（例如工具熔断）也能复用同一份失败判定，而不必反向 import ReactAgent。
``ReactAgent._is_failure_observation`` 保留为指向本函数的薄委托，公开行为不变。

``ObservationBounder`` 是内核护栏：它先于 ``after_tool_result`` 钩子链执行，
保证模型、对话历史与 trace 看到的是同一份（可能已截断的）文本。
"""

from __future__ import annotations

from typing import Any

from dm_agent.memory.context_budget import truncate_observation

from .run_state import RunContext

# 判定一条观察是否代表失败。措辞约束见 memory/context_budget 模块文档：
# 内核生成的护栏文案（截断标记、守卫拒绝、熔断拦截）都刻意避开这些词。
FAILURE_MARKERS = (
    "Tool execution failed",
    "Unknown tool",
    "Tool arguments",
    "parse failed",
    "Critic rejected",
    "Critic review failed",
    "returncode: 1",
    "error",
    "Error",
    "Traceback",
    "失败",
    "错误",
    "不存在",
)


def is_failure_observation(observation: str) -> bool:
    """观察文本里是否出现失败标记。"""
    return any(marker in observation for marker in FAILURE_MARKERS)


class ObservationBounder:
    """按字符上限截断观察，并把截断事实记进 metadata 与 trace。"""

    def __init__(self, *, max_chars: int, trace_writer: Any | None = None) -> None:
        self.max_chars = max_chars
        self.trace_writer = trace_writer

    def bound(
        self,
        observation: Any,
        *,
        action: str,
        action_input: Any,
        context: RunContext,
    ) -> str:
        """截断超长观察并记录审计信息；模型、历史与 trace 看到同一份文本。"""
        result = truncate_observation(
            str(observation),
            max_chars=self.max_chars,
            action=action,
            action_input=action_input,
        )
        if result.truncated:
            metadata = context.metadata
            metadata["truncation_count"] += 1
            metadata["truncated_chars_saved"] += result.original_chars - result.kept_chars
            if self.trace_writer:
                self.trace_writer.record(
                    "observation_truncated",
                    {
                        "step_number": context.step_number,
                        "action": action,
                        "original_chars": result.original_chars,
                        "kept_chars": result.kept_chars,
                        "original_lines": result.original_lines,
                    },
                )
        return result.text

在 dm_agent\core\agent.py 中找到 6 处匹配：

    24:     HookFailure,
    25:     LLMRequestClient,
    26:     RunEndEvent,
    27:     RunStartEvent,
    28: )
    29: from .guards import ReadBeforeEditGuard
>>> 30: from .observation import ObservationBounder, is_failure_observation
    31: from .persistence import (
    32:     RunPersistence,
    33:     agent_config_snapshot,
    34:     json_safe_metadata,
    35:     metadata_from_checkpoint,
    36:     plan_from_checkpoint,

    654:             if self.trace_writer:
    655:                 self.trace_writer.record_tool_call(
    656:                     step_number=step_num,
    657:                     action=action,
    658:                     action_input=action_input,
    659:                     observation=observation,
>>> 660:                     failed=self._is_failure_observation(observation),
    661:                 )
    662: 
    663:             # 更新计划进度（如果有计划；被守卫拦下的编辑不算完成）。
    664:             # 只有下一个待办步骤的动作匹配时才标记，消除同名动作的二义性。
    665:             if plan and self.planner and not guard_blocked:
    666:                 next_step = self.planner.get_next_step()

    671:             tool_info = f"执行工具 {action}，输入：{json.dumps(action_input, ensure_ascii=False)}\n观察：{observation}"
    672:             self._append_history("user", tool_info, kind="tool_result")
    673: 
    674:             # 调用回调函数实时输出步骤
    675:             self._publish_step(step, step_num)
    676: 
>>> 677:             if self._is_failure_observation(observation) and plan and self.planner:
    678:                 plan = self._replan_after_failure(
    679:                     task,
    680:                     plan,
    681:                     metadata,
    682:                     FailureContext(
    683:                         observation=observation,

    688:                 )
    689: 
    690:             # 检查是否调用了 task_complete 工具
    691:             if (
    692:                 action == "task_complete"
    693:                 and accepted
>>> 694:                 and not self._is_failure_observation(observation)
    695:             ):
    696:                 metadata["status"] = "success"
    697:                 metadata["failure_reason"] = ""
    698:                 metadata["duration_seconds"] = time.perf_counter() - started_at
    699:                 return finish_result(observation)
    700: 

    794:             agent_config=self._config_snapshot(max_steps=limit),
    795:             cwd=str(Path.cwd()),
    796:         )
    797:         self._persistence.save(path, checkpoint)
    798: 
    799:     @staticmethod
>>> 800:     def _is_failure_observation(observation: str) -> bool:
    801:         """委托到 ``core.observation``，让内核外的能力复用同一份失败判定。"""
    802:         return is_failure_observation(observation)
    803: 
    804:     def _replan_after_failure(
    805:         self,
    806:         task: str,

    796:         )
    797:         self._persistence.save(path, checkpoint)
    798: 
    799:     @staticmethod
    800:     def _is_failure_observation(observation: str) -> bool:
    801:         """委托到 ``core.observation``，让内核外的能力复用同一份失败判定。"""
>>> 802:         return is_failure_observation(observation)
    803: 
    804:     def _replan_after_failure(
    805:         self,
    806:         task: str,
    807:         plan: list[PlanStep],
    808:         metadata: dict[str, Any],

                    self.step_callback(step_num, step)
                if self.trace_writer:
                    self.trace_writer.record_tool_call(
                        step_number=step_num,
                        action=action,
                        action_input=action_input,
                        observation=observation,
                        failed=True,
                    )
                    self.trace_writer.record_step(step_number=step_num, step=step)
                if plan and self.planner:
                    plan = self._replan_after_failure(
                        task,
                        plan,
                        metadata,
                        FailureContext(
                            observation=observation,
                            action=action,
                            step_number=step_num,
                            error_kind="unknown_tool",
                        ),
                    )
                continue

            invocation = self._tool_invoker.invoke(
                tool,
                action=action,
                action_input=action_input,
                context=self._run_context,
            )
            action_input = invocation.arguments
            observation = invocation.observation
            error_kind = invocation.error_kind
            guard_blocked = invocation.blocked  # 被守卫拦下的编辑不算计划完成
            accepted = False
            if action == "task_complete" and invocation.tool_succeeded:
                accepted, observation = self._completion_gate.review(
                    task=task,
                    action=action,
                    completion_text=str(observation),
                    steps=steps,
                    context=self._run_context,
                )
                if not accepted:
                    error_kind = "critic_rejected"

            step = Step(
                thought=thought,
                action=action,
                action_input=action_input,
                observation=observation,
                raw=raw,
            )
            steps.append(step)
            if self.trace_writer:
                self.trace_writer.record_tool_call(
                    step_number=step_num,
                    action=action,
                    action_input=action_input,
                    observation=observation,
                    failed=self._is_failure_observation(observation),
                )

            # 更新计划进度（如果有计划；被守卫拦下的编辑不算完成）。
            # 只有下一个待办步骤的动作匹配时才标记，消除同名动作的二义性。
            if plan and self.planner and not guard_blocked:
                next_step = self.planner.get_next_step()
                if next_step and next_step.action == action:
                    self.planner.mark_completed(next_step.step_number, observation)

            # 将工具执行结果添加到历史记录
            tool_info = f"执行工具 {action}，输入：{json.dumps(action_input, ensure_ascii=False)}\n观察：{observation}"
            self._append_history("user", tool_info, kind="tool_result")

            # 调用回调函数实时输出步骤
            self._publish_step(step, step_num)

            if self._is_failure_observation(observation) and plan and self.planner:
                plan = self._replan_after_failure(
                    task,
                    plan,
                    metadata,
                    FailureContext(
                        observation=observation,
                        action=action,
                        step_number=step_num,
                        error_kind=error_kind or None,
                    ),
                )

            # 检查是否调用了 task_complete 工具
            if (
                action == "task_complete"
                and accepted
                and not self._is_failure_observation(observation)
            ):
                metadata["status"] = "success"
                metadata["failure_reason"] = ""
                metadata["duration_seconds"] = time.perf_counter() - started_at
                return finish_result(observation)

"""同一组行为断言用于历史缺陷版本与修复版本，不调用模型。"""

import inspect

import pytest

from dm_agent.core.observation import is_failure_observation


@pytest.mark.parametrize(
    "action,observation,failed",
    [
        ("read_file", "1 | from errors import NotFound", False),
        ("edit_file", "已替换 1 处。\n2 | raise ValueError('invalid')", False),
        ("list_directory", "errors.py\nmain.py", False),
        ("run_python", "stdout:\nERROR: expected fixture\nreturncode: 0", False),
        ("run_tests", "stdout:\nall tests passed\nreturncode: 0", False),
        ("run_python", "stderr:\nprocess stopped\nreturncode: 2", True),
        ("run_shell", "stderr:\nprocess stopped\nreturncode: 255", True),
        ("read_file", "文件 missing.py 不存在。", True),
        ("read_file", "路径 tests 不是文件。", True),
        ("edit_file", "Tool execution failed: permission denied", True),
        ("edit_file", "[语法检查] 未通过：invalid syntax", True),
        ("third_party_tool", "Error: extension failed", True),
    ],
)
def test_observation_status(action, observation, failed):
    # 历史接口没有 action 参数；调用真实旧接口，避免以 TypeError 冒充缺陷复现。
    kwargs = {"action": action} if "action" in inspect.signature(is_failure_observation).parameters else {}
    assert is_failure_observation(observation, **kwargs) is failed

Recorded replacement arguments, not a full repository diff. Check the tool receipt to see whether the edit was applied.

--- before excerpt+++ after excerpt@@ -1,5 +1,6 @@ from __future__ import annotations +import re from typing import Any  from dm_agent.memory.context_budget import truncate_observation@@ -24,7 +25,80 @@     "不存在", ) +# 文件/目录类工具：观察文本是工具携带的源码或目录列表，内容里出现 "error"、+# "Error"、"不存在" 等词并不代表工具失败。只有工具自身的拒绝/失败文案才算失败。+FILE_TOOL_ACTIONS = frozenset(+    {+        "read_file",+        "edit_file",+        "write_file",+        "create_file",+        "list_directory",+        "search_in_file",+        "search_files",+        "delete_file",+        "move_file",+        "copy_file",+    }+) -def is_failure_observation(observation: str) -> bool:-    """观察文本里是否出现失败标记。"""+# 进程类工具：观察文本里带 "returncode: N"，N 非 0 即失败；N 为 0 时即使+# stdout/stderr 里出现 ERROR 也不算失败。+PROCESS_TOOL_ACTIONS = frozenset(+    {+        "run_python",+        "run_shell",+        "run_tests",+        "run_command",+        "execute",+    }+)++# 文件工具自身的失败/拒绝文案（不含源码内容里可能出现的普通词）。+FILE_TOOL_FAILURE_MARKERS = (+    "Tool execution failed",+    "Unknown tool",+    "Tool arguments",+    "parse failed",+    "Critic rejected",+    "Critic review failed",+    "[语法检查] 未通过",+    "Traceback",+)++# 文件/路径类拒绝文案，例如 "文件 missing.py 不存在。"、"路径 tests 不是文件。"。+_FILE_TOOL_REFUSAL_PATTERN = re.compile(+    r"(?:文件|路径|目录|file|path|directory)\b[^\n]{0,200}?"+    r"(?:不存在|不是文件|不是目录|不是文件夹|not found|not a file|not a directory|permission denied|权限)",+    re.IGNORECASE,+)++_RETURNCODE_PATTERN = re.compile(r"returncode:\s*(-?\d+)")+++def _is_file_tool_failure(observation: str) -> bool:+    """文件/目录类工具：只认工具自身的失败/拒绝文案，忽略携带的源码内容。"""+    if any(marker in observation for marker in FILE_TOOL_FAILURE_MARKERS):+        return True+    return bool(_FILE_TOOL_REFUSAL_PATTERN.search(observation))+++def _is_process_tool_failure(observation: str) -> bool:+    """进程类工具：以 returncode 为准；缺失时回退到保守的标记判定。"""+    match = _RETURNCODE_PATTERN.search(observation)+    if match is not None:+        return int(match.group(1)) != 0     return any(marker in observation for marker in FAILURE_MARKERS)+++def is_failure_observation(observation: str, action: str | None = None) -> bool:+    """判断一条观察是否代表工具失败。++    ``action`` 是产生该观察的工具动作。带上它时按工具类型区分「工具携带的源码/输出」+    与「工具自身的状态」；缺失或属于未知扩展时保持保守的旧标记判定。+    """+    if action in FILE_TOOL_ACTIONS:+        return _is_file_tool_failure(observation)+    if action in PROCESS_TOOL_ACTIONS:+        return _is_process_tool_failure(observation)+    return any(marker in observation for marker in FAILURE_MARKERS)

已按内容替换 dm_agent\core\observation.py 的 1 处（653 -> 2685 字符）。
[编辑后] 第 8-47 行（共 154 行）：
     8 | 保证模型、对话历史与 trace 看到的是同一份（可能已截断的）文本。
     9 | """
    10 | 
>   11 | from __future__ import annotations
>   12 | 
>   13 | import re
>   14 | from typing import Any
>   15 | 
>   16 | from dm_agent.memory.context_budget import truncate_observation
>   17 | 
>   18 | from .run_state import RunContext
>   19 | 
>   20 | # 判定一条观察是否代表失败。措辞约束见 memory/context_budget 模块文档：
>   21 | # 内核生成的护栏文案（截断标记、守卫拒绝、熔断拦截）都刻意避开这些词。
>   22 | FAILURE_MARKERS = (
>   23 |     "Tool execution failed",
>   24 |     "Unknown tool",
>   25 |     "Tool arguments",
>   26 |     "parse failed",
>   27 |     "Critic rejected",
>   28 |     "Critic review failed",
>   29 |     "returncode: 1",
>   30 |     "error",
>   31 |     "Error",
>   32 |     "Traceback",
>   33 |     "失败",
>   34 |     "错误",
>   35 |     "不存在",
>   36 | )
>   37 | 
>   38 | # 文件/目录类工具：观察文本是工具携带的源码或目录列表，内容里出现 "error"、
>   39 | # "Error"、"不存在" 等词并不代表工具失败。只有工具自身的拒绝/失败文案才算失败。
>   40 | FILE_TOOL_ACTIONS = frozenset(
>   41 |     {
>   42 |         "read_file",
>   43 |         "edit_file",
>   44 |         "write_file",
>   45 |         "create_file",
>   46 |         "list_directory",
>   47 |         "search_in_file",
... 回显截断，仅显示前 40 行

Recorded replacement arguments, not a full repository diff. Check the tool receipt to see whether the edit was applied.

--- before excerpt+++ after excerpt@@ -1,4 +1,7 @@     @staticmethod-    def _is_failure_observation(observation: str) -> bool:-        """委托到 ``core.observation``，让内核外的能力复用同一份失败判定。"""-        return is_failure_observation(observation)+    def _is_failure_observation(observation: str, action: str | None = None) -> bool:+        """委托到 ``core.observation``，让内核外的能力复用同一份失败判定。++        ``action`` 是产生该观察的工具动作，用于区分工具携带的源码/输出与工具状态。+        """+        return is_failure_observation(observation, action=action)

已按内容替换 dm_agent\core\agent.py 的 1 处（184 -> 288 字符）。
[编辑后] 第 796-808 行（共 850 行）：
   796 |         )
   797 |         self._persistence.save(path, checkpoint)
   798 | 
>  799 |     @staticmethod
>  800 |     def _is_failure_observation(observation: str, action: str | None = None) -> bool:
>  801 |         """委托到 ``core.observation``，让内核外的能力复用同一份失败判定。
>  802 | 
>  803 |         ``action`` 是产生该观察的工具动作，用于区分工具携带的源码/输出与工具状态。
>  804 |         """
>  805 |         return is_failure_observation(observation, action=action)
   806 | 
   807 |     def _replan_after_failure(
   808 |         self,

Recorded replacement arguments, not a full repository diff. Check the tool receipt to see whether the edit was applied.

--- before excerpt+++ after excerpt@@ -1,3 +1,3 @@                     observation=observation,-                    failed=self._is_failure_observation(observation),+                    failed=self._is_failure_observation(observation, action=action),                 )

已按内容替换 dm_agent\core\agent.py 的 1 处（132 -> 147 字符）。
[编辑后] 第 656-664 行（共 850 行）：
   656 |                     step_number=step_num,
   657 |                     action=action,
   658 |                     action_input=action_input,
>  659 |                     observation=observation,
>  660 |                     failed=self._is_failure_observation(observation, action=action),
>  661 |                 )
   662 | 
   663 |             # 更新计划进度（如果有计划；被守卫拦下的编辑不算完成）。
   664 |             # 只有下一个待办步骤的动作匹配时才标记，消除同名动作的二义性。

Recorded replacement arguments, not a full repository diff. Check the tool receipt to see whether the edit was applied.

--- before excerpt+++ after excerpt@@ -1 +1 @@-            if self._is_failure_observation(observation) and plan and self.planner:+            if self._is_failure_observation(observation, action=action) and plan and self.planner:

已按内容替换 dm_agent\core\agent.py 的 1 处（83 -> 98 字符）。
[编辑后] 第 674-680 行（共 850 行）：
   674 |             # 调用回调函数实时输出步骤
   675 |             self._publish_step(step, step_num)
   676 | 
>  677 |             if self._is_failure_observation(observation, action=action) and plan and self.planner:
   678 |                 plan = self._replan_after_failure(
   679 |                     task,
   680 |                     plan,

Recorded replacement arguments, not a full repository diff. Check the tool receipt to see whether the edit was applied.

--- before excerpt+++ after excerpt@@ -1 +1 @@-                and not self._is_failure_observation(observation)+                and not self._is_failure_observation(observation, action=action)

已按内容替换 dm_agent\core\agent.py 的 1 处（65 -> 80 字符）。
[编辑后] 第 691-697 行（共 850 行）：
   691 |             if (
   692 |                 action == "task_complete"
   693 |                 and accepted
>  694 |                 and not self._is_failure_observation(observation, action=action)
   695 |             ):
   696 |                 metadata["status"] = "success"
   697 |                 metadata["failure_reason"] = ""

============================= test session starts =============================
platform win32 -- Python 3.13.9, pytest-9.1.1, pluggy-1.6.0
rootdir: [local path]
configfile: pyproject.toml
plugins: anyio-4.14.2, cov-7.1.0
collected 12 items

tests\test_maintenance_case.py ............                              [100%]

============================= 12 passed in 2.05s ==============================
returncode: 0

........................................................................ [ 15%]
.....................................................................F.. [ 30%]
.....F.................................................................. [ 46%]
........................................................................ [ 61%]
........................................................................ [ 77%]
........................................................................ [ 92%]
..................................                                       [100%]
================================== FAILURES ===================================
__________________ test_eval_ablation_runs_all_builtin_tasks __________________

    def test_eval_ablation_runs_all_builtin_tasks():
        tasks = get_builtin_tasks()
        report = run_suite(tasks=tasks)
    
        assert report["summary"]["total_runs"] == len(tasks) * 4
        assert "full" in report["summary"]["variants"]
>       assert report["summary"]["variants"]["full"]["success_rate"] == 1.0
E       assert 0.9285714285714286 == 1.0

tests\test_evals.py:32: AssertionError
_________ test_new_guard_and_truncation_eval_tasks_pass_all_variants __________

    def test_new_guard_and_truncation_eval_tasks_pass_all_variants():
        report = run_suite(task_ids=["truncated_read_pagination", "edit_guard_reread"])
    
>       assert report["summary"]["overall_success_rate"] == 1.0
E       assert 0.625 == 1.0

tests\test_evals.py:129: AssertionError
============================== warnings summary ===============================
tests/test_server_conversations.py::test_a_new_conversation_starts_idle_with_no_turns
  <python-environment>\Lib\site-packages\fastapi\testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
    from starlette.testclient import TestClient as TestClient  # noqa

-- Docs: [URL redacted]
=========================== short test summary info ===========================
FAILED tests/test_evals.py::test_eval_ablation_runs_all_builtin_tasks - asser...
FAILED tests/test_evals.py::test_new_guard_and_truncation_eval_tasks_pass_all_variants
2 failed, 464 passed, 1 warning in 48.00s
returncode: 1

import json

import pytest

from dm_agent.evals.cli import main as eval_main
from dm_agent.evals.gate import evaluate_gate
from dm_agent.evals.gate import main as gate_main
from dm_agent.evals.models import EvalResult
from dm_agent.evals.real_runner import get_real_tasks
from dm_agent.evals.runner import EvalVariant, run_suite, summarize_results
from dm_agent.evals.tasks import get_builtin_tasks


def test_builtin_eval_suite_runs_full_variant():
    report = run_suite(
        variants=[EvalVariant("full", True, True, True)],
        task_ids=["direct_finish", "json_repair", "tool_failure_replan"],
    )

    summary = report["summary"]["variants"]["full"]
    assert summary["tasks"] == 3
    assert summary["success_rate"] == 1.0
    assert summary["recovery_events"] >= 2


def test_eval_ablation_runs_all_builtin_tasks():
    tasks = get_builtin_tasks()
    report = run_suite(tasks=tasks)

    assert report["summary"]["total_runs"] == len(tasks) * 4
    assert "full" in report["summary"]["variants"]
    assert report["summary"]["variants"]["full"]["success_rate"] == 1.0


def test_eval_cli_writes_reports(tmp_path):
    json_path = tmp_path / "report.json"
    md_path = tmp_path / "report.md"

    exit_code = eval_main(
        [
            "--variant",
            "full",
            "--task",
            "direct_finish",
            "--output",
            str(json_path),
            "--markdown",
            str(md_path),
        ]
    )

    assert exit_code == 0
    assert json_path.exists()
    assert md_path.exists()
    assert "Ablation Summary" in md_path.read_text(encoding="utf-8")


def test_real_eval_manifest_is_explicit_and_keyless():
    tasks = get_real_tasks()

    assert tasks
    assert all(task.task_id.startswith("real_") for task in tasks)
    assert all(task.agent_responses == [] for task in tasks)
    assert any("recovery" in task.tags for task in tasks)


def test_real_eval_cli_lists_without_api_key():
    assert eval_main(["--real", "--list"]) == 0


def _eval_result(task_id, *, success, metadata=None, variant="full"):
    return EvalResult(
        task_id=task_id,
        task_name=task_id,
        variant=variant,
        success=success,
        final_answer="",
        failure_reason="" if success else "failed validation",
        actions=["finish"],
        steps_count=1,
        tool_calls=0,
        duration_seconds=0.0,
        prompt_chars=10,
        completion_chars=5,
        estimated_tokens=4,
        estimated_cost_usd=0.0,
        metadata=metadata or {},
    )


def test_summarize_results_recovery_rate_math():
    results = [
        _eval_result("a", success=True, metadata={"tool_error_count": 1}),
        _eval_result("b", success=False, metadata={"parse_error_count": 2}),
        _eval_result("c", success=True),
    ]
    summary = summarize_results(results)
    variant = summary["variants"]["full"]

    assert variant["runs_with_failures"] == 2
    assert variant["recovered_runs"] == 1
    assert variant["recovery_success_rate"] == pytest.approx(0.5)


def test_summarize_results_recovery_rate_none_without_failures():
    summary = summarize_results([_eval_result("a", success=True)])
    assert summary["variants"]["full"]["recovery_success_rate"] is None


def test_summarize_results_by_tag_uses_task_tags():
    tasks = [task for task in get_builtin_tasks() if task.task_id in {"direct_finish"}]
    results = [_eval_result("direct_finish", success=True)]
    summary = summarize_results(results, tasks=tasks)

    assert summary["by_tag"]["control"]["runs"] == 1
    assert summary["by_tag"]["control"]["success_rate"] == 1.0


def test_summarize_results_tolerates_legacy_metadata():
    # v2.0-era results lack the new counters entirely; aggregation must not raise.
    legacy = _eval_result("old", success=True, metadata={"status": "success"})
    summary = summarize_results([legacy])
    assert summary["variants"]["full"]["runs_with_failures"] == 0


def test_new_guard_and_truncation_eval_tasks_pass_all_variants():
    report = run_suite(task_ids=["truncated_read_pagination", "edit_guard_reread"])

    assert report["summary"]["overall_success_rate"] == 1.0
    assert report["summary"]["total_runs"] == 8  # 2 tasks x 4 variants


def test_gate_passes_green_report(tmp_path, capsys):
    report = run_suite(task_ids=["direct_finish"])
    path = tmp_path / "report.json"
    path.write_text(json.dumps(report), encoding="utf-8")

    assert gate_main([str(path), "--min-success-rate", "1.0"]) == 0

在 dm_agent\evals\tasks.py 中找到 3 处匹配：

    212:         tags=["recovery"],
    213:     ),
    214:     EvalTask(
>>> 215:         task_id="tool_failure_replan",
    216:         name="Recover from failed file read",
    217:         prompt="If reading missing.txt fails, create recovered.txt instead.",
    218:         planner_response=planner_response("read_file", "create_file", "task_complete"),

    250:         tags=["recovery", "arguments"],
    251:     ),
    252:     EvalTask(
>>> 253:         task_id="truncated_read_pagination",
    254:         name="Page through a truncated read",
    255:         prompt="Read big.txt, then re-read the needed range precisely.",
    256:         planner_response=planner_response("read_file", "read_file", "finish"),

    279:         tags=["long-context", "truncation"],
    280:     ),
    281:     EvalTask(
>>> 282:         task_id="edit_guard_reread",
    283:         name="Re-read before editing",
    284:         prompt="Update app.py line 1; the guard requires reading before editing.",
    285:         planner_response=planner_response("edit_file", "read_file", "edit_file", "task_complete"),