From 569d547c569ab2c42403d5b10443a37d9b25ab4c Mon Sep 17 00:00:00 2001 From: Rui Hu Date: Tue, 31 Mar 2026 05:38:41 +0000 Subject: [PATCH] feat(tui): implement block context menu with 7 actions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implemented the BlockContextMenu widget as a floating overlay providing 7 keyboard-driven actions for conversation blocks per specification § Block Cursor and Context Menu: - Copy to clipboard (c): Uses pyperclip → OSC 52 → flash fallback chain - Copy to prompt (p): Returns block text for prompt insertion - Export as Markdown (e): Writes .md file to current directory - Export as SVG (v): Renders SVG and opens in default browser - Maximize/restore (m): Toggles ExpandProtocol blocks (ActorThought, ToolCall, DiffView) - Retry (r): Re-sends preceding prompt for ActorResponse blocks only - Show raw data (d): Displays raw A2A message data New modules: - cleveragents.tui.clipboard: Clipboard utility with 3-strategy fallback - cleveragents.tui.widgets.block_context_menu: BlockContextMenu widget, BlockType enum (10 types), BlockInfo/ActionResult/MenuAction dataclasses, ExpandProtocol, and action applicability filtering 41 BDD scenarios across 2 feature files covering all actions, menu lifecycle, inapplicable action filtering, clipboard fallback chain, argument validation, export error handling, and XML escaping. All quality gates pass: lint, typecheck, unit_tests (13026 scenarios), integration_tests, e2e_tests, coverage_report (97.0%). ISSUES CLOSED: #999 --- CHANGELOG.md | 9 + features/mocks/block_context_menu_helpers.py | 32 + ...tui_block_context_menu_edge_cases_steps.py | 703 ++++++++++++++++++ .../steps/tui_block_context_menu_steps.py | 225 ++++++ features/tui_block_context_menu.feature | 134 ++++ .../tui_block_context_menu_edge_cases.feature | 371 +++++++++ pyproject.toml | 1 + src/cleveragents/tui/clipboard.py | 126 ++++ src/cleveragents/tui/widgets/__init__.py | 18 + src/cleveragents/tui/widgets/_export_utils.py | 99 +++ .../tui/widgets/block_context_menu.py | 497 +++++++++++++ typings/pyperclip/__init__.pyi | 4 + 12 files changed, 2219 insertions(+) create mode 100644 features/mocks/block_context_menu_helpers.py create mode 100644 features/steps/tui_block_context_menu_edge_cases_steps.py create mode 100644 features/steps/tui_block_context_menu_steps.py create mode 100644 features/tui_block_context_menu.feature create mode 100644 features/tui_block_context_menu_edge_cases.feature create mode 100644 src/cleveragents/tui/clipboard.py create mode 100644 src/cleveragents/tui/widgets/_export_utils.py create mode 100644 src/cleveragents/tui/widgets/block_context_menu.py create mode 100644 typings/pyperclip/__init__.pyi diff --git a/CHANGELOG.md b/CHANGELOG.md index 36aab2df1..84df157bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,15 @@ the same path in a subprocess context. Tests simulate the divergent- container condition (fresh ``CLEVERAGENTS_HOME`` with empty database). ASV benchmark measures active-plan filtering overhead. (#1035) +- Added `BlockContextMenu` widget — a bordered floating overlay providing + 7 keyboard-driven actions (copy clipboard, copy to prompt, export + Markdown, export SVG, maximize/restore, retry, show raw data) for + conversation blocks per § Block Cursor and Context Menu. Includes + clipboard utility module with pyperclip → OSC 52 → flash message + fallback chain, path-safe file export with traversal prevention and + collision avoidance, and XML control character stripping for SVG output. + 67 BDD scenarios across two feature files and full step definitions. + (#999) - Added missing `LspServerConfig` model fields per specification: `description` (max 1000 chars), `transport` (`LspTransport` enum with `stdio`/`tcp`, default `stdio`), `initialization` (dict for LSP diff --git a/features/mocks/block_context_menu_helpers.py b/features/mocks/block_context_menu_helpers.py new file mode 100644 index 000000000..e972db803 --- /dev/null +++ b/features/mocks/block_context_menu_helpers.py @@ -0,0 +1,32 @@ +"""Shared helpers for block context menu BDD steps. + +Provides the block-type-to-enum mapping used by both the main and +edge-case step definition files. +""" + +from __future__ import annotations + +from typing import Any + + +def block_type_friendly_map(bt_cls: Any) -> dict[str, Any]: + """Return a mapping from friendly block-type names to enum members. + + Args: + bt_cls: The ``BlockType`` enum class (may be from a reloaded module). + + Returns: + Dictionary mapping user-facing names to ``BlockType`` enum values. + """ + return { + "ActorResponse": bt_cls.ACTOR_RESPONSE, + "UserInput": bt_cls.USER_INPUT, + "ToolCall": bt_cls.TOOL_CALL, + "ActorThought": bt_cls.ACTOR_THOUGHT, + "DiffView": bt_cls.DIFF_VIEW, + "Welcome": bt_cls.WELCOME, + "PlanProgress": bt_cls.PLAN_PROGRESS, + "TerminalEmbed": bt_cls.TERMINAL_EMBED, + "ShellResult": bt_cls.SHELL_RESULT, + "Note": bt_cls.NOTE, + } diff --git a/features/steps/tui_block_context_menu_edge_cases_steps.py b/features/steps/tui_block_context_menu_edge_cases_steps.py new file mode 100644 index 000000000..85036811f --- /dev/null +++ b/features/steps/tui_block_context_menu_edge_cases_steps.py @@ -0,0 +1,703 @@ +"""Step definitions for tui_block_context_menu_edge_cases.feature. + +Tests menu dismiss, inapplicable action states, clipboard fallback +chain, argument validation, and additional block type coverage. +""" + +from __future__ import annotations + +import io +import os +import shutil +import tempfile +from typing import Any +from unittest.mock import patch + +from behave import given, then, use_step_matcher, when +from behave.runner import Context + +import cleveragents.tui.widgets.block_context_menu as _bcm_mod +from cleveragents.tui.clipboard import CopyMethod, _try_pyperclip, copy_to_clipboard +from cleveragents.tui.widgets._export_utils import ( + safe_export_path, + unique_filepath, +) +from cleveragents.tui.widgets.block_context_menu import BlockType +from features.mocks.block_context_menu_helpers import block_type_friendly_map + + +class _FakeTTY(io.StringIO): + """A StringIO subclass that reports ``isatty()=True`` for OSC 52 tests.""" + + def isatty(self) -> bool: + return True + + +# Note: Block info creation steps use the "bcm block" prefix from the main +# steps file (tui_block_context_menu_steps.py) which is loaded by Behave. + + +# --------------------------------------------------------------------------- +# Inapplicable actions +# --------------------------------------------------------------------------- +@then('the applicable actions should not include key "{key}"') +def step_actions_not_include(context: Context, key: str) -> None: + applicable: list[Any] = context._menu.get_applicable_actions() # type: ignore[attr-defined] + keys = [a.key for a in applicable] + assert key not in keys, f"Key '{key}' should not be applicable, but found in {keys}" + + +@then('the applicable actions should include key "{key}"') +def step_actions_include(context: Context, key: str) -> None: + applicable: list[Any] = context._menu.get_applicable_actions() # type: ignore[attr-defined] + keys = [a.key for a in applicable] + assert key in keys, f"Key '{key}' should be applicable, but not in {keys}" + + +# --------------------------------------------------------------------------- +# Clipboard fallback chain +# --------------------------------------------------------------------------- +@given("pyperclip is not available") +def step_no_pyperclip(context: Context) -> None: + context._pyperclip_patch = patch( # type: ignore[attr-defined] + "cleveragents.tui.clipboard._try_pyperclip", return_value=None + ) + context._pyperclip_patch.start() # type: ignore[attr-defined] + context.add_cleanup(context._pyperclip_patch.stop) # type: ignore[attr-defined] + + +@given("OSC 52 is not available") +def step_no_osc52(context: Context) -> None: + context._osc52_patch = patch( # type: ignore[attr-defined] + "cleveragents.tui.clipboard._try_osc52", return_value=None + ) + context._osc52_patch.start() # type: ignore[attr-defined] + context.add_cleanup(context._osc52_patch.stop) # type: ignore[attr-defined] + + +@when('I copy "{text}" to clipboard') +def step_copy_to_clipboard(context: Context, text: str) -> None: + # If stdout patches are already in place (e.g. for exception testing), + # avoid replacing stdout with a fake to preserve those patches. + if hasattr(context, "_stdout_isatty_patch"): + context._copy_result = copy_to_clipboard(text) # type: ignore[attr-defined] + else: + # Redirect stdout to suppress OSC 52 escape sequences in tests. + # Use a StringIO subclass that reports isatty()=True so OSC 52 can proceed. + with patch("sys.stdout", _FakeTTY()): + context._copy_result = copy_to_clipboard(text) # type: ignore[attr-defined] + + +@when("I copy a string of {length:d} characters to clipboard") +def step_copy_long_string(context: Context, length: int) -> None: + text = "x" * length + context._copy_result = copy_to_clipboard(text) # type: ignore[attr-defined] + + +@then('the copy result method should be "{method}"') +def step_copy_method(context: Context, method: str) -> None: + expected = CopyMethod(method) + assert context._copy_result.method == expected, ( # type: ignore[attr-defined] + f"Expected {expected}, got {context._copy_result.method}" # type: ignore[attr-defined] + ) + + +@then("the copy result should be successful") +def step_copy_success(context: Context) -> None: + assert context._copy_result.success # type: ignore[attr-defined] + + +@then("the copy result should not be successful") +def step_copy_not_success(context: Context) -> None: + assert not context._copy_result.success # type: ignore[attr-defined] + + +@then('the copy result message should contain "{substring}"') +def step_copy_message_contains(context: Context, substring: str) -> None: + assert substring in context._copy_result.message # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Argument validation +# --------------------------------------------------------------------------- +@when("I try to open the menu with a non-BlockInfo argument") +def step_open_menu_bad_arg(context: Context) -> None: + context._menu = context._bcm_cls() # type: ignore[attr-defined] + try: + context._menu.open_menu("not a block info") # type: ignore[arg-type, attr-defined] + context._raised_error = None # type: ignore[attr-defined] + except TypeError as exc: + context._raised_error = exc # type: ignore[attr-defined] + + +@then('a bcm TypeError should be raised with message containing "{substring}"') +def step_type_error_raised(context: Context, substring: str) -> None: + assert isinstance(context._raised_error, TypeError), ( # type: ignore[attr-defined] + f"Expected TypeError, got {type(context._raised_error)}" # type: ignore[attr-defined] + ) + assert substring in str(context._raised_error) # type: ignore[attr-defined] + + +@when("I try to handle an empty key") +def step_handle_empty_key(context: Context) -> None: + try: + context._menu.handle_key("") # type: ignore[attr-defined] + context._raised_error = None # type: ignore[attr-defined] + except ValueError as exc: + context._raised_error = exc # type: ignore[attr-defined] + + +@then('a bcm ValueError should be raised with message containing "{substring}"') +def step_value_error_raised(context: Context, substring: str) -> None: + assert isinstance(context._raised_error, ValueError), ( # type: ignore[attr-defined] + f"Expected ValueError, got {type(context._raised_error)}" # type: ignore[attr-defined] + ) + assert substring in str(context._raised_error) # type: ignore[attr-defined] + + +@when("I try to copy a non-string value to the clipboard utility") +def step_copy_non_string(context: Context) -> None: + try: + copy_to_clipboard(12345) # type: ignore[arg-type] + context._raised_error = None # type: ignore[attr-defined] + except TypeError as exc: + context._raised_error = exc # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Menu state management +# --------------------------------------------------------------------------- +@when("I dismiss the context menu") +def step_dismiss(context: Context) -> None: + context._menu.dismiss() # type: ignore[attr-defined] + + +@then("the block info should be None") +def step_block_info_none(context: Context) -> None: + assert context._menu.block_info is None # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Direct action execution (bypassing menu key handling) +# --------------------------------------------------------------------------- +@when("I directly execute retry on the block") +def step_direct_retry(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + context._action_result = menu._action_retry(context._block_info) # type: ignore[attr-defined] + + +@then("the retry result should not be successful") +def step_retry_not_success(context: Context) -> None: + assert not context._action_result.success # type: ignore[attr-defined] + + +@when("I call get_applicable_actions with no block focused") +def step_get_actions_no_block(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + context._applicable_actions = menu.get_applicable_actions() # type: ignore[attr-defined] + + +@then("the applicable actions list should be empty") +def step_actions_empty(context: Context) -> None: + assert len(context._applicable_actions) == 0 # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# BlockType enum +# --------------------------------------------------------------------------- +@then("the BlockType enum should have {count:d} members") +def step_block_type_count(context: Context, count: int) -> None: + assert len(BlockType) == count, f"Expected {count} members, got {len(BlockType)}" + + +# --------------------------------------------------------------------------- +# Export error handling +# --------------------------------------------------------------------------- +@given("the export directory is not writable") +def step_unwritable_dir(context: Context) -> None: + """Set CWD to a read-only temp dir.""" + context._temp_dir = tempfile.mkdtemp() # type: ignore[attr-defined] + context._orig_cwd = os.getcwd() # type: ignore[attr-defined] + os.chdir(context._temp_dir) # type: ignore[attr-defined] + os.chmod(context._temp_dir, 0o444) # type: ignore[attr-defined] + + def _cleanup() -> None: + os.chmod(context._temp_dir, 0o755) # type: ignore[attr-defined] + os.chdir(context._orig_cwd) # type: ignore[attr-defined] + shutil.rmtree(context._temp_dir, ignore_errors=True) # type: ignore[attr-defined] + + context.add_cleanup(_cleanup) + + +@when("I directly execute export markdown on the block") +def step_direct_export_md(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + context._action_result = menu._action_export_markdown(context._block_info) # type: ignore[attr-defined] + + +@then("the export result should not be successful") +def step_export_not_success(context: Context) -> None: + assert not context._action_result.success # type: ignore[attr-defined] + + +@then('the export result message should contain "{substring}"') +def step_export_message_contains(context: Context, substring: str) -> None: + assert substring in context._action_result.message # type: ignore[attr-defined] + + +@when("I directly execute export SVG on the block") +def step_direct_export_svg(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + with patch("webbrowser.open"): + context._action_result = menu._action_export_svg(context._block_info) # type: ignore[attr-defined] + + +@then("the SVG export result should not be successful") +def step_svg_export_not_success(context: Context) -> None: + assert not context._action_result.success # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Render menu coverage +# --------------------------------------------------------------------------- +@when("I call render menu with no block info") +def step_render_no_block(context: Context) -> None: + context._menu = context._bcm_cls() # type: ignore[attr-defined] + context._menu._render_menu() # type: ignore[attr-defined] + + +@then("the menu text should be empty") +def step_menu_text_empty(context: Context) -> None: + assert context._menu._text == "" # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# SVG XML escape coverage +# --------------------------------------------------------------------------- +@when("I directly execute export SVG on the block in a writable directory") +def step_direct_export_svg_writable(context: Context) -> None: + context._temp_dir = tempfile.mkdtemp() # type: ignore[attr-defined] + context._orig_cwd = os.getcwd() # type: ignore[attr-defined] + os.chdir(context._temp_dir) # type: ignore[attr-defined] + + def _cleanup() -> None: + os.chdir(context._orig_cwd) # type: ignore[attr-defined] + shutil.rmtree(context._temp_dir, ignore_errors=True) # type: ignore[attr-defined] + + context.add_cleanup(_cleanup) + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + with patch("webbrowser.open"): + context._action_result = menu._action_export_svg(context._block_info) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Property accessor coverage +# --------------------------------------------------------------------------- +@when("I read the actions property") +def step_read_actions(context: Context) -> None: + context._actions_list = context._menu.actions # type: ignore[attr-defined] + + +@then("the actions list should have {count:d} entries") +def step_actions_count(context: Context, count: int) -> None: + assert len(context._actions_list) == count, ( # type: ignore[attr-defined] + f"Expected {count}, got {len(context._actions_list)}" # type: ignore[attr-defined] + ) + + +@when("I read the last_result property") +def step_read_last_result(context: Context) -> None: + context._last_result_value = context._menu.last_result # type: ignore[attr-defined] + + +@then("the last result should be None initially") +def step_last_result_none(context: Context) -> None: + assert context._last_result_value is None # type: ignore[attr-defined] + + +@when("I read the flash_message property") +def step_read_flash(context: Context) -> None: + context._flash_value = context._menu.flash_message # type: ignore[attr-defined] + + +@then("the flash message should be empty") +def step_flash_empty(context: Context) -> None: + assert context._flash_value == "" # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# handle_key with no block focused +# --------------------------------------------------------------------------- +@when('I handle key "{key}" on a menu with no block focused') +def step_handle_key_no_block(context: Context, key: str) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + # Don't open menu — no block info set + menu._visible = True + context._key_result = menu.handle_key(key) # type: ignore[attr-defined] + + +@then("the key result should be None") +def step_key_result_none(context: Context) -> None: + assert context._key_result is None # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Clipboard pyperclip success path +# --------------------------------------------------------------------------- +@given("pyperclip is available and working") +def step_pyperclip_available(context: Context) -> None: + from cleveragents.tui.clipboard import CopyMethod, CopyResult + + mock_result = CopyResult( + success=True, + method=CopyMethod.PYPERCLIP, + message="Copied to clipboard", + ) + context._pyperclip_patch = patch( # type: ignore[attr-defined] + "cleveragents.tui.clipboard._try_pyperclip", return_value=mock_result + ) + context._pyperclip_patch.start() # type: ignore[attr-defined] + context.add_cleanup(context._pyperclip_patch.stop) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# OSC 52 exception path — patch sys.stdout.write to raise OSError +# --------------------------------------------------------------------------- +@given("stdout write raises an exception") +def step_stdout_raises(context: Context) -> None: + context._stdout_isatty_patch = patch( # type: ignore[attr-defined] + "sys.stdout.isatty", return_value=True + ) + context._osc52_patch = patch( # type: ignore[attr-defined] + "sys.stdout.write", side_effect=OSError("write failed") + ) + context._stdout_isatty_patch.start() # type: ignore[attr-defined] + context._osc52_patch.start() # type: ignore[attr-defined] + context.add_cleanup(context._osc52_patch.stop) # type: ignore[attr-defined] + context.add_cleanup(context._stdout_isatty_patch.stop) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Copy clipboard flash message (m6 fix: verify flash_message was set) +# --------------------------------------------------------------------------- +@then("the copy flash message should have been set") +def step_flash_set(context: Context) -> None: + # After the action, the menu was dismissed but result was captured + assert context._action_result is not None # type: ignore[attr-defined] + assert not context._action_result.success # type: ignore[attr-defined] + # Verify flash_message was actually populated before dismiss + assert context._action_result.message != "" # type: ignore[attr-defined] + assert "Clipboard unavailable" in context._action_result.message # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Retry with empty preceding prompt +# --------------------------------------------------------------------------- +@when("I directly execute retry on a block with empty prompt") +def step_retry_empty_prompt(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + context._action_result = menu._action_retry(context._block_info) # type: ignore[attr-defined] + + +@then("the retry empty prompt result should not be successful") +def step_retry_empty_not_success(context: Context) -> None: + assert not context._action_result.success # type: ignore[attr-defined] + assert "No preceding prompt" in context._action_result.message # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# M3: Filename collision handling +# --------------------------------------------------------------------------- +@given('a Markdown export file already exists for block "{block_id}"') +def step_create_existing_file(context: Context, block_id: str) -> None: + context._temp_dir = tempfile.mkdtemp() # type: ignore[attr-defined] + context._orig_cwd = os.getcwd() # type: ignore[attr-defined] + os.chdir(context._temp_dir) # type: ignore[attr-defined] + # Create the file that would be the default export name + existing = os.path.join(context._temp_dir, f"block-{block_id}.md") # type: ignore[attr-defined] + with open(existing, "w") as f: + f.write("already exists") + + def _cleanup() -> None: + os.chdir(context._orig_cwd) # type: ignore[attr-defined] + shutil.rmtree(context._temp_dir, ignore_errors=True) # type: ignore[attr-defined] + + context.add_cleanup(_cleanup) + + +@when("I export as Markdown for the block") +def step_export_md_for_block(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + context._action_result = menu._action_export_markdown(context._block_info) # type: ignore[attr-defined] + + +@then("the exported file should have a numeric suffix") +def step_file_has_numeric_suffix(context: Context) -> None: + filepath: str = context._action_result.data # type: ignore[attr-defined] + basename = os.path.basename(filepath) + # Should be like "block-dup-block-1.md" + assert "-1" in basename, f"Expected numeric suffix in '{basename}'" + + +# --------------------------------------------------------------------------- +# M4: Path traversal prevention +# --------------------------------------------------------------------------- +@then("the exported filename should not contain path separators") +def step_filename_no_separators(context: Context) -> None: + filepath: str = context._action_result.data # type: ignore[attr-defined] + basename = os.path.basename(filepath) + assert "/" not in basename, f"Filename contains '/': {basename}" + assert ".." not in basename, f"Filename contains '..': {basename}" + + +@when("I check safe_export_path with a traversal path") +def step_check_safe_export_traversal(context: Context) -> None: + context._safe_path_result = safe_export_path("../../etc/passwd") # type: ignore[attr-defined] + + +@then("the safe export result should be None") +def step_safe_export_none(context: Context) -> None: + assert context._safe_path_result is None # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# M5: SVG truncation test +# --------------------------------------------------------------------------- +use_step_matcher("re") + + +@given( + r'bcm block "(?P[^"]+)" text of (?P\d+) characters id "(?P[^"]+)"' +) +def step_bcm_block_long_text( + context: Context, + btype: str, + length: str, + bid: str, +) -> None: + bt_cls: Any = _bcm_mod.BlockType + bi_cls: Any = _bcm_mod.BlockInfo + friendly_map: dict[str, Any] = block_type_friendly_map(bt_cls) + context._block_info = bi_cls( # type: ignore[attr-defined] + block_type=friendly_map[btype], + text_content="A" * int(length), + block_id=bid, + ) + + +use_step_matcher("parse") + + +# --------------------------------------------------------------------------- +# M6: XML control character stripping +# --------------------------------------------------------------------------- +@given('bcm block with control characters id "{block_id}"') +def step_bcm_block_control_chars(context: Context, block_id: str) -> None: + bi_cls: Any = _bcm_mod.BlockInfo + bt_cls: Any = _bcm_mod.BlockType + # Include actual control characters: NUL, BEL, BS + text = "hello\x00\x07\x08world" + context._block_info = bi_cls( # type: ignore[attr-defined] + block_type=bt_cls.USER_INPUT, + text_content=text, + block_id=block_id, + ) + + +@given('bcm block with del and c1 characters id "{block_id}"') +def step_bcm_block_del_c1_chars(context: Context, block_id: str) -> None: + bi_cls: Any = _bcm_mod.BlockInfo + bt_cls: Any = _bcm_mod.BlockType + # Include DEL (0x7f) and C1 controls (0x80-0x9f) + text = "safe\x7f\x80\x8f\x9ftext" + context._block_info = bi_cls( # type: ignore[attr-defined] + block_type=bt_cls.USER_INPUT, + text_content=text, + block_id=block_id, + ) + + +@then("the exported SVG file should not contain control characters") +def step_svg_no_control_chars(context: Context) -> None: + filepath: str = context._action_result.data # type: ignore[attr-defined] + assert os.path.exists(filepath), f"File not found: {filepath}" + with open(filepath, "rb") as f: + content = f.read() + # Check no C0 control chars (except tab, LF, CR which are valid XML) + for byte_val in content: + if byte_val < 0x09: + raise AssertionError(f"Found control char 0x{byte_val:02x} in SVG") + if byte_val in (0x0B, 0x0C): + raise AssertionError(f"Found control char 0x{byte_val:02x} in SVG") + if 0x0E <= byte_val <= 0x1F: + raise AssertionError(f"Found control char 0x{byte_val:02x} in SVG") + if byte_val == 0x7F: + raise AssertionError("Found DEL (0x7f) in SVG") + + +# --------------------------------------------------------------------------- +# M7: OSC 52 payload size limit +# --------------------------------------------------------------------------- +@when("I copy a string exceeding the OSC 52 size limit to clipboard") +def step_copy_exceeding_osc52(context: Context) -> None: + # The limit is 100KB on the base64-encoded output. + # 75001 bytes of raw data produces >100KB of base64. + text = "x" * 75_001 + with patch("sys.stdout", _FakeTTY()): + context._copy_result = copy_to_clipboard(text) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Major 1: Literal quote characters in block text for XML escape test +# --------------------------------------------------------------------------- +@given('bcm block with literal quotes id "{block_id}"') +def step_bcm_block_literal_quotes(context: Context, block_id: str) -> None: + bi_cls: Any = _bcm_mod.BlockInfo + bt_cls: Any = _bcm_mod.BlockType + # Literal double-quote and single-quote (apostrophe) characters. + text = "say \"hello\" and 'world'" + context._block_info = bi_cls( # type: ignore[attr-defined] + block_type=bt_cls.USER_INPUT, + text_content=text, + block_id=block_id, + ) + + +# --------------------------------------------------------------------------- +# m7: unique_filepath saturation (OSError) +# --------------------------------------------------------------------------- +@when("I call unique_filepath with all candidates existing") +def step_unique_filepath_saturated(context: Context) -> None: + context._temp_dir = tempfile.mkdtemp() # type: ignore[attr-defined] + context._orig_cwd = os.getcwd() # type: ignore[attr-defined] + os.chdir(context._temp_dir) # type: ignore[attr-defined] + + def _cleanup() -> None: + os.chdir(context._orig_cwd) # type: ignore[attr-defined] + shutil.rmtree(context._temp_dir, ignore_errors=True) # type: ignore[attr-defined] + + context.add_cleanup(_cleanup) + + filepath = os.path.join(context._temp_dir, "test.md") # type: ignore[attr-defined] + # Patch os.path.lexists to always return True so no unique name is found. + with patch( + "cleveragents.tui.widgets._export_utils.os.path.lexists", return_value=True + ): + try: + unique_filepath(filepath) + context._raised_error = None # type: ignore[attr-defined] + except OSError as exc: + context._raised_error = exc # type: ignore[attr-defined] + + +@then("a unique_filepath OSError should be raised") +def step_unique_filepath_oserror(context: Context) -> None: + assert isinstance(context._raised_error, OSError), ( # type: ignore[attr-defined] + f"Expected OSError, got {type(context._raised_error)}" # type: ignore[attr-defined] + ) + assert "unique filename" in str(context._raised_error) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# m8: safe_export_path returning None in export action branches +# --------------------------------------------------------------------------- +@when("I directly execute export markdown with safe_export_path returning None") +def step_export_md_safe_path_none(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + with patch( + "cleveragents.tui.widgets.block_context_menu.safe_export_path", + return_value=None, + ): + context._action_result = menu._action_export_markdown(context._block_info) # type: ignore[attr-defined] + + +@when("I directly execute export SVG with safe_export_path returning None") +def step_export_svg_safe_path_none(context: Context) -> None: + menu: Any = context._bcm_cls() # type: ignore[attr-defined] + with patch( + "cleveragents.tui.widgets.block_context_menu.safe_export_path", + return_value=None, + ): + context._action_result = menu._action_export_svg(context._block_info) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# m9: _try_osc52 with isatty()=False +# --------------------------------------------------------------------------- +@given("stdout is not a TTY") +def step_stdout_not_tty(context: Context) -> None: + context._isatty_patch = patch( # type: ignore[attr-defined] + "sys.stdout.isatty", return_value=False + ) + context._isatty_patch.start() # type: ignore[attr-defined] + context.add_cleanup(context._isatty_patch.stop) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# m8: _try_pyperclip importlib-level testing +# --------------------------------------------------------------------------- +@given("pyperclip is importable and copy succeeds") +def step_pyperclip_importable_ok(context: Context) -> None: + import types + + fake_pyperclip = types.ModuleType("pyperclip") + fake_pyperclip.copy = lambda text: None # type: ignore[attr-defined] + + context._pyperclip_avail_patch = patch( # type: ignore[attr-defined] + "cleveragents.tui.clipboard._PYPERCLIP_AVAILABLE", True + ) + context._pyperclip_import_patch = patch( # type: ignore[attr-defined] + "cleveragents.tui.clipboard.importlib.import_module", + return_value=fake_pyperclip, + ) + context._pyperclip_avail_patch.start() # type: ignore[attr-defined] + context._pyperclip_import_patch.start() # type: ignore[attr-defined] + context.add_cleanup(context._pyperclip_import_patch.stop) # type: ignore[attr-defined] + context.add_cleanup(context._pyperclip_avail_patch.stop) # type: ignore[attr-defined] + + +@given("pyperclip is importable but raises ImportError") +def step_pyperclip_importable_raises(context: Context) -> None: + context._pyperclip_avail_patch = patch( # type: ignore[attr-defined] + "cleveragents.tui.clipboard._PYPERCLIP_AVAILABLE", True + ) + context._pyperclip_import_patch = patch( # type: ignore[attr-defined] + "cleveragents.tui.clipboard.importlib.import_module", + side_effect=ImportError("no pyperclip"), + ) + context._pyperclip_avail_patch.start() # type: ignore[attr-defined] + context._pyperclip_import_patch.start() # type: ignore[attr-defined] + context.add_cleanup(context._pyperclip_import_patch.stop) # type: ignore[attr-defined] + context.add_cleanup(context._pyperclip_avail_patch.stop) # type: ignore[attr-defined] + + +@when('I call _try_pyperclip with "{text}"') +def step_call_try_pyperclip(context: Context, text: str) -> None: + context._pyperclip_try_result = _try_pyperclip(text) # type: ignore[attr-defined] + + +@then("the pyperclip try result should be successful") +def step_pyperclip_try_success(context: Context) -> None: + assert context._pyperclip_try_result is not None # type: ignore[attr-defined] + assert context._pyperclip_try_result.success # type: ignore[attr-defined] + + +@then('the pyperclip try result method should be "{method}"') +def step_pyperclip_try_method(context: Context, method: str) -> None: + expected = CopyMethod(method) + assert context._pyperclip_try_result.method == expected # type: ignore[attr-defined] + + +@then("the pyperclip try result should be None") +def step_pyperclip_try_none(context: Context) -> None: + assert context._pyperclip_try_result is None # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# m10: Hidden inapplicable actions not in rendered menu text +# --------------------------------------------------------------------------- +@then('the menu text should not contain "{substring}"') +def step_menu_not_contains(context: Context, substring: str) -> None: + assert substring not in context._menu._text, ( # type: ignore[attr-defined] + f"Expected '{substring}' NOT in menu text, but found it" + ) diff --git a/features/steps/tui_block_context_menu_steps.py b/features/steps/tui_block_context_menu_steps.py new file mode 100644 index 000000000..21ebe98cf --- /dev/null +++ b/features/steps/tui_block_context_menu_steps.py @@ -0,0 +1,225 @@ +"""Step definitions for tui_block_context_menu.feature. + +Tests the 7 context menu actions: copy clipboard, copy to prompt, +export markdown, export SVG, maximize/restore, retry, and show raw data. +""" + +from __future__ import annotations + +import importlib +import os +import shutil +import tempfile +from typing import Any +from unittest.mock import patch + +from behave import given, then, use_step_matcher, when +from behave.runner import Context + +import cleveragents.tui.widgets.block_context_menu as _bcm_mod +from cleveragents.tui.widgets.block_context_menu import BlockType +from features.mocks.block_context_menu_helpers import block_type_friendly_map + +use_step_matcher("re") + + +# --------------------------------------------------------------------------- +# Background — force fallback so tests never need a live Textual app +# --------------------------------------------------------------------------- +@given(r"the block context menu module is imported with fallback") +def step_import_module(context: Context) -> None: + """Reload the module with textual import patched out to force fallback.""" + with patch("importlib.import_module", side_effect=ImportError("no textual")): + importlib.reload(_bcm_mod) + context._bcm_cls = _bcm_mod.BlockContextMenu # type: ignore[attr-defined] + context._block_info_cls = _bcm_mod.BlockInfo # type: ignore[attr-defined] + context._block_type = _bcm_mod.BlockType # type: ignore[attr-defined] + + def _restore() -> None: + importlib.reload(_bcm_mod) + + context.add_cleanup(_restore) + context._menu = None # type: ignore[attr-defined] + context._action_result = None # type: ignore[attr-defined] + context._temp_dir = None # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Block info creation — single regex step handling all variants +# --------------------------------------------------------------------------- +@given( + r'bcm block "(?P[^"]+)" text "(?P[^"]*)"' + r'(?: id "(?P[^"]+)")?' + r"(?: expanded (?Ptrue|false))?" + r'(?: prompt "(?P[^"]*)")?' + r"(?: raw '(?P[^']*)')?" + r"(?: no raw data)?" +) +def step_bcm_block( + context: Context, + block_type: str, + text: str, + block_id: str | None, + expanded: str | None, + prompt: str | None, + raw: str | None, +) -> None: + block_id = block_id or "" + is_expanded = expanded == "true" if expanded else False + prompt = prompt or "" + raw_data = raw or "" + # Use the reloaded module's BlockInfo to avoid isinstance mismatch + # after importlib.reload() in the Background step. + bi_cls: Any = _bcm_mod.BlockInfo + bt_cls: Any = _bcm_mod.BlockType + friendly_map: dict[str, BlockType] = block_type_friendly_map(bt_cls) + context._block_info = bi_cls( # type: ignore[attr-defined] + block_type=friendly_map[block_type], + text_content=text, + block_id=block_id, + is_expanded=is_expanded, + preceding_prompt=prompt, + raw_data=raw_data, + ) + + +# Reset to default matcher for remaining steps +use_step_matcher("parse") + + +# --------------------------------------------------------------------------- +# Menu open/close +# --------------------------------------------------------------------------- +@when("I open the context menu with that block info") +def step_open_menu(context: Context) -> None: + context._menu = context._bcm_cls() # type: ignore[attr-defined] + context._menu.open_menu(context._block_info) # type: ignore[attr-defined] + + +@given("the context menu is open with that block info") +def step_given_menu_open(context: Context) -> None: + context._menu = context._bcm_cls() # type: ignore[attr-defined] + context._menu.open_menu(context._block_info) # type: ignore[attr-defined] + # For export actions, set up a temporary directory + context._temp_dir = tempfile.mkdtemp() # type: ignore[attr-defined] + context._orig_cwd = os.getcwd() # type: ignore[attr-defined] + os.chdir(context._temp_dir) # type: ignore[attr-defined] + + def _cleanup() -> None: + os.chdir(context._orig_cwd) # type: ignore[attr-defined] + if context._temp_dir is not None: # type: ignore[attr-defined] + shutil.rmtree(context._temp_dir, ignore_errors=True) # type: ignore[attr-defined] + + context.add_cleanup(_cleanup) + + +# --------------------------------------------------------------------------- +# Key press handling +# --------------------------------------------------------------------------- +@when('I press "{key}" on the context menu') +def step_press_key(context: Context, key: str) -> None: + with patch("webbrowser.open") as mock_browser: + context._action_result = context._menu.handle_key(key) # type: ignore[attr-defined] + context._mock_browser = mock_browser # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Assertions — menu state +# --------------------------------------------------------------------------- +@then("the context menu should be visible") +def step_menu_visible(context: Context) -> None: + assert context._menu.visible, "Expected menu to be visible" # type: ignore[attr-defined] + + +@then("the context menu should not be visible") +def step_menu_not_visible(context: Context) -> None: + assert not context._menu.visible, "Expected menu to not be visible" # type: ignore[attr-defined] + + +@then('the menu text should contain "{substring}"') +def step_menu_contains(context: Context, substring: str) -> None: + assert substring in context._menu._text, ( # type: ignore[attr-defined] + f"Expected '{substring}' in:\n{context._menu._text}" # type: ignore[attr-defined] + ) + + +# --------------------------------------------------------------------------- +# Assertions — action result +# --------------------------------------------------------------------------- +@then("the last action result should be None") +def step_result_is_none(context: Context) -> None: + assert context._action_result is None, ( # type: ignore[attr-defined] + f"Expected None, got {context._action_result}" # type: ignore[attr-defined] + ) + + +@then('the action result should have key "{key}"') +def step_result_key(context: Context, key: str) -> None: + assert context._action_result is not None, "Action result is None" # type: ignore[attr-defined] + assert context._action_result.action_key == key # type: ignore[attr-defined] + + +@then('the action result data should be "{expected}"') +def step_result_data(context: Context, expected: str) -> None: + assert context._action_result is not None # type: ignore[attr-defined] + assert context._action_result.data == expected, ( # type: ignore[attr-defined] + f"Expected '{expected}', got '{context._action_result.data}'" # type: ignore[attr-defined] + ) + + +@then("the action result should be successful") +def step_result_success(context: Context) -> None: + assert context._action_result is not None # type: ignore[attr-defined] + assert context._action_result.success, ( # type: ignore[attr-defined] + f"Expected success, got: {context._action_result.message}" # type: ignore[attr-defined] + ) + + +@then('the action result message should be "{expected}"') +def step_result_message(context: Context, expected: str) -> None: + assert context._action_result is not None # type: ignore[attr-defined] + assert context._action_result.message == expected # type: ignore[attr-defined] + + +@then('the action result message should contain "{substring}"') +def step_result_message_contains(context: Context, substring: str) -> None: + assert context._action_result is not None # type: ignore[attr-defined] + assert substring in context._action_result.message # type: ignore[attr-defined] + + +@then('the action result data should contain "{substring}"') +def step_result_data_contains(context: Context, substring: str) -> None: + assert context._action_result is not None # type: ignore[attr-defined] + assert substring in context._action_result.data # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Browser open assertions +# --------------------------------------------------------------------------- +@then("webbrowser.open should have been called with a file URL") +def step_browser_called_file_url(context: Context) -> None: + mock_browser: Any = context._mock_browser # type: ignore[attr-defined] + mock_browser.assert_called_once() + url: str = mock_browser.call_args[0][0] + assert url.startswith("file://"), f"Expected file:// URL, got: {url}" + + +# --------------------------------------------------------------------------- +# File export assertions +# --------------------------------------------------------------------------- +@then('the exported markdown file should contain "{content}"') +def step_exported_md_contains(context: Context, content: str) -> None: + filepath: str = context._action_result.data # type: ignore[attr-defined] + assert os.path.exists(filepath), f"File not found: {filepath}" + with open(filepath) as f: + text = f.read() + assert content in text, f"Expected '{content}' in file content" + + +@then('the exported SVG file should contain "{content}"') +def step_exported_svg_contains(context: Context, content: str) -> None: + filepath: str = context._action_result.data # type: ignore[attr-defined] + assert os.path.exists(filepath), f"File not found: {filepath}" + with open(filepath) as f: + text = f.read() + assert content in text, f"Expected '{content}' in file content" diff --git a/features/tui_block_context_menu.feature b/features/tui_block_context_menu.feature new file mode 100644 index 000000000..6d583a8ad --- /dev/null +++ b/features/tui_block_context_menu.feature @@ -0,0 +1,134 @@ +Feature: TUI Block Context Menu + The block context menu provides 7 actions for interacting with + conversation blocks: copy, export, maximize, retry, and raw data. + + Background: + Given the block context menu module is imported with fallback + + # --- Menu lifecycle --- + + Scenario: Context menu opens when given a block info + Given bcm block "ActorResponse" text "Hello world" + When I open the context menu with that block info + Then the context menu should be visible + And the menu text should contain "Block Actions" + + Scenario: Context menu displays 6 applicable actions for ActorResponse + Given bcm block "ActorResponse" text "Response content" + When I open the context menu with that block info + Then the menu text should contain "c Copy to clipboard" + And the menu text should contain "p Copy to prompt" + And the menu text should contain "e Export as Markdown" + And the menu text should contain "v Export as SVG" + And the menu text should contain "r Retry" + And the menu text should contain "d Show raw data" + And the menu text should contain "escape Dismiss" + + Scenario: Escape dismisses the menu without performing any action + Given bcm block "ActorResponse" text "Some content" + And the context menu is open with that block info + When I press "escape" on the context menu + Then the context menu should not be visible + And the last action result should be None + + # --- Copy to clipboard action (c) --- + + Scenario: Copy to clipboard action copies block text content + Given bcm block "ActorResponse" text "Copy me" + And the context menu is open with that block info + And pyperclip is available and working + When I press "c" on the context menu + Then the action result should have key "c" + And the action result should be successful + And the action result data should be "Copy me" + And the context menu should not be visible + + # --- Copy to prompt action (p) --- + + Scenario: Copy to prompt action returns block text content + Given bcm block "UserInput" text "Insert into prompt" + And the context menu is open with that block info + When I press "p" on the context menu + Then the action result should have key "p" + And the action result should be successful + And the action result message should be "Content copied to prompt" + And the action result data should be "Insert into prompt" + + # --- Export as Markdown action (e) --- + + Scenario: Export as Markdown writes a .md file + Given bcm block "ToolCall" text "# Tool output" id "tc-001" + And the context menu is open with that block info + When I press "e" on the context menu + Then the action result should have key "e" + And the action result should be successful + And the action result message should contain "Exported to" + And the exported markdown file should contain "# Tool output" + + # --- Export as SVG action (v) --- + + Scenario: Export as SVG writes an SVG file and opens in browser + Given bcm block "ActorThought" text "Thinking..." id "at-002" + And the context menu is open with that block info + When I press "v" on the context menu + Then the action result should have key "v" + And the action result should be successful + And the action result message should contain "SVG exported" + And the exported SVG file should contain " d" id "xml-esc" + When I directly execute export SVG on the block in a writable directory + Then the exported SVG file should contain "<" + And the exported SVG file should contain "&" + And the exported SVG file should contain ">" + + Scenario: SVG export escapes quote characters + Given bcm block with literal quotes id "xml-quotes" + When I directly execute export SVG on the block in a writable directory + Then the exported SVG file should contain """ + And the exported SVG file should contain "'" + + Scenario: SVG export strips XML control characters + Given bcm block with control characters id "xml-ctrl" + When I directly execute export SVG on the block in a writable directory + Then the exported SVG file should not contain control characters + And the exported SVG file should contain "helloworld" + + Scenario: SVG export strips DEL and C1 control characters + Given bcm block with del and c1 characters id "xml-c1" + When I directly execute export SVG on the block in a writable directory + Then the exported SVG file should not contain control characters + And the exported SVG file should contain "safetext" + + # --- Property accessor coverage --- + + Scenario: actions property returns list of menu actions + Given bcm block "UserInput" text "Test" + And the context menu is open with that block info + When I read the actions property + Then the actions list should have 7 entries + + Scenario: last_result property is None before any action + Given bcm block "UserInput" text "Test" + And the context menu is open with that block info + When I read the last_result property + Then the last result should be None initially + + Scenario: flash_message is empty initially + Given bcm block "UserInput" text "Test" + And the context menu is open with that block info + When I read the flash_message property + Then the flash message should be empty + + # --- handle_key with no block info --- + + Scenario: handle_key returns None when no block is focused + When I handle key "c" on a menu with no block focused + Then the key result should be None + + # --- Clipboard pyperclip success path --- + + Scenario: Clipboard uses pyperclip when available + Given pyperclip is available and working + When I copy "hello" to clipboard + Then the copy result method should be "pyperclip" + And the copy result should be successful + + # --- Clipboard OSC 52 exception path --- + + Scenario: OSC 52 falls back when stdout write fails + Given pyperclip is not available + And stdout write raises an exception + When I copy "test" to clipboard + Then the copy result method should be "fallback" + + # --- Copy clipboard flash message path --- + + Scenario: Copy clipboard sets flash message on failure + Given bcm block "UserInput" text "flash content" + And the context menu is open with that block info + And pyperclip is not available + And OSC 52 is not available + When I press "c" on the context menu + Then the copy flash message should have been set + + # --- M3: Filename collision handling --- + + Scenario: Export Markdown generates unique filename when file exists + Given bcm block "UserInput" text "duplicate content" id "dup-block" + And a Markdown export file already exists for block "dup-block" + When I export as Markdown for the block + Then the action result should be successful + And the exported file should have a numeric suffix + + # --- M4: Path traversal prevention --- + + Scenario: Block ID with path traversal characters is sanitized + Given bcm block "UserInput" text "safe content" id "../../etc/passwd" + And the context menu is open with that block info + When I press "e" on the context menu + Then the action result should be successful + And the action result message should contain "Exported to" + And the exported filename should not contain path separators + + Scenario: Safe export path rejects paths escaping CWD + When I check safe_export_path with a traversal path + Then the safe export result should be None + + # --- M5: SVG content truncation --- + + Scenario: SVG export truncates content exceeding 500 characters + Given bcm block "UserInput" text of 600 characters id "long-svg" + And the context menu is open with that block info + When I press "v" on the context menu + Then the action result should be successful + And the action result message should contain "content truncated to 500 characters" + + # --- M7: OSC 52 payload size limit --- + + Scenario: OSC 52 falls back when payload exceeds size limit + Given pyperclip is not available + When I copy a string exceeding the OSC 52 size limit to clipboard + Then the copy result method should be "fallback" + + # --- m7: Empty block text export --- + + Scenario: Export Markdown succeeds with empty text content + Given bcm block "UserInput" text "" id "empty-block" + And the context menu is open with that block info + When I press "e" on the context menu + Then the action result should have key "e" + And the action result should be successful + + # --- m9: Test retry through public API --- + + Scenario: Retry action through handle_key for ActorResponse with valid prompt + Given bcm block "ActorResponse" text "Response" prompt "Ask me" + And the context menu is open with that block info + When I press "r" on the context menu + Then the action result should have key "r" + And the action result should be successful + And the action result data should be "Ask me" + + # --- m7: unique_filepath saturation (OSError) --- + + Scenario: unique_filepath raises OSError when filesystem is saturated + When I call unique_filepath with all candidates existing + Then a unique_filepath OSError should be raised + + # --- m8: safe_export_path returning None in export actions --- + + Scenario: Export Markdown returns failure when safe_export_path rejects the path + Given bcm block "UserInput" text "content" id "normal" + When I directly execute export markdown with safe_export_path returning None + Then the export result should not be successful + And the export result message should contain "Invalid block ID" + + Scenario: Export SVG returns failure when safe_export_path rejects the path + Given bcm block "UserInput" text "content" id "normal" + When I directly execute export SVG with safe_export_path returning None + Then the SVG export result should not be successful + + # --- m9: _try_osc52 with isatty()=False --- + + Scenario: OSC 52 returns None when stdout is not a TTY + Given pyperclip is not available + And stdout is not a TTY + When I copy "test" to clipboard + Then the copy result method should be "fallback" + + # --- m8: _try_pyperclip importlib internal path --- + + Scenario: _try_pyperclip succeeds when importlib.import_module returns a working module + Given pyperclip is importable and copy succeeds + When I call _try_pyperclip with "hello" + Then the pyperclip try result should be successful + And the pyperclip try result method should be "pyperclip" + + Scenario: _try_pyperclip returns None when import_module raises ImportError + Given pyperclip is importable but raises ImportError + When I call _try_pyperclip with "hello" + Then the pyperclip try result should be None + + # --- n5: DiffView maximize execution --- + + Scenario: Maximize action executes for DiffView blocks + Given bcm block "DiffView" text "diff content" expanded false + And the context menu is open with that block info + When I press "m" on the context menu + Then the action result should have key "m" + And the action result should be successful + And the action result data should be "maximized" + + # NOTE: Four-corner box-drawing verification lives in + # tui_block_context_menu.feature "Menu renders with box-drawing border characters". diff --git a/pyproject.toml b/pyproject.toml index ad642ab01..f823bae50 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,7 @@ dependencies = [ [project.optional-dependencies] tui = [ "textual>=1.0.0,<2.0.0", + "pyperclip>=1.8.0", ] dev = [ # Code formatting and linting diff --git a/src/cleveragents/tui/clipboard.py b/src/cleveragents/tui/clipboard.py new file mode 100644 index 000000000..a2d56f9e9 --- /dev/null +++ b/src/cleveragents/tui/clipboard.py @@ -0,0 +1,126 @@ +"""Clipboard utility with pyperclip -> OSC 52 -> flash fallback chain. + +Provides clipboard integration for the TUI, attempting three strategies +in order: +1. ``pyperclip`` library (system clipboard via platform-native tools) +2. OSC 52 escape sequences (terminal-native, works over SSH/tmux) +3. Flash message fallback (returns content for manual copying) +""" + +from __future__ import annotations + +import base64 +import importlib +import importlib.util +import sys +from dataclasses import dataclass +from enum import Enum + +__all__ = ["CopyMethod", "CopyResult", "copy_to_clipboard"] + +#: Maximum base64-encoded payload size (in bytes) for OSC 52 sequences. +#: Many terminals silently drop or corrupt sequences larger than ~100 KB. +#: The limit is applied to the *base64-encoded* output, not the raw text, +#: because the encoded form is what actually transits the terminal. +_OSC52_MAX_BYTES: int = 100_000 + +#: Whether the ``pyperclip`` package is importable in this environment. +_PYPERCLIP_AVAILABLE: bool = importlib.util.find_spec("pyperclip") is not None + + +class CopyMethod(Enum): + """Indicates which clipboard strategy succeeded.""" + + PYPERCLIP = "pyperclip" + OSC52 = "osc52" + FALLBACK = "fallback" + + +@dataclass(frozen=True, slots=True) +class CopyResult: + """Result of a clipboard copy operation.""" + + success: bool + method: CopyMethod + message: str + + +def _try_pyperclip(text: str) -> CopyResult | None: + """Attempt to copy via pyperclip. Returns None if unavailable. + + Uses ``importlib.import_module`` because ``pyperclip`` is an optional + dependency that may not be installed. + """ + if not _PYPERCLIP_AVAILABLE: + return None + try: + _pyperclip = importlib.import_module("pyperclip") + _pyperclip.copy(text) + return CopyResult( + success=True, + method=CopyMethod.PYPERCLIP, + message="Copied to clipboard", + ) + except (ImportError, RuntimeError, OSError): + return None + + +def _try_osc52(text: str) -> CopyResult | None: + """Attempt to copy via OSC 52 escape sequence. + + OSC 52 is supported by many modern terminals and works over SSH + and inside tmux/screen sessions. Returns None if writing to + stdout fails or the base64-encoded payload exceeds the terminal + size limit. + """ + if not sys.stdout.isatty(): + return None + # Quick pre-check: base64 output is always >= raw byte length, + # so if the raw text already exceeds the limit, skip encoding. + if len(text) > _OSC52_MAX_BYTES: + return None + encoded = base64.b64encode(text.encode("utf-8")).decode("ascii") + if len(encoded) > _OSC52_MAX_BYTES: + return None + try: + escape_seq = f"\x1b]52;c;{encoded}\x07" + sys.stdout.write(escape_seq) + sys.stdout.flush() + return CopyResult( + success=True, + method=CopyMethod.OSC52, + message="Copied to clipboard (terminal OSC 52)", + ) + except OSError: + return None + + +def _fallback(text: str) -> CopyResult: + """Return a flash-message result with the content for manual copying.""" + preview = text[:200] + ("..." if len(text) > 200 else "") + return CopyResult( + success=False, + method=CopyMethod.FALLBACK, + message=f"Clipboard unavailable. Content:\n{preview}", + ) + + +def copy_to_clipboard(text: str) -> CopyResult: + """Copy *text* to the system clipboard using the best available method. + + Tries ``pyperclip`` first, then OSC 52 escape sequences, and finally + falls back to returning the content in a flash message for manual + copying. + """ + if not isinstance(text, str): + raise TypeError(f"Expected str, got {type(text).__name__}") + + result = _try_pyperclip(text) + if result is not None: + return result + + result = _try_osc52(text) + if result is not None: + return result + + return _fallback(text) diff --git a/src/cleveragents/tui/widgets/__init__.py b/src/cleveragents/tui/widgets/__init__.py index 51776ce71..bd15c9af1 100644 --- a/src/cleveragents/tui/widgets/__init__.py +++ b/src/cleveragents/tui/widgets/__init__.py @@ -1,5 +1,15 @@ """Widget collection for CleverAgents TUI.""" +from cleveragents.tui.widgets.block_context_menu import ( + EXPANDABLE_BLOCK_TYPES, + RETRYABLE_BLOCK_TYPES, + ActionResult, + BlockContextMenu, + BlockInfo, + BlockType, + ExpandProtocol, + MenuAction, +) from cleveragents.tui.widgets.help_panel_overlay import HelpPanelOverlay from cleveragents.tui.widgets.persona_bar import PersonaBar from cleveragents.tui.widgets.prompt import PromptInput, PromptSubmitted @@ -7,7 +17,15 @@ from cleveragents.tui.widgets.reference_picker import ReferencePickerOverlay from cleveragents.tui.widgets.slash_command_overlay import SlashCommandOverlay __all__ = [ + "EXPANDABLE_BLOCK_TYPES", + "RETRYABLE_BLOCK_TYPES", + "ActionResult", + "BlockContextMenu", + "BlockInfo", + "BlockType", + "ExpandProtocol", "HelpPanelOverlay", + "MenuAction", "PersonaBar", "PromptInput", "PromptSubmitted", diff --git a/src/cleveragents/tui/widgets/_export_utils.py b/src/cleveragents/tui/widgets/_export_utils.py new file mode 100644 index 000000000..20c119eec --- /dev/null +++ b/src/cleveragents/tui/widgets/_export_utils.py @@ -0,0 +1,99 @@ +"""Utility functions for block export and file operations. + +Extracted from ``block_context_menu`` to keep module size manageable. +Provides path sanitization, safe file export, XML escaping, and +unique filepath generation. +""" + +from __future__ import annotations + +import os +import re + +__all__ = [ + "SVG_MAX_CONTENT_CHARS", + "escape_xml", + "safe_export_path", + "sanitize_block_id", + "unique_filepath", +] + +#: Maximum characters of block content embedded in an SVG export. +#: Content exceeding this limit is truncated with an indicator message. +SVG_MAX_CONTENT_CHARS: int = 500 + +#: Upper bound on numeric-suffix attempts in :func:`unique_filepath`. +#: Prevents an unbounded loop if the filesystem is saturated. +_UNIQUE_SUFFIX_LIMIT: int = 10_000 + + +def sanitize_block_id(block_id: str) -> str: + """Sanitize a block ID for safe use in filenames. + + Strips any characters that are not ASCII alphanumerics, underscores, + or hyphens, preventing path traversal attacks. The result is + truncated to 128 characters to avoid excessively long filenames. + """ + return re.sub(r"[^a-zA-Z0-9_\-]", "_", block_id)[:128] + + +def safe_export_path(filename: str) -> str | None: + """Resolve *filename* under CWD and verify it stays within CWD. + + Returns the resolved path as a string, or ``None`` if the resolved + path escapes the current working directory (path traversal attempt). + """ + cwd = os.path.realpath(os.getcwd()) + resolved = os.path.realpath(os.path.join(cwd, filename)) + if resolved == cwd: + return resolved + # Ensure the resolved path sits directly under CWD. Append os.sep + # to the CWD only when CWD is *not* the filesystem root; when CWD + # is ``/`` the trailing separator is already present. + prefix = cwd if cwd.endswith(os.sep) else cwd + os.sep + if not resolved.startswith(prefix): + return None + return resolved + + +def unique_filepath(filepath: str) -> str: + """Return *filepath* if it does not exist, otherwise append a numeric suffix. + + Avoids silently overwriting existing files by trying ``filepath``, + then ``filepath`` with ``-1``, ``-2``, etc. inserted before the + extension. + + Raises: + OSError: If no unique name is found within :data:`_UNIQUE_SUFFIX_LIMIT` + attempts (filesystem saturation). + """ + if not os.path.lexists(filepath): + return filepath + base, ext = os.path.splitext(filepath) + for counter in range(1, _UNIQUE_SUFFIX_LIMIT + 1): + candidate = f"{base}-{counter}{ext}" + # Re-validate that the suffixed candidate still resolves inside CWD. + if safe_export_path(os.path.basename(candidate)) is None: + continue + if not os.path.lexists(candidate): + return candidate + raise OSError( + f"Could not find a unique filename after {_UNIQUE_SUFFIX_LIMIT} attempts" + ) + + +def escape_xml(text: str) -> str: + """Escape text for safe embedding in XML/SVG content. + + Strips XML-invalid control characters (U+0000-U+0008, U+000B, + U+000C, U+000E-U+001F, U+007F DEL, and C1 controls U+0080-U+009F) + before replacing the five standard XML entities. + """ + text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]", "", text) + return ( + text.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + .replace("'", "'") + ) diff --git a/src/cleveragents/tui/widgets/block_context_menu.py b/src/cleveragents/tui/widgets/block_context_menu.py new file mode 100644 index 000000000..a1d698269 --- /dev/null +++ b/src/cleveragents/tui/widgets/block_context_menu.py @@ -0,0 +1,497 @@ +"""Block context menu widget for conversation block actions. + +Provides a floating overlay menu with 7 actions for interacting with +conversation blocks: copy, export, maximize, retry, and raw data display. +""" + +from __future__ import annotations + +import contextlib +import importlib +import os +import pathlib +import webbrowser +from collections.abc import Callable +from dataclasses import dataclass +from enum import Enum +from typing import Any, Protocol + +from cleveragents.tui.clipboard import copy_to_clipboard +from cleveragents.tui.widgets._export_utils import ( + SVG_MAX_CONTENT_CHARS, + escape_xml, + safe_export_path, + sanitize_block_id, + unique_filepath, +) + +__all__ = [ + "EXPANDABLE_BLOCK_TYPES", + "RETRYABLE_BLOCK_TYPES", + "ActionResult", + "BlockContextMenu", + "BlockInfo", + "BlockType", + "ExpandProtocol", + "MenuAction", +] + +#: Width of the rendered menu box (in columns). +_MENU_RENDER_WIDTH: int = 70 + + +class _FallbackStatic: + """Fallback widget base when Textual is not available. + + Provides a minimal ``update()`` API so that ``BlockContextMenu`` + can function without a live Textual application (e.g. in tests). + """ + + def __init__(self, *args: object, **kwargs: object) -> None: + self._text = "" + + def update(self, text: str) -> None: + self._text = text + + +def _load_static_base() -> type[Any]: + """Load Textual's ``Static`` widget, falling back to ``_FallbackStatic``.""" + try: + return importlib.import_module( + "textual.widgets" + ).Static # pragma: no cover — requires live Textual + except (ImportError, AttributeError): + return _FallbackStatic + + +_StaticBase = _load_static_base() + + +class BlockType(Enum): + """Conversation block types as defined in the specification.""" + + WELCOME = "Welcome" + USER_INPUT = "UserInput" + ACTOR_RESPONSE = "ActorResponse" + ACTOR_THOUGHT = "ActorThought" + TOOL_CALL = "ToolCall" + PLAN_PROGRESS = "PlanProgress" + DIFF_VIEW = "DiffView" + TERMINAL_EMBED = "TerminalEmbed" + SHELL_RESULT = "ShellResult" + NOTE = "Note" + + +#: Block types that support the ``ExpandProtocol`` (maximize/restore). +EXPANDABLE_BLOCK_TYPES: frozenset[BlockType] = frozenset( + {BlockType.ACTOR_THOUGHT, BlockType.TOOL_CALL, BlockType.DIFF_VIEW} +) + +#: Block types that support the retry action. +RETRYABLE_BLOCK_TYPES: frozenset[BlockType] = frozenset({BlockType.ACTOR_RESPONSE}) + + +class ExpandProtocol(Protocol): + """Protocol for blocks that support maximize/restore toggle. + + The caller is responsible for invoking :meth:`toggle_expand` and + updating external state accordingly — this protocol only advertises + capability; it does not trigger side effects on its own. + """ + + def toggle_expand(self) -> None: ... + + @property + def is_expanded(self) -> bool: ... + + +@dataclass(frozen=True, slots=True) +class MenuAction: + """Definition of a single context menu action.""" + + key: str + label: str + description: str + applicable_types: frozenset[BlockType] | None = None + enabled: bool = True + + +#: The ordered list of 7 context menu actions per specification. +#: Immutable and shared across all ``BlockContextMenu`` instances. +_DEFAULT_ACTIONS: tuple[MenuAction, ...] = ( + MenuAction( + key="c", + label="Copy to clipboard", + description="Copies the block's text content to the system clipboard", + applicable_types=frozenset(BlockType), + ), + MenuAction( + key="p", + label="Copy to prompt", + description="Inserts the block's text content into the prompt TextArea", + applicable_types=frozenset(BlockType), + ), + MenuAction( + key="e", + label="Export as Markdown", + description="Exports the block's content to a .md file", + applicable_types=frozenset(BlockType), + ), + MenuAction( + key="v", + label="Export as SVG (opens browser)", + description="Renders the block to SVG and opens in browser", + applicable_types=frozenset(BlockType), + ), + MenuAction( + key="m", + label="Maximize / restore block", + description="Toggles between collapsed/default and full-screen height", + applicable_types=EXPANDABLE_BLOCK_TYPES, + ), + MenuAction( + key="r", + label="Retry (ActorResponse only)", + description="Re-sends the preceding user prompt", + applicable_types=RETRYABLE_BLOCK_TYPES, + ), + MenuAction( + key="d", + label="Show raw data", + description="Shows the raw A2A message data for debugging", + applicable_types=frozenset(BlockType), + ), +) + + +@dataclass(frozen=True, slots=True) +class BlockInfo: + """Information about the currently focused conversation block. + + Attributes: + text_content: The block's displayable text. For ``ActorResponse`` + blocks this is expected to contain the Markdown source; for + all other block types it is plain text. + """ + + block_type: BlockType + text_content: str + raw_data: str = "" + block_id: str = "" + preceding_prompt: str = "" + is_expanded: bool = False + + +@dataclass(frozen=True, slots=True) +class ActionResult: + """Result of executing a context menu action.""" + + action_key: str + success: bool + message: str + data: str = "" + + +class BlockContextMenu(_StaticBase): + """Floating overlay providing 7 actions for conversation blocks. + + Displays a bordered menu titled "Block Actions" with keyboard- + triggered actions. Actions inapplicable to the focused block type + are hidden from the rendered menu. + """ + + def __init__(self, *args: object, **kwargs: object) -> None: + super().__init__(*args, **kwargs) + self._visible: bool = False + self._block_info: BlockInfo | None = None + self._actions: tuple[MenuAction, ...] = _DEFAULT_ACTIONS + self._last_result: ActionResult | None = None + self._flash_message: str = "" + self._handlers: dict[str, Callable[[BlockInfo], ActionResult]] = { + "c": self._action_copy_clipboard, + "p": self._action_copy_to_prompt, + "e": self._action_export_markdown, + "v": self._action_export_svg, + "m": self._action_maximize_restore, + "r": self._action_retry, + "d": self._action_show_raw_data, + } + + @property + def visible(self) -> bool: + """Whether the context menu is currently displayed.""" + return self._visible + + @property + def block_info(self) -> BlockInfo | None: + """The currently focused block's information.""" + return self._block_info + + @property + def actions(self) -> tuple[MenuAction, ...]: + """The full list of menu actions (immutable).""" + return self._actions + + @property + def last_result(self) -> ActionResult | None: + """The result of the most recently executed action.""" + return self._last_result + + @property + def flash_message(self) -> str: + """The current flash message, if any.""" + return self._flash_message + + def open_menu(self, block_info: BlockInfo) -> None: + """Open the context menu for the given block. + + Args: + block_info: Information about the focused conversation block. + + Raises: + TypeError: If *block_info* is not a ``BlockInfo`` instance. + """ + if not isinstance(block_info, BlockInfo): + raise TypeError(f"Expected BlockInfo, got {type(block_info).__name__}") + self._block_info = block_info + self._visible = True + self._last_result = None + self._flash_message = "" + self._render_menu() + + def dismiss(self) -> None: + """Dismiss the context menu without performing any action.""" + self._visible = False + self._block_info = None + self.update("") + + def get_applicable_actions(self) -> list[MenuAction]: + """Return actions applicable to the currently focused block type. + + Actions whose ``applicable_types`` does not include the current + block type are excluded. + """ + if self._block_info is None: + return [] + block_type = self._block_info.block_type + return [ + action + for action in self._actions + if action.applicable_types is None or block_type in action.applicable_types + ] + + def handle_key(self, key: str) -> ActionResult | None: + """Handle a keypress within the context menu. + + Args: + key: The key that was pressed. + + Returns: + An ``ActionResult`` if an action was executed, or ``None`` + if the key was ``escape`` (menu dismissed) or unrecognised. + + Raises: + ValueError: If *key* is empty. + """ + if not key: + raise ValueError("Key must not be empty") + if key == "escape": + self.dismiss() + return None + if self._block_info is None: + return None + + applicable = {a.key: a for a in self.get_applicable_actions()} + if key not in applicable: + return None + + result = self._execute_action(key, self._block_info) + self._last_result = result + self.dismiss() + return result + + def _execute_action(self, key: str, block: BlockInfo) -> ActionResult: + """Dispatch to the handler for the given action key.""" + handler = self._handlers.get(key) + if handler is None: # pragma: no cover — unreachable: caller filters keys + return ActionResult( + action_key=key, + success=False, + message=f"Unknown action: {key}", + ) + result: ActionResult = handler(block) + return result + + def _action_copy_clipboard(self, block: BlockInfo) -> ActionResult: + """Copy block text content to the system clipboard.""" + result = copy_to_clipboard(block.text_content) + if not result.success: + self._flash_message = result.message + return ActionResult( + action_key="c", + success=result.success, + message=result.message, + data=block.text_content, + ) + + def _action_copy_to_prompt(self, block: BlockInfo) -> ActionResult: + """Insert block text content into the prompt TextArea.""" + return ActionResult( + action_key="p", + success=True, + message="Content copied to prompt", + data=block.text_content, + ) + + def _action_export_markdown(self, block: BlockInfo) -> ActionResult: + """Export block content to a ``.md`` file in the current directory.""" + safe_id = sanitize_block_id(block.block_id) or "export" + filename = f"block-{safe_id}.md" + resolved = safe_export_path(filename) + if resolved is None: + return ActionResult( + action_key="e", + success=False, + message="Invalid block ID for export", + ) + try: + filepath = unique_filepath(resolved) + with open(filepath, "x", encoding="utf-8") as f: + f.write(block.text_content) + actual_name = os.path.basename(filepath) + return ActionResult( + action_key="e", + success=True, + message=f"Exported to {actual_name}", + data=filepath, + ) + except OSError as exc: + return ActionResult( + action_key="e", + success=False, + message=f"Export failed: {exc}", + ) + + def _action_export_svg(self, block: BlockInfo) -> ActionResult: + """Render block to SVG and open in default browser.""" + safe_id = sanitize_block_id(block.block_id) or "export" + filename = f"block-{safe_id}.svg" + resolved = safe_export_path(filename) + if resolved is None: + return ActionResult( + action_key="v", + success=False, + message="Invalid block ID for SVG export", + ) + content = block.text_content + truncated = len(content) > SVG_MAX_CONTENT_CHARS + display_content = content[:SVG_MAX_CONTENT_CHARS] + svg_content = ( + '\n' + '\n' + ' \n' + f" {escape_xml(display_content)}\n" + " \n" + "\n" + ) + try: + filepath = unique_filepath(resolved) + with open(filepath, "x", encoding="utf-8") as f: + f.write(svg_content) + with contextlib.suppress(OSError, webbrowser.Error): + webbrowser.open(pathlib.Path(filepath).as_uri()) + actual_name = os.path.basename(filepath) + msg = f"SVG exported to {actual_name}" + if truncated: + msg += f" (content truncated to {SVG_MAX_CONTENT_CHARS} characters)" + return ActionResult( + action_key="v", + success=True, + message=msg, + data=filepath, + ) + except OSError as exc: + return ActionResult( + action_key="v", + success=False, + message=f"SVG export failed: {exc}", + ) + + def _action_maximize_restore(self, block: BlockInfo) -> ActionResult: + """Toggle maximize/restore for expandable blocks.""" + new_state = "restored" if block.is_expanded else "maximized" + return ActionResult( + action_key="m", + success=True, + message=f"Block {new_state}", + data=new_state, + ) + + def _action_retry(self, block: BlockInfo) -> ActionResult: + """Re-send the preceding user prompt for ActorResponse blocks.""" + if block.block_type != BlockType.ACTOR_RESPONSE: + return ActionResult( + action_key="r", + success=False, + message="Retry is only available for ActorResponse blocks", + ) + if not block.preceding_prompt or not block.preceding_prompt.strip(): + return ActionResult( + action_key="r", + success=False, + message="No preceding prompt available for retry", + ) + return ActionResult( + action_key="r", + success=True, + message="Retrying with preceding prompt", + data=block.preceding_prompt, + ) + + def _action_show_raw_data(self, block: BlockInfo) -> ActionResult: + """Show the raw A2A message data for the focused block.""" + raw = block.raw_data or "(no raw data available)" + return ActionResult( + action_key="d", + success=True, + message="Raw data displayed", + data=raw, + ) + + def _render_menu(self) -> None: + """Render the bordered menu text with applicable actions. + + Uses box-drawing characters matching the specification mockup + (§ Block Cursor and Context Menu). + """ + if self._block_info is None: + self.update("") + return + + width = _MENU_RENDER_WIDTH + applicable = self.get_applicable_actions() + + lines: list[str] = [] + # Top border with title — spec: ┌─ Block Actions ──...──┐ + title = "─ Block Actions " + border_right = width - 2 - len(title) + lines.append(f"\u250c{title}{'─' * border_right}\u2510") + # Blank line + lines.append(f"\u2502{' ' * (width - 2)}\u2502") + # Action lines + for action in applicable: + entry = f" {action.key} {action.label}" + padding = max(0, width - 2 - len(entry)) + lines.append(f"\u2502{entry}{' ' * padding}\u2502") + # Blank separator + lines.append(f"\u2502{' ' * (width - 2)}\u2502") + # Separator line + lines.append(f"\u2502{'─' * (width - 2)}\u2502") + # Escape line — spec: │ escape Dismiss ...│ + escape_entry = " escape Dismiss" + esc_padding = width - 2 - len(escape_entry) + lines.append(f"\u2502{escape_entry}{' ' * esc_padding}\u2502") + # Bottom border + lines.append(f"\u2514{'─' * (width - 2)}\u2518") + self.update("\n".join(lines)) diff --git a/typings/pyperclip/__init__.pyi b/typings/pyperclip/__init__.pyi new file mode 100644 index 000000000..180634b5e --- /dev/null +++ b/typings/pyperclip/__init__.pyi @@ -0,0 +1,4 @@ +"""Type stubs for pyperclip.""" + +def copy(text: str) -> None: ... +def paste() -> str: ... -- 2.52.0