feat(03-02): implement ApprovalHandler class

- Add ApprovalState dataclass for approval workflow state
- Add ApprovalHandler class with:
  - APPROVAL_PATTERNS for detection (AskUserQuestion, ExitPlanMode, 需要确认, Please confirm)
  - detect_approval_request() to find approval requests in output
  - create_approval_card() with confirm/reject buttons using CardButton
  - handle_callback() to process approval actions
  - In-memory pending_approvals cache
- Fix test assertions to use kwargs instead of positional args

All 8 tests pass for approval handling.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-11 16:03:54 +08:00
parent efd4892f2f
commit ddd0477498
2 changed files with 249 additions and 6 deletions
@@ -0,0 +1,243 @@
"""
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)
+6 -6
View File
@@ -60,8 +60,8 @@ class TestApprovalHandlerCardCreation:
card = handler.create_approval_card(context, "session-123", "test-project")
formatter.format_info.assert_called_once()
call_args = formatter.format_info.call_args
assert call_args[0][0] == "需要确认"
call_kwargs = formatter.format_info.call_args.kwargs
assert call_kwargs.get("title") == "需要确认"
def test_create_approval_card_has_confirm_button(self):
"""create_approval_card has confirm button with action='approve'."""
@@ -79,8 +79,8 @@ class TestApprovalHandlerCardCreation:
card = handler.create_approval_card(context, "session-123", "test-project")
# Check that format_info was called with buttons
call_args = formatter.format_info.call_args
buttons = call_args[1].get('buttons') or (call_args[0][2] if len(call_args[0]) > 2 else None)
call_kwargs = formatter.format_info.call_args.kwargs
buttons = call_kwargs.get('buttons')
assert buttons is not None
# Find confirm button
confirm_btn = next((b for b in buttons if b.value.get("action") == "approve"), None)
@@ -103,8 +103,8 @@ class TestApprovalHandlerCardCreation:
card = handler.create_approval_card(context, "session-123", "test-project")
# Check that format_info was called with buttons
call_args = formatter.format_info.call_args
buttons = call_args[1].get('buttons') or (call_args[0][2] if len(call_args[0]) > 2 else None)
call_kwargs = formatter.format_info.call_args.kwargs
buttons = call_kwargs.get('buttons')
assert buttons is not None
# Find reject button
reject_btn = next((b for b in buttons if b.value.get("action") == "reject"), None)