feat: P0+P1 optimization — circuit breaker, gate confidence, experience replay, session state

- P0-1: Verifier structured error output (error_type/file/line/message extraction)
- P0-2: Circuit breaker + scope narrowing (syntax/import fast-break, 3x same-error hard-break, auto-narrow on 2nd failure)
- P0-3: Gate confidence estimation (keyword signal scoring, low-confidence retry reduction)
- P1-1: Experience Replay via BM25 trajectory similarity search
- P1-2: SessionState multi-turn coherence + attention decay countermeasure
- P1-3: Locator minimal context extraction (function boundary detection, comment stripping, 60-line cap)
- Add CONTRIBUTING.md with architecture red lines and PR standards
- Update README contributing section with quick reference

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

CONTRIBUTING.md

@@ -0,0 +1,161 @@
+# KWCode 贡献指南
+
+感谢你的贡献。请在提交 PR 前阅读本文档。
+
+---
+
+## 快速开始
+
+```bash
+git clone https://github.com/val1813/kwcode.git
+cd kwcode
+pip install -e ".[dev]"
+python -m pytest kaiwu/tests/ -v --ignore=kaiwu/tests/bench_tasks
+# 全部绿才能提 PR
+```
+
+---
+
+## 架构红线（违反即拒绝）
+
+kwcode 的核心设计原则是**确定性流水线**，任何 PR 不得违反：
+
+| 红线 | 说明 |
+|------|------|
+| **RED-1** | Gate 必须输出结构化 JSON，不得用字符串解析分类结果 |
+| **RED-2** | LLM 只做分类和生成，不得在流水线 step 之间让 LLM 决定下一步 |
+| **RED-3** | 每个专家有独立上下文窗口，不得继承上一个专家的对话历史 |
+| **RED-4** | 新增依赖必须离线可用，不得引入需要外部服务才能运行的包 |
+| **RED-5** | 重试次数必须有硬上限，不得无限循环 |
+
+**一票否决的改动类型**：
+- 引入向量数据库（Mem0、Chroma、Pinecone、Weaviate 等）
+- 引入企业级安全库（LLM-Guard、Guardrails 等）
+- 引入需要云服务的依赖（非离线可用）
+- 在确定性流水线的 step 之间插入 LLM routing 调用
+- 多 Agent 并行框架（现阶段串行流水线，不做并行）
+- 自动修改 Gate 路由规则的逻辑（漂移难追踪）
+
+---
+
+## 欢迎的贡献类型
+
+按优先级排序：
+
+**P0 — 最欢迎**
+- 新增预置专家（bugfix/refactor/testgen 等领域的 SKILL.md + YAML）
+- 修复已知 bug（附上能复现 bug 的测试用例）
+- 多语言 AST 支持（JS/TS/Go/Rust/Java 调用图）
+- 性能优化（Locator 定位速度、ContextPruner 压缩质量）
+
+**P1 — 欢迎**
+- Verifier 结构化输出改进（error_type 分类更精准）
+- Experience Replay / trajectory_collector 检索能力
+- Session 连贯性改进（System Reminders、RULES.md）
+- README / 文档改进（中英文均可）
+- CI/CD 改进（GitHub Actions、测试覆盖率）
+
+**P2 — 需讨论后再做**
+- 新增 CLI 命令（先开 Issue 讨论）
+- Gate 分类逻辑改动（影响所有任务路由）
+- Orchestrator 流程改动（影响核心流水线）
+- 新增 llm backend 支持
+
+---
+
+## PR 标准
+
+### 必须满足
+
+**1. 测试全绿**
+```bash
+python -m pytest kaiwu/tests/ -v --ignore=kaiwu/tests/bench_tasks
+```
+所有现有测试必须通过，不得删除已有测试。
+
+**2. 新功能必须有测试**
+- 改动了 `kaiwu/core/`（gate/orchestrator/verifier）→ 必须有对应测试
+- 改动了 `kaiwu/experts/` → 必须有对应测试
+- 改动了 `kaiwu/flywheel/` → 必须有对应测试
+- 只改 README / .gitignore / 文档 → 不需要测试
+
+**3. 新增依赖需说明**
+在 PR 描述里说明：
+- 为什么需要这个包
+- 是否离线可用
+- 包大小和主要依赖
+
+**4. 文档和实现同步**
+README 里提到的功能必须已经实现，不得在文档里描述未实现的功能。
+
+### PR 描述模板
+
+```
+## 改动内容
+<!-- 一句话说明这个 PR 做了什么 -->
+
+## 改动类型
+- [ ] Bug 修复
+- [ ] 新增专家 / SKILL.md
+- [ ] 性能优化
+- [ ] 文档 / CI 改进
+- [ ] 其他（请说明）
+
+## 测试
+- [ ] 现有测试全部通过
+- [ ] 新增了对应测试
+- [ ] 只改文档，无需测试
+
+## 新增依赖（如有）
+| 包名 | 版本 | 用途 | 离线可用 |
+|------|------|------|---------|
+|      |      |      |         |
+
+## 验证方式
+<!-- 说明如何验证这个改动有效 -->
+```
+
+---
+
+## 新增专家（最简单的贡献方式）
+
+1. 复制 `kaiwu/builtin_experts/bugfix/` 目录结构
+2. 编辑 `SKILL.md`（领域知识，越详细越好）
+3. 用真实项目测试 ≥ 5 个任务，通过率 ≥ 80%
+4. 在 PR 里附上测试结果截图
+
+急需认领的专家：
+- `Vue3Expert` / `ReactExpert`
+- `DjangoExpert` / `FastAPIExpert`
+- `GoGinExpert` / `RustActixExpert`
+- `K8sExpert` / `DockerExpert`
+- `MySQLExpert` / `RedisExpert`
+
+---
+
+## 代码风格
+
+- Python 3.10+，类型注解尽量完整
+- 非阻塞路径的异常必须 `logger.debug/warning`，不得静默吞掉
+- 新模块加模块级 docstring，说明设计意图
+- 中文注释可以，英文也可以，同一文件保持一致
+
+---
+
+## 开 Issue 还是直接 PR？
+
+| 情况 | 建议 |
+|------|------|
+| 发现 bug | 直接 PR（附复现步骤和测试） |
+| 新增专家 | 直接 PR |
+| 改动 Gate / Orchestrator 逻辑 | 先开 Issue 讨论 |
+| 新增 CLI 命令 | 先开 Issue 讨论 |
+| 不确定方向对不对 | 先开 Issue |
+
+---
+
+## 行为准则
+
+- 代码审查的反馈是针对代码，不是针对人
+- 中英文交流均可
+- 不接受的 PR 会说明原因，欢迎根据反馈修改后重新提交

README.md

@@ -30,6 +30,7 @@
 
 | 日期 | 内容 |
 |------|------|
+| 05-06 | **v1.1.0** 熔断器+智能重试(syntax/import快速熔断+scope缩小) + Gate置信度 + Verifier结构化错误 + Experience Replay(BM25历史轨迹) + Session多轮连贯 + Locator精准裁剪 |
 | 04-30 | 三层上下文架构 + SSH持久会话 + Gate/路由优化 + PCED-Lite多源聚合 + 搜索site:自动限定 + qwen3:8b 20题真实验证 + 13项bug修复 |
 | 04-29 | 5元专家体系定稿 + 15个SKILL.md渐进加载 + DAG多任务编排 + Debug Subagent + Token预算/Guardrails/可观测性 |
 
@@ -55,7 +56,7 @@ KWCode 的思路不同：**LLM 只做分类和生成，确定性流水线做决
 
 小模型修 bug 失败后，用同样的方式再试一遍，三次机会全浪费在同一个错误上。
 
-> KWCode 解法：**三阶段重试 + Reflection + Debug Subagent**——第一次正常描述，第二次从错误信息出发（注入运行时调试数据），第三次最小化修改。每次重试前先做 Reflection（LLM 分析上次为什么失败）+ Debug Subagent（sys.settrace 捕获真实变量值），绝不重复同样的错。
+> KWCode 解法：**三阶段重试 + Reflection + Debug Subagent + 智能熔断**——第一次正常描述，第二次从错误信息出发（注入运行时调试数据），第三次最小化修改。语法错误/缺依赖自动熔断不浪费重试；同类错误3次自动停止；第2次失败自动缩小修改范围。
 
 **痛点三：不能调用工具**
 
@@ -546,37 +547,49 @@ kaiwu/
 
 ## 参与贡献
 
-**KWCode 是中国开发者做的，欢迎 fork 后自由修改优化。**
+**KWCode 是中国开发者做的，欢迎贡献代码。** 详细规范见 [CONTRIBUTING.md](CONTRIBUTING.md)。
 
-### 推荐方式
-
-1. **Fork 本仓库**，在你自己的分支上修改
-2. 跑通测试：`python -m pytest kaiwu/tests/ --ignore=kaiwu/tests/bench_tasks`
-3. 提交 PR 或直接在你的 fork 上用
-
-### 可以做的事
-
-**新增领域知识**（最简单，创建一个 SKILL.md 目录）：
+### 快速开始
 
 ```bash
-# 在 kaiwu/builtin_experts/ 下创建新目录
-mkdir kaiwu/builtin_experts/vue3
-# 编辑 SKILL.md（参考现有专家格式）
+git clone https://github.com/val1813/kwcode.git
+cd kwcode
+pip install -e ".[dev]"
+python -m pytest kaiwu/tests/ -v --ignore=kaiwu/tests/bench_tasks
+# 全部绿才能提 PR
 ```
 
-急需的领域知识：Vue3 · Django · Go Gin · Rust Actix · K8s · Docker · Redis · MySQL · React · Next.js
+### 架构红线（违反即拒绝）
 
-**其他方向**：
-- 多语言 AST 支持（JavaScript/TypeScript/Java/Go）
-- bench_tasks 补齐（bugfix 类、跨文件类）
-- 新的确定性脚本（`scripts/` 目录下，不进 LLM context）
-- 飞轮优化规则的质量验证
+| 红线 | 说明 |
+|------|------|
+| RED-1 | Gate 必须输出结构化 JSON，不得字符串解析 |
+| RED-2 | LLM 只做分类和生成，不得让 LLM 决定流水线下一步 |
+| RED-3 | 每个专家独立上下文，不继承上一个专家的对话历史 |
+| RED-4 | 新增依赖必须离线可用 |
+| RED-5 | 重试次数必须有硬上限 |
 
-### 不建议改的
+**一票否决**：向量数据库、企业级安全库、需要云服务的依赖、多 Agent 并行框架、自动修改 Gate 路由规则。
 
-- 5 个元专家的接口和流水线顺序（架构已定稿）
-- Gate 的 JSON 输出格式
-- SKILL.md 的 frontmatter 字段定义
+### 最欢迎的贡献
+
+| 类型 | 说明 |
+|------|------|
+| 新增专家 | 创建 `kaiwu/builtin_experts/<name>/SKILL.md`，最简单的贡献方式 |
+| Bug 修复 | 附复现步骤和测试用例 |
+| 多语言 AST | JS/TS/Go/Rust/Java 调用图支持 |
+| 性能优化 | Locator 定位速度、ContextPruner 压缩质量 |
+
+急需认领的专家：Vue3 · React · Django · FastAPI · Go Gin · Rust Actix · K8s · Docker · MySQL · Redis
+
+### PR 要求
+
+1. **测试全绿**：`python -m pytest kaiwu/tests/ --ignore=kaiwu/tests/bench_tasks`
+2. **新功能必须有测试**（改 core/experts/flywheel 目录时）
+3. **新增依赖需说明**用途和离线可用性
+4. **改 Gate/Orchestrator 逻辑**请先开 Issue 讨论
+
+完整 PR 模板和代码风格规范见 [CONTRIBUTING.md](CONTRIBUTING.md)。
 
 ---
 

STATUS.md

@@ -7,9 +7,44 @@
 
 ---
 
-## 当前状态：v0.9.0 (2026-04-29)
+## 当前状态：v1.1.0 (2026-05-06)
 
-292/292 测试全绿。Python专家系统已移除，改为正确方向。
+328/328 测试全绿（不含bench_tasks存根）。P0+P1优化全部完成。
+
+### v1.1.0 新增：P0+P1 优化（7文件 +362行）
+
+**P0-1: Verifier结构化输出** (`experts/verifier.py`)
+- `_classify_error()` 纯正则提取 error_type/error_file/error_line/error_message/failed_tests
+- 5种错误类型：syntax/assertion/import/runtime/patch_apply
+- 所有错误路径统一返回结构化字段，DebugSubagent不再需要自己解析
+
+**P0-2: 熔断器+缩小scope** (`core/orchestrator.py`)
+- syntax错误1次后直接熔断（重试无意义）
+- import错误立即熔断+提示安装依赖
+- 同类error_type连续3次→硬熔断
+- 第2次失败自动缩小scope到第一个文件+函数
+- 低置信度(<0.6)任务自动减少重试预算
+
+**P0-3: Gate置信度输出** (`core/gate.py`)
+- `_estimate_confidence()` 关键词信号强度评分（0.92/0.75/0.55三档）
+- 不覆盖expert_registry已有的confidence
+- orchestrator消费：低置信度减少max_retries
+
+**P1-1: Experience Replay** (`flywheel/trajectory_collector.py` + `core/orchestrator.py` + `core/context.py`)
+- `find_similar()` BM25检索历史成功轨迹（复用已有rank-bm25依赖）
+- orchestrator.run()开头自动调用，结果存入ctx.similar_trajectories
+- 飞轮闭环：同类任务不走冷启动
+
+**P1-2: Session内多轮连贯** (`cli/main.py`)
+- SessionState类：跟踪tasks/files_touched/turn_count
+- `to_reminder()` 生成System Reminder注入Gate memory_context
+- 每5轮重新注入KWCODE.md核心规则（注意力衰减对抗）
+
+**P1-3: Locator最小上下文裁剪** (`experts/locator.py`)
+- 函数边界识别（indent-based，def到下一个同级def）
+- 去掉纯注释行，docstring限3行
+- 60行/函数上限，gap marker标记不连续区域
+- 文件路径header + 行号前缀
 
 ### v0.9.0 新增
 

kaiwu/cli/main.py

@@ -445,6 +445,44 @@ REPL_COMMANDS = {
 }
 
 
+class SessionState:
+    """Tracks session state for multi-turn coherence and System Reminders."""
+
+    def __init__(self):
+        self.tasks_this_session: list[dict] = []
+        self.files_touched: set = set()
+        self.turn_count: int = 0
+
+    def record_task(self, user_input: str, success: bool, files: list[str], elapsed: float):
+        """Record a completed task."""
+        self.turn_count += 1
+        self.tasks_this_session.append({
+            "input": user_input[:100],
+            "success": success,
+            "files": files[:5],
+            "elapsed": elapsed,
+        })
+        self.files_touched.update(files)
+
+    def to_reminder(self) -> str:
+        """Generate System Reminder text for injection into Gate memory_context."""
+        if not self.tasks_this_session:
+            return ""
+        recent = self.tasks_this_session[-3:]
+        lines = ["[本次会话已完成]"]
+        for t in recent:
+            status = "OK" if t["success"] else "FAIL"
+            files_str = ", ".join(t["files"][:2]) if t["files"] else ""
+            line = f"- [{status}] {t['input'][:40]}"
+            if files_str:
+                line += f" → {files_str}"
+            lines.append(line)
+        if self.files_touched:
+            touched = list(self.files_touched)[:5]
+            lines.append(f"[已修改文件] {', '.join(touched)}")
+        return "\n".join(lines)
+
+
 VERSION = "0.9.0"
 
 # ── Shadow/重影大字 KAIWU ──
@@ -538,6 +576,7 @@ def _repl(model_path, ollama_url, ollama_model, project_root, verbose, no_search
     tps_estimator = TokPerSecEstimator()
     pruner = ContextPruner(max_tokens=status.ctx_max)
     conversation_history: list[dict] = []
+    session_state = SessionState()
 
     # Background VRAM watcher
     vram_watcher = VRAMWatcher(status)
@@ -776,6 +815,16 @@ def _repl(model_path, ollama_url, ollama_model, project_root, verbose, no_search
         t0 = time.perf_counter()
         # P2: Small model forces plan mode (问题6修复：用户可通过 no_search 间接控制)
         effective_plan = plan_next or (model_strategy.force_plan_mode and not no_search)
+
+        # Inject session reminder into memory context for Gate
+        session_reminder = session_state.to_reminder()
+        if session_reminder and session_state.turn_count % 5 == 0:
+            # Every 5 turns, also re-inject KWCODE.md core rules (attention decay countermeasure)
+            from kaiwu.core.kwcode_md import load_kwcode_md, build_kwcode_system
+            kwcode_sections = load_kwcode_md(project_root)
+            if kwcode_sections and "all" in kwcode_sections:
+                session_reminder += f"\n\n[项目规则提醒]\n{kwcode_sections['all'][:500]}"
+
         success = _run_task(
             task=user_input,
             gate=gate,
@@ -795,6 +844,19 @@ def _repl(model_path, ollama_url, ollama_model, project_root, verbose, no_search
         elapsed = time.perf_counter() - t0
         plan_next = False  # Reset plan flag
 
+        # Record task in session state
+        task_files = []
+        if success:
+            try:
+                last_result = getattr(orchestrator, '_last_result', None)
+                if last_result and last_result.get("context"):
+                    ctx = last_result["context"]
+                    if ctx.generator_output:
+                        task_files = [p.get("file", "") for p in ctx.generator_output.get("patches", [])]
+            except Exception:
+                pass
+        session_state.record_task(user_input, success, task_files, elapsed)
+
         # Update tok/s estimator (rough: use elapsed as proxy)
         tps_estimator.record("x" * int(elapsed * 15), elapsed)  # ~15 tok/s estimate
         status.tok_per_sec = tps_estimator.value

kaiwu/core/context.py

@@ -67,3 +67,6 @@ class TaskContext:
 
     # 上游依赖结果摘要（Active Context，≤2K tokens，供Gate/Generator看）
     upstream_summary: str = ""
+
+    # Experience Replay: similar successful trajectories from history
+    similar_trajectories: list = field(default_factory=list)

kaiwu/core/gate.py

@@ -154,6 +154,10 @@ class Gate:
                     result["route_type"] = "general_with_expert"
                     # 不覆盖pipeline，让orchestrator用通用的EXPERT_SEQUENCES
 
+        # Inject confidence if not already set by expert registry
+        if "confidence" not in result:
+            result["confidence"] = self._estimate_confidence(result, user_input)
+
         return result
 
     @staticmethod
@@ -224,3 +228,31 @@ class Gate:
         if start != -1 and end != -1 and end > start:
             return text[start:end + 1]
         return text.strip()
+
+    def _estimate_confidence(self, result: dict, user_input: str) -> float:
+        """
+        Estimate classification confidence based on keyword signal strength.
+        High confidence = user input contains clear type-signal words.
+        No LLM call, pure rules, <1ms.
+        """
+        et = result.get("expert_type", "")
+        lower = user_input.lower()
+
+        STRONG_SIGNALS = {
+            "locator_repair": ["修复", "fix", "bug", "报错", "错误", "失败", ".py:", "line "],
+            "codegen":        ["写一个", "创建", "生成", "新建", "from scratch", "写个"],
+            "refactor":       ["重构", "优化", "整理", "拆分", "extract", "rename"],
+            "office":         [".xlsx", ".docx", ".pptx", "excel", "幻灯片", "演示文稿"],
+            "chat":           ["你好", "什么是", "解释", "为什么", "怎么理解"],
+            "vision":         ["图片", "截图", "设计图", "image", "screenshot"],
+        }
+
+        signals = STRONG_SIGNALS.get(et, [])
+        matched = sum(1 for s in signals if s in lower)
+
+        if matched >= 2:
+            return 0.92
+        elif matched == 1:
+            return 0.75
+        else:
+            return 0.55

kaiwu/core/orchestrator.py

@@ -214,6 +214,18 @@ class PipelineOrchestrator:
 
         self._emit(on_status, "gate", f"任务类型：{expert_type} | 难度：{gate_result.get('difficulty', '?')}")
 
+        # ── Experience Replay: find similar successful trajectories ──
+        if self.trajectory_collector and expert_type not in ("chat", "office", "vision"):
+            try:
+                similar = self.trajectory_collector.find_similar(user_input, expert_type, k=3)
+                if similar:
+                    ctx.similar_trajectories = similar
+                    best = similar[0]
+                    self._emit(on_status, "replay",
+                               f"发现相似成功案例：{best.get('user_input', '')[:40]}")
+            except Exception as e:
+                logger.debug("Experience replay failed (non-blocking): %s", e)
+
         # codegen任务如果涉及实时数据，首次就触发搜索（不等失败重试）
         if expert_type == "codegen" and not no_search and self._needs_realtime_data(user_input):
             self._emit(on_status, "search", "检测到实时数据需求，预搜索...")
@@ -233,6 +245,13 @@ class PipelineOrchestrator:
         # Dynamic retry budget based on task difficulty
         max_retries = self._get_max_retries(gate_result)
 
+        # Low confidence: reduce retry budget (not worth many attempts)
+        confidence = gate_result.get("confidence", 1.0)
+        if confidence < 0.6 and expert_type not in ("chat", "office", "vision"):
+            max_retries = min(max_retries, 2)
+            self._emit(on_status, "low_confidence",
+                       f"任务分类置信度较低({confidence:.0%})，减少重试次数")
+
         while ctx.retry_count < max_retries:
             success = self._run_sequence(sequence, ctx, on_status)
 
@@ -277,6 +296,51 @@ class PipelineOrchestrator:
             # Save failure info for retry strategy
             ctx.previous_failure = error_detail
 
+            # ── Circuit breaker: same error_type streak ──
+            current_error_type = ""
+            if ctx.verifier_output:
+                current_error_type = ctx.verifier_output.get("error_type", "unknown")
+
+            if not hasattr(ctx, '_error_type_streak'):
+                ctx._error_type_streak = {"type": "", "count": 0}
+
+            if current_error_type and current_error_type == ctx._error_type_streak["type"]:
+                ctx._error_type_streak["count"] += 1
+            else:
+                ctx._error_type_streak = {"type": current_error_type, "count": 1}
+
+            # Fast circuit break: syntax errors don't improve with retries
+            if current_error_type == "syntax" and ctx.retry_count >= 1:
+                self._emit(on_status, "circuit_break", "语法错误重试无效，模型能力不足以完成此任务")
+                break
+            # Fast circuit break: missing imports need user action
+            if current_error_type == "import":
+                missing = ctx.verifier_output.get("error_message", "") if ctx.verifier_output else ""
+                self._emit(on_status, "circuit_break", f"缺少依赖：{missing}，请先安装")
+                break
+            # Hard circuit break: same error type 3 times in a row
+            if ctx._error_type_streak["count"] >= 3:
+                self._emit(on_status, "circuit_break",
+                           f"同类错误({current_error_type})连续{ctx._error_type_streak['count']}次，停止重试")
+                break
+
+            # ── 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._emit(on_status, "retry", f"第{ctx.retry_count}次尝试失败：{error_detail[:100]}")
 
             # Reflection before 2nd retry: ask LLM why the patch failed

kaiwu/experts/locator.py

@@ -177,7 +177,7 @@ class LocatorExpert:
             content = self.tools.read_file(fpath)
             if content.startswith("[ERROR]"):
                 continue
-            snippet = self._extract_snippet(content, relevant_functions)
+            snippet = self._extract_snippet(content, relevant_functions, file_path=fpath)
             if snippet:
                 code_snippets[fpath] = snippet
 
@@ -234,7 +234,7 @@ class LocatorExpert:
             content = self.tools.read_file(fpath)
             if content.startswith("[ERROR]"):
                 continue
-            snippet = self._extract_snippet(content, all_functions)
+            snippet = self._extract_snippet(content, all_functions, file_path=fpath)
             if snippet:
                 code_snippets[fpath] = snippet
 
@@ -356,8 +356,9 @@ class LocatorExpert:
         verified_locs = [f"{file_path}:{f}" for f in verified_funcs]
         return verified_funcs, verified_locs
 
-    def _extract_snippet(self, content: str, functions: list[str]) -> str:
-        """Extract code around target functions (+-20 lines)."""
+    def _extract_snippet(self, content: str, functions: list[str], file_path: str = "") -> str:
+        """Extract minimal context around target functions. Identifies function boundaries,
+        strips comments/docstrings, caps at 60 lines per function."""
         if not functions:
             return content[:2000]
 
@@ -365,23 +366,105 @@ class LocatorExpert:
         collected = set()
 
         for func_name in functions:
-            # Strip class prefix for matching
             short_name = func_name.split(".")[-1] if "." in func_name else func_name
+            func_start = None
+            func_indent = -1
+
             for i, line in enumerate(lines):
-                if f"def {short_name}" in line or f"class {short_name}" in line:
-                    start = max(0, i - 5)
-                    end = min(len(lines), i + 40)
-                    for j in range(start, end):
+                stripped = line.lstrip()
+                if (f"def {short_name}" in line or f"class {short_name}" in line) and stripped.startswith(("def ", "class ")):
+                    func_start = i
+                    func_indent = len(line) - len(stripped)
+                    break
+
+            if func_start is None:
+                # Fallback: grab +-20 lines around keyword match
+                for i, line in enumerate(lines):
+                    if short_name in line:
+                        start = max(0, i - 5)
+                        end = min(len(lines), i + 25)
+                        for j in range(start, end):
+                            collected.add(j)
+                        break
+                continue
+
+            # Find function end: next def/class at same or lower indent level
+            func_end = len(lines)
+            for i in range(func_start + 1, len(lines)):
+                line = lines[i]
+                if not line.strip():
+                    continue
+                current_indent = len(line) - len(line.lstrip())
+                if current_indent <= func_indent and line.lstrip().startswith(("def ", "class ", "@")):
+                    func_end = i
+                    break
+
+            # Include decorators above function (up to 3 lines)
+            decorator_start = func_start
+            for i in range(func_start - 1, max(func_start - 4, -1), -1):
+                if i >= 0 and lines[i].lstrip().startswith("@"):
+                    decorator_start = i
+                else:
+                    break
+
+            # Collect lines, skip pure comment blocks in the middle
+            start = decorator_start
+            end = min(func_end, func_start + 60)  # Cap at 60 lines
+            in_docstring = False
+            docstring_lines = 0
+
+            for j in range(start, end):
+                line = lines[j]
+                stripped = line.strip()
+
+                # Track docstring boundaries
+                if '"""' in stripped or "'''" in stripped:
+                    if in_docstring:
+                        in_docstring = False
+                        docstring_lines += 1
+                        if docstring_lines <= 3:
+                            collected.add(j)
+                        continue
+                    else:
+                        in_docstring = True
+                        docstring_lines = 0
+                        if stripped.endswith('"""') and stripped.count('"""') == 2:
+                            # Single-line docstring
+                            in_docstring = False
+                            collected.add(j)
+                            continue
                         collected.add(j)
+                        continue
+
+                if in_docstring:
+                    docstring_lines += 1
+                    if docstring_lines <= 3:
+                        collected.add(j)
+                    continue
+
+                # Skip pure comment lines (but keep inline comments)
+                if stripped.startswith("#") and j > func_start + 1:
+                    # Keep comments that look structural
+                    if any(kw in stripped for kw in ("TODO", "FIXME", "NOTE", "HACK", "──")):
+                        collected.add(j)
+                    continue
+
+                collected.add(j)
 
         if not collected:
             return content[:2000]
 
         sorted_lines = sorted(collected)
         result = []
+        prev_idx = -2
         for idx in sorted_lines:
+            if idx - prev_idx > 1 and prev_idx >= 0:
+                result.append("     | ...")  # Gap marker
             result.append(f"{idx + 1:4d} | {lines[idx]}")
-        return "\n".join(result)
+            prev_idx = idx
+
+        header = f"# {file_path}\n" if file_path else ""
+        return header + "\n".join(result)
 
     @staticmethod
     def _parse_file_list(raw: str) -> list[str]:

kaiwu/experts/verifier.py

@@ -6,6 +6,7 @@ RED-3: Independent context window, does not inherit Generator history.
 
 import json
 import logging
+import re
 from typing import Optional
 
 from kaiwu.core.context import TaskContext
@@ -81,6 +82,11 @@ class VerifierExpert:
                 "tests_passed": 0,
                 "tests_total": 0,
                 "error_detail": "All patches failed to apply",
+                "error_type": "patch_apply",
+                "error_file": "",
+                "error_line": 0,
+                "error_message": "All patches failed to apply",
+                "failed_tests": [],
             }
             ctx.verifier_output = result
             return result
@@ -100,12 +106,19 @@ class VerifierExpert:
 
         if not syntax_ok:
             self._rollback(backups)
+            error_msg = f"Syntax errors: {'; '.join(syntax_errors)}"
+            error_info = self._classify_error(error_msg)
             result = {
                 "passed": False,
                 "syntax_ok": False,
                 "tests_passed": 0,
                 "tests_total": 0,
-                "error_detail": f"Syntax errors: {'; '.join(syntax_errors)}",
+                "error_detail": error_msg,
+                "error_type": "syntax",
+                "error_file": error_info["error_file"],
+                "error_line": error_info["error_line"],
+                "error_message": error_msg[:200],
+                "failed_tests": [],
             }
             ctx.verifier_output = result
             return result
@@ -121,16 +134,67 @@ class VerifierExpert:
         if not passed:
             self._rollback(backups)
 
+        error_info = self._classify_error(test_error) if not passed else {
+            "error_type": "", "error_file": "", "error_line": 0,
+            "error_message": "", "failed_tests": []}
         result = {
             "passed": passed,
             "syntax_ok": syntax_ok,
             "tests_passed": tests_passed,
             "tests_total": tests_total,
             "error_detail": test_error if not passed else "",
+            "error_type": error_info["error_type"],
+            "error_file": error_info["error_file"],
+            "error_line": error_info["error_line"],
+            "error_message": error_info["error_message"],
+            "failed_tests": error_info["failed_tests"],
         }
         ctx.verifier_output = result
         return result
 
+    def _classify_error(self, error_detail: str) -> dict:
+        """Extract structured error info from pytest/compile output. Pure regex, no LLM."""
+        info = {"error_type": "unknown", "error_file": "", "error_line": 0,
+                "error_message": "", "failed_tests": []}
+
+        if not error_detail:
+            return info
+
+        # error_type classification
+        if "SyntaxError" in error_detail:
+            info["error_type"] = "syntax"
+        elif "AssertionError" in error_detail:
+            info["error_type"] = "assertion"
+        elif "ModuleNotFoundError" in error_detail or "ImportError" in error_detail:
+            info["error_type"] = "import"
+        elif "patch" in error_detail.lower() and "failed" in error_detail.lower():
+            info["error_type"] = "patch_apply"
+        elif any(exc in error_detail for exc in ("TypeError", "ValueError", "KeyError",
+                 "AttributeError", "NameError", "IndexError", "RuntimeError")):
+            info["error_type"] = "runtime"
+
+        # Extract file and line: File "xxx.py", line 42
+        file_match = re.search(r'File "([^"]+)", line (\d+)', error_detail)
+        if file_match:
+            info["error_file"] = file_match.group(1)
+            info["error_line"] = int(file_match.group(2))
+
+        # Extract failed test names: "FAILED tests/test_xxx.py::test_func"
+        info["failed_tests"] = re.findall(r'FAILED\s+(\S+::\S+)', error_detail)
+
+        # Extract error message
+        lines = [l.strip() for l in error_detail.splitlines() if l.strip()]
+        if lines:
+            for line in reversed(lines):
+                exc_match = re.match(r'(\w+Error|\w+Exception):\s*(.+)', line)
+                if exc_match:
+                    info["error_message"] = exc_match.group(2)[:200]
+                    break
+            if not info["error_message"] and lines:
+                info["error_message"] = lines[-1][:200]
+
+        return info
+
     def _run_tests(self, ctx: TaskContext) -> tuple[int, int, str]:
         """Run project tests. Returns (passed, total, error_detail)."""
         # Try to detect test runner
@@ -168,7 +232,6 @@ class VerifierExpert:
     @staticmethod
     def _parse_test_output(output: str) -> tuple[int, int]:
         """Parse test counts from pytest/unittest output."""
-        import re
 
         # pytest format: "5 passed" or "3 passed, 2 failed"
         passed_match = re.search(r"(\d+) passed", output)

kaiwu/flywheel/trajectory_collector.py

@@ -131,3 +131,46 @@ class TrajectoryCollector:
             t for t in self._load_all()
             if t.gate_result.get("expert_name") == expert_name
         ]
+
+    def find_similar(self, user_input: str, expert_type: str, k: int = 3) -> list[dict]:
+        """
+        BM25 retrieval of similar successful trajectories.
+        Returns: [{"user_input": str, "pipeline": list, "files_modified": list, "latency_s": float}]
+        """
+        candidates = self._load_successful(expert_type)
+        if not candidates:
+            return []
+
+        try:
+            from rank_bm25 import BM25Okapi
+        except ImportError:
+            logger.debug("[trajectory] rank_bm25 not available, skipping similarity search")
+            return []
+
+        corpus = [t["user_input"].split() for t in candidates]
+        bm25 = BM25Okapi(corpus)
+        scores = bm25.get_scores(user_input.split())
+        top_k = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:k]
+        return [candidates[i] for i in top_k if scores[i] > 0.1]
+
+    def _load_successful(self, expert_type: str, limit: int = 200) -> list[dict]:
+        """Load successful trajectories of given type, most recent first, up to limit."""
+        if not os.path.isdir(self._dir):
+            return []
+
+        trajs = []
+        fnames = sorted(os.listdir(self._dir), reverse=True)
+        for fname in fnames[:500]:
+            if not fname.endswith(".json"):
+                continue
+            path = os.path.join(self._dir, fname)
+            try:
+                with open(path, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                if data.get("success") and data.get("expert_used") == expert_type:
+                    trajs.append(data)
+                    if len(trajs) >= limit:
+                        break
+            except Exception:
+                continue
+        return trajs