diff --git a/features/steps/tui_materializer_steps.py b/features/steps/tui_materializer_steps.py new file mode 100644 index 000000000..6f7a2244d --- /dev/null +++ b/features/steps/tui_materializer_steps.py @@ -0,0 +1,535 @@ +"""Step definitions for tui_materializer.feature.""" + +from __future__ import annotations + +import threading +from typing import Any + +from behave import given, then, when # type: ignore[import-untyped] + + +# --------------------------------------------------------------------------- +# Background +# --------------------------------------------------------------------------- + + +@given("the tui materializer module is imported") +def step_import_tui_materializer(context: Any) -> None: + from cleveragents.tui.materializer import ( + TuiMaterializer, + TuiWidgetEvent, + TuiWidgetEventType, + render_element_for_tui, + ) + + context.TuiMaterializer = TuiMaterializer + context.TuiWidgetEvent = TuiWidgetEvent + context.TuiWidgetEventType = TuiWidgetEventType + context.render_element_for_tui = render_element_for_tui + context.materializer = None + context.session = None + context.panel_handle = None + context.status_handle = None + context.rendered_text = "" + context.callback_events: list[TuiWidgetEvent] = [] + context.last_event: TuiWidgetEvent | None = None + + +# --------------------------------------------------------------------------- +# Module exports +# --------------------------------------------------------------------------- + + +@then("TuiMaterializer should be importable from tui.materializer") +def step_check_tui_materializer_importable(context: Any) -> None: + from cleveragents.tui.materializer import TuiMaterializer + + assert TuiMaterializer is not None + + +@then("TuiWidgetEvent should be importable from tui.materializer") +def step_check_tui_widget_event_importable(context: Any) -> None: + from cleveragents.tui.materializer import TuiWidgetEvent + + assert TuiWidgetEvent is not None + + +@then("TuiWidgetEventType should be importable from tui.materializer") +def step_check_tui_widget_event_type_importable(context: Any) -> None: + from cleveragents.tui.materializer import TuiWidgetEventType + + assert TuiWidgetEventType is not None + + +@then("render_element_for_tui should be importable from tui.materializer") +def step_check_render_element_for_tui_importable(context: Any) -> None: + from cleveragents.tui.materializer import render_element_for_tui + + assert render_element_for_tui is not None + + +# --------------------------------------------------------------------------- +# TuiMaterializer instantiation +# --------------------------------------------------------------------------- + + +@when("I create a TuiMaterializer without a callback") +def step_create_materializer_no_callback(context: Any) -> None: + from cleveragents.tui.materializer import TuiMaterializer + + context.materializer = TuiMaterializer() + + +@when("I create a TuiMaterializer with a callback") +def step_create_materializer_with_callback(context: Any) -> None: + from cleveragents.tui.materializer import TuiMaterializer, TuiWidgetEvent + + context.callback_events = [] + + def _callback(event: TuiWidgetEvent) -> None: + context.callback_events.append(event) + + context.materializer = TuiMaterializer(on_event=_callback) + + +@then('the materializer strategy_name should be "tui"') +def step_check_strategy_name(context: Any) -> None: + assert context.materializer.strategy_name == "tui" + + +@then("the materializer supports_incremental_updates should be True") +def step_check_supports_incremental(context: Any) -> None: + assert context.materializer.supports_incremental_updates is True + + +@then("the materializer events list should be empty") +def step_check_events_empty(context: Any) -> None: + assert context.materializer.events == [] + + +# --------------------------------------------------------------------------- +# on_session_begin +# --------------------------------------------------------------------------- + + +@when("I call on_session_begin with a mock session") +def step_call_on_session_begin(context: Any) -> None: + context.materializer.on_session_begin(None) # type: ignore[arg-type] + + +@then("no error should be raised") +def step_no_error_raised(context: Any) -> None: + pass # If we got here, no exception was raised + + +# --------------------------------------------------------------------------- +# OutputSession integration +# --------------------------------------------------------------------------- + + +@when("I use the materializer with an OutputSession") +def step_use_with_output_session(context: Any) -> None: + from cleveragents.cli.output.session import OutputSession + + context.session = OutputSession(strategy=context.materializer) + + +@when('I create a panel handle titled "{title}"') +def step_create_panel_handle(context: Any, title: str) -> None: + context.panel_handle = context.session.panel(title) + + +@when('I create a table handle titled "{title}"') +def step_create_table_handle(context: Any, title: str) -> None: + context.table_handle = context.session.table(title) + + +@when('I create a status handle with message "{message}"') +def step_create_status_handle(context: Any, message: str) -> None: + context.status_handle = context.session.status(message) + + +@when('I create a progress handle with label "{label}"') +def step_create_progress_handle(context: Any, label: str) -> None: + context.progress_handle = context.session.progress(label, indeterminate=True) + + +@when('I create a code handle with content "{content}" and language "{language}"') +def step_create_code_handle(context: Any, content: str, language: str) -> None: + context.code_handle = context.session.code(content, language=language) + + +@when("I create a separator handle") +def step_create_separator_handle(context: Any) -> None: + context.separator_handle = context.session.separator() + + +@when('I create an action hint handle with command "{command}"') +def step_create_action_hint_handle(context: Any, command: str) -> None: + context.action_hint_handle = context.session.action_hint([command]) + + +@when("I close the panel handle") +def step_close_panel_handle(context: Any) -> None: + context.panel_handle.close() + + +@when("I close the status handle") +def step_close_status_handle(context: Any) -> None: + context.status_handle.close() + + +@when("I close the session") +def step_close_session(context: Any) -> None: + context.session.close() + + +@when('I create and close a status handle with message "{message}"') +def step_create_and_close_status_handle(context: Any, message: str) -> None: + handle = context.session.status(message) + handle.close() + + +# --------------------------------------------------------------------------- +# Event assertions +# --------------------------------------------------------------------------- + + +@then("an element_created event should be emitted") +def step_check_element_created_event(context: Any) -> None: + events = context.materializer.events + created_events = [e for e in events if e.event_type == "element_created"] + assert len(created_events) > 0, f"No element_created events found. Events: {events}" + context.last_event = created_events[-1] + + +@then('the event element_kind should be "{kind}"') +def step_check_event_element_kind(context: Any, kind: str) -> None: + assert context.last_event is not None + assert context.last_event.element_kind == kind, ( + f"Expected element_kind={kind!r}, got {context.last_event.element_kind!r}" + ) + + +@then('the event rendered_text should contain "{text}"') +def step_check_event_rendered_text_contains(context: Any, text: str) -> None: + assert context.last_event is not None + assert text in context.last_event.rendered_text, ( + f"Expected {text!r} in rendered_text={context.last_event.rendered_text!r}" + ) + + +@then("an element_closed event should be emitted") +def step_check_element_closed_event(context: Any) -> None: + events = context.materializer.events + closed_events = [e for e in events if e.event_type == "element_closed"] + assert len(closed_events) > 0, f"No element_closed events found. Events: {events}" + context.last_closed_event = closed_events[-1] + + +@then('the closed event rendered_text should contain "{text}"') +def step_check_closed_event_rendered_text(context: Any, text: str) -> None: + assert context.last_closed_event is not None + assert text in context.last_closed_event.rendered_text, ( + f"Expected {text!r} in rendered_text={context.last_closed_event.rendered_text!r}" + ) + + +@then("a session_end event should be emitted") +def step_check_session_end_event(context: Any) -> None: + events = context.materializer.events + end_events = [e for e in events if e.event_type == "session_end"] + assert len(end_events) > 0, f"No session_end events found. Events: {events}" + + +# --------------------------------------------------------------------------- +# Callback assertions +# --------------------------------------------------------------------------- + + +@then("the callback should have been called") +def step_check_callback_called(context: Any) -> None: + assert len(context.callback_events) > 0, "Callback was not called" + + +@then('the callback event type should be "element_created"') +def step_check_callback_event_type_created(context: Any) -> None: + created = [e for e in context.callback_events if e.event_type == "element_created"] + assert len(created) > 0, ( + f"No element_created callback events. Got: {[e.event_type for e in context.callback_events]}" + ) + + +@then('the callback should have been called with event_type "{event_type}"') +def step_check_callback_event_type(context: Any, event_type: str) -> None: + matching = [e for e in context.callback_events if e.event_type == event_type] + assert len(matching) > 0, ( + f"No {event_type!r} callback events. Got: {[e.event_type for e in context.callback_events]}" + ) + + +# --------------------------------------------------------------------------- +# rendered_output property +# --------------------------------------------------------------------------- + + +@then("the materializer rendered_output should be empty") +def step_check_rendered_output_empty(context: Any) -> None: + assert context.materializer.rendered_output == "", ( + f"Expected empty rendered_output, got: {context.materializer.rendered_output!r}" + ) + + +@then('the materializer rendered_output should contain "{text}"') +def step_check_rendered_output_contains(context: Any, text: str) -> None: + output = context.materializer.rendered_output + assert text in output, f"Expected {text!r} in rendered_output={output!r}" + + +# --------------------------------------------------------------------------- +# render_element_for_tui +# --------------------------------------------------------------------------- + + +@when('I render a Panel element with title "{title}" and entry "{key}" "{value}"') +def step_render_panel_element(context: Any, title: str, key: str, value: str) -> None: + from cleveragents.cli.output.handles._models import Panel, PanelEntry + from cleveragents.tui.materializer import render_element_for_tui + + element = Panel(title=title, entries=[PanelEntry(key=key, value=value)]) + context.rendered_text = render_element_for_tui(element) + + +@when('I render a Table element with title "{title}" and column "{column}"') +def step_render_table_element(context: Any, title: str, column: str) -> None: + from cleveragents.cli.output.handles._models import ColumnDef, Table + from cleveragents.tui.materializer import render_element_for_tui + + element = Table(title=title, columns=[ColumnDef(name=column)]) + context.rendered_text = render_element_for_tui(element) + + +@when('I render a StatusMessage element with level "{level}" and message "{message}"') +def step_render_status_element(context: Any, level: str, message: str) -> None: + from cleveragents.cli.output.handles._models import StatusMessage + from cleveragents.tui.materializer import render_element_for_tui + + element = StatusMessage(message=message, level=level) + context.rendered_text = render_element_for_tui(element) + + +@when('I render an indeterminate ProgressIndicator with label "{label}"') +def step_render_indeterminate_progress(context: Any, label: str) -> None: + from cleveragents.cli.output.handles._models import ProgressIndicator + from cleveragents.tui.materializer import render_element_for_tui + + element = ProgressIndicator(label=label, indeterminate=True) + context.rendered_text = render_element_for_tui(element) + + +@when( + 'I render a determinate ProgressIndicator with label "{label}" current {current:d} total {total:d}' +) +def step_render_determinate_progress( + context: Any, label: str, current: int, total: int +) -> None: + from cleveragents.cli.output.handles._models import ProgressIndicator + from cleveragents.tui.materializer import render_element_for_tui + + element = ProgressIndicator(label=label, current=current, total=total) + context.rendered_text = render_element_for_tui(element) + + +@when('I render a Tree element with root "{root}" and child "{child}"') +def step_render_tree_element(context: Any, root: str, child: str) -> None: + from cleveragents.cli.output.handles._models import Tree, TreeNode + from cleveragents.tui.materializer import render_element_for_tui + + root_node = TreeNode(label=root, children=[TreeNode(label=child)]) + element = Tree(root=root_node) + context.rendered_text = render_element_for_tui(element) + + +@when( + 'I render a CodeBlock element with language "{language}" and content "{content}"' +) +def step_render_code_element(context: Any, language: str, content: str) -> None: + from cleveragents.cli.output.handles._models import CodeBlock + from cleveragents.tui.materializer import render_element_for_tui + + element = CodeBlock(content=content, language=language) + context.rendered_text = render_element_for_tui(element) + + +@when( + 'I render a DiffBlock element with file_a "{file_a}" and file_b "{file_b}"' +) +def step_render_diff_element(context: Any, file_a: str, file_b: str) -> None: + from cleveragents.cli.output.handles._models import DiffBlock + from cleveragents.tui.materializer import render_element_for_tui + + element = DiffBlock(file_a=file_a, file_b=file_b) + context.rendered_text = render_element_for_tui(element) + + +@when('I render a Separator element with style "{style}"') +def step_render_separator_element(context: Any, style: str) -> None: + from cleveragents.cli.output.handles._models import Separator + from cleveragents.tui.materializer import render_element_for_tui + + element = Separator(style=style) + context.rendered_text = render_element_for_tui(element) + + +@when('I render an ActionHint element with command "{command}"') +def step_render_action_hint_element(context: Any, command: str) -> None: + from cleveragents.cli.output.handles._models import ActionHint + from cleveragents.tui.materializer import render_element_for_tui + + element = ActionHint(commands=[command]) + context.rendered_text = render_element_for_tui(element) + + +@when('I render a TextBlock element with content "{content}"') +def step_render_text_element(context: Any, content: str) -> None: + from cleveragents.cli.output.handles._models import TextBlock + from cleveragents.tui.materializer import render_element_for_tui + + element = TextBlock(content=content) + context.rendered_text = render_element_for_tui(element) + + +@when('I render a TextBlock element with content "{content}" and indent {indent:d}') +def step_render_text_element_with_indent( + context: Any, content: str, indent: int +) -> None: + from cleveragents.cli.output.handles._models import TextBlock + from cleveragents.tui.materializer import render_element_for_tui + + element = TextBlock(content=content, indent=indent) + context.rendered_text = render_element_for_tui(element) + + +@then('the rendered text should contain "{text}"') +def step_check_rendered_text_contains(context: Any, text: str) -> None: + assert text in context.rendered_text, ( + f"Expected {text!r} in rendered_text={context.rendered_text!r}" + ) + + +@then("the rendered text should be empty") +def step_check_rendered_text_empty(context: Any) -> None: + assert context.rendered_text == "", ( + f"Expected empty rendered_text, got: {context.rendered_text!r}" + ) + + +# --------------------------------------------------------------------------- +# A2A event routing +# --------------------------------------------------------------------------- + + +@when( + 'I route a permission request for "{file_path}" with type "{request_type}"' +) +def step_route_permission_request( + context: Any, file_path: str, request_type: str +) -> None: + from cleveragents.domain.models.core.inline_permission_question import ( + InlinePermissionQuestion, + PermissionRequestType, + ) + + question = InlinePermissionQuestion( + file_path=file_path, + request_type=PermissionRequestType(request_type), + ) + context.permission_question = question + context.materializer.route_permission_request(question) + + +@then("a permission_request event should be emitted") +def step_check_permission_request_event(context: Any) -> None: + events = context.materializer.events + perm_events = [e for e in events if e.event_type == "permission_request"] + assert len(perm_events) > 0, f"No permission_request events found. Events: {events}" + context.last_perm_event = perm_events[-1] + + +@then("the permission event extra should be the InlinePermissionQuestion") +def step_check_permission_event_extra(context: Any) -> None: + assert context.last_perm_event.extra is context.permission_question + + +@then('the permission event rendered_text should contain "{text}"') +def step_check_permission_event_rendered_text(context: Any, text: str) -> None: + assert text in context.last_perm_event.rendered_text, ( + f"Expected {text!r} in rendered_text={context.last_perm_event.rendered_text!r}" + ) + + +@when('I route a thought block with content "{content}"') +def step_route_thought_block(context: Any, content: str) -> None: + from cleveragents.domain.models.thought.thought_block import ThoughtBlock + + thought = ThoughtBlock(content=content) + context.thought_block = thought + context.materializer.route_thought_block(thought) + + +@then("a thought_block event should be emitted") +def step_check_thought_block_event(context: Any) -> None: + events = context.materializer.events + thought_events = [e for e in events if e.event_type == "thought_block"] + assert len(thought_events) > 0, f"No thought_block events found. Events: {events}" + context.last_thought_event = thought_events[-1] + + +@then("the thought event extra should be the ThoughtBlock") +def step_check_thought_event_extra(context: Any) -> None: + assert context.last_thought_event.extra is context.thought_block + + +@then('the thought event rendered_text should contain "{text}"') +def step_check_thought_event_rendered_text(context: Any, text: str) -> None: + assert text in context.last_thought_event.rendered_text, ( + f"Expected {text!r} in rendered_text={context.last_thought_event.rendered_text!r}" + ) + + +# --------------------------------------------------------------------------- +# Thread safety +# --------------------------------------------------------------------------- + + +@when("I create multiple status handles concurrently") +def step_create_multiple_status_handles_concurrently(context: Any) -> None: + errors: list[Exception] = [] + + def _create_handle(i: int) -> None: + try: + handle = context.session.status(f"Message {i}") + handle.close() + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=_create_handle, args=(i,)) for i in range(10)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.thread_errors = errors + + +@then("all events should be recorded without data corruption") +def step_check_thread_safety(context: Any) -> None: + assert context.thread_errors == [], ( + f"Thread errors: {context.thread_errors}" + ) + events = context.materializer.events + # We should have at least 10 element_created + 10 element_closed events + created = [e for e in events if e.event_type == "element_created"] + closed = [e for e in events if e.event_type == "element_closed"] + assert len(created) >= 10, f"Expected >= 10 created events, got {len(created)}" + assert len(closed) >= 10, f"Expected >= 10 closed events, got {len(closed)}" diff --git a/features/tui_materializer.feature b/features/tui_materializer.feature new file mode 100644 index 000000000..fd32883cd --- /dev/null +++ b/features/tui_materializer.feature @@ -0,0 +1,233 @@ +Feature: TUI Materializer + As a TUI developer + I want a TuiMaterializer that bridges the Output Rendering Framework to Textual widgets + So that all CLI command producers can render in the TUI without modification + + Background: + Given the tui materializer module is imported + + # ── Module exports ──────────────────────────────────────────────────────── + + Scenario: TuiMaterializer is importable from tui.materializer + Then TuiMaterializer should be importable from tui.materializer + And TuiWidgetEvent should be importable from tui.materializer + And TuiWidgetEventType should be importable from tui.materializer + And render_element_for_tui should be importable from tui.materializer + + # ── TuiMaterializer instantiation ──────────────────────────────────────── + + Scenario: TuiMaterializer can be instantiated without arguments + When I create a TuiMaterializer without a callback + Then the materializer strategy_name should be "tui" + And the materializer supports_incremental_updates should be True + And the materializer events list should be empty + + Scenario: TuiMaterializer can be instantiated with a callback + When I create a TuiMaterializer with a callback + Then the materializer strategy_name should be "tui" + + # ── MaterializationStrategy protocol ───────────────────────────────────── + + Scenario: TuiMaterializer implements on_session_begin + When I create a TuiMaterializer without a callback + And I call on_session_begin with a mock session + Then no error should be raised + + Scenario: TuiMaterializer emits element_created event for panel + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a panel handle titled "Test Panel" + Then an element_created event should be emitted + And the event element_kind should be "panel" + And the event rendered_text should contain "Test Panel" + + Scenario: TuiMaterializer emits element_created event for table + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a table handle titled "Results" + Then an element_created event should be emitted + And the event element_kind should be "table" + And the event rendered_text should contain "Results" + + Scenario: TuiMaterializer emits element_created event for status + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a status handle with message "Operation complete" + Then an element_created event should be emitted + And the event element_kind should be "status" + And the event rendered_text should contain "Operation complete" + + Scenario: TuiMaterializer emits element_created event for progress + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a progress handle with label "Loading" + Then an element_created event should be emitted + And the event element_kind should be "progress" + And the event rendered_text should contain "Loading" + + Scenario: TuiMaterializer emits element_created event for code block + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a code handle with content "print('hello')" and language "python" + Then an element_created event should be emitted + And the event element_kind should be "code" + And the event rendered_text should contain "print('hello')" + + Scenario: TuiMaterializer emits element_created event for separator + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a separator handle + Then an element_created event should be emitted + And the event element_kind should be "separator" + + Scenario: TuiMaterializer emits element_created event for action hint + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create an action hint handle with command "agents plan list" + Then an element_created event should be emitted + And the event element_kind should be "action_hint" + And the event rendered_text should contain "agents plan list" + + Scenario: TuiMaterializer emits element_closed event when handle is closed + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a panel handle titled "Closing Panel" + And I close the panel handle + Then an element_closed event should be emitted + And the closed event rendered_text should contain "Closing Panel" + + Scenario: TuiMaterializer emits session_end event when session ends + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create a status handle with message "Done" + And I close the session + Then a session_end event should be emitted + + # ── Callback invocation ─────────────────────────────────────────────────── + + Scenario: TuiMaterializer invokes callback on element_created + When I create a TuiMaterializer with a callback + And I use the materializer with an OutputSession + And I create a status handle with message "Callback test" + Then the callback should have been called + And the callback event type should be "element_created" + + Scenario: TuiMaterializer invokes callback on element_closed + When I create a TuiMaterializer with a callback + And I use the materializer with an OutputSession + And I create a status handle with message "Close callback test" + And I close the status handle + Then the callback should have been called with event_type "element_closed" + + Scenario: TuiMaterializer invokes callback on session_end + When I create a TuiMaterializer with a callback + And I use the materializer with an OutputSession + And I close the session + Then the callback should have been called with event_type "session_end" + + # ── rendered_output property ────────────────────────────────────────────── + + Scenario: rendered_output returns empty string when no elements + When I create a TuiMaterializer without a callback + Then the materializer rendered_output should be empty + + Scenario: rendered_output accumulates closed element text + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create and close a status handle with message "Accumulated" + Then the materializer rendered_output should contain "Accumulated" + + # ── render_element_for_tui ──────────────────────────────────────────────── + + Scenario: render_element_for_tui renders Panel with title and entries + When I render a Panel element with title "Info" and entry "key" "value" + Then the rendered text should contain "Info" + And the rendered text should contain "key" + And the rendered text should contain "value" + + Scenario: render_element_for_tui renders Table with columns and rows + When I render a Table element with title "Data" and column "Name" + Then the rendered text should contain "Data" + And the rendered text should contain "Name" + + Scenario: render_element_for_tui renders StatusMessage ok level + When I render a StatusMessage element with level "ok" and message "Success" + Then the rendered text should contain "ok" + And the rendered text should contain "Success" + + Scenario: render_element_for_tui renders StatusMessage error level + When I render a StatusMessage element with level "error" and message "Failed" + Then the rendered text should contain "error" + And the rendered text should contain "Failed" + + Scenario: render_element_for_tui renders indeterminate ProgressIndicator + When I render an indeterminate ProgressIndicator with label "Thinking" + Then the rendered text should contain "Thinking" + + Scenario: render_element_for_tui renders determinate ProgressIndicator + When I render a determinate ProgressIndicator with label "Loading" current 5 total 10 + Then the rendered text should contain "Loading" + And the rendered text should contain "50%" + + Scenario: render_element_for_tui renders Tree with root and children + When I render a Tree element with root "root" and child "child1" + Then the rendered text should contain "root" + And the rendered text should contain "child1" + + Scenario: render_element_for_tui renders CodeBlock with language + When I render a CodeBlock element with language "python" and content "x = 1" + Then the rendered text should contain "python" + And the rendered text should contain "x = 1" + + Scenario: render_element_for_tui renders DiffBlock with hunks + When I render a DiffBlock element with file_a "old.py" and file_b "new.py" + Then the rendered text should contain "old.py" + And the rendered text should contain "new.py" + + Scenario: render_element_for_tui renders Separator line style + When I render a Separator element with style "line" + Then the rendered text should contain "-" + + Scenario: render_element_for_tui renders Separator blank style + When I render a Separator element with style "blank" + Then the rendered text should be empty + + Scenario: render_element_for_tui renders Separator double style + When I render a Separator element with style "double" + Then the rendered text should contain "=" + + Scenario: render_element_for_tui renders ActionHint with commands + When I render an ActionHint element with command "agents plan list" + Then the rendered text should contain "agents plan list" + + Scenario: render_element_for_tui renders TextBlock content + When I render a TextBlock element with content "Hello world" + Then the rendered text should contain "Hello world" + + Scenario: render_element_for_tui renders TextBlock with indent + When I render a TextBlock element with content "indented" and indent 4 + Then the rendered text should contain " indented" + + # ── A2A event routing ───────────────────────────────────────────────────── + + Scenario: route_permission_request emits permission_request event + When I create a TuiMaterializer without a callback + And I route a permission request for "src/main.py" with type "file_write" + Then a permission_request event should be emitted + And the permission event extra should be the InlinePermissionQuestion + And the permission event rendered_text should contain "src/main.py" + + Scenario: route_thought_block emits thought_block event + When I create a TuiMaterializer without a callback + And I route a thought block with content "I am thinking" + Then a thought_block event should be emitted + And the thought event extra should be the ThoughtBlock + And the thought event rendered_text should contain "I am thinking" + + # ── Thread safety ───────────────────────────────────────────────────────── + + Scenario: TuiMaterializer events list is thread-safe + When I create a TuiMaterializer without a callback + And I use the materializer with an OutputSession + And I create multiple status handles concurrently + Then all events should be recorded without data corruption diff --git a/src/cleveragents/tui/materializer.py b/src/cleveragents/tui/materializer.py new file mode 100644 index 000000000..231b31310 --- /dev/null +++ b/src/cleveragents/tui/materializer.py @@ -0,0 +1,559 @@ +"""TuiMaterializer — A2A integration layer bridging the Output Rendering +Framework to Textual UI widgets. + +The ``TuiMaterializer`` implements the ``MaterializationStrategy`` protocol +and maps ``ElementHandle`` events to Textual widget operations, enabling all +CLI command producers to render in the TUI without modification. + +Based on ADR-044 §TuiMaterializer — Output Rendering Framework Integration +and the M8 specification (v3.7.0). + +Widget mapping (per ADR-044): + ++-------------------+------------------------------------------+ +| ElementHandle | Textual Widget | ++===================+==========================================+ +| PanelHandle | Static container with Collapsible | +| TableHandle | DataTable | +| TreeHandle | Tree | +| ProgressHandle | ProgressBar (det.) / Throbber (indet.) | +| StatusHandle | Label with semantic CSS class | +| CodeHandle | Read-only TextArea with syntax highlight | +| DiffHandle | DiffView (custom Static subclass) | +| SeparatorHandle | Rule | +| ActionHintHandle | Static with muted text | ++-------------------+------------------------------------------+ +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable +from typing import TYPE_CHECKING, Any + +from cleveragents.cli.output.handles import ( + ElementClosed, + ElementCreated, + ElementUpdated, + SessionEnd, +) +from cleveragents.cli.output.handles._models import ( + ActionHint, + CodeBlock, + DiffBlock, + ElementSnapshot, + Panel, + ProgressIndicator, + Separator, + StatusMessage, + Table, + TextBlock, + Tree, +) + +if TYPE_CHECKING: + from cleveragents.cli.output.session import OutputSession + from cleveragents.domain.models.core.inline_permission_question import ( + InlinePermissionQuestion, + ) + from cleveragents.domain.models.thought.thought_block import ThoughtBlock + +__all__ = [ + "TuiMaterializer", + "TuiWidgetEvent", + "TuiWidgetEventType", + "render_element_for_tui", +] + + +# --------------------------------------------------------------------------- +# Event types emitted by TuiMaterializer to the host application +# --------------------------------------------------------------------------- + + +class TuiWidgetEventType: + """String constants for TUI widget event types.""" + + ELEMENT_CREATED: str = "element_created" + ELEMENT_UPDATED: str = "element_updated" + ELEMENT_CLOSED: str = "element_closed" + SESSION_END: str = "session_end" + PERMISSION_REQUEST: str = "permission_request" + THOUGHT_BLOCK: str = "thought_block" + + +class TuiWidgetEvent: + """An event emitted by the TuiMaterializer to the host application. + + The host application (e.g. ``_TextualCleverAgentsTuiApp``) subscribes + to these events to update the conversation widget incrementally. + + Attributes: + event_type: One of the ``TuiWidgetEventType`` constants. + handle_id: The handle ID of the element that triggered the event. + element_kind: The element kind (e.g. ``"panel"``, ``"table"``). + rendered_text: Plain-text rendering of the element for display. + element_snapshot: The element snapshot at the time of the event. + extra: Optional extra data (e.g. permission question, thought block). + """ + + __slots__ = ( + "element_kind", + "element_snapshot", + "event_type", + "extra", + "handle_id", + "rendered_text", + ) + + def __init__( + self, + *, + event_type: str, + handle_id: str, + element_kind: str, + rendered_text: str = "", + element_snapshot: ElementSnapshot | None = None, + extra: Any = None, + ) -> None: + self.event_type = event_type + self.handle_id = handle_id + self.element_kind = element_kind + self.rendered_text = rendered_text + self.element_snapshot = element_snapshot + self.extra = extra + + def __repr__(self) -> str: + return ( + f"TuiWidgetEvent(" + f"event_type={self.event_type!r}, " + f"handle_id={self.handle_id!r}, " + f"element_kind={self.element_kind!r})" + ) + + +# --------------------------------------------------------------------------- +# Plain-text rendering helpers for TUI display +# --------------------------------------------------------------------------- + + +def _render_panel(element: Panel) -> str: + """Render a Panel element as plain text.""" + lines: list[str] = [f"-- {element.title} --"] + for entry in element.entries: + icon_prefix = f"{entry.icon} " if entry.icon else "" + lines.append(f" {icon_prefix}{entry.key}: {entry.value}") + return "\n".join(lines) + + +def _render_table(element: Table) -> str: + """Render a Table element as plain text.""" + lines: list[str] = [] + if element.title: + lines.append(f"-- {element.title} --") + if element.columns: + header = " | ".join(col.name for col in element.columns) + lines.append(header) + lines.append("-" * len(header)) + for row in element.rows: + if element.columns: + cells = [str(row.get(col.name, "")) for col in element.columns] + else: + cells = [str(v) for v in row.values()] + lines.append(" | ".join(cells)) + if element.summary: + lines.append(f"Summary: {element.summary}") + return "\n".join(lines) + + +def _render_status(element: StatusMessage) -> str: + """Render a StatusMessage element as plain text.""" + level_icons: dict[str, str] = { + "ok": "[ok]", + "info": "[info]", + "warn": "[warn]", + "error": "[error]", + } + icon = level_icons.get(element.level, "[*]") + text = f"{icon} {element.message}" + if element.detail: + text += f"\n {element.detail}" + return text + + +def _render_progress(element: ProgressIndicator) -> str: + """Render a ProgressIndicator element as plain text.""" + if element.indeterminate: + return f"[...] {element.label}" + if ( + element.total is not None + and element.total > 0 + and element.current is not None + ): + pct = int(element.current / element.total * 100) + bar_width = 20 + filled = int(bar_width * element.current / element.total) + bar = "#" * filled + "." * (bar_width - filled) + return f"{element.label} [{bar}] {pct}%" + if element.steps: + lines = [f"{element.label}:"] + step_icons: dict[str, str] = { + "done": "[done]", + "running": "[run]", + "pending": "[wait]", + } + for step in element.steps: + status_icon = step_icons.get(step.status, "[*]") + lines.append(f" {status_icon} {step.label}") + return "\n".join(lines) + return f"[...] {element.label}" + + +def _render_tree(element: Tree) -> str: + """Render a Tree element as plain text.""" + lines: list[str] = [] + + def _render_node(node: Any, prefix: str, is_last: bool) -> None: + connector = "L-- " if is_last else "+-- " + lines.append(f"{prefix}{connector}{node.label}") + child_prefix = prefix + (" " if is_last else "| ") + for i, child in enumerate(node.children): + _render_node(child, child_prefix, i == len(node.children) - 1) + + root = element.root + lines.append(root.label) + for i, child in enumerate(root.children): + _render_node(child, "", i == len(root.children) - 1) + return "\n".join(lines) + + +def _render_code(element: CodeBlock) -> str: + """Render a CodeBlock element as plain text.""" + lang_hint = element.language or "" + header = f"```{lang_hint}" + footer = "```" + return f"{header}\n{element.content}\n{footer}" + + +def _render_diff(element: DiffBlock) -> str: + """Render a DiffBlock element as plain text.""" + lines: list[str] = [] + if element.file_a or element.file_b: + lines.append(f"--- {element.file_a or '/dev/null'}") + lines.append(f"+++ {element.file_b or '/dev/null'}") + for hunk in element.hunks: + lines.append(hunk.header) + for diff_line in hunk.lines: + prefix = {"add": "+", "remove": "-", "context": " "}.get( + diff_line.line_type, " " + ) + lines.append(f"{prefix}{diff_line.content}") + return "\n".join(lines) + + +def _render_separator(element: Separator) -> str: + """Render a Separator element as plain text.""" + if element.style == "blank": + return "" + if element.style == "double": + return "=" * 40 + return "-" * 40 + + +def _render_action_hint(element: ActionHint) -> str: + """Render an ActionHint element as plain text.""" + lines: list[str] = [] + if element.description: + lines.append(element.description) + for cmd in element.commands: + lines.append(f" $ {cmd}") + return "\n".join(lines) + + +def _render_text(element: TextBlock) -> str: + """Render a TextBlock element as plain text.""" + if element.indent > 0: + indent_str = " " * element.indent + return "\n".join( + indent_str + line for line in element.content.splitlines() + ) + return element.content + + +def render_element_for_tui(element: ElementSnapshot) -> str: + """Render any element snapshot to a plain-text string for TUI display. + + This is the TUI equivalent of ``render_element_plain`` from the CLI + output framework, adapted for Textual widget display. + + Args: + element: The element snapshot to render. + + Returns: + A plain-text string suitable for display in a Textual Static widget. + """ + if isinstance(element, Panel): + return _render_panel(element) + if isinstance(element, Table): + return _render_table(element) + if isinstance(element, StatusMessage): + return _render_status(element) + if isinstance(element, ProgressIndicator): + return _render_progress(element) + if isinstance(element, Tree): + return _render_tree(element) + if isinstance(element, CodeBlock): + return _render_code(element) + if isinstance(element, DiffBlock): + return _render_diff(element) + if isinstance(element, Separator): + return _render_separator(element) + if isinstance(element, ActionHint): + return _render_action_hint(element) + if isinstance(element, TextBlock): + return _render_text(element) + return str(element) # pragma: no cover + + +# --------------------------------------------------------------------------- +# TuiMaterializer +# --------------------------------------------------------------------------- + + +class TuiMaterializer: + """A2A integration layer bridging the Output Rendering Framework to + Textual UI widgets. + + The ``TuiMaterializer`` implements the ``MaterializationStrategy`` + protocol and maps ``ElementHandle`` events to Textual widget operations. + It enables all CLI command producers to render in the TUI without + modification, fulfilling the architectural promise of ADR-021 and + ADR-044. + + Usage:: + + def on_event(event: TuiWidgetEvent) -> None: + # Update the conversation widget + conversation.update(event.rendered_text) + + materializer = TuiMaterializer(on_event=on_event) + with OutputSession(strategy=materializer) as session: + panel = session.panel("Result") + panel.set_entry("status", "ok") + + The materializer is thread-safe. All event callbacks are invoked + from the thread that triggers the element lifecycle event. + + Attributes: + strategy_name: Always ``"tui"``. + supports_incremental_updates: Always ``True`` — the TUI supports + live streaming updates. + """ + + strategy_name: str = "tui" + supports_incremental_updates: bool = True + + def __init__( + self, + *, + on_event: Callable[[TuiWidgetEvent], None] | None = None, + ) -> None: + """Initialise the TuiMaterializer. + + Args: + on_event: Optional callback invoked for every ``TuiWidgetEvent``. + When ``None``, events are accumulated internally and + accessible via :attr:`events`. + """ + self._on_event = on_event + self._session: OutputSession | None = None + self._events: list[TuiWidgetEvent] = [] + self._index_map: dict[str, int] = {} + self._rendered: dict[int, str] = {} + self._lock: threading.Lock = threading.Lock() + + # ------------------------------------------------------------------ + # Public properties + # ------------------------------------------------------------------ + + @property + def events(self) -> list[TuiWidgetEvent]: + """Return a snapshot of all events emitted so far (thread-safe).""" + with self._lock: + return list(self._events) + + @property + def rendered_output(self) -> str: + """Return all rendered element text joined by newlines (thread-safe). + + Useful for testing and headless inspection. + """ + with self._lock: + parts = [ + text + for _, text in sorted(self._rendered.items()) + if text + ] + return "\n\n".join(parts) + + # ------------------------------------------------------------------ + # MaterializationStrategy protocol implementation + # ------------------------------------------------------------------ + + def on_session_begin(self, session: OutputSession) -> None: + """Called when a new output session begins.""" + self._session = session + + def on_element_created(self, event: ElementCreated) -> None: + """Called when a new element handle is created. + + Renders the initial element state and emits a ``TuiWidgetEvent`` + with ``event_type="element_created"``. + """ + rendered = "" + if event.initial_state is not None: + rendered = render_element_for_tui(event.initial_state) + + with self._lock: + self._index_map[event.handle_id] = event.declaration_index + self._rendered[event.declaration_index] = rendered + + tui_event = TuiWidgetEvent( + event_type=TuiWidgetEventType.ELEMENT_CREATED, + handle_id=event.handle_id, + element_kind=event.element_kind, + rendered_text=rendered, + element_snapshot=event.initial_state, + ) + self._emit(tui_event) + + def on_element_updated(self, event: ElementUpdated) -> None: + """Called when data is written to an element handle. + + Re-renders the element and emits a ``TuiWidgetEvent`` with + ``event_type="element_updated"``. + """ + rendered = "" + if event.element_snapshot is not None: + rendered = render_element_for_tui(event.element_snapshot) + + with self._lock: + idx = self._index_map.get(event.handle_id) + if idx is not None: + self._rendered[idx] = rendered + + tui_event = TuiWidgetEvent( + event_type=TuiWidgetEventType.ELEMENT_UPDATED, + handle_id=event.handle_id, + element_kind=event.element_kind, + rendered_text=rendered, + element_snapshot=event.element_snapshot, + ) + self._emit(tui_event) + + def on_element_closed(self, event: ElementClosed) -> None: + """Called when an element handle is closed (finalised). + + Renders the final element state and emits a ``TuiWidgetEvent`` + with ``event_type="element_closed"``. + """ + rendered = "" + if event.final_state is not None: + rendered = render_element_for_tui(event.final_state) + + with self._lock: + idx = self._index_map.get(event.handle_id) + if idx is not None: + self._rendered[idx] = rendered + + tui_event = TuiWidgetEvent( + event_type=TuiWidgetEventType.ELEMENT_CLOSED, + handle_id=event.handle_id, + element_kind=event.element_kind, + rendered_text=rendered, + element_snapshot=event.final_state, + ) + self._emit(tui_event) + + def on_session_end(self, event: SessionEnd) -> None: + """Called when the output session ends. + + Emits a ``TuiWidgetEvent`` with ``event_type="session_end"``. + """ + tui_event = TuiWidgetEvent( + event_type=TuiWidgetEventType.SESSION_END, + handle_id=event.handle_id, + element_kind=event.element_kind, + rendered_text=self.rendered_output, + ) + self._emit(tui_event) + + # ------------------------------------------------------------------ + # A2A event routing + # ------------------------------------------------------------------ + + def route_permission_request( + self, + question: InlinePermissionQuestion, + ) -> TuiWidgetEvent: + """Route a permission request event to the TUI. + + Creates a ``TuiWidgetEvent`` with ``event_type="permission_request"`` + and the ``InlinePermissionQuestion`` in the ``extra`` field. The + host application should mount a ``PermissionQuestionWidget`` (for + single-file requests) or push the ``PermissionsScreen`` (for + multi-file requests). + + Args: + question: The permission question to route. + + Returns: + The emitted ``TuiWidgetEvent``. + """ + tui_event = TuiWidgetEvent( + event_type=TuiWidgetEventType.PERMISSION_REQUEST, + handle_id="", + element_kind="permission_request", + rendered_text=f"Permission required: {question.file_path}", + extra=question, + ) + self._emit(tui_event) + return tui_event + + def route_thought_block( + self, + thought: ThoughtBlock, + ) -> TuiWidgetEvent: + """Route a thought block event to the TUI. + + Creates a ``TuiWidgetEvent`` with ``event_type="thought_block"`` + and the ``ThoughtBlock`` in the ``extra`` field. The host + application should mount a ``ThoughtBlockWidget`` in the + conversation stream. + + Args: + thought: The thought block to route. + + Returns: + The emitted ``TuiWidgetEvent``. + """ + tui_event = TuiWidgetEvent( + event_type=TuiWidgetEventType.THOUGHT_BLOCK, + handle_id="", + element_kind="thought_block", + rendered_text=thought.rendered_text(), + extra=thought, + ) + self._emit(tui_event) + return tui_event + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _emit(self, event: TuiWidgetEvent) -> None: + """Emit a TuiWidgetEvent to the callback and accumulate it.""" + with self._lock: + self._events.append(event) + if self._on_event is not None: + self._on_event(event)