feat: v1.6.2 初始测试失败注入Generator + Wink润滑增强

- Generator: 首次生成时注入initial_test_failure到prompt（LLM第一次就看到具体报错）
- Wink: 润滑规则增强
- Orchestrator: 重试逻辑优化
- pyproject.toml: 1.6.2

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

CHANGELOG.md

@@ -4,6 +4,30 @@ All notable changes to KWCode are documented here.
 
 ---
 
+## [1.6.2] - 2026-05-07
+
+### Retry优化 + 润滑/免疫机制
+
+**核心理念**：让LLM在擅长的地方发挥最大效能，在不擅长的地方不让它发力。给最好的原材料，而不是更多的retry机制。
+
+### Changed
+
+- **Generator首次生成注入initial_test_failure**：LLM第一次就能看到具体哪些测试在失败、期望什么输出，不再盲目生成
+- **retry_hint携带具体失败测试名**：从verifier_output提取failed_tests列表注入hint，LLM重试时精确知道哪些测试还没过
+- **Reviewer在tests_total==0时跳过**：无测试证据时Reviewer会幻觉reject，现在直接放行
+- **think_escalate改为条件触发**：只在assertion错误（逻辑错误）时升级深度推理，syntax/patch_apply等错误想更久没用
+- **MISSING_TOOLCHAIN快速熔断**：工具链缺失时直接返回失败告知用户，不浪费retry
+
+### Removed
+
+- **scope narrowing**：删除第2次失败时缩小到第一个文件+函数的逻辑。第一次定位错了缩小只会更错，让Locator自然重新搜索
+
+### Added
+
+- **WinkMonitor免疫机制**：新增`tests_no_progress` pattern，检测连续retry后通过率未提升，注入"换方向"纠正hint
+
+---
+
 ## [1.5.1] - 2026-05-06
 
 ### 三飞轮 + 遥测 + 模型自适应 + 前沿算法

kaiwu/core/orchestrator.py

@@ -310,6 +310,18 @@ class PipelineOrchestrator:
         # Gate 3: AB测试
         ab_candidate_name, ab_used_new, gate_result = self._setup_ab_test(gate_result, expert_type, on_status)
 
+        # MISSING_TOOLCHAIN快速熔断：工具链缺失时不进retry循环，直接告知用户
+        if ctx.gap and ctx.gap.gap_type == GapType.MISSING_TOOLCHAIN:
+            self._emit(on_status, "circuit_break",
+                       f"工具链缺失：{ctx.gap.error_msg[:100]}，请手动安装后重试")
+            self.bus.emit("circuit_break", {"msg": "missing_toolchain"})
+            elapsed = time.time() - start_time
+            _watchdog.cancel()
+            checkpoint = Checkpoint(project_root)
+            return self._record_failure_result(ctx, project_root, gate_result,
+                                               None, False, 0,
+                                               elapsed, checkpoint, False, on_status)
+
         # 优先使用专家注册表的自定义pipeline，否则用默认
         if gate_result.get("route_type") == "expert_registry" and "pipeline" in gate_result:
             sequence = gate_result["pipeline"]
@@ -441,24 +453,6 @@ class PipelineOrchestrator:
             if wink_hint:
                 ctx.retry_hint = (ctx.retry_hint + "\n" + wink_hint).strip() if ctx.retry_hint else wink_hint
 
-            # Scope narrowing: on 2nd failure, reduce to first file+function
-            if ctx.retry_count == 2 and ctx.locator_output:
-                files = ctx.locator_output.get("relevant_files", [])
-                funcs = ctx.locator_output.get("relevant_functions", [])
-                if len(files) > 1 and funcs:
-                    ctx.locator_output = {
-                        "relevant_files": [files[0]],
-                        "relevant_functions": [funcs[0]],
-                        "edit_locations": ctx.locator_output.get("edit_locations", [])[:1],
-                        "method": "scope_narrowed",
-                    }
-                    if ctx.relevant_code_snippets:
-                        ctx.relevant_code_snippets = {
-                            files[0]: ctx.relevant_code_snippets.get(files[0], "")
-                        }
-                    self._emit(on_status, "scope_narrow", f"缩小范围：只修 {funcs[0]}()")
-                    self.bus.emit("scope_narrow", {"msg": f"只修 {funcs[0]}()"})
-
             self._emit(on_status, "retry", f"第{ctx.retry_count}次尝试失败：{error_detail[:100]}")
             self.bus.emit("retry", {"count": ctx.retry_count, "error": error_detail[:100]})
 
@@ -473,15 +467,14 @@ class PipelineOrchestrator:
             # 设置重试策略：每次重试用不同方法
             ctx.retry_strategy = ctx.retry_count  # 0→1→2
 
-            # Fast/Slow双阶段：首次用fast(think=off)，失败后升级slow(think=on+高预算)
-            if ctx.retry_count == 1 and not ctx.think_config.get("think"):
-                # 第一次失败：从fast升级到slow think
-                ctx.think_config = {"think": True, "budget": 2048}
-                self._emit(on_status, "think_escalate", "升级到深度推理模式")
-            elif ctx.retry_count >= 2 and ctx.think_config.get("budget", 0) < 4096:
-                # 第二次失败：最大think预算
-                ctx.think_config = {"think": True, "budget": 4096}
-                self._emit(on_status, "think_escalate", "最大推理预算")
+            # Think escalation：只在assertion错误时升级（逻辑错误需要深度推理，其他错误想更久没用）
+            if current_error_type == "assertion":
+                if ctx.retry_count == 1 and not ctx.think_config.get("think"):
+                    ctx.think_config = {"think": True, "budget": 2048}
+                    self._emit(on_status, "think_escalate", "逻辑错误，升级到深度推理")
+                elif ctx.retry_count >= 2 and ctx.think_config.get("budget", 0) < 4096:
+                    ctx.think_config = {"think": True, "budget": 4096}
+                    self._emit(on_status, "think_escalate", "最大推理预算")
 
             # 错误驱动搜索：按失败类型决定是否搜索（网络保护：异常不阻塞）
             if self._should_search(current_error_type, ctx.retry_count) and not ctx.search_triggered and not no_search:
@@ -674,13 +667,13 @@ class PipelineOrchestrator:
                         ab_candidate_name, ab_used_new: bool, elapsed: float,
                         checkpoint, on_status) -> dict:
         """Record success: memory, registry, trajectory, AB, value, milestone, reflection."""
-        # Reviewer: 需求对齐审查 — 测试全部通过时跳过（测试是最终裁判，Reviewer会幻觉）
+        # Reviewer: 需求对齐审查 — 测试全部通过或无测试结果时跳过（无证据时Reviewer会幻觉）
         v = ctx.verifier_output or {}
         tests_passed = v.get("tests_passed", 0)
         tests_total = v.get("tests_total", 0)
-        all_tests_pass = tests_total > 0 and tests_passed == tests_total
+        skip_review = (tests_total > 0 and tests_passed == tests_total) or tests_total == 0
 
-        if not all_tests_pass:
+        if not skip_review:
             # 测试没全通过才需要Reviewer审查
             review_result = self._do_review(ctx, on_status)
             if review_result and not review_result.get("aligned") and review_result.get("confidence", 0) >= 0.7:
@@ -1086,6 +1079,11 @@ class PipelineOrchestrator:
         if last_code:
             hint += f"\n\n上次生成的代码（有问题）：\n{last_code}\n\n请不要重复同样的错误。"
 
+        # 携带具体失败的测试名，让LLM精确定位
+        failed_tests = (ctx.verifier_output or {}).get("failed_tests", [])
+        if failed_tests:
+            hint += "\n\n仍然失败的测试：\n" + "\n".join(f"  - {t}" for t in failed_tests[:5])
+
         return hint
 
     def _should_search(self, error_type: str, retry_count: int) -> bool:

kaiwu/core/wink.py

@@ -5,6 +5,7 @@ Wink 自修复监控：轨迹监控 + 偏离检测 + 课程纠正。
 - Specification Drift：偏离用户原始意图（scope creep）
 - Reasoning Problems：同类错误反复（原地打转）
 - Tool Call Failures：patch 持续失败
+- Progress Stall：重试未提升通过率（免疫机制）
 
 理论来源：
 - Wink: Recovering from Misbehaviors in Coding Agents（arXiv:2602.17037）
@@ -65,6 +66,17 @@ class WinkMonitor:
             ),
             "hint": "Generator 未产出有效 patch，尝试简化任务描述或缩小修改范围",
         },
+        # 免疫机制：重试未提升通过率（结疤，不改变结构）
+        {
+            "name": "tests_no_progress",
+            "detect": lambda ctx: (
+                ctx.retry_count >= 2 and
+                ctx.verifier_output and
+                hasattr(ctx, '_prev_tests_passed') and
+                ctx.verifier_output.get("tests_passed", 0) <= getattr(ctx, '_prev_tests_passed', 0)
+            ),
+            "hint": "连续重试未提升通过率，尝试完全不同的实现方式，不要在同一个方向上继续",
+        },
     ]
 
     def check(self, ctx, bus: Optional[EventBus] = None) -> Optional[str]:
@@ -72,6 +84,10 @@ class WinkMonitor:
         检查当前 context 是否有偏离，返回纠正 hint 或 None。
         非阻塞，任何异常静默忽略。
         """
+        # 记录本次tests_passed供下次比较（免疫机制的记忆）
+        if ctx.verifier_output:
+            ctx._prev_tests_passed = ctx.verifier_output.get("tests_passed", 0)
+
         for pattern in self.DRIFT_PATTERNS:
             try:
                 if pattern["detect"](ctx):

kaiwu/experts/generator.py

@@ -391,6 +391,11 @@ class GeneratorExpert:
         if ctx.doc_context:
             prompt += f"\n\n## 相关文档参考\n{ctx.doc_context[:800]}"
 
+        # 注入初始测试失败信息（让LLM第一次就看到具体报错）
+        initial_failure = getattr(ctx, 'initial_test_failure', '')
+        if initial_failure and ctx.retry_count == 0:
+            prompt += f"\n\n## 当前测试失败\n{initial_failure[:500]}"
+
         # Inject retry_hint if available
         if ctx.retry_hint:
             prompt += f"\n\n## 重试提示\n{ctx.retry_hint}"
@@ -435,6 +440,11 @@ class GeneratorExpert:
         while "\n\n\n" in prompt:
             prompt = prompt.replace("\n\n\n", "\n\n")
 
+        # 注入初始测试失败信息
+        initial_failure = getattr(ctx, 'initial_test_failure', '')
+        if initial_failure and ctx.retry_count == 0:
+            prompt += f"\n\n## 当前测试失败\n{initial_failure[:500]}"
+
         if ctx.retry_hint:
             prompt += f"\n\n## 重试提示\n{ctx.retry_hint}"
 
@@ -825,6 +835,11 @@ class GeneratorExpert:
                 f"实现所有pass存根函数，保持已有实现不变。"
             )
 
+            # 注入初始测试失败信息
+            initial_failure = getattr(ctx, 'initial_test_failure', '')
+            if initial_failure and ctx.retry_count == 0:
+                prompt += f"\n\n## 当前测试失败\n{initial_failure[:500]}"
+
             if ctx.retry_hint:
                 prompt += f"\n\n## 重试提示\n{ctx.retry_hint}"
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "kwcode"
-version = "1.6.1"
+version = "1.6.2"
 description = "KwCode - Local-model coding agent with MoE expert pipeline"
 requires-python = ">=3.10"
 readme = "README.md"