fix: resolve 12 bugs in v0.2 codebase

Bug fixes included:

Compose Module:
- fix(compose): correct typo in Next.input field description (#1)
- fix(workflow): defer next_pointer cleanup until first step succeeds (#2)
- fix(team): enhance task group exception handling (#11)

Core Module:
- fix(react-engine): notify user on streaming loop exit reason (#3)
- fix(memory): use lazy initialization for SummaryTokenMemory lock (#4)
- fix(session): add debounce mechanism for auto-save (#5)
- fix(toolbox): ensure thread-safe stats iteration (#6)
- fix(structure): add length limits to JSON extraction (#10)
- fix(events): use id() for robust handler deduplication (#12)

Plugins Module:
- fix(storage): add missing @register_storage to SQLiteStorage (#7)
- fix(models): improve tokenizer preloading for non-OpenAI models (#8)
- fix(knowledge): use Pydantic PrivateAttr for RetrievalTool (#9)

Breaking Changes: None
Migration Required: None

Tested on: Python 3.10, 3.11, 3.12

Author: xuemzhan

docs/system design document.md

@@ -35,7 +35,7 @@ def check_score(score: int):
     """
     node: str = Field(..., description="下一个节点的名称")
     input: Optional[Any] = Field(default=None, 
-        iption="传递给下一个节点的输入数据。如果为 None，则保持上下文中的 last_output 不变。"
+        description="传递给下一个节点的输入数据。如果为 None，则保持上下文中的 last_output 不变。"
     ) # type: ignore
     # [New] 允许在跳转时更新 Context.state # type: ignore
     update_state: Dict[str, Any] = Field(

gecko/compose/nodes.py

@@ -105,12 +105,17 @@ async def run(self, context_or_input: Any) -> List[MemberResult]:
         )
 
         # 2. 初始化容器
-        # 使用 MemberResult 占位，初始状态设为失败，防止未执行的情况
+        # 注释: results[idx] 的写入是索引隔离的，每个 worker 只写自己的索引
+        # 这种模式在 Python 中是并发安全的，因为列表元素的赋值是原子操作
         results: List[Optional[MemberResult]] = [None] * member_count
 
         # 3. 准备并发控制
         semaphore = anyio.Semaphore(self.max_concurrent) if self.max_concurrent > 0 else None
 
+        # 修复: 追踪是否有任务失败（用于日志）
+        failed_indices: List[int] = []
+        failed_lock = anyio.Lock()  # 用于保护 failed_indices 的并发写入
+
         # 4. 定义 Worker
         async def _worker(idx: int, member: Any):
             if semaphore:
@@ -139,15 +144,30 @@ async def _worker(idx: int, member: Any):
                     error=str(e),
                     is_success=False
                 )
+                # 修复: 安全地记录失败索引
+                async with failed_lock:
+                    failed_indices.append(idx)
             finally:
                 if semaphore:
                     semaphore.release()
 
         # 5. 启动并发任务组
-        async with anyio.create_task_group() as tg:
-            for idx, member in enumerate(self.members):
-                tg.start_soon(_worker, idx, member)
-
+        # 注释: anyio.create_task_group 会等待所有任务完成
+        # 如果任何任务抛出未捕获的异常，会取消其他任务
+        # 但我们在 _worker 中已经捕获了所有异常，所以不会发生取消
+        try:
+            async with anyio.create_task_group() as tg:
+                for idx, member in enumerate(self.members):
+                    tg.start_soon(_worker, idx, member)
+        except ExceptionGroup as eg:
+            # 修复: Python 3.11+ 的 ExceptionGroup 处理
+            logger.error(
+                "Team execution encountered exceptions",
+                team=self.name,
+                exception_count=len(eg.exceptions)
+            )
+            # 异常已在 worker 中处理并记录到 results，这里只记录日志
+            
         # 6. 结果整理
         # 理论上 task_group 结束时所有 results 都已被赋值，这里做一次非空断言过滤
         final_results = [r for r in results if r is not None]

gecko/compose/team.py

@@ -409,6 +409,8 @@ async def execute(self, input_data: Any, session_id: Optional[str] = None) -> An
     async def resume(self, session_id: str) -> Any:
         """
         从存储中恢复执行
+        
+        修复: 延迟清除 next_pointer，确保首步执行成功后再清除
         """
         if not self.storage:
             raise ValueError("Cannot resume: Storage not configured")
@@ -431,8 +433,9 @@ async def resume(self, session_id: str) -> Any:
 
         # 3. 确定下一步
         next_node = None
+        # ✅ 修复: 使用标记位追踪是否从 next_pointer 恢复，而不是立即清除
+        resumed_from_pointer = False
 
-        # [优化] 优先检查是否存在动态跳转指针
         if context.next_pointer:
             logger.info("Resuming from dynamic Next pointer", target=context.next_pointer.get("target_node"))
             next_node = context.next_pointer.get("target_node")
@@ -441,17 +444,15 @@ async def resume(self, session_id: str) -> Any:
             if context.next_pointer.get("input"):
                 context.state["_next_input"] = context.next_pointer["input"]
 
-            # 消费指针 (已使用，清除以避免重复)
-            context.clear_next_pointer()
+            # ✅ 修复: 标记而不是立即清除，让 _execute_loop 在成功执行后清除
+            resumed_from_pointer = True
 
         elif last_node:
-            # 只有在没有动态指针时，才回退到基于静态图的推导
             next_node = await self._find_next_node(last_node, context)
             if not next_node:
                 logger.info("Workflow already completed (no next node)", session_id=session_id)
                 return context.get_last_output()
         else:
-            # 这是一个全新的会话（或者刚初始化未执行）
             next_node = self._entry_point
 
         # 4. 继续执行循环
@@ -460,7 +461,8 @@ async def resume(self, session_id: str) -> Any:
                 context, 
                 session_id, 
                 start_node=next_node, 
-                start_step=current_step
+                start_step=current_step,
+                clear_pointer_after_first_step=resumed_from_pointer  # ✅ 新增参数
             )
 
             # 最终保存
@@ -473,39 +475,39 @@ async def resume(self, session_id: str) -> Any:
             logger.exception("Resume execution failed")
             raise
 
-    async def _execute_loop(self, 
-                            context: WorkflowContext,  
-                            session_id: Optional[str], 
-                            start_node: Optional[str],
-                            start_step: int):
+    async def _execute_loop(
+        self, 
+        context: WorkflowContext,  
+        session_id: Optional[str], 
+        start_node: Optional[str],
+        start_step: int,
+        clear_pointer_after_first_step: bool = False  # ✅ 新增参数
+    ):
         """核心执行循环"""
         current_node = start_node
         steps = start_step
+        is_first_step = True  # ✅ 追踪是否为首步
 
         while current_node and steps < self.max_steps:
             steps += 1
 
-            # 1. 如果是从 next_pointer 恢复的（Resume 场景），跳过执行，直接流转
-            # 但这里逻辑比较绕，更清晰的是：如果 next_pointer 存在，说明上一步是 Next 指令，
-            # 且已经持久化了，我们应该直接使用 next_pointer 指向的节点作为 current_node。
-            # 这在 resume() 方法中处理更合适，这里保持循环逻辑。
-
             # 执行节点
             logger.debug("Executing step", step=steps, node=current_node)
-            # 执行节点逻辑
             result = await self._execute_node_safe(current_node, context)
 
+            # ✅ 修复: 首步成功后清除 pointer
+            if is_first_step and clear_pointer_after_first_step:
+                context.clear_next_pointer()
+                is_first_step = False
+
             # 准备持久化所需的临时变量
-            # 记录当前节点为“已完成节点”
             persist_node = current_node 
             next_target = None
 
             # 处理流转逻辑
             if isinstance(result, Next):
-                # === 动态跳转处理 ===
                 next_target = result.node
 
-                # 更新 Input / State
                 if result.input is not None:
                     normalized = self._normalize_result(result.input)
                     context.history["last_output"] = normalized
@@ -514,50 +516,26 @@ async def _execute_loop(self,
                 if result.update_state:
                     context.state.update(result.update_state)
 
-                # [关键优化] 记录动态指针，确保持久化时包含此信息
                 context.next_pointer = {
                     "target_node": next_target,
                     "input": context.state.get("_next_input")
                 }
 
-                # 即使是跳转，也需要在 history 中留痕，证明此节点已执行完毕
-                # 这里记录一个特殊的标识，方便调试
                 context.history[current_node] = f"<Next -> {next_target}>"
 
             else:
-                # === 静态流转处理 ===
                 normalized = self._normalize_result(result)
                 context.history[current_node] = normalized
                 context.history["last_output"] = normalized
 
-                # 既然走了静态流程，确保清除之前的指针（防御性编程）
                 context.clear_next_pointer()
 
-                # 基于静态图寻找下一跳
                 next_target = await self._find_next_node(current_node, context)
 
-            # [优化] 立即持久化 (Atomic Checkpoint)
-            # 此时 context 包含了最新的 history 和 next_pointer
-            # 即使下一秒 Crash，resume 时也能通过 next_pointer 找到 next_target
+            # 持久化
             if self.storage and session_id:
                 await self._persist_state(session_id, steps, persist_node, context)
 
-            # [新增] 关键修正：状态推进
-            # 一旦完成了持久化，next_pointer 的使命（防Crash）在当前步已完成。
-            # 进入下一步前，如果那是基于 next_pointer 的跳转，理论上应在内存中清除，
-            # 以免在 B 的 Pre-Commit 中还带着 "A->B" 的指针。
-            # 但是，如果我们在 B 执行前 Crash，Resume 时加载的是 A 执行后的状态（含指针），这是对的。
-            # 如果我们在 B 执行中 Crash，Resume 加载的是 B 的 Pre-Commit 状态。
-            # B 的 Pre-Commit 状态如果包含 "A->B" 指针，Resume 会再次尝试跳转到 B。
-            # 此时 last_node="B", next_pointer={"target":"B"}。
-            # Resume 逻辑：优先 next_pointer -> target="B"。结果一样。
-            
-            # 问题的根源是测试用例的查找逻辑太宽泛了。它只要找到包含 next_pointer 的记录就认为那是 A 的记录。
-            # 实际上 B 的 Pre-Commit 记录也包含了它。
-            
-            # 让我们在测试用例中更精确地定位。
-
-            # 推进到下一个节点
             current_node = next_target
 
         if steps >= self.max_steps:

gecko/compose/workflow.py

@@ -385,14 +385,15 @@ async def _run_reasoning_loop(
             tool_calls=last_msg.tool_calls or [],
         )
 
-    # [修改方法] 重构流式循环，使用 _process_turn_results
+    
     async def _run_streaming_loop(
         self, context: ExecutionContext, llm_params: Dict[str, Any]
     ) -> AsyncIterator[str]:
         """
         ReAct 流式循环 (递归模式)
+        
+        修复: 在检测到死循环或错误退出时，向用户输出提示信息
         """
-        # 循环控制：只要 turn 未达上限，且 should_continue 为 True，就一直循环
         while context.turn < self.max_turns:
             context.turn += 1
 
@@ -402,30 +403,23 @@ async def _run_streaming_loop(
 
             messages_payload = [m.to_openai_format() for m in context.messages]
 
-            # 状态累积器 (每轮开始前重置)
             collected_content = []
             tool_calls_data: List[Dict[str, Any]] = []
 
-            # 1. 消费流 (Inner Loop: Streaming Consumer)
-            # 负责将 LLM 的 Token 实时透传给用户，并累积工具调用信息
             async for chunk in self.model.astream(messages=messages_payload, **llm_params): # type: ignore
                 delta = self._extract_delta(chunk)
 
-                # A. 文本内容：实时 Yield
                 content = delta.get("content")
                 if content:
                     collected_content.append(content)
                     yield content
 
-                # B. 工具调用：后台累积
                 if delta.get("tool_calls"):
                     self._accumulate_tool_chunks(tool_calls_data, delta["tool_calls"])
 
-            # 2. 组装完整消息 (Turn Completion)
             final_text = "".join(collected_content)
             assistant_msg = Message.assistant(content=final_text)
 
-            # 清洗并组装工具调用
             if tool_calls_data:
                 valid_calls = [
                     tc for tc in tool_calls_data 
@@ -434,20 +428,47 @@ async def _run_streaming_loop(
                 if valid_calls:
                     assistant_msg.tool_calls = valid_calls
 
-            # 3. 处理回合逻辑 (Decision Making)
-            # 复用基类的 _process_turn_results 方法
-            # 返回 True 表示 "工具已执行完毕，状态已更新，请继续下一轮 LLM 推理"
-            # 返回 False 表示 "任务完成" 或 "检测到死循环/无需工具"，应退出循环
             should_continue = await self._process_turn_results(
                 context, assistant_msg, response_model=None
             )
 
-            # 如果不需要继续，跳出 while 循环，结束流式生成
             if not should_continue:
+                # ✅ 修复: 检查退出原因并通知用户
+                exit_reason = self._get_exit_reason(context, assistant_msg)
+                if exit_reason:
+                    yield f"\n\n[System: {exit_reason}]"
                 break
-            
-            # 如果 should_continue 为 True，while 循环会自动进入下一轮
-            # context.turn 增加，context.messages 已包含工具结果
+
+    def _get_exit_reason(self, context: ExecutionContext, last_message: Message) -> Optional[str]:
+        """
+        ✅ 新增: 确定流式循环退出的原因
+        """
+        # 检查是否因死循环退出
+        if last_message.tool_calls:
+            if context.last_tool_calls_hash is not None:
+                try:
+                    calls_dump = json.dumps(
+                        [
+                            {
+                                "name": tc["function"]["name"],
+                                "args": tc["function"]["arguments"],
+                            }
+                            for tc in last_message.tool_calls
+                        ],
+                        sort_keys=True,
+                    )
+                    current_hash = hash(calls_dump)
+                    if context.last_tool_calls_hash == current_hash:
+                        return "Execution stopped due to detected infinite tool loop."
+                except Exception:
+                    pass
+        
+        # 检查是否因连续错误退出
+        if context.consecutive_tool_error_count >= 3:
+            return "Execution stopped due to repeated tool errors."
+        
+        # 正常结束
+        return None
 
     # ===================== 辅助逻辑 =====================
 

gecko/core/engine/react.py

@@ -81,12 +81,22 @@ async def publish(self, event: BaseEvent, wait: bool = False):
             return
 
         # 2. 获取订阅者
-        handlers = self._subscribers.get(event.type, []) + self._subscribers.get("*", [])
-        if not handlers:
+        type_handlers = self._subscribers.get(event.type, [])
+        wildcard_handlers = self._subscribers.get("*", [])
+        
+        if not type_handlers and not wildcard_handlers:
             return
-
-        # 3. 执行处理（去重）
-        unique_handlers = list(dict.fromkeys(handlers))
+        
+        # 3. 修复: 使用 id() 进行更健壮的去重
+        # dict.fromkeys() 依赖对象的 __hash__，对于方法和 lambda 可能不可靠
+        seen_ids = set()
+        unique_handlers = []
+
+        for h in type_handlers + wildcard_handlers:
+            handler_id = id(h)
+            if handler_id not in seen_ids:
+                seen_ids.add(handler_id)
+                unique_handlers.append(h)
 
         # 创建执行协程
         tasks = [self._execute_handler(h, event) for h in unique_handlers]