diff --git a/kaiwu/core/context.py b/kaiwu/core/context.py index 20ebde9..ac48e0f 100644 --- a/kaiwu/core/context.py +++ b/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 = "" diff --git a/kaiwu/core/orchestrator.py b/kaiwu/core/orchestrator.py index b59dbad..56a9234 100644 --- a/kaiwu/core/orchestrator.py +++ b/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.""" diff --git a/kaiwu/experts/debug_subagent.py b/kaiwu/experts/debug_subagent.py new file mode 100644 index 0000000..9fe089e --- /dev/null +++ b/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] = "" + 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 "" diff --git a/kaiwu/experts/generator.py b/kaiwu/experts/generator.py index cff4598..fe76c0d 100644 --- a/kaiwu/experts/generator.py +++ b/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}" diff --git a/kaiwu/tests/test_debug_subagent.py b/kaiwu/tests/test_debug_subagent.py new file mode 100644 index 0000000..d911e36 --- /dev/null +++ b/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, "", "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 == ""