refactor

Change-Id: I26c29a20e94e16e58e2f7d9dea026335cb1e6830

Author: yinsu.zs

src/code/agent/services/gateway/gateways/history_gateway.py

@@ -153,9 +153,6 @@ def _get_all_persisted_history(self, limit=None):
                                 # 降级方案：使用文件修改时间
                                 if sort_number is None:
                                     sort_number = file_mtime
-                                    log("DEBUG", f"Using file mtime for {prompt_id[:12]}... (no prompt number)")
-                                else:
-                                    log("DEBUG", f"Using prompt number {sort_number} for {prompt_id[:12]}...")
 
                                 timestamp = sort_number
 

src/code/agent/services/gateway/status/poller.py

@@ -70,11 +70,19 @@ def _poll_loop(self):
 
         while self.is_running and not task_completed:
             try:
+                # 时间监控：轮询周期开始
+                poll_start = time.time()
+                
                 # 使用增量文件读取，仅获取新增状态
+                read_start = time.time()
                 new_statuses = self.storage_service.get_status_incremental(self.task_id)
+                read_cost = (time.time() - read_start) * 1000
 
                 if new_statuses:
+                    log("INFO", f"[Perf][{self.task_id}] File read cost: {read_cost:.1f}ms, got {len(new_statuses)} statuses")
+                    
                     # 调用回调函数处理新状态
+                    callback_start = time.time()
                     for status in new_statuses:
                         if self.on_status_update:
                             self.on_status_update(self.task_id, status)
@@ -87,6 +95,10 @@ def _poll_loop(self):
                             task_completed = True
                             break
 
+                    callback_cost = (time.time() - callback_start) * 1000
+                    poll_total = (time.time() - poll_start) * 1000
+                    log("INFO", f"[Perf][{self.task_id}] Poll cycle: read={read_cost:.1f}ms, callback={callback_cost:.1f}ms, total={poll_total:.1f}ms")
+                    
             except Exception as e:
                 log("ERROR", f"Error polling task {self.task_id}: {e}")
                 from traceback import print_exception

src/code/agent/services/gateway/status/storage.py

@@ -87,15 +87,33 @@ def get_status_incremental(self, task_id: str):
             return []
 
         try:
-            # 使用 os.open + os.fstat 绕过 Python 的 stat 缓存
-            # os.fstat(fd) 直接从文件描述符获取
+            import time as perf_time
+            perf_start = perf_time.time()
+            
+            # 强制刷新 NFS 目录缓存：先 listdir 触发目录元数据更新
+            listdir_start = perf_time.time()
+            try:
+                dir_path = os.path.dirname(file_path)
+                os.listdir(dir_path)  # 强制刷新目录缓存
+            except Exception as e:
+                log("DEBUG", f"listdir failed for {task_id}: {e}")
+            listdir_cost = (perf_time.time() - listdir_start) * 1000
+            log("DEBUG", f"[Perf][{task_id}] listdir cost: {listdir_cost:.1f}ms")
+            
+            # 关键：使用 os.open + os.fstat 绕过 Python 的 stat 缓存
+            # os.stat() 可能读取缓存的元数据，而 os.fstat(fd) 直接从文件描述符获取
+            fstat_start = perf_time.time()
             try:
                 fd = os.open(file_path, os.O_RDONLY)
                 file_stat = os.fstat(fd)
                 os.close(fd)
             except Exception as e:
                 log("DEBUG", f"Fallback to regular stat for {task_id}: {e}")
                 file_stat = os.stat(file_path)
+            
+            stat_time = (perf_time.time() - fstat_start) * 1000
+            log("DEBUG", f"[Perf][{task_id}] fstat cost: {stat_time:.1f}ms")
+            
             current_size = file_stat.st_size
             current_modified = file_stat.st_mtime
 
@@ -120,6 +138,8 @@ def get_status_incremental(self, task_id: str):
 
                 # 读取新增内容 - 使用无缓冲模式
                 new_statuses = []
+                read_start = perf_time.time()
+                
                 # 使用 buffering=0 和 binary 模式读取，绕过 Python 的缓冲层
                 with open(file_path, 'rb', buffering=0) as f:
                     # 定位到上次读取位置
@@ -128,30 +148,40 @@ def get_status_incremental(self, task_id: str):
                     # 读取新增内容（二进制模式）
                     new_content_bytes = f.read()
                     new_position = f.tell()
-                    
-                    # 解码为文本
-                    new_content = new_content_bytes.decode('utf-8', errors='ignore')
-                    
-                    if new_content.strip():
-                        # 按行解析JSON
-                        for line in new_content.strip().split('\n'):
-                            if line.strip():
-                                try:
-                                    status_data = json.loads(line)
-                                    new_statuses.append(status_data)
-                                except json.JSONDecodeError as e:
-                                    log("WARNING", f"Failed to parse status line for task {task_id}: {line[:100]}... Error: {e}")
-                                    continue
+                
+                read_cost = (perf_time.time() - read_start) * 1000
+                log("DEBUG", f"[Perf][{task_id}] file read cost: {read_cost:.1f}ms, bytes={len(new_content_bytes)}")
+                
+                parse_start = perf_time.time()
+                # 解码为文本
+                new_content = new_content_bytes.decode('utf-8', errors='ignore')
+                
+                if new_content.strip():
+                    # 按行解析JSON
+                    for line in new_content.strip().split('\n'):
+                        if line.strip():
+                            try:
+                                status_data = json.loads(line)
+                                new_statuses.append(status_data)
+                            except json.JSONDecodeError as e:
+                                log("WARNING", f"Failed to parse status line for task {task_id}: {line[:100]}... Error: {e}")
+                                continue
+                
+                parse_cost = (perf_time.time() - parse_start) * 1000
+                log("DEBUG", f"[Perf][{task_id}] parse cost: {parse_cost:.1f}ms")
 
                 # 更新缓存
                 self._file_read_cache[task_id] = {
                     "position": new_position,
                     "last_modified": current_modified
                 }
 
+                total_cost = (perf_time.time() - perf_start) * 1000
+                
                 if new_statuses:
-                    log("DEBUG", f"Read {len(new_statuses)} new status updates for task {task_id} "
-                          f"(position: {last_position} -> {new_position})")
+                    log("INFO", f"[Perf][{task_id}] get_status_incremental total: {total_cost:.1f}ms "
+                          f"(listdir={listdir_cost:.1f}ms, fstat={stat_time:.1f}ms, read={read_cost:.1f}ms, parse={parse_cost:.1f}ms), "
+                          f"got {len(new_statuses)} statuses, position: {last_position} -> {new_position}")
 
                 return new_statuses