自动同步: 2026-03-12 03:00

skills/gsd-openclaw/tests/conftest.py

@@ -90,3 +90,32 @@ def sample_state() -> SessionState:
         last_output_snapshot="Test output snapshot",
         display_name="Test Project Display Name"
     )
+
+
+@pytest.fixture
+def tmp_users_root(tmp_path: Path) -> Path:
+    """Temporary root for user context testing.
+
+    This ensures tests don't pollute the real ~/gsd-workspaces/.users directory.
+
+    Args:
+        tmp_path: Pytest's built-in temporary directory fixture.
+
+    Returns:
+        Path to a temporary .users directory.
+    """
+    return tmp_path / "gsd-workspaces" / ".users"
+
+
+@pytest.fixture
+def user_context_manager(tmp_users_root: Path) -> "UserContextManager":
+    """UserContextManager with temporary root for isolation.
+
+    Args:
+        tmp_users_root: Fixture providing temporary user context root.
+
+    Returns:
+        UserContextManager instance configured for testing.
+    """
+    from gsd_openclaw.user_context import UserContextManager
+    return UserContextManager("test-user", tmp_users_root)

skills/gsd-openclaw/tests/test_approval_handler.py

@@ -0,0 +1,178 @@
+"""
+Tests for approval_handler module - ApprovalHandler and ApprovalState classes.
+
+TDD Task 1: Test scaffold for approval workflow handling.
+"""
+import pytest
+from unittest.mock import Mock, MagicMock
+import time
+
+# These imports will fail initially (RED phase)
+from gsd_openclaw.response_formatter import FeishuResponseFormatter, CardButton
+
+
+class TestApprovalHandlerDetection:
+    """Test approval request detection from output."""
+
+    def test_detect_approval_request_finds_ask_user_question(self):
+        """detect_approval_request returns context for 'AskUserQuestion' output."""
+        from gsd_openclaw.approval_handler import ApprovalHandler
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        handler = ApprovalHandler(formatter)
+
+        output = "Some text\nAskUserQuestion: Do you want to proceed?\nMore text"
+        result = handler.detect_approval_request(output)
+
+        assert result is not None
+        assert "AskUserQuestion" in result.get("type", "")
+
+    def test_detect_approval_request_finds_chinese_confirm(self):
+        """detect_approval_request returns context for '需要确认' output."""
+        from gsd_openclaw.approval_handler import ApprovalHandler
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        handler = ApprovalHandler(formatter)
+
+        output = "正在执行...\n需要确认: 是否继续?\n请回复"
+        result = handler.detect_approval_request(output)
+
+        assert result is not None
+        assert "需要确认" in result.get("type", "")
+
+
+class TestApprovalHandlerCardCreation:
+    """Test approval card creation with buttons."""
+
+    def test_create_approval_card_has_confirm_title(self):
+        """create_approval_card has title '需要确认'."""
+        from gsd_openclaw.approval_handler import ApprovalHandler
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        formatter.format_info = Mock(return_value={
+            "schema": "2.0",
+            "header": {"title": {"content": "需要确认"}, "template": "blue"},
+            "elements": []
+        })
+        handler = ApprovalHandler(formatter)
+
+        context = {"question": "是否继续?"}
+        card = handler.create_approval_card(context, "session-123", "test-project")
+
+        formatter.format_info.assert_called_once()
+        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'."""
+        from gsd_openclaw.approval_handler import ApprovalHandler
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        formatter.format_info = Mock(return_value={
+            "schema": "2.0",
+            "header": {"title": {"content": "需要确认"}, "template": "blue"},
+            "elements": []
+        })
+        handler = ApprovalHandler(formatter)
+
+        context = {"question": "是否继续?"}
+        card = handler.create_approval_card(context, "session-123", "test-project")
+
+        # Check that format_info was called with buttons
+        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)
+        assert confirm_btn is not None
+        assert confirm_btn.button_type == "primary"
+
+    def test_create_approval_card_has_reject_button(self):
+        """create_approval_card has reject button with action='reject'."""
+        from gsd_openclaw.approval_handler import ApprovalHandler
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        formatter.format_info = Mock(return_value={
+            "schema": "2.0",
+            "header": {"title": {"content": "需要确认"}, "template": "blue"},
+            "elements": []
+        })
+        handler = ApprovalHandler(formatter)
+
+        context = {"question": "是否继续?"}
+        card = handler.create_approval_card(context, "session-123", "test-project")
+
+        # Check that format_info was called with buttons
+        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)
+        assert reject_btn is not None
+        assert reject_btn.button_type == "danger"
+
+
+class TestApprovalHandlerCallback:
+    """Test approval callback handling."""
+
+    def test_handle_callback_returns_approved_for_approve_action(self):
+        """handle_callback returns 'approved' for approve action."""
+        from gsd_openclaw.approval_handler import ApprovalHandler, ApprovalState
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        handler = ApprovalHandler(formatter)
+
+        # Create a pending approval first
+        context = {"question": "是否继续?"}
+        handler.create_approval_card(context, "session-123", "test-project")
+
+        # Get the approval_id from pending_approvals
+        approval_id = list(handler.pending_approvals.keys())[0]
+
+        callback_data = {
+            "action": "approve",
+            "approval_id": approval_id,
+            "project": "test-project"
+        }
+        result = handler.handle_callback(callback_data)
+
+        assert result == "approved"
+
+    def test_handle_callback_returns_rejected_for_reject_action(self):
+        """handle_callback returns 'rejected' for reject action."""
+        from gsd_openclaw.approval_handler import ApprovalHandler
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        handler = ApprovalHandler(formatter)
+
+        # Create a pending approval first
+        context = {"question": "是否继续?"}
+        handler.create_approval_card(context, "session-123", "test-project")
+
+        # Get the approval_id from pending_approvals
+        approval_id = list(handler.pending_approvals.keys())[0]
+
+        callback_data = {
+            "action": "reject",
+            "approval_id": approval_id,
+            "project": "test-project"
+        }
+        result = handler.handle_callback(callback_data)
+
+        assert result == "rejected"
+
+    def test_handle_callback_returns_error_for_unknown_approval_id(self):
+        """handle_callback returns error for unknown approval_id."""
+        from gsd_openclaw.approval_handler import ApprovalHandler
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        handler = ApprovalHandler(formatter)
+
+        callback_data = {
+            "action": "approve",
+            "approval_id": "nonexistent-id",
+            "project": "test-project"
+        }
+        result = handler.handle_callback(callback_data)
+
+        assert "error" in result.lower()

skills/gsd-openclaw/tests/test_command_executor.py

@@ -0,0 +1,394 @@
+"""Tests for command execution and output capture.
+
+This module tests the CommandExecutor and OutputCapture classes which provide
+the foundational I/O layer for all GSD command bridging.
+"""
+import pytest
+import time
+from unittest.mock import MagicMock, patch, PropertyMock
+from dataclasses import fields
+
+from gsd_openclaw.command_executor import (
+    CommandResult,
+    OutputCapture,
+    CommandExecutor,
+)
+
+
+class TestOutputCaptureIsPrompt:
+    """Tests for OutputCapture._is_prompt method."""
+
+    def test_is_prompt_detects_dollar_prompt(self):
+        """_is_prompt should return True for '$ ' prompt."""
+        capture = OutputCapture()
+        assert capture._is_prompt("$ ") is True
+
+    def test_is_prompt_detects_dollar_with_space_only(self):
+        """_is_prompt should return True for '$' followed by only whitespace."""
+        capture = OutputCapture()
+        assert capture._is_prompt("$  ") is True
+
+    def test_is_prompt_detects_unicode_arrow_prompt(self):
+        """_is_prompt should return True for Unicode arrow prompts like '❯ '."""
+        capture = OutputCapture()
+        assert capture._is_prompt("❯ ") is True
+
+    def test_is_prompt_detects_triangle_arrow_prompt(self):
+        """_is_prompt should return True for triangle arrow prompts like '➜ '."""
+        capture = OutputCapture()
+        assert capture._is_prompt("➜ ") is True
+
+    def test_is_prompt_returns_false_for_non_prompt_lines(self):
+        """_is_prompt should return False for non-prompt lines."""
+        capture = OutputCapture()
+        assert capture._is_prompt("some command output") is False
+        assert capture._is_prompt("$ not a prompt (has content after)") is False
+        assert capture._is_prompt("") is False
+        assert capture._is_prompt("error: something failed") is False
+
+
+class TestOutputCaptureCaptureUntilComplete:
+    """Tests for OutputCapture.capture_until_complete method."""
+
+    @pytest.fixture
+    def mock_pane(self):
+        """Create a mock tmux pane."""
+        pane = MagicMock()
+        pane.pane_id = "%100"
+        return pane
+
+    def test_capture_until_complete_returns_output_and_true_when_prompt_detected(self, mock_pane):
+        """capture_until_complete should return (output, True) when prompt is detected."""
+        capture = OutputCapture()
+
+        # Simulate output that ends with prompt
+        mock_pane.capture_pane.return_value = [
+            "Running command...",
+            "Output line 1",
+            "Output line 2",
+            "$ ",  # Prompt appears
+        ]
+
+        output, completed = capture.capture_until_complete(mock_pane, timeout=1.0)
+
+        assert completed is True
+        assert "Running command..." in output
+        assert "Output line 1" in output
+
+    def test_capture_until_complete_returns_true_when_output_stabilizes(self, mock_pane):
+        """capture_until_complete should return (output, True) when output stabilizes."""
+        capture = OutputCapture()
+
+        # Simulate stable output (same for 3 consecutive polls)
+        stable_output = ["Line 1", "Line 2", "Line 3"]
+        mock_pane.capture_pane.return_value = stable_output
+
+        output, completed = capture.capture_until_complete(
+            mock_pane, timeout=1.0, poll_interval=0.01
+        )
+
+        assert completed is True
+        assert "Line 1" in output
+        assert "Line 2" in output
+
+    def test_capture_until_complete_returns_false_on_timeout(self, mock_pane):
+        """capture_until_complete should return (output, False) on timeout."""
+        capture = OutputCapture()
+
+        # Simulate output that keeps changing (never stabilizes, no prompt)
+        call_count = [0]
+
+        def changing_output():
+            call_count[0] += 1
+            return [f"Line {call_count[0]}"]
+
+        mock_pane.capture_pane.side_effect = changing_output
+
+        output, completed = capture.capture_until_complete(
+            mock_pane, timeout=0.1, poll_interval=0.02
+        )
+
+        assert completed is False
+
+
+class TestCommandExecutorDetectErrors:
+    """Tests for CommandExecutor._detect_errors method."""
+
+    def test_detect_errors_returns_true_for_error_keyword(self):
+        """_detect_errors should return (True, 'error') for 'Error: something'."""
+        executor = CommandExecutor()
+        has_error, error_type = executor._detect_errors("Error: something went wrong")
+
+        assert has_error is True
+        assert error_type == "error"
+
+    def test_detect_errors_returns_true_for_failed_keyword(self):
+        """_detect_errors should return (True, 'failed') for 'Process failed'."""
+        executor = CommandExecutor()
+        has_error, error_type = executor._detect_errors("Process failed to start")
+
+        assert has_error is True
+        assert error_type == "failed"
+
+    def test_detect_errors_returns_true_for_exception_keyword(self):
+        """_detect_errors should return (True, 'exception') for exception in output."""
+        executor = CommandExecutor()
+        has_error, error_type = executor._detect_errors("ValueError: An exception occurred")
+
+        assert has_error is True
+        assert error_type == "exception"
+
+    def test_detect_errors_returns_true_for_traceback_keyword(self):
+        """_detect_errors should return (True, 'traceback') for traceback in output."""
+        executor = CommandExecutor()
+        has_error, error_type = executor._detect_errors("Traceback (most recent call last):")
+
+        assert has_error is True
+        assert error_type == "traceback"
+
+    def test_detect_errors_returns_true_for_chinese_error_keywords(self):
+        """_detect_errors should detect Chinese error keywords."""
+        executor = CommandExecutor()
+
+        # Test Chinese error
+        has_error, error_type = executor._detect_errors("发生错误：无法连接")
+        assert has_error is True
+        assert error_type == "错误"
+
+        # Test Chinese failure
+        has_error, error_type = executor._detect_errors("操作失败")
+        assert has_error is True
+        assert error_type == "失败"
+
+    def test_detect_errors_returns_false_for_normal_output(self):
+        """_detect_errors should return (False, None) for normal output."""
+        executor = CommandExecutor()
+
+        normal_outputs = [
+            "Command completed successfully",
+            "All tests passed",
+            "Files created: 5",
+            "Done!",
+        ]
+
+        for output in normal_outputs:
+            has_error, error_type = executor._detect_errors(output)
+            assert has_error is False
+            assert error_type is None
+
+
+class TestCommandExecutorExecute:
+    """Tests for CommandExecutor.execute method."""
+
+    @pytest.fixture
+    def mock_session(self):
+        """Create a mock TmuxSession with active_pane."""
+        session = MagicMock()
+        session.session_id = "test-123"
+        session.project_name = "test-project"
+        session.tmux_session_name = "gsd-test-project"
+
+        # Mock the tmux_session.active_pane
+        mock_pane = MagicMock()
+        mock_pane.pane_id = "%100"
+        mock_pane.send_keys = MagicMock()
+
+        mock_tmux_session = MagicMock()
+        mock_tmux_session.active_pane = mock_pane
+        session.tmux_session = mock_tmux_session
+
+        return session
+
+    def test_execute_returns_command_result_with_all_fields(self, mock_session):
+        """execute should return CommandResult with all required fields."""
+        executor = CommandExecutor()
+
+        # Mock capture to return immediately with prompt
+        mock_session.tmux_session.active_pane.capture_pane.return_value = [
+            "Command output",
+            "$ ",
+        ]
+
+        result = executor.execute(mock_session, "echo hello", timeout=1.0)
+
+        # Verify CommandResult has all required fields
+        assert hasattr(result, 'output')
+        assert hasattr(result, 'completed')
+        assert hasattr(result, 'has_error')
+        assert hasattr(result, 'error_type')
+
+    def test_execute_calls_send_keys_with_enter(self, mock_session):
+        """execute should call send_keys with the command."""
+        executor = CommandExecutor()
+
+        # Mock capture to return immediately
+        mock_session.tmux_session.active_pane.capture_pane.return_value = ["$ "]
+
+        executor.execute(mock_session, "echo hello", timeout=1.0)
+
+        # Verify send_keys was called
+        mock_session.tmux_session.active_pane.send_keys.assert_called_once()
+
+    def test_execute_sets_has_error_true_when_error_keywords_detected(self, mock_session):
+        """execute should set has_error=True when error keywords detected."""
+        executor = CommandExecutor()
+
+        # Mock output with error
+        mock_session.tmux_session.active_pane.capture_pane.return_value = [
+            "Error: command failed",
+            "$ ",
+        ]
+
+        result = executor.execute(mock_session, "bad-command", timeout=1.0)
+
+        assert result.has_error is True
+        assert result.error_type == "error"
+
+    def test_execute_respects_timeout_parameter(self, mock_session):
+        """execute should respect the timeout parameter."""
+        executor = CommandExecutor()
+
+        # Mock output that never stabilizes
+        call_count = [0]
+
+        def changing_output():
+            call_count[0] += 1
+            return [f"Line {call_count[0]}"]
+
+        mock_session.tmux_session.active_pane.capture_pane.side_effect = changing_output
+
+        start_time = time.time()
+        result = executor.execute(mock_session, "slow-command", timeout=0.1, poll_interval=0.02)
+        elapsed = time.time() - start_time
+
+        assert result.completed is False
+        assert elapsed < 0.3  # Should timeout around 0.1s, not much longer
+
+    def test_execute_uses_default_timeout_of_60_seconds(self, mock_session):
+        """execute should use default timeout of 60 seconds."""
+        executor = CommandExecutor()
+
+        # Verify DEFAULT_TIMEOUT is set
+        assert executor.DEFAULT_TIMEOUT == 60.0
+
+
+class TestCommandResultDataclass:
+    """Tests for CommandResult dataclass."""
+
+    def test_command_result_stores_all_fields(self):
+        """CommandResult should store all required fields."""
+        result = CommandResult(
+            output="test output",
+            completed=True,
+            has_error=False,
+            error_type=None,
+        )
+
+        assert result.output == "test output"
+        assert result.completed is True
+        assert result.has_error is False
+        assert result.error_type is None
+
+    def test_command_result_with_error(self):
+        """CommandResult should store error information correctly."""
+        result = CommandResult(
+            output="Error: something failed",
+            completed=True,
+            has_error=True,
+            error_type="error",
+        )
+
+        assert result.has_error is True
+        assert result.error_type == "error"
+
+    def test_command_result_incomplete_execution(self):
+        """CommandResult should handle incomplete execution."""
+        result = CommandResult(
+            output="Partial output...",
+            completed=False,
+            has_error=False,
+            error_type=None,
+        )
+
+        assert result.completed is False
+
+
+class TestCommandExecutorSendCommand:
+    """Tests for CommandExecutor.send_command method."""
+
+    def test_send_command_calls_pane_send_keys(self):
+        """send_command should call pane.send_keys with the command."""
+        executor = CommandExecutor()
+
+        mock_pane = MagicMock()
+        mock_pane.send_keys = MagicMock()
+
+        executor.send_command(mock_pane, "echo hello")
+
+        mock_pane.send_keys.assert_called_once()
+
+    def test_send_command_passes_enter_true(self):
+        """send_command should send keys with enter=True to execute the command."""
+        executor = CommandExecutor()
+
+        mock_pane = MagicMock()
+        mock_pane.send_keys = MagicMock()
+
+        executor.send_command(mock_pane, "echo hello")
+
+        # Verify it was called - the exact implementation should include enter
+        call_args = mock_pane.send_keys.call_args
+        assert call_args is not None
+
+
+class TestLongCommands:
+    """Tests for long command detection in CommandExecutor."""
+
+    def test_long_commands_set_contains_expected_commands(self):
+        """LONG_COMMANDS should contain execute-phase, plan-phase, discuss-phase, new-project."""
+        executor = CommandExecutor()
+
+        assert 'execute-phase' in executor.LONG_COMMANDS
+        assert 'plan-phase' in executor.LONG_COMMANDS
+        assert 'discuss-phase' in executor.LONG_COMMANDS
+        assert 'new-project' in executor.LONG_COMMANDS
+
+    def test_is_long_command_returns_true_for_long_commands(self):
+        """_is_long_command should return True for commands in LONG_COMMANDS."""
+        executor = CommandExecutor()
+
+        assert executor._is_long_command('/gsd:execute-phase') is True
+        assert executor._is_long_command('/gsd:plan-phase') is True
+        assert executor._is_long_command('/gsd:discuss-phase') is True
+        assert executor._is_long_command('/gsd:new-project') is True
+
+    def test_is_long_command_returns_false_for_regular_commands(self):
+        """_is_long_command should return False for commands not in LONG_COMMANDS."""
+        executor = CommandExecutor()
+
+        assert executor._is_long_command('/gsd:status') is False
+        assert executor._is_long_command('/gsd:list') is False
+        assert executor._is_long_command('/gsd:help') is False
+        assert executor._is_long_command('/gsd:use') is False
+
+    def test_is_long_command_handles_commands_with_arguments(self):
+        """_is_long_command should handle commands with arguments correctly."""
+        executor = CommandExecutor()
+
+        # Long commands with args should still be detected
+        assert executor._is_long_command('/gsd:execute-phase 01-core') is True
+        assert executor._is_long_command('/gsd:plan-phase feature-x') is True
+        assert executor._is_long_command('/gsd:discuss-phase about architecture') is True
+        assert executor._is_long_command('/gsd:new-project my-project') is True
+
+        # Regular commands with args should not be detected as long
+        assert executor._is_long_command('/gsd:status --verbose') is False
+        assert executor._is_long_command('/gsd:use my-project') is False
+
+    def test_is_long_command_returns_false_for_non_gsd_commands(self):
+        """_is_long_command should return False for non-/gsd: commands."""
+        executor = CommandExecutor()
+
+        assert executor._is_long_command('status') is False
+        assert executor._is_long_command('/help') is False
+        assert executor._is_long_command('regular message') is False

skills/gsd-openclaw/tests/test_progress_monitor.py

@@ -0,0 +1,131 @@
+"""
+Tests for progress_monitor module - ProgressMonitor and ProgressInfo classes.
+
+TDD Task 1: Test scaffold for progress detection and update logic.
+"""
+import pytest
+from unittest.mock import Mock, MagicMock
+import time
+
+# These imports will fail initially (RED phase)
+from gsd_openclaw.response_formatter import FeishuResponseFormatter
+
+
+class TestProgressMonitorExtraction:
+    """Test progress extraction from output."""
+
+    def test_extract_progress_returns_percentage(self):
+        """extract_progress returns percentage for 'Progress: 75%'."""
+        from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        monitor = ProgressMonitor(formatter)
+
+        output = "Working on task...\nProgress: 75%\nAlmost done"
+        result = monitor.extract_progress(output)
+
+        assert result is not None
+        assert result.type == "percentage"
+        assert result.value == 75
+
+    def test_extract_progress_returns_fraction(self):
+        """extract_progress returns fraction for '[3/5 tasks]'."""
+        from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        monitor = ProgressMonitor(formatter)
+
+        output = "Processing...\n[3/5 tasks] completed\nContinuing..."
+        result = monitor.extract_progress(output)
+
+        assert result is not None
+        assert result.type == "fraction"
+        assert result.current == 3
+        assert result.total == 5
+
+    def test_extract_progress_returns_phase(self):
+        """extract_progress returns phase for 'Phase 2 complete'."""
+        from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        monitor = ProgressMonitor(formatter)
+
+        output = "Starting work...\nPhase 2 complete\nMoving to next"
+        result = monitor.extract_progress(output)
+
+        assert result is not None
+        assert result.type == "phase"
+        assert result.value == 2
+
+
+class TestProgressMonitorThrottling:
+    """Test progress update throttling."""
+
+    def test_should_send_update_returns_false_immediately(self):
+        """should_send_update returns False immediately after creation."""
+        from gsd_openclaw.progress_monitor import ProgressMonitor
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        monitor = ProgressMonitor(formatter, update_interval=60.0)
+
+        result = monitor.should_send_update()
+
+        assert result is False
+
+    def test_should_send_update_returns_true_after_interval(self):
+        """should_send_update returns True after 60 seconds."""
+        from gsd_openclaw.progress_monitor import ProgressMonitor
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        monitor = ProgressMonitor(formatter, update_interval=60.0)
+
+        # Simulate time passing by modifying last_update_time
+        monitor.last_update_time = time.time() - 61.0
+
+        result = monitor.should_send_update()
+
+        assert result is True
+
+
+class TestProgressMonitorCardCreation:
+    """Test progress card creation."""
+
+    def test_create_progress_card_includes_percentage(self):
+        """create_progress_card includes progress percentage in content."""
+        from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        formatter.format_info = Mock(return_value={
+            "schema": "2.0",
+            "header": {"title": {"content": "正在执行..."}, "template": "blue"},
+            "elements": []
+        })
+        monitor = ProgressMonitor(formatter)
+
+        progress = ProgressInfo(type="percentage", value=75)
+        card = monitor.create_progress_card(progress, "Processing data...")
+
+        formatter.format_info.assert_called_once()
+        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):
+        """create_progress_card uses title '正在执行...'."""
+        from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
+
+        formatter = Mock(spec=FeishuResponseFormatter)
+        formatter.format_info = Mock(return_value={
+            "schema": "2.0",
+            "header": {"title": {"content": "正在执行..."}, "template": "blue"},
+            "elements": []
+        })
+        monitor = ProgressMonitor(formatter)
+
+        progress = ProgressInfo(type="percentage", value=50)
+        card = monitor.create_progress_card(progress, "Working...")
+
+        formatter.format_info.assert_called_once()
+        call_kwargs = formatter.format_info.call_args.kwargs
+        title = call_kwargs.get("title", "")
+        assert title == "正在执行..."

skills/gsd-openclaw/tests/test_user_context.py

@@ -0,0 +1,389 @@
+"""Tests for UserContextManager - per-user project context tracking."""
+import json
+from pathlib import Path
+
+import pytest
+
+from gsd_openclaw.user_context import UserContext, UserContextManager
+
+
+class TestUserContext:
+    """Tests for UserContext dataclass."""
+
+    def test_create_with_all_fields(self) -> None:
+        """UserContext can be created with all fields."""
+        context = UserContext(
+            schema_version="1.0",
+            user_id="user-123",
+            current_project="my-project",
+            recent_projects=["proj1", "proj2"],
+            last_active_at="2026-03-11T10:00:00Z",
+            preferences={"theme": "dark"}
+        )
+        assert context.schema_version == "1.0"
+        assert context.user_id == "user-123"
+        assert context.current_project == "my-project"
+        assert context.recent_projects == ["proj1", "proj2"]
+        assert context.last_active_at == "2026-03-11T10:00:00Z"
+        assert context.preferences == {"theme": "dark"}
+
+    def test_to_dict_returns_complete_dict(self) -> None:
+        """UserContext.to_dict() returns complete dict with all fields."""
+        context = UserContext(
+            schema_version="1.0",
+            user_id="user-456",
+            current_project="test-project",
+            recent_projects=["a", "b", "c"],
+            last_active_at="2026-03-11T11:00:00Z",
+            preferences={"key": "value"}
+        )
+        result = context.to_dict()
+        assert isinstance(result, dict)
+        assert result["schema_version"] == "1.0"
+        assert result["user_id"] == "user-456"
+        assert result["current_project"] == "test-project"
+        assert result["recent_projects"] == ["a", "b", "c"]
+        assert result["last_active_at"] == "2026-03-11T11:00:00Z"
+        assert result["preferences"] == {"key": "value"}
+
+    def test_from_dict_reconstructs_from_dict(self) -> None:
+        """UserContext.from_dict() reconstructs from dict, ignoring unknown fields."""
+        data = {
+            "schema_version": "1.0",
+            "user_id": "user-789",
+            "current_project": "reconstructed-project",
+            "recent_projects": ["x", "y"],
+            "last_active_at": "2026-03-11T12:00:00Z",
+            "preferences": {"setting": True},
+            "unknown_field": "should be ignored",
+            "another_unknown": 12345
+        }
+        context = UserContext.from_dict(data)
+        assert context.schema_version == "1.0"
+        assert context.user_id == "user-789"
+        assert context.current_project == "reconstructed-project"
+        assert context.recent_projects == ["x", "y"]
+        assert context.last_active_at == "2026-03-11T12:00:00Z"
+        assert context.preferences == {"setting": True}
+        # Unknown fields should not cause issues
+        assert not hasattr(context, "unknown_field")
+
+
+class TestUserContextManager:
+    """Tests for UserContextManager class."""
+
+    def test_init_sets_users_root_and_creates_context_path(self, tmp_users_root: Path) -> None:
+        """UserContextManager.__init__ sets users_root and creates context_path correctly."""
+        manager = UserContextManager("user-abc", tmp_users_root)
+        assert manager.users_root == tmp_users_root
+        assert manager.context_path == tmp_users_root / "user-abc" / "context.json"
+
+    def test_init_with_default_users_root(self) -> None:
+        """UserContextManager uses default USERS_ROOT when not specified."""
+        manager = UserContextManager("user-default")
+        expected_root = Path.home() / "gsd-workspaces" / ".users"
+        assert manager.users_root == expected_root
+        assert manager.context_path == expected_root / "user-default" / "context.json"
+
+    def test_load_returns_none_for_new_users(self, tmp_users_root: Path) -> None:
+        """UserContextManager.load() returns None for new users (no existing context)."""
+        manager = UserContextManager("new-user-no-context", tmp_users_root)
+        result = manager.load()
+        assert result is None
+
+    def test_load_returns_usercontext_for_existing_file(self, tmp_users_root: Path) -> None:
+        """UserContextManager.load() returns UserContext for existing context file."""
+        user_id = "existing-user"
+        context_path = tmp_users_root / user_id / "context.json"
+        context_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Write a valid context file
+        data = {
+            "schema_version": "1.0",
+            "user_id": user_id,
+            "current_project": "loaded-project",
+            "recent_projects": ["proj-a"],
+            "last_active_at": "2026-03-11T09:00:00Z",
+            "preferences": {}
+        }
+        with open(context_path, "w", encoding="utf-8") as f:
+            json.dump(data, f)
+
+        manager = UserContextManager(user_id, tmp_users_root)
+        result = manager.load()
+
+        assert result is not None
+        assert isinstance(result, UserContext)
+        assert result.user_id == user_id
+        assert result.current_project == "loaded-project"
+
+
+class TestAtomicWrites:
+    """Tests for atomic write pattern in UserContextManager.save()."""
+
+    def test_save_creates_file_atomically(self, tmp_users_root: Path) -> None:
+        """save() creates file atomically (no temp files remain after save)."""
+        manager = UserContextManager("atomic-user", tmp_users_root)
+        context = UserContext(
+            user_id="atomic-user",
+            current_project="atomic-test",
+            last_active_at="2026-03-11T10:00:00Z"
+        )
+        manager.save(context)
+
+        # Main file should exist
+        assert manager.context_path.exists()
+
+        # No temp files should remain
+        temp_files = list(manager.context_path.parent.glob("*.tmp"))
+        assert len(temp_files) == 0
+
+    def test_save_creates_parent_directories(self, tmp_users_root: Path) -> None:
+        """save() creates parent directories if missing."""
+        manager = UserContextManager("nested-user", tmp_users_root)
+        assert not manager.context_path.parent.exists()
+
+        context = UserContext(user_id="nested-user", last_active_at="2026-03-11T10:00:00Z")
+        manager.save(context)
+
+        assert manager.context_path.parent.exists()
+        assert manager.context_path.exists()
+
+    def test_save_rotates_backup(self, tmp_users_root: Path) -> None:
+        """save() rotates backup (context.json -> context.json.bak)."""
+        manager = UserContextManager("backup-user", tmp_users_root)
+
+        # First save
+        context1 = UserContext(
+            user_id="backup-user",
+            current_project="first-project",
+            last_active_at="2026-03-11T10:00:00Z"
+        )
+        manager.save(context1)
+
+        # No backup yet
+        backup_path = manager.context_path.with_suffix(".json.bak")
+        assert not backup_path.exists()
+
+        # Second save
+        context2 = UserContext(
+            user_id="backup-user",
+            current_project="second-project",
+            last_active_at="2026-03-11T11:00:00Z"
+        )
+        manager.save(context2)
+
+        # Backup should now exist
+        assert backup_path.exists()
+
+        # Backup should contain first project
+        with open(backup_path, "r", encoding="utf-8") as f:
+            backup_data = json.load(f)
+        assert backup_data["current_project"] == "first-project"
+
+    def test_save_overwrites_old_backup(self, tmp_users_root: Path) -> None:
+        """save() overwrites old backup on subsequent saves."""
+        manager = UserContextManager("overwrite-user", tmp_users_root)
+        backup_path = manager.context_path.with_suffix(".json.bak")
+
+        # Save three times
+        for i, project in enumerate(["proj1", "proj2", "proj3"]):
+            context = UserContext(
+                user_id="overwrite-user",
+                current_project=project,
+                last_active_at=f"2026-03-11T{10+i}:00:00Z"
+            )
+            manager.save(context)
+
+        # Backup should contain proj2 (the one before last)
+        with open(backup_path, "r", encoding="utf-8") as f:
+            backup_data = json.load(f)
+        assert backup_data["current_project"] == "proj2"
+
+        # Main file should contain proj3
+        with open(manager.context_path, "r", encoding="utf-8") as f:
+            main_data = json.load(f)
+        assert main_data["current_project"] == "proj3"
+
+    def test_save_load_roundtrip_preserves_data(self, tmp_users_root: Path) -> None:
+        """save/load roundtrip preserves all data."""
+        manager = UserContextManager("roundtrip-user", tmp_users_root)
+
+        original = UserContext(
+            schema_version="1.0",
+            user_id="roundtrip-user",
+            current_project="roundtrip-project",
+            recent_projects=["r1", "r2", "r3"],
+            last_active_at="2026-03-11T12:30:45Z",
+            preferences={"setting1": "value1", "setting2": 42}
+        )
+        manager.save(original)
+
+        loaded = manager.load()
+        assert loaded is not None
+        assert loaded.schema_version == original.schema_version
+        assert loaded.user_id == original.user_id
+        assert loaded.current_project == original.current_project
+        assert loaded.recent_projects == original.recent_projects
+        assert loaded.last_active_at == original.last_active_at
+        assert loaded.preferences == original.preferences
+
+    def test_load_falls_back_to_backup_when_main_corrupted(self, tmp_users_root: Path) -> None:
+        """load() falls back to backup when main file is corrupted."""
+        manager = UserContextManager("fallback-user", tmp_users_root)
+
+        # Create valid context and save twice to create backup
+        context = UserContext(
+            user_id="fallback-user",
+            current_project="first-project",
+            last_active_at="2026-03-11T09:00:00Z"
+        )
+        manager.save(context)
+
+        # Save again to create backup (backup only exists after second save)
+        context.current_project = "valid-project"
+        context.last_active_at = "2026-03-11T10:00:00Z"
+        manager.save(context)
+
+        # Backup should now exist with "first-project"
+        backup_path = manager.context_path.with_suffix(".json.bak")
+        assert backup_path.exists()
+
+        # Corrupt the main file
+        with open(manager.context_path, "w", encoding="utf-8") as f:
+            f.write("not valid json {{{")
+
+        # Load should fall back to backup
+        loaded = manager.load()
+        assert loaded is not None
+        assert loaded.current_project == "first-project"
+
+
+class TestMultiUser:
+    """Tests for multi-user isolation."""
+
+    def test_two_users_have_isolated_context_files(self, tmp_users_root: Path) -> None:
+        """Two users have completely isolated context files."""
+        manager1 = UserContextManager("user-alice", tmp_users_root)
+        manager2 = UserContextManager("user-bob", tmp_users_root)
+
+        # Different context paths
+        assert manager1.context_path != manager2.context_path
+        assert "user-alice" in str(manager1.context_path)
+        assert "user-bob" in str(manager2.context_path)
+
+    def test_changes_to_one_user_do_not_affect_another(self, tmp_users_root: Path) -> None:
+        """Changes to one user don't affect another."""
+        manager1 = UserContextManager("user-isolated-1", tmp_users_root)
+        manager2 = UserContextManager("user-isolated-2", tmp_users_root)
+
+        # Set context for user 1
+        context1 = UserContext(
+            user_id="user-isolated-1",
+            current_project="alice-project",
+            recent_projects=["alice-recent"],
+            last_active_at="2026-03-11T10:00:00Z"
+        )
+        manager1.save(context1)
+
+        # Set context for user 2
+        context2 = UserContext(
+            user_id="user-isolated-2",
+            current_project="bob-project",
+            recent_projects=["bob-recent"],
+            last_active_at="2026-03-11T11:00:00Z"
+        )
+        manager2.save(context2)
+
+        # Load both and verify isolation
+        loaded1 = manager1.load()
+        loaded2 = manager2.load()
+
+        assert loaded1 is not None
+        assert loaded2 is not None
+        assert loaded1.current_project == "alice-project"
+        assert loaded2.current_project == "bob-project"
+        assert loaded1.recent_projects == ["alice-recent"]
+        assert loaded2.recent_projects == ["bob-recent"]
+
+    def test_get_current_project_returns_current_or_none(self, tmp_users_root: Path) -> None:
+        """get_current_project() returns current_project or None."""
+        manager = UserContextManager("get-project-user", tmp_users_root)
+
+        # New user has no context
+        assert manager.get_current_project() is None
+
+        # Set a project
+        context = UserContext(
+            user_id="get-project-user",
+            current_project="my-current-project",
+            last_active_at="2026-03-11T10:00:00Z"
+        )
+        manager.save(context)
+
+        assert manager.get_current_project() == "my-current-project"
+
+    def test_set_current_project_updates_project_and_recent(self, tmp_users_root: Path) -> None:
+        """set_current_project() updates current_project and adds to recent_projects."""
+        manager = UserContextManager("set-project-user", tmp_users_root)
+
+        manager.set_current_project("new-project")
+
+        loaded = manager.load()
+        assert loaded is not None
+        assert loaded.current_project == "new-project"
+        assert "new-project" in loaded.recent_projects
+
+    def test_set_current_project_maintains_max_five_recent(self, tmp_users_root: Path) -> None:
+        """set_current_project() maintains only last 5 recent projects."""
+        manager = UserContextManager("max-five-user", tmp_users_root)
+
+        # Add 7 projects
+        projects = [f"project-{i}" for i in range(7)]
+        for project in projects:
+            manager.set_current_project(project)
+
+        loaded = manager.load()
+        assert loaded is not None
+
+        # Should have exactly 5 recent projects
+        assert len(loaded.recent_projects) == 5
+
+        # Should be the most recent 5 (2-6), excluding project-0 and project-1
+        expected = ["project-6", "project-5", "project-4", "project-3", "project-2"]
+        assert loaded.recent_projects == expected
+
+    def test_get_recent_projects_returns_list(self, tmp_users_root: Path) -> None:
+        """get_recent_projects() returns list of recent projects."""
+        manager = UserContextManager("recent-list-user", tmp_users_root)
+
+        # New user has empty list
+        assert manager.get_recent_projects() == []
+
+        # Add projects
+        manager.set_current_project("first")
+        manager.set_current_project("second")
+
+        recent = manager.get_recent_projects()
+        assert "second" in recent
+        assert "first" in recent
+
+    def test_new_user_gets_default_context_on_first_access(self, tmp_users_root: Path) -> None:
+        """New user gets default context created on first access."""
+        manager = UserContextManager("default-context-user", tmp_users_root)
+
+        # No context file exists yet
+        assert not manager.context_path.exists()
+
+        # set_current_project should create default context
+        manager.set_current_project("first-project")
+
+        # Now context should exist with proper defaults
+        loaded = manager.load()
+        assert loaded is not None
+        assert loaded.schema_version == "1.0"
+        assert loaded.user_id == "default-context-user"
+        assert loaded.current_project == "first-project"
+        assert loaded.recent_projects == ["first-project"]
+        assert loaded.preferences == {}
+        assert loaded.last_active_at != ""  # Should be set