openclaw/skills/gsd-openclaw/.planning/phases/03-orchestration/03-RESEARCH.md

Phase 3: Orchestration - Research

Researched: 2026-03-11 Domain: Command routing, output capture, approval workflow, progress monitoring Confidence: HIGH

Summary

Phase 3 implements the orchestration layer that connects user chat messages to GSD slash commands via tmux sessions. The orchestrator routes commands, captures tmux output, handles approval workflows through Feishu Card button callbacks, and provides progress updates during long-running operations.

Primary recommendation: Build orchestrator.py as the central command router that integrates TmuxSessionManager (Phase 1), FeishuResponseFormatter (Phase 2), and StateManager (Phase 1) to provide bidirectional chat-to-GSD bridging with approval workflow support.

<user_constraints>

User Constraints (from CONTEXT.md)

Implementation Decisions

Command Routing Strategy:

• Command parsing: Direct parsing (regex or string splitting), validated by orchestrator.py
• Session requirement: All commands require tmux session (unified flow, simplified implementation)
• Execution timeout: 60 seconds (most GSD commands complete in reasonable time)
• No session handling: Auto-create session and execute command (reference Phase 2 CONTEXT.md design)

Approval Workflow Integration

• Approval detection: Parse Claude tool call markers (e.g., AskUserQuestion, ExitPlanMode)
• Callback handling: Use Feishu Card button callbacks (value field stores context)
• Approval timeout: No timeout limit (wait for user response or session end)

Progress Update Mechanism

• Progress detection: Output pattern matching (monitor percentage, phase names, etc.)
• Update frequency: Send progress update every 1 minute
• Completion detection: tmux prompt appears (Claude CLI output complete, session returns to prompt)

Error Handling Strategy

• Error detection: Keyword matching (monitor output for error, failed, exception, etc.)
• Error presentation: Display raw error directly (use Phase 2's format_error() method)
• Session loss handling: Auto-recover session and retry command (CLI already supports session recovery)

Claude's Discretion

• Specific regex patterns for progress matching
• Complete list of error keywords
• Button text and styling for approval buttons
• Progress update card format

Deferred Ideas (OUT OF SCOPE)

• WeChat/Telegram/Discord channel support (Phase 2 deferred, this phase still Feishu only)
• Multi-language support (Chinese/English switching)
• Command history and search
• Batch command execution

</user_constraints>

<phase_requirements>

Phase Requirements

ID	Description	Research Support
BRDG-01	User can send any GSD slash command through chat	orchestrator.py command routing + TmuxSessionManager.send_keys()
BRDG-02	Command output captured and returned through chat	libtmux Pane.capture_pane() + FeishuResponseFormatter
BRDG-03	Clear error messages when commands fail	Error keyword detection + format_error() from Phase 2
BRDG-04	User can approve/reject GSD workflow gates through chat	Feishu Card button callbacks + approval state tracking
BRDG-05	Periodic progress updates during long operations	Output pattern matching + interval-based progress cards
OCLW-03	orchestrator.py routes commands to appropriate handlers	Command router pattern with handler dispatch
OCLW-05	Skill integrates with OpenClaw security model (dmPolicy, allowlists)	OpenClaw dmPolicy/allowFrom configuration
OCLW-06	Multi-channel support works (all channels)	SKILL.md triggers + channel-agnostic orchestrator interface

</phase_requirements>

Standard Stack

Core

Library	Version	Purpose	Why Standard
re	stdlib	Command parsing, pattern matching	Python built-in, sufficient for GSD command patterns
threading	stdlib	Background output capture	Non-blocking I/O without asyncio complexity
time	stdlib	Timing for progress updates, timeouts	Standard library

Existing (from Phase 1 & 2)

Library	Version	Purpose	When to Use
libtmux	0.46.2	tmux session control	send_keys(), capture_pane() for all CLI I/O
FeishuResponseFormatter	Phase 2	Card formatting	format_success(), format_error(), format_info()
TmuxSessionManager	Phase 1	Session lifecycle	get_session(), create_session(), resume_session()
StateManager	Phase 1	State persistence	Load/save session state, project context
CardButton	Phase 2	Interactive buttons	Approval/reject buttons with callback values

Alternatives Considered

Instead of	Could Use	Tradeoff
threading	asyncio	asyncio adds complexity for minimal benefit; tmux operations are synchronous
re module	argparse subparsers	argparse overkill for simple slash command parsing; regex is more flexible
Manual polling	tmux hooks	Hooks require tmux config changes; polling is simpler and more portable

Installation: No new dependencies - all required libraries are from stdlib or existing Phase 1/2 code.

Architecture Patterns

Recommended Project Structure

src/gsd_openclaw/
├── orchestrator.py        # Main command router (NEW - this phase)
├── command_executor.py    # tmux command execution + output capture (NEW - this phase)
├── approval_handler.py    # Approval workflow state machine (NEW - this phase)
├── progress_monitor.py    # Progress detection + update logic (NEW - this phase)
├── tmux_session.py        # (Phase 1) Session lifecycle
├── state_manager.py       # (Phase 1) State persistence
├── response_formatter.py  # (Phase 2) Feishu Card formatting
└── prompts/               # (Phase 2) Prompt templates

Pattern 1: Command Router

What: Central routing logic that parses user input and dispatches to handlers. When to use: All incoming user messages from OpenClaw. Example:

# Source: Project architecture (ARCHITECTURE.md)
class CommandRouter:
    """Routes /gsd:* commands to appropriate handlers."""

    COMMAND_PATTERNS = {
        r'^/gsd:new-project\s+(.+)$': 'handle_new_project',
        r'^/gsd:use\s+(.+)$': 'handle_use_project',
        r'^/gsd:status$': 'handle_status',
        r'^/gsd:plan$': 'handle_plan',
        r'^/gsd:ship$': 'handle_ship',
        r'^/gsd:help$': 'handle_help',
    }

    def route(self, message: str, context: dict) -> CommandResult:
        for pattern, handler_name in self.COMMAND_PATTERNS.items():
            match = re.match(pattern, message.strip())
            if match:
                handler = getattr(self, handler_name)
                return handler(match.groups(), context)
        return CommandResult(error="Unknown command. Use /gsd:help for available commands.")

Pattern 2: Output Capture with Completion Detection

What: Poll tmux pane output, detect when Claude CLI returns to prompt. When to use: After sending any command to tmux session. Example:

# Source: libtmux API (pane.py capture_pane/send_keys)
import re
import time
from typing import Optional

PROMPT_PATTERNS = [
    r'^\$\s*$',           # Basic shell prompt
    r'^❯\s*$',            # Claude CLI prompt
    r'^➜\s*$',            # Alternative prompt
]

class OutputCapture:
    """Captures tmux output until command completes."""

    def capture_until_complete(
        self,
        pane,
        timeout: float = 60.0,
        poll_interval: float = 0.5
    ) -> tuple[str, bool]:
        """Capture output until prompt reappears or timeout.

        Returns:
            (output, completed) - output text and whether command completed
        """
        start_time = time.time()
        output_lines = []
        last_line_count = 0
        stable_count = 0

        while time.time() - start_time < timeout:
            lines = pane.capture_pane()
            current_output = '\n'.join(lines)

            # Check for prompt (completion)
            if lines and self._is_prompt(lines[-1]):
                return current_output, True

            # Check for stable output (no changes for 3 polls)
            if len(lines) == last_line_count:
                stable_count += 1
                if stable_count >= 3:
                    return current_output, True
            else:
                stable_count = 0
                last_line_count = len(lines)

            time.sleep(poll_interval)

        return current_output, False  # Timeout

    def _is_prompt(self, line: str) -> bool:
        for pattern in PROMPT_PATTERNS:
            if re.match(pattern, line.strip()):
                return True
        return False

Pattern 3: Approval Workflow with Card Callbacks

What: Detect approval requests in output, create Feishu Card with buttons, handle callbacks. When to use: When Claude CLI asks for user approval (AskUserQuestion, ExitPlanMode). Example:

# Source: Feishu Card API (Phase 2 response_formatter.py)
from gsd_openclaw.response_formatter import FeishuResponseFormatter, CardButton

APPROVAL_PATTERNS = [
    r'AskUserQuestion',
    r'ExitPlanMode',
    r'需要您的确认',
    r'Please confirm',
]

class ApprovalHandler:
    """Handles approval workflow state and callbacks."""

    def __init__(self, formatter: FeishuResponseFormatter):
        self.formatter = formatter
        self.pending_approvals = {}  # session_id -> approval_state

    def detect_approval_request(self, output: str) -> Optional[dict]:
        """Check if output contains approval request."""
        for pattern in APPROVAL_PATTERNS:
            if re.search(pattern, output, re.IGNORECASE):
                return self._extract_approval_context(output)
        return None

    def create_approval_card(self, context: dict, session_id: str) -> dict:
        """Create Feishu Card with approval buttons."""
        approval_id = f"approval-{session_id}-{time.time_ns()}"

        # Store pending approval
        self.pending_approvals[approval_id] = {
            "session_id": session_id,
            "context": context,
            "created_at": time.time(),
        }

        buttons = [
            CardButton(
                text="确认",
                value={"action": "approve", "approval_id": approval_id},
                button_type="primary"
            ),
            CardButton(
                text="拒绝",
                value={"action": "reject", "approval_id": approval_id},
                button_type="danger"
            ),
        ]

        return self.formatter.format_info(
            title="需要确认",
            content=context.get("question", "请确认此操作"),
            buttons=buttons
        )

    def handle_callback(self, callback_data: dict) -> str:
        """Handle button callback from Feishu."""
        action = callback_data.get("action")
        approval_id = callback_data.get("approval_id")

        if approval_id not in self.pending_approvals:
            return "Approval request expired or not found"

        approval = self.pending_approvals.pop(approval_id)

        if action == "approve":
            # Send approval to tmux session
            return "approved"
        elif action == "reject":
            return "rejected"

        return "Unknown action"

Pattern 4: Progress Monitoring

What: Detect progress indicators in output, send periodic updates. When to use: Long-running operations (planning, shipping phases). Example:

# Source: CONTEXT.md decisions
PROGRESS_PATTERNS = [
    (r'(\d+)%', 'percentage'),
    (r'\[(\d+)/(\d+)\]', 'fraction'),
    (r'Phase\s+(\d+)', 'phase'),
    (r'完成|completed|done', 'completion'),
]

class ProgressMonitor:
    """Monitors output for progress indicators."""

    def __init__(self, formatter: FeishuResponseFormatter, update_interval: float = 60.0):
        self.formatter = formatter
        self.update_interval = update_interval
        self.last_update_time = 0

    def extract_progress(self, output: str) -> Optional[dict]:
        """Extract progress information from output."""
        for pattern, ptype in PROGRESS_PATTERNS:
            match = re.search(pattern, output, re.IGNORECASE)
            if match:
                if ptype == 'percentage':
                    return {"type": "percentage", "value": int(match.group(1))}
                elif ptype == 'fraction':
                    return {"type": "fraction", "current": int(match.group(1)), "total": int(match.group(2))}
                elif ptype == 'phase':
                    return {"type": "phase", "value": int(match.group(1))}
        return None

    def should_send_update(self) -> bool:
        """Check if enough time has passed for update."""
        now = time.time()
        if now - self.last_update_time >= self.update_interval:
            self.last_update_time = now
            return True
        return False

    def create_progress_card(self, progress: dict, message: str) -> dict:
        """Create progress update card."""
        if progress["type"] == "percentage":
            content = f"{message}\n\n进度: {progress['value']}%"
        elif progress["type"] == "fraction":
            content = f"{message}\n\n进度: [{progress['current']}/{progress['total']}]"
        else:
            content = message

        return self.formatter.format_info(
            title="正在执行...",
            content=content
        )

Anti-Patterns to Avoid

• Blocking on output capture: Use threading, not synchronous wait that blocks Gateway
• Hardcoded prompt patterns: Use configurable patterns, different shells/CLIs have different prompts
• Approval state in memory only: Persist to StateManager for crash recovery
• Assuming immediate completion: GSD operations can take minutes; always implement timeout + progress

Don't Hand-Roll

Problem	Don't Build	Use Instead	Why
Command parsing	Complex parser	re module with patterns	GSD commands are simple slash patterns
Output capture	Custom tmux integration	libtmux Pane.capture_pane()	Handles edge cases, escape sequences
Card formatting	Raw JSON construction	FeishuResponseFormatter	Already implemented in Phase 2
State persistence	Custom database	StateManager (Phase 1)	Already implemented with backup rotation
Session management	Direct subprocess calls	TmuxSessionManager (Phase 1)	Already handles lifecycle, recovery

Key insight: This phase is integration work - the building blocks exist from Phase 1 and 2. Focus on orchestration logic, not reimplementing infrastructure.

Common Pitfalls

Pitfall 1: Race Condition in Output Capture

What goes wrong: capture_pane() called before command output appears Why it happens: tmux output is not instantaneous; send_keys() returns before output renders How to avoid: Add small delay (0.1-0.2s) after send_keys(), then poll with stable_output detection Warning signs: Missing output, incomplete captures, "empty" responses

Pitfall 2: Approval State Loss

What goes wrong: User clicks approval button but session state was lost Why it happens: Approval state kept only in memory; server restart loses pending approvals How to avoid: Persist approval state to StateManager immediately when creating approval card Warning signs: Button clicks return "not found" errors after server restart

Pitfall 3: Incorrect Completion Detection

What goes wrong: Command marked complete but Claude still processing Why it happens: Output contains prompt-like text that matches pattern (e.g., $ in code) How to avoid: Use stable output detection (no changes for N polls) in addition to prompt matching Warning signs: Commands "complete" with partial output, truncated responses

Pitfall 4: Timeout Too Short

What goes wrong: Complex GSD operations timeout before completion Why it happens: 60s timeout too aggressive for planning/shipping phases How to avoid: Make timeout configurable per command type; planning commands may need 300s+ Warning signs: Frequent timeout errors during complex operations

Pitfall 5: Missing Error Keywords

What goes wrong: Errors not detected, user sees raw output instead of error card Why it happens: Error keyword list incomplete How to avoid: Include common patterns: error, failed, exception, trace, error:, failed: Warning signs: Users report "weird output" that should be error cards

Code Examples

Complete Orchestrator Flow

# Source: Project architecture synthesis
from gsd_openclaw.tmux_session import TmuxSessionManager, TmuxSessionError
from gsd_openclaw.state_manager import StateManager
from gsd_openclaw.response_formatter import FeishuResponseFormatter, CardButton

class Orchestrator:
    """Main command orchestrator for gsd-openclaw skill."""

    ERROR_KEYWORDS = ['error', 'failed', 'exception', 'traceback', '错误', '失败']

    def __init__(self):
        self.tmux_manager = TmuxSessionManager()
        self.state_manager = StateManager  # Per-project
        self.formatter = FeishuResponseFormatter()

    def handle_message(self, user_id: str, message: str, channel: str, context: dict) -> dict:
        """Main entry point for OpenClaw skill messages.

        Returns Feishu Card JSON for response.
        """
        project_name = context.get('project', 'default')

        try:
            # 1. Get or create session
            session = self._ensure_session(project_name)

            # 2. Send command to tmux
            pane = session.tmux_session.active_pane
            pane.send_keys(message, enter=True)

            # 3. Capture output with timeout
            output, completed = self._capture_output(pane, timeout=60.0)

            # 4. Check for approval request
            approval = self._detect_approval(output)
            if approval:
                return self._create_approval_card(approval, session.session_id)

            # 5. Check for errors
            if self._has_errors(output):
                return self.formatter.format_error(
                    title="命令执行失败",
                    content=output[-2000:]  # Truncate for card limits
                )

            # 6. Return success with output
            return self.formatter.format_success(
                title="执行完成",
                content=output[-2000:] if len(output) > 2000 else output
            )

        except TmuxSessionError as e:
            return self.formatter.format_error(
                title="会话错误",
                content=str(e)
            )

    def _ensure_session(self, project_name: str) -> TmuxSession:
        """Get existing or create new tmux session."""
        session = self.tmux_manager.get_session(project_name)
        if session:
            return session
        return self.tmux_manager.create_session(project_name)

    def _capture_output(self, pane, timeout: float) -> tuple[str, bool]:
        """Capture output until command completes or timeout."""
        import time
        time.sleep(0.2)  # Wait for initial output

        lines = pane.capture_pane()
        return '\n'.join(lines), True  # Simplified; see Pattern 2 for full impl

    def _detect_approval(self, output: str) -> Optional[dict]:
        """Check for approval request patterns."""
        patterns = ['AskUserQuestion', 'ExitPlanMode', '需要确认']
        for p in patterns:
            if p in output:
                return {"question": "请确认此操作"}
        return None

    def _has_errors(self, output: str) -> bool:
        """Check for error indicators."""
        output_lower = output.lower()
        return any(kw in output_lower for kw in self.ERROR_KEYWORDS)

Callback Handler for Approval

# Source: Feishu callback integration pattern
class CallbackHandler:
    """Handles Feishu Card button callbacks."""

    def __init__(self, orchestrator: Orchestrator):
        self.orchestrator = orchestrator

    def handle(self, callback_data: dict) -> dict:
        """Process button callback and return response card."""
        action = callback_data.get('action')
        approval_id = callback_data.get('approval_id')

        if action == 'approve':
            # Send approval to tmux and continue
            return self._process_approval(approval_id, 'approve')

        elif action == 'reject':
            # Send rejection to tmux
            return self._process_approval(approval_id, 'reject')

        return self.orchestrator.formatter.format_error(
            title="未知操作",
            content=f"Unknown action: {action}"
        )

    def _process_approval(self, approval_id: str, action: str) -> dict:
        """Send approval/rejection to tmux session."""
        # Load approval state
        state = self.orchestrator.state_manager.load()
        approval = state.pending_approvals.get(approval_id)

        if not approval:
            return self.orchestrator.formatter.format_error(
                title="审批已过期",
                content="请重新执行命令"
            )

        # Get session and send response
        session = self.orchestrator.tmux_manager.get_session(approval['project'])
        if session:
            pane = session.tmux_session.active_pane
            response = "y" if action == 'approve' else "n'
            pane.send_keys(response, enter=True)

        # Clear pending approval
        del state.pending_approvals[approval_id]
        self.orchestrator.state_manager.save(state)

        return self.orchestrator.formatter.format_success(
            title="已处理",
            content=f"已{'确认' if action == 'approve' else '拒绝'}操作"
        )

State of the Art

Old Approach	Current Approach	When Changed	Impact
Blocking subprocess calls	libtmux ORM-style API	Phase 1 (2026-03-10)	Type-safe tmux control, better error handling
Raw JSON card construction	FeishuResponseFormatter class	Phase 2 (2026-03-11)	Consistent card structure, theme support
In-memory state	StateManager with JSON persistence	Phase 1 (2026-03-10)	Crash recovery, session continuity

Deprecated/outdated:

• Direct subprocess calls to tmux: Use libtmux instead
• Manual JSON construction for cards: Use FeishuResponseFormatter
• Global state variables: Use StateManager per-project

Open Questions

1. Exact approval detection patterns

   • What we know: Claude CLI uses AskUserQuestion, ExitPlanMode for approvals
   • What's unclear: Exact output format for these tool calls
   • Recommendation: Start with patterns in CONTEXT.md, refine during implementation based on actual output

2. Feishu callback URL registration

   • What we know: Feishu sends HTTP POST for button callbacks
   • What's unclear: How OpenClaw exposes callback handling to skills
   • Recommendation: Research OpenClaw skill callback API; may need gateway-level handler

3. Progress update delivery

   • What we know: Progress updates every 1 minute (CONTEXT.md decision)
   • What's unclear: How to deliver updates to same chat thread vs new message
   • Recommendation: Use message update API if available; otherwise send new message with reference

Validation Architecture

Test Framework

Property	Value
Framework	pytest (existing from Phase 1)
Config file	tests/conftest.py (existing)
Quick run command	pytest tests/ -v -x
Full suite command	pytest tests/ -v --cov=src/gsd_openclaw

Phase Requirements -> Test Map

Req ID	Behavior	Test Type	Automated Command	File Exists?
BRDG-01	Send GSD slash command through chat	unit	pytest tests/test_orchestrator.py::test_route_command -v	Wave 0
BRDG-02	Capture output and return	unit	pytest tests/test_command_executor.py::test_capture_output -v	Wave 0
BRDG-03	Error message formatting	unit	pytest tests/test_orchestrator.py::test_error_detection -v	Wave 0
BRDG-04	Approval workflow	unit	pytest tests/test_approval_handler.py -v	Wave 0
BRDG-05	Progress updates	unit	pytest tests/test_progress_monitor.py -v	Wave 0
OCLW-03	Command routing	unit	pytest tests/test_orchestrator.py::test_command_router -v	Wave 0
OCLW-05	Security integration	integration	Manual verification	Wave 0
OCLW-06	Multi-channel support	integration	Manual verification	Wave 0

Sampling Rate

• Per task commit: pytest tests/ -v -x
• Per wave merge: pytest tests/ -v --cov=src/gsd_openclaw
• Phase gate: Full suite green before /gsd:verify-work

Wave 0 Gaps

• tests/test_orchestrator.py - covers BRDG-01, BRDG-03, OCLW-03
• tests/test_command_executor.py - covers BRDG-02
• tests/test_approval_handler.py - covers BRDG-04
• tests/test_progress_monitor.py - covers BRDG-05
• Mock fixtures for tmux pane operations

(If no gaps: "None - existing test infrastructure covers all phase requirements")

Sources

Primary (HIGH confidence)

• libtmux pane.py - capture_pane(), send_keys() API documentation (verified from installed package)
• Phase 1 TmuxSessionManager - session lifecycle patterns (verified from src/gsd_openclaw/tmux_session.py)
• Phase 2 FeishuResponseFormatter - card formatting API (verified from src/gsd_openclaw/response_formatter.py)
• Project ARCHITECTURE.md - orchestrator patterns and data flow

Secondary (MEDIUM confidence)

• OpenClaw GitHub - dmPolicy, allowlist, skill structure
• libtmux documentation - pane API reference

Tertiary (LOW confidence)

• Feishu callback structure - web reader couldn't extract full content; verify during implementation

Metadata

Confidence breakdown:

• Standard stack: HIGH - All libraries are stdlib or existing Phase 1/2 code
• Architecture: HIGH - Patterns from established project architecture document
• Pitfalls: MEDIUM - Based on general tmux/CLI experience, needs real-world validation

Research date: 2026-03-11 Valid until: 30 days (stable patterns, no external API dependencies)