From 12180c4f23b5941919a1243b8f81f17aa15e3a62 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Tue, 28 Apr 2026 08:24:07 +0000 Subject: [PATCH 1/2] feat(tui): implement TuiMaterializer A2A integration layer Implements the TuiMaterializer class that bridges the Output Rendering Framework to Textual UI widgets per ADR-044. The materializer maps all ElementHandle event types (Panel, Table, Status, Progress, Tree, Code, Diff, Separator, ActionHint, TextBlock) to plain-text renderings for display in the TUI conversation widget. Also implements A2A event routing for PermissionRequest and ThoughtBlock events, enabling the TUI to display permission questions and actor reasoning traces. Adds BDD feature file and step definitions covering all element types, callback invocation, rendered_output property, A2A routing, and thread safety. ISSUES CLOSED: #8442 --- features/steps/tui_materializer_steps.py | 535 ++++++++++++++++++++++ features/tui_materializer.feature | 233 ++++++++++ src/cleveragents/tui/materializer.py | 559 +++++++++++++++++++++++ 3 files changed, 1327 insertions(+) create mode 100644 features/steps/tui_materializer_steps.py create mode 100644 features/tui_materializer.feature create mode 100644 src/cleveragents/tui/materializer.py 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) -- 2.52.0 From f8c255bed916696fc5578a13a5abbf06dff24cbd Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Fri, 29 May 2026 22:44:41 -0400 Subject: [PATCH 2/2] chore: re-trigger CI [controller] -- 2.52.0