feat: add Debug Subagent for runtime debugging on retry

Based on Debug2Fix (Microsoft, 2026): weak model + debugger > strong model.

- Add kaiwu/experts/debug_subagent.py: sys.settrace-based variable capture,
  LLM-guided debug strategy, pytest --tb=long fallback
- Add debug_info field to TaskContext
- Orchestrator calls _do_debug() after verifier failure (non-blocking)
- Generator retry prompts (strategy 1+2) inject ctx.debug_info
- 15 new tests (292 total, all passing)

Flow: verifier fails → LLM decides breakpoint location + variables →
sys.settrace captures runtime values → injected into next generator retry

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

kaiwu/core/context.py

@@ -47,5 +47,8 @@ class TaskContext:
     # Document context (populated by DocReader via Locator)
     doc_context: str = ""
 
+    # Debug Subagent output (populated by orchestrator on retry)
+    debug_info: str = ""
+
     # KWCODE.md injected rules (populated by orchestrator)
     kwcode_rules: str = ""

kaiwu/core/orchestrator.py

@@ -57,6 +57,7 @@ class PipelineOrchestrator:
         trajectory_collector: TrajectoryCollector | None = None,
         ab_tester: ABTester | None = None,
         chat_expert: ChatExpert | None = None,
+        debug_subagent=None,
     ):
         self.locator = locator
         self.generator = generator
@@ -70,6 +71,7 @@ class PipelineOrchestrator:
         self.trajectory_collector = trajectory_collector
         self._pattern_detector = PatternDetector(trajectory_collector) if trajectory_collector else None
         self.ab_tester = ab_tester
+        self.debug_subagent = debug_subagent
         self._value_tracker = ValueTracker()
         self._notifier = FlywheelNotifier()
 
@@ -226,6 +228,10 @@ class PipelineOrchestrator:
             if ctx.retry_count == 1 and ctx.verifier_output and ctx.generator_output:
                 self._do_reflection(ctx, on_status)
 
+            # Debug Subagent: capture runtime info on failure (test failures only)
+            if ctx.retry_count >= 1 and ctx.verifier_output:
+                self._do_debug(ctx, on_status)
+
             # Set retry strategy: each retry uses a different approach
             ctx.retry_strategy = ctx.retry_count  # 0→1→2
 
@@ -405,6 +411,21 @@ class PipelineOrchestrator:
         except Exception as e:
             logger.debug("Reflection failed (non-blocking): %s", e)
 
+    def _do_debug(self, ctx: TaskContext, on_status):
+        """Debug Subagent: capture runtime info after test failure (non-blocking)."""
+        if not self.debug_subagent:
+            return
+        try:
+            self._emit(on_status, "debug", "调试子代理：采集运行时信息...")
+            debug_info = self.debug_subagent.investigate(ctx)
+            if debug_info:
+                ctx.debug_info = debug_info
+                self._emit(on_status, "debug_done", f"调试信息：{debug_info[:80]}")
+            else:
+                self._emit(on_status, "debug_done", "未获取到额外调试信息")
+        except Exception as e:
+            logger.debug("Debug subagent failed (non-blocking): %s", e)
+
     @staticmethod
     def _emit(callback, stage: str, detail: str):
         """Emit status update if callback provided."""

kaiwu/experts/debug_subagent.py

@@ -0,0 +1,266 @@
+"""
+Debug Subagent: 运行时调试信息采集。
+论文基础：Debug2Fix (Microsoft, 2026) — 弱模型+调试器 > 强模型裸跑。
+
+核心思路：verifier 失败后，用 sys.settrace 非侵入式捕获目标行的变量值，
+或用 pytest --tb=long 获取完整异常堆栈，为 generator 重试提供真实运行时数据。
+
+约束：
+- 不引入新依赖（sys.settrace/subprocess 是标准库）
+- 超时 30s，失败返回空字符串，不中断主流程
+- 只在有 pytest 输出的失败场景触发（语法错误不触发）
+"""
+
+import json
+import logging
+import os
+import re
+import tempfile
+from typing import Optional
+
+from kaiwu.core.context import TaskContext
+from kaiwu.llm.llama_backend import LLMBackend
+from kaiwu.tools.executor import ToolExecutor
+
+logger = logging.getLogger(__name__)
+
+DEBUG_STRATEGY_PROMPT = """根据以下测试失败信息，决定需要检查什么运行时数据。
+
+失败信息：
+{error_detail}
+
+修改的文件：{modified_file}
+修改的代码片段：
+```
+{modified_snippet}
+```
+
+回答以下问题（JSON格式，不要解释）：
+1. 哪个文件的哪一行最可能是问题所在？
+2. 需要检查哪些变量的值？（最多5个）
+3. 是什么类型的错误？(exception/assertion/logic)
+
+格式：{{"file": "path/to/file.py", "line": 42, "variables": ["var1", "var2"], "error_type": "exception"}}"""
+
+TRACE_SCRIPT_TEMPLATE = '''import sys, json, os
+
+captured = {{"variables": {{}}, "exception": None, "reached": False}}
+target_file = {target_file!r}
+target_line = {target_line}
+variables = {variables!r}
+
+def tracer(frame, event, arg):
+    fname = frame.f_code.co_filename
+    if not fname.endswith(target_file):
+        return tracer
+    if event == "line" and frame.f_lineno == target_line:
+        captured["reached"] = True
+        for var in variables:
+            if var in frame.f_locals:
+                try:
+                    captured["variables"][var] = repr(frame.f_locals[var])[:200]
+                except Exception:
+                    captured["variables"][var] = "<repr failed>"
+    if event == "exception":
+        exc_type, exc_value, _ = arg
+        if exc_type is not None:
+            captured["exception"] = f"{{exc_type.__name__}}: {{exc_value}}"
+    return tracer
+
+sys.settrace(tracer)
+os.chdir({project_root!r})
+try:
+    import pytest
+    pytest.main([{test_path!r}, "-x", "-q", "--tb=no", "--no-header"])
+except SystemExit:
+    pass
+except Exception as e:
+    captured["exception"] = f"{{type(e).__name__}}: {{e}}"
+sys.settrace(None)
+print("__DEBUG_JSON__" + json.dumps(captured, ensure_ascii=False))
+'''
+
+
+class DebugSubagent:
+    """
+    运行时调试子代理。
+    verifier 失败后调用 investigate()，返回结构化调试信息。
+    """
+
+    TIMEOUT = 30  # seconds
+
+    def __init__(self, llm: LLMBackend, tool_executor: ToolExecutor):
+        self.llm = llm
+        self.tools = tool_executor
+
+    def investigate(self, ctx: TaskContext) -> str:
+        """
+        主入口。分析 verifier 失败，获取运行时信息。
+        返回人类可读的调试结论字符串，失败返回空字符串。
+        """
+        try:
+            # 前置条件：必须有 verifier 失败输出
+            if not ctx.verifier_output:
+                return ""
+            error_detail = ctx.verifier_output.get("error_detail", "")
+            if not error_detail or "Syntax error" in error_detail:
+                return ""  # 语法错误不需要运行时调试
+
+            # Step 1: LLM 决定调试策略
+            strategy = self._plan_debug_strategy(ctx, error_detail)
+            if not strategy:
+                return ""
+
+            # Step 2: 生成 trace 脚本并执行
+            runtime_data = self._execute_trace(ctx, strategy)
+            if not runtime_data:
+                # Fallback: 用 pytest --tb=long 获取详细堆栈
+                return self._fallback_detailed_traceback(ctx)
+
+            # Step 3: 格式化结果
+            return self._format_results(strategy, runtime_data)
+
+        except Exception as e:
+            logger.warning("[debug_subagent] investigate failed: %s", e)
+            return ""
+
+    def _plan_debug_strategy(self, ctx: TaskContext, error_detail: str) -> Optional[dict]:
+        """用 LLM 决定要检查哪个文件、哪一行、哪些变量。"""
+        patches = ctx.generator_output.get("patches", []) if ctx.generator_output else []
+        modified_file = patches[0].get("file", "") if patches else ""
+        modified_snippet = patches[0].get("modified", "")[:300] if patches else ""
+
+        prompt = DEBUG_STRATEGY_PROMPT.format(
+            error_detail=error_detail[:500],
+            modified_file=modified_file,
+            modified_snippet=modified_snippet,
+        )
+
+        try:
+            response = self.llm.generate(prompt=prompt, max_tokens=200, temperature=0.1)
+            # 提取 JSON
+            json_match = re.search(r'\{[^}]+\}', response)
+            if not json_match:
+                return None
+            strategy = json.loads(json_match.group())
+            # 验证必要字段
+            if "file" not in strategy or "line" not in strategy:
+                return None
+            strategy.setdefault("variables", [])
+            strategy.setdefault("error_type", "unknown")
+            # 限制变量数量
+            strategy["variables"] = strategy["variables"][:5]
+            return strategy
+        except Exception as e:
+            logger.warning("[debug_subagent] strategy planning failed: %s", e)
+            return None
+
+    def _execute_trace(self, ctx: TaskContext, strategy: dict) -> Optional[dict]:
+        """生成并执行 sys.settrace 脚本，捕获运行时变量。"""
+        target_file = strategy["file"]
+        target_line = int(strategy["line"])
+        variables = strategy["variables"]
+
+        # 找到测试文件
+        test_path = self._find_test_file(ctx.project_root)
+        if not test_path:
+            return None
+
+        script = TRACE_SCRIPT_TEMPLATE.format(
+            target_file=target_file,
+            target_line=target_line,
+            variables=variables,
+            project_root=ctx.project_root,
+            test_path=test_path,
+        )
+
+        # 写入临时文件执行（避免命令行转义问题）
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w", suffix=".py", delete=False, encoding="utf-8"
+            ) as f:
+                f.write(script)
+                script_path = f.name
+
+            result = self.tools.run_bash(
+                f'python "{script_path}"',
+                timeout=self.TIMEOUT,
+            )
+
+            # 清理临时文件
+            try:
+                os.unlink(script_path)
+            except OSError:
+                pass
+
+            if not result or not result.get("stdout"):
+                return None
+
+            # 提取 JSON 结果
+            stdout = result["stdout"]
+            marker = "__DEBUG_JSON__"
+            if marker in stdout:
+                json_str = stdout.split(marker)[-1].strip()
+                return json.loads(json_str)
+
+            return None
+
+        except Exception as e:
+            logger.warning("[debug_subagent] trace execution failed: %s", e)
+            return None
+
+    def _fallback_detailed_traceback(self, ctx: TaskContext) -> str:
+        """Fallback：用 pytest --tb=long 获取详细异常堆栈。"""
+        test_path = self._find_test_file(ctx.project_root)
+        if not test_path:
+            return ""
+
+        try:
+            result = self.tools.run_bash(
+                f'cd "{ctx.project_root}" && python -m pytest "{test_path}" -x --tb=long -q 2>&1 | tail -40',
+                timeout=self.TIMEOUT,
+            )
+            if result and result.get("stdout"):
+                output = result["stdout"][:800]
+                return f"[详细堆栈]\n{output}"
+        except Exception as e:
+            logger.warning("[debug_subagent] fallback traceback failed: %s", e)
+        return ""
+
+    @staticmethod
+    def _find_test_file(project_root: str) -> Optional[str]:
+        """找到项目的测试文件路径。"""
+        candidates = [
+            "tests/",
+            "test/",
+            ".",
+        ]
+        for candidate in candidates:
+            test_dir = os.path.join(project_root, candidate)
+            if os.path.isdir(test_dir):
+                for fname in os.listdir(test_dir):
+                    if fname.startswith("test_") and fname.endswith(".py"):
+                        return os.path.join(candidate, fname)
+                    if fname.endswith("_test.py"):
+                        return os.path.join(candidate, fname)
+        return None
+
+    @staticmethod
+    def _format_results(strategy: dict, runtime_data: dict) -> str:
+        """把运行时数据格式化为人类可读的调试结论。"""
+        parts = []
+
+        if runtime_data.get("exception"):
+            parts.append(f"异常: {runtime_data['exception']}")
+
+        if not runtime_data.get("reached"):
+            parts.append(f"注意: 执行未到达 {strategy['file']}:{strategy['line']}")
+        else:
+            parts.append(f"断点命中: {strategy['file']}:{strategy['line']}")
+
+        variables = runtime_data.get("variables", {})
+        if variables:
+            var_lines = [f"  {k} = {v}" for k, v in variables.items()]
+            parts.append("变量值:\n" + "\n".join(var_lines))
+
+        return "\n".join(parts) if parts else ""

kaiwu/experts/generator.py

@@ -353,8 +353,9 @@ class GeneratorExpert:
         elif strategy == 1:
             error = ctx.previous_failure or "验证失败"
             reflection_line = f"\n失败分析：{ctx.reflection}" if ctx.reflection else ""
+            debug_line = f"\n运行时调试信息：{ctx.debug_info}" if ctx.debug_info else ""
             return (
-                f"上次修改失败了。错误信息：\n{error[:500]}{reflection_line}\n\n"
+                f"上次修改失败了。错误信息：\n{error[:500]}{reflection_line}{debug_line}\n\n"
                 f"原始代码（来自 {fpath}）：\n```\n{original}\n```\n\n"
                 f"{search_line}"
                 f"直接修复这个错误。只输出修改后的完整函数代码，不要解释。"
@@ -363,8 +364,9 @@ class GeneratorExpert:
         else:
             error = ctx.previous_failure or "验证失败"
             reflection_line = f"\n上次失败原因：{ctx.reflection}" if ctx.reflection else ""
+            debug_line = f"\n运行时调试信息：{ctx.debug_info}" if ctx.debug_info else ""
             return (
-                f"只修改以下代码的最小必要部分，其他代码一行都不要动。{reflection_line}\n\n"
+                f"只修改以下代码的最小必要部分，其他代码一行都不要动。{reflection_line}{debug_line}\n\n"
                 f"需要修复的错误：{error[:300]}\n\n"
                 f"原始代码（来自 {fpath}）：\n```\n{original}\n```\n\n"
                 f"{search_line}"

kaiwu/tests/test_debug_subagent.py

@@ -0,0 +1,272 @@
+"""
+Tests for DebugSubagent: runtime debugging info capture.
+Uses mock LLM and mock tool_executor since Ollama may not be running.
+"""
+
+import json
+import os
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from kaiwu.core.context import TaskContext
+from kaiwu.experts.debug_subagent import DebugSubagent, TRACE_SCRIPT_TEMPLATE
+
+
+# ── Fixtures ──
+
+def _make_mock_llm(response: str = '{"file": "src/calc.py", "line": 10, "variables": ["x", "y"], "error_type": "exception"}'):
+    llm = MagicMock()
+    llm.generate = MagicMock(return_value=response)
+    return llm
+
+
+def _make_mock_tools(bash_stdout: str = ""):
+    tools = MagicMock()
+    tools.run_bash = MagicMock(return_value={"stdout": bash_stdout, "stderr": "", "returncode": 0})
+    return tools
+
+
+def _make_ctx_with_failure(project_root: str = "/tmp/test") -> TaskContext:
+    ctx = TaskContext(
+        user_input="fix the calculator bug",
+        project_root=project_root,
+        gate_result={"expert_type": "locator_repair"},
+        verifier_output={
+            "passed": False,
+            "syntax_ok": True,
+            "tests_passed": 0,
+            "tests_total": 3,
+            "error_detail": "FAILED tests/test_calc.py::test_add - AssertionError: assert 5 == 3",
+        },
+        generator_output={
+            "patches": [{"file": "src/calc.py", "original": "return x - y", "modified": "return x + y"}],
+            "explanation": "Fixed addition",
+        },
+    )
+    return ctx
+
+
+# ── Unit Tests ──
+
+class TestDebugSubagentInvestigate:
+    """Test the main investigate() method."""
+
+    def test_returns_string_on_success(self):
+        """investigate() should return a non-empty string with debug info."""
+        llm = _make_mock_llm()
+        debug_json = json.dumps({
+            "variables": {"x": "3", "y": "2"},
+            "exception": None,
+            "reached": True,
+        })
+        tools = _make_mock_tools(bash_stdout=f"__DEBUG_JSON__{debug_json}")
+        subagent = DebugSubagent(llm, tools)
+
+        # Create a temp dir with a test file
+        with tempfile.TemporaryDirectory() as tmpdir:
+            os.makedirs(os.path.join(tmpdir, "tests"))
+            with open(os.path.join(tmpdir, "tests", "test_calc.py"), "w") as f:
+                f.write("def test_add(): pass\n")
+
+            ctx = _make_ctx_with_failure(project_root=tmpdir)
+            result = subagent.investigate(ctx)
+
+        assert isinstance(result, str)
+        assert len(result) > 0
+        assert "x = 3" in result or "y = 2" in result
+
+    def test_returns_empty_on_no_verifier_output(self):
+        """investigate() should return '' when verifier_output is None."""
+        llm = _make_mock_llm()
+        tools = _make_mock_tools()
+        subagent = DebugSubagent(llm, tools)
+
+        ctx = TaskContext(user_input="test", project_root="/tmp")
+        ctx.verifier_output = None
+
+        result = subagent.investigate(ctx)
+        assert result == ""
+
+    def test_returns_empty_on_syntax_error(self):
+        """investigate() should skip syntax errors (not runtime issues)."""
+        llm = _make_mock_llm()
+        tools = _make_mock_tools()
+        subagent = DebugSubagent(llm, tools)
+
+        ctx = TaskContext(user_input="test", project_root="/tmp")
+        ctx.verifier_output = {
+            "passed": False,
+            "error_detail": "Syntax error in src/main.py: invalid syntax (line 5)",
+        }
+
+        result = subagent.investigate(ctx)
+        assert result == ""
+
+    def test_fallback_on_trace_failure(self):
+        """When trace script fails, should fallback to pytest --tb=long."""
+        llm = _make_mock_llm()
+        # First call (trace) returns empty, second call (fallback) returns traceback
+        tools = MagicMock()
+        tools.run_bash = MagicMock(side_effect=[
+            {"stdout": "", "stderr": "", "returncode": 1},  # trace fails
+            {"stdout": "FAILED test_calc.py\nAssertionError: 5 != 3\n  File calc.py line 10", "stderr": "", "returncode": 1},  # fallback
+        ])
+        subagent = DebugSubagent(llm, tools)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            os.makedirs(os.path.join(tmpdir, "tests"))
+            with open(os.path.join(tmpdir, "tests", "test_calc.py"), "w") as f:
+                f.write("def test_add(): pass\n")
+
+            ctx = _make_ctx_with_failure(project_root=tmpdir)
+            result = subagent.investigate(ctx)
+
+        assert "详细堆栈" in result or "AssertionError" in result
+
+    def test_no_crash_on_llm_failure(self):
+        """investigate() should not crash if LLM returns garbage."""
+        llm = MagicMock()
+        llm.generate = MagicMock(return_value="I don't understand the question")
+        tools = _make_mock_tools()
+        subagent = DebugSubagent(llm, tools)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            os.makedirs(os.path.join(tmpdir, "tests"))
+            with open(os.path.join(tmpdir, "tests", "test_x.py"), "w") as f:
+                f.write("def test_x(): pass\n")
+
+            ctx = _make_ctx_with_failure(project_root=tmpdir)
+            result = subagent.investigate(ctx)
+
+        # Should not crash, returns empty or fallback
+        assert isinstance(result, str)
+
+
+class TestDebugSubagentStrategy:
+    """Test the _plan_debug_strategy method."""
+
+    def test_parses_valid_json(self):
+        """Should extract strategy from LLM JSON response."""
+        llm = _make_mock_llm('Here is the analysis: {"file": "app.py", "line": 25, "variables": ["data", "result"], "error_type": "assertion"}')
+        tools = _make_mock_tools()
+        subagent = DebugSubagent(llm, tools)
+
+        ctx = _make_ctx_with_failure()
+        strategy = subagent._plan_debug_strategy(ctx, "AssertionError: 5 != 3")
+
+        assert strategy is not None
+        assert strategy["file"] == "app.py"
+        assert strategy["line"] == 25
+        assert "data" in strategy["variables"]
+
+    def test_returns_none_on_invalid_json(self):
+        """Should return None if LLM doesn't produce valid JSON."""
+        llm = _make_mock_llm("I think the bug is in the add function")
+        tools = _make_mock_tools()
+        subagent = DebugSubagent(llm, tools)
+
+        ctx = _make_ctx_with_failure()
+        strategy = subagent._plan_debug_strategy(ctx, "some error")
+
+        assert strategy is None
+
+    def test_limits_variables_to_5(self):
+        """Should cap variables at 5 even if LLM suggests more."""
+        many_vars = '{"file": "x.py", "line": 1, "variables": ["a","b","c","d","e","f","g"], "error_type": "logic"}'
+        llm = _make_mock_llm(many_vars)
+        tools = _make_mock_tools()
+        subagent = DebugSubagent(llm, tools)
+
+        ctx = _make_ctx_with_failure()
+        strategy = subagent._plan_debug_strategy(ctx, "error")
+
+        assert len(strategy["variables"]) == 5
+
+
+class TestTraceScriptGeneration:
+    """Test that generated trace scripts are valid Python."""
+
+    def test_script_is_valid_python(self):
+        """Generated trace script should compile without syntax errors."""
+        script = TRACE_SCRIPT_TEMPLATE.format(
+            target_file="src/calc.py",
+            target_line=10,
+            variables=["x", "y", "result"],
+            project_root="/tmp/myproject",
+            test_path="tests/test_calc.py",
+        )
+        # Should not raise SyntaxError
+        compile(script, "<trace_script>", "exec")
+
+    def test_script_contains_marker(self):
+        """Script output should contain __DEBUG_JSON__ marker."""
+        script = TRACE_SCRIPT_TEMPLATE.format(
+            target_file="app.py",
+            target_line=5,
+            variables=["data"],
+            project_root="/tmp/proj",
+            test_path="tests/test_app.py",
+        )
+        assert "__DEBUG_JSON__" in script
+
+
+class TestFormatResults:
+    """Test _format_results static method."""
+
+    def test_formats_exception(self):
+        strategy = {"file": "x.py", "line": 10, "variables": []}
+        data = {"exception": "TypeError: unsupported operand", "reached": False, "variables": {}}
+        result = DebugSubagent._format_results(strategy, data)
+        assert "TypeError" in result
+        assert "未到达" in result
+
+    def test_formats_variables(self):
+        strategy = {"file": "x.py", "line": 10, "variables": ["a", "b"]}
+        data = {"exception": None, "reached": True, "variables": {"a": "None", "b": "42"}}
+        result = DebugSubagent._format_results(strategy, data)
+        assert "a = None" in result
+        assert "b = 42" in result
+        assert "断点命中" in result
+
+
+class TestOrchestratorIntegration:
+    """Test that orchestrator correctly calls debug subagent."""
+
+    def test_do_debug_injects_info(self):
+        """_do_debug should write to ctx.debug_info."""
+        from kaiwu.core.orchestrator import PipelineOrchestrator
+
+        # Create minimal orchestrator with mock debug subagent
+        mock_debug = MagicMock()
+        mock_debug.investigate = MagicMock(return_value="异常: TypeError at line 10\n变量值:\n  x = None")
+
+        orch = MagicMock(spec=PipelineOrchestrator)
+        orch.debug_subagent = mock_debug
+        orch._emit = MagicMock()
+
+        # Call _do_debug directly
+        ctx = _make_ctx_with_failure()
+        PipelineOrchestrator._do_debug(orch, ctx, None)
+
+        assert ctx.debug_info == "异常: TypeError at line 10\n变量值:\n  x = None"
+        mock_debug.investigate.assert_called_once_with(ctx)
+
+    def test_do_debug_no_crash_without_subagent(self):
+        """_do_debug should be a no-op when debug_subagent is None."""
+        from kaiwu.core.orchestrator import PipelineOrchestrator
+
+        orch = MagicMock(spec=PipelineOrchestrator)
+        orch.debug_subagent = None
+
+        ctx = _make_ctx_with_failure()
+        PipelineOrchestrator._do_debug(orch, ctx, None)
+
+        assert ctx.debug_info == ""
+
+    def test_context_has_debug_info_field(self):
+        """TaskContext should have debug_info field."""
+        ctx = TaskContext()
+        assert hasattr(ctx, "debug_info")
+        assert ctx.debug_info == ""