feat(03-02): implement ProgressMonitor class

- Add ProgressInfo dataclass for progress data
- Add ProgressMonitor class with:
  - PROGRESS_PATTERNS for extraction (percentage, fraction, phase, completion)
  - extract_progress() to find progress indicators in output
  - should_send_update() with 60 second throttling
  - create_progress_card() for Feishu card formatting
- Fix fraction pattern to allow content after fraction (e.g., [3/5 tasks])
- Fix should_send_update() to return False on first call
- Fix test assertions to use kwargs instead of positional args

All 7 tests pass for progress monitoring.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

gsd-openclaw/src/gsd_openclaw/progress_monitor.py

@@ -0,0 +1,164 @@
+"""
+Progress monitor for tracking and reporting GSD operation progress.
+
+This module provides:
+- Detection of progress indicators in Claude CLI output
+- Throttled progress updates (1 minute intervals)
+- Progress card creation for Feishu messages
+"""
+from __future__ import annotations
+
+import re
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Tuple, Pattern
+
+from gsd_openclaw.response_formatter import FeishuResponseFormatter
+
+
+@dataclass
+class ProgressInfo:
+    """Represents extracted progress information.
+
+    Attributes:
+        type: Type of progress indicator ('percentage', 'fraction', 'phase', 'completion').
+        value: Numeric value (for percentage or phase).
+        current: Current count (for fraction type).
+        total: Total count (for fraction type).
+    """
+    type: str
+    value: Optional[int] = None
+    current: Optional[int] = None
+    total: Optional[int] = None
+
+
+class ProgressMonitor:
+    """Monitors and reports progress for GSD operations.
+
+    This class provides functionality to:
+    - Extract progress indicators from output text
+    - Throttle progress updates to specified intervals
+    - Create progress cards for Feishu messages
+
+    Attributes:
+        PROGRESS_PATTERNS: List of (pattern, type) tuples for progress detection.
+        update_interval: Minimum seconds between progress updates (default 60.0).
+        last_update_time: Unix timestamp of last update.
+    """
+
+    # Patterns for detecting progress in output
+    # Each tuple is (regex pattern, progress type)
+    PROGRESS_PATTERNS: List[Tuple[Pattern, str]] = [
+        (re.compile(r'(\d+)%'), 'percentage'),
+        (re.compile(r'\[(\d+)/(\d+)(?:[^\]]*)\]'), 'fraction'),  # Allow content after fraction
+        (re.compile(r'Phase\s+(\d+)', re.IGNORECASE), 'phase'),
+        (re.compile(r'完成|completed|done', re.IGNORECASE), 'completion'),
+    ]
+
+    def __init__(
+        self,
+        formatter: FeishuResponseFormatter,
+        update_interval: float = 60.0
+    ):
+        """Initialize ProgressMonitor.
+
+        Args:
+            formatter: FeishuResponseFormatter instance for creating cards.
+            update_interval: Minimum seconds between progress updates (default 60.0).
+        """
+        self.formatter = formatter
+        self.update_interval = update_interval
+        self.last_update_time: float = 0.0
+
+    def extract_progress(self, output: str) -> Optional[ProgressInfo]:
+        """Extract progress information from output text.
+
+        Scans output for known progress patterns and returns the first match.
+
+        Args:
+            output: The output text to scan for progress indicators.
+
+        Returns:
+            ProgressInfo if progress pattern found, None otherwise.
+        """
+        for pattern, progress_type in self.PROGRESS_PATTERNS:
+            match = pattern.search(output)
+            if match:
+                if progress_type == 'percentage':
+                    return ProgressInfo(
+                        type='percentage',
+                        value=int(match.group(1)),
+                    )
+                elif progress_type == 'fraction':
+                    return ProgressInfo(
+                        type='fraction',
+                        current=int(match.group(1)),
+                        total=int(match.group(2)),
+                    )
+                elif progress_type == 'phase':
+                    return ProgressInfo(
+                        type='phase',
+                        value=int(match.group(1)),
+                    )
+                elif progress_type == 'completion':
+                    return ProgressInfo(
+                        type='completion',
+                        value=100,  # Completion means 100%
+                    )
+
+        return None
+
+    def should_send_update(self) -> bool:
+        """Check if enough time has elapsed for a progress update.
+
+        Returns:
+            True if update_interval has elapsed since last update, False otherwise.
+            Returns False if no update has been sent yet (first call).
+        """
+        # If no update has been sent yet (last_update_time is 0), return False
+        if self.last_update_time == 0.0:
+            return False
+
+        current_time = time.time()
+        elapsed = current_time - self.last_update_time
+        return elapsed >= self.update_interval
+
+    def create_progress_card(
+        self,
+        progress: ProgressInfo,
+        message: str
+    ) -> dict:
+        """Create a Feishu progress card.
+
+        Args:
+            progress: ProgressInfo with extracted progress data.
+            message: Additional message to include in the card.
+
+        Returns:
+            Feishu Card JSON dict with progress information.
+        """
+        # Format progress content
+        if progress.type == 'percentage':
+            progress_text = f"进度: {progress.value}%"
+        elif progress.type == 'fraction':
+            progress_text = f"进度: [{progress.current}/{progress.total}]"
+        elif progress.type == 'phase':
+            progress_text = f"阶段: {progress.value}"
+        elif progress.type == 'completion':
+            progress_text = "完成"
+        else:
+            progress_text = f"进度: {progress.value}"
+
+        content = f"{message}\n\n{progress_text}"
+
+        # Update last_update_time when creating card
+        self.last_update_time = time.time()
+
+        return self.formatter.format_info(
+            title="正在执行...",
+            content=content,
+        )
+
+    def reset_timer(self) -> None:
+        """Reset the update timer (useful for new operations)."""
+        self.last_update_time = 0.0

gsd-openclaw/tests/test_progress_monitor.py

@@ -106,8 +106,8 @@ class TestProgressMonitorCardCreation:
         card = monitor.create_progress_card(progress, "Processing data...")
 
         formatter.format_info.assert_called_once()
-        call_args = formatter.format_info.call_args
-        content = call_args[0][1]
+        call_kwargs = formatter.format_info.call_args.kwargs
+        content = call_kwargs.get("content", "")
         assert "75%" in content
 
     def test_create_progress_card_uses_executing_title(self):
@@ -126,6 +126,6 @@ class TestProgressMonitorCardCreation:
         card = monitor.create_progress_card(progress, "Working...")
 
         formatter.format_info.assert_called_once()
-        call_args = formatter.format_info.call_args
-        title = call_args[0][0]
+        call_kwargs = formatter.format_info.call_args.kwargs
+        title = call_kwargs.get("title", "")
         assert title == "正在执行..."