feat(telemetry): elapsed-ratio ETA with per-step history blending and Feishu heartbeat

Author: YoTro

AWS/src/core/telemetry/tracker.py

@@ -1,62 +1,176 @@
-from __future__ import annotations
-import time
-import logging
-from typing import Optional, List
-
-logger = logging.getLogger(__name__)
-
-class TimeEstimator:
-    """
-    Provides static ETA estimations based on heuristics.
-    In a production system, this could query a database of historical execution times.
-    """
-    
-    WORKFLOW_AVERAGES = {
-        "amazon_bsr": 30.0,  # Baseline 30 seconds
-        # Add others as needed
-    }
-    
-    AGENT_BASELINE_PER_ITERATION = 6.0  # Approx 12s per ReAct loop (LLM + Tool)
-    
-    @classmethod
-    def estimate_workflow(cls, workflow_name: str, params: dict = None) -> str:
-        base_time = cls.WORKFLOW_AVERAGES.get(workflow_name, 30.0)
-        # E.g., if we know pagination params, we could multiply base_time
-        return f"~{int(base_time)}秒"
-
-    @classmethod
-    def estimate_agent(cls, max_iterations: int = 5) -> str:
-        # Agent is highly variable, give a conservative range
-        min_time = int(cls.AGENT_BASELINE_PER_ITERATION * 1)
-        max_time = int(cls.AGENT_BASELINE_PER_ITERATION * (max_iterations * 0.6)) # Assume it rarely hits max
-        return f"{min_time}~{max_time}秒"
-
-
-class TelemetryTracker:
-    """
-    Tracks dynamic progress and calculates moving averages for remaining ETA.
-    """
-    def __init__(self, total_steps: int):
-        self.total_steps = total_steps
-        self.current_step = 0
-        self.start_time = time.monotonic()
-        self.step_times: List[float] = []
-        self.last_step_time = self.start_time
-
-    def record_step(self):
-        """Record the completion of a step and track its duration."""
-        now = time.monotonic()
-        duration = now - self.last_step_time
-        self.step_times.append(duration)
-        self.last_step_time = now
-        self.current_step += 1
-
-    def get_dynamic_eta(self) -> Optional[int]:
-        """Returns the estimated remaining time in seconds."""
-        if self.current_step == 0 or self.current_step >= self.total_steps:
-            return None
-        
-        # Simple Average for remaining steps
-        avg_time_per_step = sum(self.step_times) / len(self.step_times)
-        remaining_steps = self.total_steps - self.current_step
-        return int(avg_time_per_step * remaining_steps)
+from __future__ import annotations
+import json
+import logging
+import os
+import time
+from typing import Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Per-step history stored here; keyed by step_name, value = list of durations (s)
+_HISTORY_PATH = os.path.join(os.path.dirname(__file__), "step_history.json")
+_HISTORY_MAX_SAMPLES = 20   # rolling window per step
+
+
+def _load_history() -> Dict[str, List[float]]:
+    try:
+        with open(_HISTORY_PATH, encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def _save_history(history: Dict[str, List[float]]) -> None:
+    try:
+        with open(_HISTORY_PATH, "w", encoding="utf-8") as f:
+            json.dump(history, f)
+    except Exception as e:
+        logger.debug(f"Could not save step history: {e}")
+
+
+class TimeEstimator:
+    """
+    Static ETA estimations based on heuristics (kept for backward compat).
+    """
+
+    WORKFLOW_AVERAGES = {
+        "amazon_bsr": 30.0,
+    }
+    AGENT_BASELINE_PER_ITERATION = 6.0
+
+    @classmethod
+    def estimate_workflow(cls, workflow_name: str, params: dict = None) -> str:
+        base_time = cls.WORKFLOW_AVERAGES.get(workflow_name, 30.0)
+        return f"~{int(base_time)}秒"
+
+    @classmethod
+    def estimate_agent(cls, max_iterations: int = 5) -> str:
+        min_time = int(cls.AGENT_BASELINE_PER_ITERATION * 1)
+        max_time = int(cls.AGENT_BASELINE_PER_ITERATION * (max_iterations * 0.6))
+        return f"{min_time}~{max_time}秒"
+
+
+class TelemetryTracker:
+    """
+    Tracks dynamic progress and calculates ETA using two complementary methods:
+
+    1. Elapsed-ratio (always available, no history needed):
+          eta = elapsed_total × remaining_steps / completed_steps
+       Same algorithm used by curl, wget, and GitHub Actions progress bars.
+       Robust because it doesn't assume equal step durations — it uses actual
+       wall time.
+
+    2. Per-step historical baseline (learned from past runs):
+          eta = Σ historical_avg(step_i) for remaining steps
+       Stored in step_history.json.  When N ≥ 3 samples exist for a step,
+       the estimate is blended: 40 % elapsed-ratio + 60 % historical.
+
+    Confidence tiers shown in the UI:
+       🔴  < 2 completed steps     — wild guess, elapsed-ratio only
+       🟡  2–4 completed steps     — improving estimate
+       🟢  history N ≥ 3 per step  — data-backed estimate
+    """
+
+    def __init__(self, total_steps: int, workflow_name: str = ""):
+        self.total_steps = total_steps
+        self.workflow_name = workflow_name
+        self.current_step = 0
+        self.start_time = time.monotonic()
+        self.step_times: List[float] = []
+        self.step_names: List[str] = []
+        self.last_step_time = self.start_time
+        self._history: Dict[str, List[float]] = _load_history()
+
+    def record_step(self, step_name: str = "") -> None:
+        now = time.monotonic()
+        duration = now - self.last_step_time
+        self.step_times.append(duration)
+        self.step_names.append(step_name)
+        self.last_step_time = now
+        self.current_step += 1
+
+        # Persist to rolling history
+        if step_name:
+            key = f"{self.workflow_name}:{step_name}" if self.workflow_name else step_name
+            samples = self._history.setdefault(key, [])
+            samples.append(duration)
+            if len(samples) > _HISTORY_MAX_SAMPLES:
+                samples.pop(0)
+            _save_history(self._history)
+
+    # ── ETA methods ──────────────────────────────────────────────────────
+
+    def _elapsed_ratio_eta(self) -> Optional[float]:
+        """eta = elapsed × remaining / completed  (download-bar method)."""
+        if self.current_step == 0:
+            return None
+        elapsed = time.monotonic() - self.start_time
+        remaining = self.total_steps - self.current_step
+        return elapsed * remaining / self.current_step
+
+    def _historical_eta(self, remaining_step_names: List[str]) -> Optional[float]:
+        """Sum of historical averages for the remaining steps."""
+        if not remaining_step_names:
+            return None
+        total = 0.0
+        covered = 0
+        for name in remaining_step_names:
+            key = f"{self.workflow_name}:{name}" if self.workflow_name else name
+            samples = self._history.get(key, [])
+            if len(samples) >= 3:
+                total += sum(samples) / len(samples)
+                covered += 1
+        if covered == 0:
+            return None
+        # Scale up to cover steps with no history (proportional)
+        total *= len(remaining_step_names) / covered
+        return total
+
+    def _min_history_samples(self, remaining_step_names: List[str]) -> int:
+        """Minimum number of samples across remaining steps (confidence proxy)."""
+        mins = []
+        for name in remaining_step_names:
+            key = f"{self.workflow_name}:{name}" if self.workflow_name else name
+            mins.append(len(self._history.get(key, [])))
+        return min(mins) if mins else 0
+
+    def get_dynamic_eta(
+        self,
+        remaining_step_names: Optional[List[str]] = None,
+    ) -> Optional[str]:
+        """
+        Return a human-readable ETA string, or None when estimate is unavailable.
+
+        Parameters
+        ----------
+        remaining_step_names:
+            Optional list of future step names (enables historical blending).
+        """
+        if self.current_step == 0 or self.current_step >= self.total_steps:
+            return None
+
+        er_eta = self._elapsed_ratio_eta()
+        hist_eta = self._historical_eta(remaining_step_names or [])
+        min_samples = self._min_history_samples(remaining_step_names or [])
+
+        # Blend: weight historical more heavily when data is solid
+        if hist_eta is not None and min_samples >= 3:
+            eta = 0.4 * (er_eta or hist_eta) + 0.6 * hist_eta
+            confidence = "🟢"
+        elif er_eta is not None and self.current_step >= 2:
+            eta = er_eta
+            confidence = "🟡"
+        elif er_eta is not None:
+            eta = er_eta
+            confidence = "🔴"
+        else:
+            return None
+
+        secs = max(1, int(eta))
+        if secs >= 60:
+            mins, s = divmod(secs, 60)
+            eta_str = f"{mins}分{s:02d}秒" if s else f"{mins}分钟"
+        else:
+            eta_str = f"{secs}秒"
+
+        return f"{confidence} 预计剩余 {eta_str}"

AWS/src/jobs/callbacks/feishu.py

@@ -12,8 +12,9 @@
 import asyncio
 import functools
 import logging
+import time
 from datetime import datetime
-from typing import Set
+from typing import Optional, Set
 
 
 def _to_thread(func, *args, **kwargs):
@@ -32,6 +33,9 @@ def _to_thread(func, *args, **kwargs):
 _CIRCUIT_OPEN_THRESHOLD = 3     # consecutive failures before opening circuit
 _CIRCUIT_COOLDOWN_STEPS = 5     # steps to skip before retrying
 
+_HEARTBEAT_INTERVAL = 10        # seconds between heartbeat card updates
+_SPINNER_FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
+
 
 class FeishuCallback(JobCallback):
     """
@@ -69,6 +73,11 @@ def __init__(
         self._consecutive_failures = 0
         self._cooldown_remaining = 0
 
+        # Heartbeat state
+        self._heartbeat_task: Optional[asyncio.Task] = None
+        self._heartbeat_base_text: str = ""
+        self._heartbeat_start: float = 0.0
+
     @property
     def feishu(self):
         if self._feishu is None:
@@ -100,6 +109,32 @@ def _is_circuit_open(self) -> bool:
             return True
         return False
 
+    # ── Heartbeat ────────────────────────────────────────────────────────
+
+    def _cancel_heartbeat(self) -> None:
+        if self._heartbeat_task and not self._heartbeat_task.done():
+            self._heartbeat_task.cancel()
+        self._heartbeat_task = None
+
+    async def _heartbeat_loop(self) -> None:
+        """Periodically update the progress card while a step is running."""
+        frame_idx = 0
+        try:
+            while True:
+                await asyncio.sleep(_HEARTBEAT_INTERVAL)
+                if self._is_circuit_open():
+                    continue
+                elapsed = int(time.monotonic() - self._heartbeat_start)
+                spinner = _SPINNER_FRAMES[frame_idx % len(_SPINNER_FRAMES)]
+                frame_idx += 1
+                heartbeat_text = (
+                    f"{self._heartbeat_base_text}\n"
+                    f"{spinner} 正在执行中… 已等待 {elapsed}s"
+                )
+                asyncio.create_task(self._send_progress(heartbeat_text))
+        except asyncio.CancelledError:
+            pass
+
     # ── Progress (fire-and-forget, non-blocking) ─────────────────────────
 
     async def on_progress(
@@ -108,7 +143,7 @@ async def on_progress(
         if self._tracker.total_steps != total_steps:
             self._tracker.total_steps = total_steps
 
-        self._tracker.record_step()
+        self._tracker.record_step(step_name)
 
         if self._is_circuit_open():
             return
@@ -122,7 +157,13 @@ async def on_progress(
 
         eta = self._tracker.get_dynamic_eta()
         if eta is not None:
-            text += f"\n⏳ 动态预计剩余: {eta}秒"
+            text += f"\n⏳ {eta}"
+
+        # Cancel previous heartbeat and start a fresh one for this step
+        self._cancel_heartbeat()
+        self._heartbeat_base_text = text
+        self._heartbeat_start = time.monotonic()
+        self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
 
         # Fire-and-forget: do not block the workflow on Feishu I/O
         asyncio.create_task(self._send_progress(text))
@@ -320,6 +361,7 @@ async def _send_interaction_card(self, signal: dict) -> None:
             logger.info(f"Feishu send_raw_card response: {send_res}")
 
     async def on_complete(self, result) -> None:
+        self._cancel_heartbeat()
         try:
             import os
             items = result.final_items if hasattr(result, "final_items") else []
@@ -476,6 +518,7 @@ async def on_complete(self, result) -> None:
                 pass
 
     async def on_error(self, error: Exception, job_id: str = None) -> None:
+        self._cancel_heartbeat()
         lines = [f"❌ Workflow failed: {error}"]
         if job_id:
             lines.append(f"Job ID: `{job_id}`")