v0.0.3-sse关闭逻辑优化

- 测试unit_review 事件推送是否消失提交

core/base/progress_manager.py

@@ -55,6 +55,8 @@ class SSECallbackManager:
         if callback_task_id in self._callbacks:
             del self._callbacks[callback_task_id]
             logger.info(f"强制关闭SSE连接: {callback_task_id}")
+        else:
+            logger.warning(f"SSE连接已不存在，无需关闭: {callback_task_id}")
 
 sse_callback_manager = SSECallbackManager()
 
@@ -317,11 +319,13 @@ class ProgressManager:
             )
 
             logger.info(f"取消注册任务: {callback_task_id}")
-            # 强制关闭SSE连接
+            # 先取消注册，再强制关闭，确保彻底清理
+            sse_callback_manager.unregister_callback(callback_task_id)
+
+            # 强制关闭SSE连接（防止残留）
             sse_callback_manager.force_close_sse(callback_task_id)
 
-            # 取消SSE回调注册，避免重复推送
-            sse_callback_manager.unregister_callback(callback_task_id)
+            logger.info(f"SSE连接已彻底关闭: {callback_task_id}")
 
 
 

core/construction_review/workflows/ai_review_workflow.py

@@ -243,7 +243,7 @@ class AIReviewWorkflow:
     
     async def _ai_review_node(self, state: AIReviewState) -> AIReviewState:
         """
-        AI审查节点 
+        AI审查节点
 
         Args:
             state: AI审查状态
@@ -253,15 +253,24 @@ class AIReviewWorkflow:
         """
         try:
             state["current_stage"] = "ai_review"
+            logger.info(f"AI审查节点开始执行，任务ID: {state['callback_task_id']}")
 
             # 1. 准备审查单元数据
             review_chunks, total_units, total_all_units = await self.core_fun._prepare_review_units(state)
+            logger.info(f"准备审查单元完成，总单元数: {total_units}, 实际审查: {len(review_chunks)}")
+
+            if not review_chunks:
+                logger.warning(f"没有可审查的单元，任务ID: {state['callback_task_id']}")
+                state["error_message"] = "没有可审查的单元"
+                return state
 
             # 2. 发送开始审查进度
             await self.core_fun._send_start_review_progress(state, total_units)
 
             # 3. 执行并发审查
+            logger.info(f"开始执行并发审查，任务ID: {state['callback_task_id']}")
             successful_results = await self.core_fun._execute_concurrent_reviews(review_chunks, total_units, state)
+            logger.info(f"并发审查完成，成功结果: {len(successful_results)}, 任务ID: {state['callback_task_id']}")
 
             # 4. 汇总结果
             summary = self.inter_tool._aggregate_results(successful_results)
@@ -282,10 +291,11 @@ class AIReviewWorkflow:
                 content=f"AI审查完成，共处理{total_units}个单元，成功{len(successful_results)}个"
             ))
 
+            logger.info(f"AI审查节点执行成功，任务ID: {state['callback_task_id']}")
             return state
 
         except Exception as e:
-            logger.error(f"AI审查失败: {str(e)}")
+            logger.error(f"AI审查节点执行失败，任务ID: {state['callback_task_id']}, 错误: {str(e)}", exc_info=True)
             state["error_message"] = str(e)
             state["messages"].append(AIMessage(content=f"AI审查失败: {str(e)}"))
             return state