kb-skills/gsd-openclaw/src/gsd_openclaw/approval_handler.py

"""
Approval workflow handler for interactive GSD operations.

This module provides:
- Detection of approval requests in Claude CLI output
- Creation of approval cards with confirm/reject buttons
- Callback handling for approval actions
- Persistence of approval state for crash recovery
"""
from __future__ import annotations

import re
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Pattern

from gsd_openclaw.response_formatter import FeishuResponseFormatter, CardButton


@dataclass
class ApprovalState:
    """Represents the state of a pending approval.

    Attributes:
        approval_id: Unique identifier for this approval.
        session_id: ID of the GSD session this approval belongs to.
        project_name: Name of the project.
        context: Additional context about the approval request.
        created_at: Unix timestamp when the approval was created.
    """
    approval_id: str
    session_id: str
    project_name: str
    context: Dict[str, Any]
    created_at: float = field(default_factory=time.time)

    def to_dict(self) -> Dict[str, Any]:
        """Convert ApprovalState to a dictionary for persistence."""
        return {
            "approval_id": self.approval_id,
            "session_id": self.session_id,
            "project_name": self.project_name,
            "context": self.context,
            "created_at": self.created_at,
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> ApprovalState:
        """Create ApprovalState from a dictionary."""
        return cls(
            approval_id=data["approval_id"],
            session_id=data["session_id"],
            project_name=data["project_name"],
            context=data.get("context", {}),
            created_at=data.get("created_at", time.time()),
        )


class ApprovalHandler:
    """Handles approval workflow for interactive GSD operations.

    This class provides functionality to:
    - Detect approval requests in Claude CLI output
    - Create Feishu cards with confirm/reject buttons
    - Handle button callbacks
    - Persist approval state for crash recovery

    Attributes:
        APPROVAL_PATTERNS: List of patterns that indicate an approval request.
        pending_approvals: In-memory cache of pending approvals.
    """

    # Patterns that indicate Claude is asking for user confirmation
    APPROVAL_PATTERNS: List[str] = [
        'AskUserQuestion',
        'ExitPlanMode',
        '需要确认',
        'Please confirm',
    ]

    def __init__(self, formatter: FeishuResponseFormatter):
        """Initialize ApprovalHandler.

        Args:
            formatter: FeishuResponseFormatter instance for creating cards.
        """
        self.formatter = formatter
        self.pending_approvals: Dict[str, ApprovalState] = {}

    def detect_approval_request(self, output: str) -> Optional[Dict[str, Any]]:
        """Detect if output contains an approval request.

        Scans the output for known approval patterns and returns context
        about the approval request if found.

        Args:
            output: The output text to scan for approval patterns.

        Returns:
            Dictionary with approval context if found, None otherwise.
            The dictionary includes:
            - type: The matched pattern type
            - matched_text: The text that was matched
        """
        for pattern in self.APPROVAL_PATTERNS:
            if pattern in output:
                # Extract some context around the match
                idx = output.find(pattern)
                start = max(0, idx - 50)
                end = min(len(output), idx + len(pattern) + 50)
                matched_text = output[start:end].strip()

                return {
                    "type": pattern,
                    "matched_text": matched_text,
                }

        return None

    def create_approval_card(
        self,
        context: Dict[str, Any],
        session_id: str,
        project_name: str
    ) -> dict:
        """Create a Feishu approval card with confirm/reject buttons.

        Args:
            context: Context dictionary from detect_approval_request.
            session_id: ID of the GSD session.
            project_name: Name of the project.

        Returns:
            Feishu Card JSON dict with confirm/reject buttons.
        """
        # Generate unique approval ID
        approval_id = str(uuid.uuid4())

        # Create approval state and store it
        approval_state = ApprovalState(
            approval_id=approval_id,
            session_id=session_id,
            project_name=project_name,
            context=context,
        )
        self.pending_approvals[approval_id] = approval_state

        # Extract question text for display
        question_text = context.get("matched_text", "是否继续?")

        # Create buttons
        confirm_button = CardButton(
            text="确认",
            value={
                "action": "approve",
                "approval_id": approval_id,
                "project": project_name,
            },
            button_type="primary",
        )
        reject_button = CardButton(
            text="取消",
            value={
                "action": "reject",
                "approval_id": approval_id,
                "project": project_name,
            },
            button_type="danger",
        )

        # Format card using formatter
        card = self.formatter.format_info(
            title="需要确认",
            content=question_text,
            buttons=[confirm_button, reject_button],
        )

        return card

    def handle_callback(self, callback_data: Dict[str, Any]) -> str:
        """Handle a button callback from the approval card.

        Args:
            callback_data: Dictionary containing:
                - action: "approve" or "reject"
                - approval_id: The ID of the approval
                - project: The project name

        Returns:
            "approved" if action was approve,
            "rejected" if action was reject,
            "error: ..." if approval not found or invalid.
        """
        action = callback_data.get("action", "")
        approval_id = callback_data.get("approval_id", "")

        # Check if approval exists
        if approval_id not in self.pending_approvals:
            return "error: approval not found or expired"

        # Remove the approval from pending (one-time use)
        approval = self.pending_approvals.pop(approval_id)

        if action == "approve":
            return "approved"
        elif action == "reject":
            return "rejected"
        else:
            return f"error: unknown action '{action}'"

    def persist_approval(self, state_manager, approval: ApprovalState) -> None:
        """Persist approval state to StateManager for crash recovery.

        Args:
            state_manager: StateManager instance for persistence.
            approval: The ApprovalState to persist.
        """
        # Load current state
        current_state = state_manager.load()
        if current_state is None:
            return  # No session to persist to

        # Store approval in state (would need SessionState extension)
        # For now, we keep approvals in memory only as per plan
        pass

    def load_approval(
        self,
        state_manager,
        approval_id: str
    ) -> Optional[ApprovalState]:
        """Load approval state from StateManager.

        Args:
            state_manager: StateManager instance for persistence.
            approval_id: The approval ID to load.

        Returns:
            ApprovalState if found, None otherwise.
        """
        # For now, return from in-memory cache
        return self.pending_approvals.get(approval_id)