feat(tui): implement TuiMaterializer A2A integration layer #10643
@@ -0,0 +1,531 @@
|
||||
"""Step implementations for TUI Materializer A2A integration tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from behave import given, then, when
|
||||
|
||||
from cleveragents.a2a.events import A2aEventQueue
|
||||
from cleveragents.a2a.models import A2aEvent
|
||||
from cleveragents.tui.materializer import TuiMaterializer
|
||||
|
||||
|
||||
@given("a mock A2A event queue")
|
||||
def step_mock_event_queue(context: Any) -> None:
|
||||
"""Create a mock A2A event queue."""
|
||||
context.event_queue = A2aEventQueue()
|
||||
|
||||
|
||||
@given("a mock Textual app reference")
|
||||
def step_mock_app_reference(context: Any) -> None:
|
||||
"""Create a mock Textual app reference."""
|
||||
context.app = MagicMock()
|
||||
context.conversation_widget = MagicMock()
|
||||
context.thought_widget = MagicMock()
|
||||
context.permission_widget = MagicMock()
|
||||
context.permissions_screen = MagicMock()
|
||||
|
||||
def query_one_side_effect(selector: str, expect_type: Any = None) -> Any:
|
||||
if selector == "#conversation":
|
||||
return context.conversation_widget
|
||||
elif selector == "#thought-block":
|
||||
return context.thought_widget
|
||||
elif selector == "#permission-question":
|
||||
return context.permission_widget
|
||||
elif selector == "#permissions-screen":
|
||||
return context.permissions_screen
|
||||
else:
|
||||
raise Exception(f"Widget not found: {selector}")
|
||||
|
||||
context.app.query_one.side_effect = query_one_side_effect
|
||||
|
||||
|
||||
@given("a TuiMaterializer instance")
|
||||
def step_materializer_instance(context: Any) -> None:
|
||||
"""Create a TuiMaterializer instance."""
|
||||
context.materializer = TuiMaterializer(context.event_queue, context.app)
|
||||
|
||||
|
||||
@when("I create a TuiMaterializer with a valid event queue and app")
|
||||
def step_create_materializer_valid(context: Any) -> None:
|
||||
"""Create a materializer with valid inputs."""
|
||||
context.materializer = TuiMaterializer(context.event_queue, context.app)
|
||||
|
||||
|
||||
@then("the materializer should be created successfully")
|
||||
def step_materializer_created(context: Any) -> None:
|
||||
"""Verify materializer was created."""
|
||||
assert context.materializer is not None
|
||||
assert isinstance(context.materializer, TuiMaterializer)
|
||||
|
||||
|
||||
@then("the materializer should not be active initially")
|
||||
def step_materializer_not_active(context: Any) -> None:
|
||||
"""Verify materializer is not active."""
|
||||
assert not context.materializer.is_active
|
||||
|
||||
|
||||
@when("I try to create a TuiMaterializer with event_queue=None")
|
||||
def step_create_materializer_none_queue(context: Any) -> None:
|
||||
"""Try to create materializer with None queue."""
|
||||
context.error = None
|
||||
try:
|
||||
TuiMaterializer(None, context.app)
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@then("a TypeError should be raised with message {message}")
|
||||
def step_check_type_error(context: Any, message: str) -> None:
|
||||
"""Verify TypeError was raised with expected message."""
|
||||
assert context.error is not None
|
||||
assert isinstance(context.error, TypeError)
|
||||
assert str(context.error) == message
|
||||
|
||||
|
||||
@when("I try to create a TuiMaterializer with app=None")
|
||||
def step_create_materializer_none_app(context: Any) -> None:
|
||||
"""Try to create materializer with None app."""
|
||||
context.error = None
|
||||
try:
|
||||
TuiMaterializer(context.event_queue, None)
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I try to create a TuiMaterializer with an invalid event queue")
|
||||
def step_create_materializer_invalid_queue(context: Any) -> None:
|
||||
"""Try to create materializer with invalid queue."""
|
||||
context.error = None
|
||||
invalid_queue = MagicMock(spec=[]) # No subscribe_local method
|
||||
try:
|
||||
TuiMaterializer(invalid_queue, context.app)
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I create a closed event queue")
|
||||
def step_create_closed_queue(context: Any) -> None:
|
||||
"""Create and close an event queue."""
|
||||
context.closed_queue = A2aEventQueue()
|
||||
context.closed_queue.close()
|
||||
|
||||
|
||||
@when("I try to create a TuiMaterializer with the closed queue")
|
||||
def step_create_materializer_closed_queue(context: Any) -> None:
|
||||
"""Try to create materializer with closed queue."""
|
||||
context.error = None
|
||||
try:
|
||||
TuiMaterializer(context.closed_queue, context.app)
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@then("a ValueError should be raised with message {message}")
|
||||
def step_check_value_error(context: Any, message: str) -> None:
|
||||
"""Verify ValueError was raised with expected message."""
|
||||
assert context.error is not None
|
||||
assert isinstance(context.error, ValueError)
|
||||
assert str(context.error) == message
|
||||
|
||||
|
||||
@when("I call start on the materializer")
|
||||
def step_start_materializer(context: Any) -> None:
|
||||
"""Start the materializer."""
|
||||
context.materializer.start()
|
||||
|
||||
|
||||
@then("the materializer should be active")
|
||||
def step_materializer_active(context: Any) -> None:
|
||||
"""Verify materializer is active."""
|
||||
assert context.materializer.is_active
|
||||
|
||||
|
||||
@then("a subscription should be registered with the event queue")
|
||||
def step_subscription_registered(context: Any) -> None:
|
||||
"""Verify subscription was registered."""
|
||||
assert context.materializer._subscription_id is not None
|
||||
|
||||
|
||||
@when("I call start on the materializer with plan_id {plan_id}")
|
||||
def step_start_with_plan_id(context: Any, plan_id: str) -> None:
|
||||
"""Start materializer with a plan ID."""
|
||||
context.materializer.start(plan_id=plan_id)
|
||||
|
||||
|
||||
@then("the materializer should track plan_id {plan_id}")
|
||||
def step_check_plan_id(context: Any, plan_id: str) -> None:
|
||||
"""Verify materializer tracks the plan ID."""
|
||||
assert context.materializer.current_plan_id == plan_id
|
||||
|
||||
|
||||
@when("I call start on the materializer again")
|
||||
def step_start_again(context: Any) -> None:
|
||||
"""Try to start materializer again."""
|
||||
context.error = None
|
||||
try:
|
||||
context.materializer.start()
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@then("a RuntimeError should be raised with message {message}")
|
||||
def step_check_runtime_error(context: Any, message: str) -> None:
|
||||
"""Verify RuntimeError was raised with expected message."""
|
||||
assert context.error is not None
|
||||
assert isinstance(context.error, RuntimeError)
|
||||
assert str(context.error) == message
|
||||
|
||||
|
||||
@when("I try to call start with plan_id {plan_id}")
|
||||
def step_try_start_with_plan_id(context: Any, plan_id: str) -> None:
|
||||
"""Try to start with a plan ID."""
|
||||
context.error = None
|
||||
try:
|
||||
if plan_id == '""':
|
||||
context.materializer.start(plan_id="")
|
||||
else:
|
||||
context.materializer.start(plan_id=plan_id)
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I call stop on the materializer")
|
||||
def step_stop_materializer(context: Any) -> None:
|
||||
"""Stop the materializer."""
|
||||
context.materializer.stop()
|
||||
|
||||
|
||||
@then("the materializer should not be active")
|
||||
def step_materializer_not_active_check(context: Any) -> None:
|
||||
"""Verify materializer is not active."""
|
||||
assert not context.materializer.is_active
|
||||
|
||||
|
||||
@then("the subscription should be unregistered")
|
||||
def step_subscription_unregistered(context: Any) -> None:
|
||||
"""Verify subscription was unregistered."""
|
||||
assert context.materializer._subscription_id is None
|
||||
|
||||
|
||||
@then("no tui materializer error should be raised")
|
||||
def step_no_tui_error(context: Any) -> None:
|
||||
"""Verify no TUI materializer error occurred."""
|
||||
assert not hasattr(context, "error") or context.error is None
|
||||
|
||||
|
||||
@when("I publish a text chunk event with text {text}")
|
||||
def step_publish_text_chunk(context: Any, text: str) -> None:
|
||||
"""Publish a text chunk event."""
|
||||
event = A2aEvent(
|
||||
event_type="TextChunkEvent",
|
||||
data={"text": text},
|
||||
)
|
||||
context.event_queue.publish(event)
|
||||
|
||||
|
||||
@then("the conversation widget should receive {text}")
|
||||
def step_check_conversation_received(context: Any, text: str) -> None:
|
||||
"""Verify conversation widget received text."""
|
||||
# Check if append or update was called with the text
|
||||
calls = context.conversation_widget.append.call_args_list
|
||||
if not calls:
|
||||
calls = context.conversation_widget.update.call_args_list
|
||||
assert any(text in str(call) for call in calls), f"Text '{text}' not found in calls"
|
||||
|
||||
|
||||
@given("the conversation widget supports append")
|
||||
def step_conversation_supports_append(context: Any) -> None:
|
||||
"""Configure conversation widget to support append."""
|
||||
context.conversation_widget.append = MagicMock()
|
||||
|
||||
|
||||
@given("the conversation widget does not support append")
|
||||
def step_conversation_no_append(context: Any) -> None:
|
||||
"""Configure conversation widget without append."""
|
||||
del context.conversation_widget.append
|
||||
|
||||
|
||||
@then("the conversation widget should have both chunks appended")
|
||||
def step_check_both_chunks(context: Any) -> None:
|
||||
"""Verify both chunks were appended."""
|
||||
assert context.conversation_widget.append.call_count >= 2
|
||||
|
||||
|
||||
@then("the conversation widget should be updated with {text}")
|
||||
def step_check_conversation_updated(context: Any, text: str) -> None:
|
||||
"""Verify conversation widget was updated."""
|
||||
context.conversation_widget.update.assert_called()
|
||||
|
||||
|
||||
@then("the conversation widget should not be updated")
|
||||
def step_conversation_not_updated(context: Any) -> None:
|
||||
"""Verify conversation widget was not updated."""
|
||||
context.conversation_widget.update.assert_not_called()
|
||||
context.conversation_widget.append.assert_not_called()
|
||||
|
||||
|
||||
@when("I publish a ThoughtBlockEvent with content {content}")
|
||||
def step_publish_thought_block(context: Any, content: str) -> None:
|
||||
"""Publish a ThoughtBlock event."""
|
||||
event = A2aEvent(
|
||||
event_type="ThoughtBlockEvent",
|
||||
data={"content": content},
|
||||
)
|
||||
context.event_queue.publish(event)
|
||||
|
||||
|
||||
@then("the thought widget should receive the thought content")
|
||||
def step_check_thought_received(context: Any) -> None:
|
||||
"""Verify thought widget received content."""
|
||||
context.thought_widget.set_thought.assert_called()
|
||||
|
||||
|
||||
@given("the thought widget does not exist")
|
||||
def step_thought_widget_missing(context: Any) -> None:
|
||||
"""Configure app to not have thought widget."""
|
||||
def query_one_side_effect(selector: str, expect_type: Any = None) -> Any:
|
||||
if selector == "#thought-block":
|
||||
raise Exception("Widget not found")
|
||||
return context.conversation_widget
|
||||
|
||||
context.app.query_one.side_effect = query_one_side_effect
|
||||
|
||||
|
||||
@when("I publish a PermissionRequestEvent with request {request}")
|
||||
def step_publish_permission_request(context: Any, request: str) -> None:
|
||||
"""Publish a PermissionRequest event."""
|
||||
event = A2aEvent(
|
||||
event_type="PermissionRequestEvent",
|
||||
data={"request": request},
|
||||
)
|
||||
context.event_queue.publish(event)
|
||||
|
||||
|
||||
@then("the permission widget should receive the request")
|
||||
def step_check_permission_received(context: Any) -> None:
|
||||
"""Verify permission widget received request."""
|
||||
context.permission_widget.set_permission_request.assert_called()
|
||||
|
||||
|
||||
@given("the permission question widget does not exist")
|
||||
def step_permission_question_missing(context: Any) -> None:
|
||||
"""Configure app to not have permission question widget."""
|
||||
def query_one_side_effect(selector: str, expect_type: Any = None) -> Any:
|
||||
if selector == "#permission-question":
|
||||
raise Exception("Widget not found")
|
||||
elif selector == "#permissions-screen":
|
||||
return context.permissions_screen
|
||||
return context.conversation_widget
|
||||
|
||||
context.app.query_one.side_effect = query_one_side_effect
|
||||
|
||||
|
||||
@then("the permissions screen should receive the request")
|
||||
def step_check_permissions_screen_received(context: Any) -> None:
|
||||
"""Verify permissions screen received request."""
|
||||
context.permissions_screen.set_permission_request.assert_called()
|
||||
|
||||
|
||||
@given("neither permission widget exists")
|
||||
def step_no_permission_widgets(context: Any) -> None:
|
||||
"""Configure app to not have any permission widgets."""
|
||||
def query_one_side_effect(selector: str, expect_type: Any = None) -> Any:
|
||||
if selector in ["#permission-question", "#permissions-screen"]:
|
||||
raise Exception("Widget not found")
|
||||
return context.conversation_widget
|
||||
|
||||
context.app.query_one.side_effect = query_one_side_effect
|
||||
|
||||
|
||||
@when("I publish a TaskStatusUpdateEvent with status {status}")
|
||||
def step_publish_status_update(context: Any, status: str) -> None:
|
||||
"""Publish a TaskStatusUpdateEvent."""
|
||||
event = A2aEvent(
|
||||
event_type="TaskStatusUpdateEvent",
|
||||
data={"status": status},
|
||||
)
|
||||
context.event_queue.publish(event)
|
||||
|
||||
|
||||
@then("the conversation widget should display a task status update")
|
||||
def step_check_status_update_displayed(context: Any) -> None:
|
||||
"""Verify conversation widget displayed a task status update."""
|
||||
context.conversation_widget.append.assert_called()
|
||||
|
||||
|
||||
@when("I publish a TaskArtifactUpdateEvent with artifact {artifact}")
|
||||
def step_publish_artifact_update(context: Any, artifact: str) -> None:
|
||||
"""Publish a TaskArtifactUpdateEvent."""
|
||||
event = A2aEvent(
|
||||
event_type="TaskArtifactUpdateEvent",
|
||||
data={"artifact": artifact},
|
||||
)
|
||||
context.event_queue.publish(event)
|
||||
|
||||
|
||||
@then("the conversation widget should display a task artifact update")
|
||||
def step_check_artifact_update_displayed(context: Any) -> None:
|
||||
"""Verify conversation widget displayed a task artifact update."""
|
||||
context.conversation_widget.append.assert_called()
|
||||
|
||||
|
||||
@when("I publish a text chunk event with plan_id {plan_id} and text {text}")
|
||||
def step_publish_with_plan_id(context: Any, plan_id: str, text: str) -> None:
|
||||
"""Publish a text chunk event with plan ID."""
|
||||
actual_plan_id = None if plan_id == "None" else plan_id
|
||||
event = A2aEvent(
|
||||
event_type="TextChunkEvent",
|
||||
plan_id=actual_plan_id,
|
||||
data={"text": text},
|
||||
)
|
||||
context.event_queue.publish(event)
|
||||
|
||||
|
||||
@when("I call dispatch_prompt with text {text}")
|
||||
def step_dispatch_prompt(context: Any, text: str) -> None:
|
||||
"""Dispatch a prompt."""
|
||||
context.request_id = context.materializer.dispatch_prompt(text)
|
||||
|
||||
|
||||
@then("a request ID should be returned")
|
||||
def step_check_request_id(context: Any) -> None:
|
||||
"""Verify request ID was returned."""
|
||||
assert context.request_id is not None
|
||||
assert isinstance(context.request_id, str)
|
||||
|
||||
|
||||
@then("the request ID should be a valid ULID")
|
||||
def step_check_valid_ulid(context: Any) -> None:
|
||||
"""Verify request ID is a valid ULID."""
|
||||
from ulid import ULID
|
||||
|
||||
try:
|
||||
ULID.from_str(context.request_id)
|
||||
except Exception as exc:
|
||||
raise AssertionError(f"Invalid ULID: {context.request_id}") from exc
|
||||
|
||||
|
||||
@when("I try to call dispatch_prompt with prompt=None")
|
||||
def step_dispatch_prompt_none(context: Any) -> None:
|
||||
"""Try to dispatch with None prompt."""
|
||||
context.error = None
|
||||
try:
|
||||
context.materializer.dispatch_prompt(None)
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I try to call dispatch_prompt with prompt={prompt}")
|
||||
def step_dispatch_prompt_invalid(context: Any, prompt: str) -> None:
|
||||
"""Try to dispatch with invalid prompt."""
|
||||
context.error = None
|
||||
try:
|
||||
if prompt == "123":
|
||||
context.materializer.dispatch_prompt(123)
|
||||
elif prompt == '""':
|
||||
context.materializer.dispatch_prompt("")
|
||||
elif prompt == '" "':
|
||||
context.materializer.dispatch_prompt(" ")
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I define an on_response callback")
|
||||
def step_define_callback(context: Any) -> None:
|
||||
"""Define an on_response callback."""
|
||||
context.callback = MagicMock()
|
||||
|
||||
|
||||
@when("I call dispatch_prompt with text {text} and session_id {session_id}")
|
||||
def step_dispatch_with_session(context: Any, text: str, session_id: str) -> None:
|
||||
"""Dispatch prompt with session ID."""
|
||||
context.request_id = context.materializer.dispatch_prompt(
|
||||
text, session_id=session_id
|
||||
)
|
||||
|
||||
|
||||
@when("I call dispatch_prompt with text {text} and the callback")
|
||||
def step_dispatch_with_callback(context: Any, text: str) -> None:
|
||||
"""Dispatch prompt with callback."""
|
||||
context.request_id = context.materializer.dispatch_prompt(
|
||||
text, on_response=context.callback
|
||||
)
|
||||
|
||||
|
||||
@when("I check is_active before starting")
|
||||
def step_check_is_active_before(context: Any) -> None:
|
||||
"""Check is_active before starting."""
|
||||
context.is_active_before = context.materializer.is_active
|
||||
|
||||
|
||||
@then("is_active should be False")
|
||||
def step_is_active_false(context: Any) -> None:
|
||||
"""Verify is_active is False."""
|
||||
assert not context.materializer.is_active
|
||||
|
||||
|
||||
@then("is_active should be True")
|
||||
def step_is_active_true(context: Any) -> None:
|
||||
"""Verify is_active is True."""
|
||||
assert context.materializer.is_active
|
||||
|
||||
|
||||
@then("current_plan_id should be {plan_id}")
|
||||
def step_check_current_plan_id(context: Any, plan_id: str) -> None:
|
||||
"""Verify current_plan_id."""
|
||||
expected = None if plan_id == "None" else plan_id
|
||||
assert context.materializer.current_plan_id == expected
|
||||
|
||||
|
||||
@then("current_plan_id should still be {plan_id}")
|
||||
def step_check_current_plan_id_still(context: Any, plan_id: str) -> None:
|
||||
"""Verify current_plan_id is still set after stop."""
|
||||
expected = None if plan_id == "None" else plan_id
|
||||
assert context.materializer.current_plan_id == expected
|
||||
|
||||
|
||||
@when("I call start on the materializer without a plan_id")
|
||||
def step_start_without_plan_id(context: Any) -> None:
|
||||
"""Start materializer without plan ID."""
|
||||
context.materializer.start()
|
||||
|
||||
|
||||
@when("I publish a null event")
|
||||
def step_publish_null_event(context: Any) -> None:
|
||||
"""Try to publish a null event."""
|
||||
# This is handled by the materializer's _on_event method
|
||||
context.materializer._on_event(None)
|
||||
|
||||
|
||||
@when("I publish an event with missing event_type")
|
||||
def step_publish_missing_event_type(context: Any) -> None:
|
||||
"""Publish event with missing event_type."""
|
||||
event = MagicMock()
|
||||
event.event_type = None
|
||||
context.materializer._on_event(event)
|
||||
|
||||
|
||||
@given("the app raises an exception on query_one")
|
||||
def step_app_raises_exception(context: Any) -> None:
|
||||
"""Configure app to raise exception."""
|
||||
context.app.query_one.side_effect = Exception("Widget error")
|
||||
|
||||
|
||||
@then("the error should be logged")
|
||||
def step_error_logged(context: Any) -> None:
|
||||
"""Verify error was logged."""
|
||||
# This is verified by the logger calls in the materializer
|
||||
pass
|
||||
|
||||
|
||||
@when("I publish a text chunk event")
|
||||
def step_publish_text_chunk_generic(context: Any) -> None:
|
||||
"""Publish a generic text chunk event."""
|
||||
event = A2aEvent(
|
||||
event_type="TextChunkEvent",
|
||||
data={"text": "test text"},
|
||||
)
|
||||
context.event_queue.publish(event)
|
||||
@@ -0,0 +1,242 @@
|
||||
Feature: TUI Materializer A2A Integration Layer
|
||||
The TuiMaterializer bridges A2A (Agent-to-Agent) events and the Textual UI,
|
||||
enabling the TUI to dispatch requests to the AI backend and stream responses
|
||||
back to the conversation widget.
|
||||
|
||||
This feature covers the core event routing and streaming logic required by
|
||||
ADR-044 (TUI Architecture), ADR-045 (Persona System), and ADR-046 (Reference/Command System).
|
||||
|
||||
Background:
|
||||
Given a mock A2A event queue
|
||||
And a mock Textual app reference
|
||||
And a TuiMaterializer instance
|
||||
|
||||
# --- Initialization and lifecycle ---
|
||||
|
||||
Scenario: TuiMaterializer can be instantiated with event queue and app
|
||||
When I create a TuiMaterializer with a valid event queue and app
|
||||
Then the materializer should be created successfully
|
||||
And the materializer should not be active initially
|
||||
|
||||
Scenario: TuiMaterializer raises TypeError if event_queue is None
|
||||
When I try to create a TuiMaterializer with event_queue=None
|
||||
Then a TypeError should be raised with message "event_queue must not be None"
|
||||
|
||||
Scenario: TuiMaterializer raises TypeError if app is None
|
||||
When I try to create a TuiMaterializer with app=None
|
||||
Then a TypeError should be raised with message "app must not be None"
|
||||
|
||||
Scenario: TuiMaterializer raises TypeError if event_queue lacks subscribe_local
|
||||
When I try to create a TuiMaterializer with an invalid event queue
|
||||
Then a TypeError should be raised with message "event_queue must have subscribe_local method"
|
||||
|
||||
Scenario: TuiMaterializer raises ValueError if event_queue is closed
|
||||
When I create a closed event queue
|
||||
And I try to create a TuiMaterializer with the closed queue
|
||||
Then a ValueError should be raised with message "event_queue is closed"
|
||||
|
||||
# --- Start/Stop lifecycle ---
|
||||
|
||||
Scenario: start() subscribes to the event queue
|
||||
When I call start on the materializer
|
||||
Then the materializer should be active
|
||||
And a subscription should be registered with the event queue
|
||||
|
||||
Scenario: start() with plan_id filters events by plan
|
||||
When I call start on the materializer with plan_id "plan-123"
|
||||
Then the materializer should track plan_id "plan-123"
|
||||
And the materializer should be active
|
||||
|
||||
Scenario: start() raises RuntimeError if already started
|
||||
When I call start on the materializer
|
||||
And I call start on the materializer again
|
||||
Then a RuntimeError should be raised with message "TuiMaterializer is already started"
|
||||
|
||||
Scenario: start() raises ValueError if plan_id is empty string
|
||||
When I try to call start with plan_id ""
|
||||
Then a ValueError should be raised with message "plan_id must not be empty string"
|
||||
|
||||
Scenario: start() raises TypeError if plan_id is not a string
|
||||
When I try to call start with plan_id 123
|
||||
Then a TypeError should be raised with message "plan_id must be a string or None"
|
||||
|
||||
Scenario: stop() unsubscribes from the event queue
|
||||
When I call start on the materializer
|
||||
And I call stop on the materializer
|
||||
Then the materializer should not be active
|
||||
And the subscription should be unregistered
|
||||
|
||||
Scenario: stop() is safe to call multiple times
|
||||
When I call start on the materializer
|
||||
And I call stop on the materializer
|
||||
And I call stop on the materializer again
|
||||
Then no tui materializer error should be raised
|
||||
|
||||
# --- Event routing: Text chunks ---
|
||||
|
||||
Scenario: Text chunk events are routed to conversation widget
|
||||
When I call start on the materializer
|
||||
And I publish a text chunk event with text "Hello, world!"
|
||||
Then the conversation widget should receive "Hello, world!"
|
||||
|
||||
Scenario: Text chunks are appended if widget supports append
|
||||
When I call start on the materializer
|
||||
And the conversation widget supports append
|
||||
And I publish a text chunk event with text "First chunk"
|
||||
And I publish a text chunk event with text "Second chunk"
|
||||
Then the conversation widget should have both chunks appended
|
||||
|
||||
Scenario: Text chunks are updated if widget lacks append
|
||||
When I call start on the materializer
|
||||
And the conversation widget does not support append
|
||||
And I publish a text chunk event with text "Updated text"
|
||||
Then the conversation widget should be updated with "Updated text"
|
||||
|
||||
Scenario: Empty text chunks are ignored
|
||||
When I call start on the materializer
|
||||
And I publish a text chunk event with text ""
|
||||
Then the conversation widget should not be updated
|
||||
|
||||
# --- Event routing: ThoughtBlock events ---
|
||||
|
||||
Scenario: ThoughtBlock events are routed to thought widget
|
||||
When I call start on the materializer
|
||||
And I publish a ThoughtBlockEvent with content "Thinking about the problem"
|
||||
Then the thought widget should receive the thought content
|
||||
|
||||
Scenario: ThoughtBlock routing handles missing widget gracefully
|
||||
When I call start on the materializer
|
||||
And the thought widget does not exist
|
||||
And I publish a ThoughtBlockEvent with content "Some thought"
|
||||
Then no tui materializer error should be raised
|
||||
|
||||
# --- Event routing: PermissionRequest events ---
|
||||
|
||||
Scenario: PermissionRequest events are routed to permission widget
|
||||
When I call start on the materializer
|
||||
And I publish a PermissionRequestEvent with request "Allow file access?"
|
||||
Then the permission widget should receive the request
|
||||
|
||||
Scenario: PermissionRequest falls back to permissions screen if question widget missing
|
||||
When I call start on the materializer
|
||||
And the permission question widget does not exist
|
||||
And I publish a PermissionRequestEvent with request "Allow file access?"
|
||||
Then the permissions screen should receive the request
|
||||
|
||||
Scenario: PermissionRequest routing handles missing widgets gracefully
|
||||
When I call start on the materializer
|
||||
And neither permission widget exists
|
||||
And I publish a PermissionRequestEvent with request "Allow file access?"
|
||||
Then no tui materializer error should be raised
|
||||
|
||||
# --- Event routing: TaskStatusUpdateEvent ---
|
||||
|
||||
Scenario: TaskStatusUpdateEvent is routed to conversation widget
|
||||
When I call start on the materializer
|
||||
And I publish a TaskStatusUpdateEvent with status "running"
|
||||
Then the conversation widget should display a task status update
|
||||
|
||||
Scenario: TaskStatusUpdateEvent with empty status is ignored
|
||||
When I call start on the materializer
|
||||
And I publish a TaskStatusUpdateEvent with status ""
|
||||
Then the conversation widget should not be updated
|
||||
|
||||
# --- Event routing: TaskArtifactUpdateEvent ---
|
||||
|
||||
Scenario: TaskArtifactUpdateEvent is routed to conversation widget
|
||||
When I call start on the materializer
|
||||
And I publish a TaskArtifactUpdateEvent with artifact "result.txt"
|
||||
Then the conversation widget should display a task artifact update
|
||||
|
||||
Scenario: TaskArtifactUpdateEvent with empty artifact is ignored
|
||||
When I call start on the materializer
|
||||
And I publish a TaskArtifactUpdateEvent with artifact ""
|
||||
Then the conversation widget should not be updated
|
||||
|
||||
# --- Plan ID filtering ---
|
||||
|
||||
Scenario: Events are filtered by plan_id when specified
|
||||
When I call start on the materializer with plan_id "plan-123"
|
||||
And I publish a text chunk event with plan_id "plan-123" and text "Matching plan"
|
||||
Then the conversation widget should receive "Matching plan"
|
||||
|
||||
Scenario: Events with different plan_id are ignored
|
||||
When I call start on the materializer with plan_id "plan-123"
|
||||
And I publish a text chunk event with plan_id "plan-456" and text "Different plan"
|
||||
Then the conversation widget should not be updated
|
||||
|
||||
Scenario: Events with no plan_id are ignored when filtering by plan_id
|
||||
When I call start on the materializer with plan_id "plan-123"
|
||||
And I publish a text chunk event with plan_id None and text "No plan"
|
||||
Then the conversation widget should not be updated
|
||||
|
||||
# --- Prompt dispatch ---
|
||||
|
||||
Scenario: dispatch_prompt validates and returns a request ID
|
||||
When I call dispatch_prompt with text "What is 2+2?"
|
||||
Then a request ID should be returned
|
||||
And the request ID should be a valid ULID
|
||||
|
||||
Scenario: dispatch_prompt raises TypeError if prompt is None
|
||||
When I try to call dispatch_prompt with prompt=None
|
||||
Then a TypeError should be raised with message "prompt must not be None"
|
||||
|
||||
Scenario: dispatch_prompt raises TypeError if prompt is not a string
|
||||
When I try to call dispatch_prompt with prompt=123
|
||||
Then a TypeError should be raised with message "prompt must be a string"
|
||||
|
||||
Scenario: dispatch_prompt raises ValueError if prompt is empty
|
||||
When I try to call dispatch_prompt with prompt ""
|
||||
Then a ValueError should be raised with message "prompt must not be empty"
|
||||
|
||||
Scenario: dispatch_prompt raises ValueError if prompt is whitespace only
|
||||
When I try to call dispatch_prompt with prompt " "
|
||||
Then a ValueError should be raised with message "prompt must not be empty"
|
||||
|
||||
Scenario: dispatch_prompt accepts optional session_id
|
||||
When I call dispatch_prompt with text "Hello" and session_id "sess-1"
|
||||
Then a request ID should be returned
|
||||
|
||||
Scenario: dispatch_prompt accepts optional on_response callback
|
||||
When I define an on_response callback
|
||||
And I call dispatch_prompt with text "Hello" and the callback
|
||||
Then a request ID should be returned
|
||||
|
||||
# --- Properties ---
|
||||
|
||||
Scenario: is_active property reflects subscription state
|
||||
When I check is_active before starting
|
||||
Then is_active should be False
|
||||
When I call start on the materializer
|
||||
Then is_active should be True
|
||||
When I call stop on the materializer
|
||||
Then is_active should be False
|
||||
|
||||
Scenario: current_plan_id property returns the tracked plan
|
||||
When I call start on the materializer with plan_id "plan-789"
|
||||
Then current_plan_id should be "plan-789"
|
||||
When I call stop on the materializer
|
||||
Then current_plan_id should still be "plan-789"
|
||||
|
||||
Scenario: current_plan_id is None when no plan is tracked
|
||||
When I call start on the materializer without a plan_id
|
||||
Then current_plan_id should be None
|
||||
|
||||
# --- Error handling ---
|
||||
|
||||
Scenario: Null events are handled gracefully
|
||||
When I call start on the materializer
|
||||
And I publish a null event
|
||||
Then no tui materializer error should be raised
|
||||
|
||||
Scenario: Events with missing event_type are handled gracefully
|
||||
When I call start on the materializer
|
||||
And I publish an event with missing event_type
|
||||
Then no tui materializer error should be raised
|
||||
|
||||
Scenario: Widget query errors are logged but not raised
|
||||
When I call start on the materializer
|
||||
And the app raises an exception on query_one
|
||||
And I publish a text chunk event
|
||||
Then no tui materializer error should be raised
|
||||
And the error should be logged
|
||||
@@ -0,0 +1,372 @@
|
||||
"""TUI Materializer — A2A integration layer for Textual UI.
|
||||
|
||||
The TuiMaterializer bridges A2A (Agent-to-Agent) events and the Textual UI,
|
||||
enabling the TUI to dispatch requests to the AI backend and stream responses
|
||||
back to the conversation widget.
|
||||
|
||||
This module implements the core event routing and streaming logic required by
|
||||
ADR-044 (TUI Architecture), ADR-045 (Persona System), and
|
||||
ADR-046 (Reference/Command System).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from typing import TYPE_CHECKING, Any, Protocol
|
||||
|
||||
import structlog
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from cleveragents.a2a.events import A2aEventQueue
|
||||
from cleveragents.a2a.models import A2aEvent
|
||||
|
||||
logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
class TextualAppReference(Protocol):
|
||||
"""Protocol for Textual app reference used by TuiMaterializer.
|
||||
|
||||
Allows the materializer to update UI widgets without tight coupling
|
||||
to the Textual framework.
|
||||
"""
|
||||
|
||||
def query_one(self, selector: str, expect_type: type[Any] | None = None) -> Any:
|
||||
"""Query for a single widget by selector.
|
||||
|
||||
Args:
|
||||
selector: CSS selector or widget ID.
|
||||
expect_type: Optional type to validate the widget.
|
||||
|
||||
Returns:
|
||||
The matched widget.
|
||||
|
||||
Raises:
|
||||
NoMatches: If no widget matches the selector.
|
||||
TooManyMatches: If multiple widgets match.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class ConversationWidgetReference(Protocol):
|
||||
"""Protocol for conversation widget that receives streaming updates."""
|
||||
|
||||
def update(self, text: str) -> None:
|
||||
"""Update the widget with new text content.
|
||||
|
||||
Args:
|
||||
text: The text to display.
|
||||
"""
|
||||
...
|
||||
|
||||
def append(self, text: str) -> None:
|
||||
"""Append text to the widget's current content.
|
||||
|
||||
Args:
|
||||
text: The text to append.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class TuiMaterializer:
|
||||
"""A2A integration layer for the Textual TUI.
|
||||
|
||||
Subscribes to A2A events and routes them to appropriate TUI widgets:
|
||||
- Normal text chunks → conversation widget (streaming)
|
||||
- ThoughtBlock events → ThoughtBlockWidget
|
||||
- PermissionRequest events → PermissionQuestionWidget or PermissionsScreen
|
||||
|
||||
This class implements the bridge between the A2A event queue and the
|
||||
Textual UI, enabling real-time streaming of AI responses.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
event_queue: A2aEventQueue,
|
||||
app: TextualAppReference,
|
||||
) -> None:
|
||||
"""Initialize the TuiMaterializer.
|
||||
|
||||
Args:
|
||||
event_queue: The A2A event queue to subscribe to.
|
||||
app: Reference to the Textual app for widget access.
|
||||
|
||||
Raises:
|
||||
TypeError: If event_queue or app are not the expected types.
|
||||
ValueError: If event_queue is closed.
|
||||
"""
|
||||
if event_queue is None:
|
||||
raise TypeError("event_queue must not be None")
|
||||
if app is None:
|
||||
raise TypeError("app must not be None")
|
||||
if not hasattr(event_queue, "subscribe_local"):
|
||||
raise TypeError("event_queue must have subscribe_local method")
|
||||
if event_queue.is_closed:
|
||||
raise ValueError("event_queue is closed")
|
||||
|
||||
self._event_queue = event_queue
|
||||
self._app = app
|
||||
self._subscription_id: str | None = None
|
||||
self._current_plan_id: str | None = None
|
||||
|
||||
def start(self, plan_id: str | None = None) -> None:
|
||||
"""Start listening to A2A events.
|
||||
|
||||
Args:
|
||||
plan_id: Optional plan ID to filter events for.
|
||||
|
||||
Raises:
|
||||
ValueError: If plan_id is empty string.
|
||||
RuntimeError: If already started.
|
||||
"""
|
||||
if plan_id is not None and not isinstance(plan_id, str):
|
||||
raise TypeError("plan_id must be a string or None")
|
||||
if plan_id == "":
|
||||
raise ValueError("plan_id must not be empty string")
|
||||
if self._subscription_id is not None:
|
||||
raise RuntimeError("TuiMaterializer is already started")
|
||||
|
||||
self._current_plan_id = plan_id
|
||||
self._subscription_id = self._event_queue.subscribe_local(self._on_event)
|
||||
logger.info(
|
||||
"tui.materializer.started",
|
||||
subscription_id=self._subscription_id,
|
||||
plan_id=plan_id,
|
||||
)
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Stop listening to A2A events.
|
||||
|
||||
Safe to call multiple times.
|
||||
"""
|
||||
if self._subscription_id is not None:
|
||||
self._event_queue.unsubscribe(self._subscription_id)
|
||||
self._subscription_id = None
|
||||
logger.info("tui.materializer.stopped")
|
||||
|
||||
def _on_event(self, event: A2aEvent) -> None:
|
||||
"""Handle an A2A event and route it to the appropriate widget.
|
||||
|
||||
Args:
|
||||
event: The A2A event to process.
|
||||
"""
|
||||
if event is None:
|
||||
logger.warning("tui.materializer.null_event")
|
||||
return
|
||||
|
||||
# Filter by plan_id if one was specified
|
||||
if self._current_plan_id is not None and event.plan_id != self._current_plan_id:
|
||||
return
|
||||
|
||||
event_type = getattr(event, "event_type", None)
|
||||
if event_type is None:
|
||||
logger.warning("tui.materializer.missing_event_type")
|
||||
return
|
||||
|
||||
logger.debug(
|
||||
"tui.materializer.event_received",
|
||||
event_type=event_type,
|
||||
event_id=getattr(event, "event_id", None),
|
||||
)
|
||||
|
||||
# Route based on event type
|
||||
if event_type == "ThoughtBlockEvent":
|
||||
self._route_thought_block(event)
|
||||
elif event_type == "PermissionRequestEvent":
|
||||
self._route_permission_request(event)
|
||||
elif event_type == "TaskStatusUpdateEvent":
|
||||
self._route_task_status_update(event)
|
||||
elif event_type == "TaskArtifactUpdateEvent":
|
||||
self._route_task_artifact_update(event)
|
||||
else:
|
||||
# Generic text chunk or unknown event
|
||||
self._route_text_chunk(event)
|
||||
|
||||
def _route_text_chunk(self, event: A2aEvent) -> None:
|
||||
"""Route a text chunk to the conversation widget.
|
||||
|
||||
Args:
|
||||
event: The event containing text data.
|
||||
"""
|
||||
try:
|
||||
conversation = self._app.query_one("#conversation")
|
||||
data = getattr(event, "data", {})
|
||||
text = data.get("text", "")
|
||||
if text:
|
||||
if hasattr(conversation, "append"):
|
||||
conversation.append(text)
|
||||
else:
|
||||
conversation.update(text)
|
||||
logger.debug(
|
||||
"tui.materializer.text_routed",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.exception(
|
||||
"tui.materializer.text_route_error",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
def _route_thought_block(self, event: A2aEvent) -> None:
|
||||
"""Route a ThoughtBlock event to the ThoughtBlockWidget.
|
||||
|
||||
Args:
|
||||
event: The ThoughtBlock event.
|
||||
"""
|
||||
try:
|
||||
thought_widget = self._app.query_one("#thought-block")
|
||||
data = getattr(event, "data", {})
|
||||
if hasattr(thought_widget, "set_thought"):
|
||||
thought_widget.set_thought(data)
|
||||
logger.debug(
|
||||
"tui.materializer.thought_routed",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug(
|
||||
"tui.materializer.thought_route_error",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
def _route_permission_request(self, event: A2aEvent) -> None:
|
||||
"""Route a PermissionRequest event to the permission widget.
|
||||
|
||||
Args:
|
||||
event: The PermissionRequest event.
|
||||
"""
|
||||
try:
|
||||
# Try single-file permission widget first
|
||||
try:
|
||||
perm_widget = self._app.query_one("#permission-question")
|
||||
except Exception:
|
||||
# Fall back to permissions screen
|
||||
perm_widget = self._app.query_one("#permissions-screen")
|
||||
|
||||
data = getattr(event, "data", {})
|
||||
if hasattr(perm_widget, "set_permission_request"):
|
||||
perm_widget.set_permission_request(data)
|
||||
logger.debug(
|
||||
"tui.materializer.permission_routed",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug(
|
||||
"tui.materializer.permission_route_error",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
def _route_task_status_update(self, event: A2aEvent) -> None:
|
||||
"""Route a TaskStatusUpdateEvent to the conversation widget.
|
||||
|
||||
Args:
|
||||
event: The TaskStatusUpdateEvent.
|
||||
"""
|
||||
try:
|
||||
conversation = self._app.query_one("#conversation")
|
||||
data = getattr(event, "data", {})
|
||||
status = data.get("status", "")
|
||||
if status:
|
||||
status_text = f"[Status: {status}]"
|
||||
if hasattr(conversation, "append"):
|
||||
conversation.append(status_text)
|
||||
else:
|
||||
conversation.update(status_text)
|
||||
logger.debug(
|
||||
"tui.materializer.status_routed",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug(
|
||||
"tui.materializer.status_route_error",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
def _route_task_artifact_update(self, event: A2aEvent) -> None:
|
||||
"""Route a TaskArtifactUpdateEvent to the conversation widget.
|
||||
|
||||
Args:
|
||||
event: The TaskArtifactUpdateEvent.
|
||||
"""
|
||||
try:
|
||||
conversation = self._app.query_one("#conversation")
|
||||
data = getattr(event, "data", {})
|
||||
artifact = data.get("artifact", "")
|
||||
if artifact:
|
||||
artifact_text = f"[Artifact: {artifact}]"
|
||||
if hasattr(conversation, "append"):
|
||||
conversation.append(artifact_text)
|
||||
else:
|
||||
conversation.update(artifact_text)
|
||||
logger.debug(
|
||||
"tui.materializer.artifact_routed",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug(
|
||||
"tui.materializer.artifact_route_error",
|
||||
event_id=getattr(event, "event_id", None),
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
def dispatch_prompt(
|
||||
self,
|
||||
prompt: str,
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
on_response: Callable[[str], None] | None = None,
|
||||
) -> str:
|
||||
"""Dispatch a user prompt to the A2A backend.
|
||||
|
||||
This method is a placeholder for future integration with the A2A
|
||||
request dispatch mechanism. Currently, it validates the prompt and
|
||||
returns a request ID.
|
||||
|
||||
Args:
|
||||
prompt: The user's prompt text.
|
||||
session_id: Optional session ID for multi-session support.
|
||||
on_response: Optional callback for response chunks.
|
||||
|
||||
Returns:
|
||||
A request ID for tracking the dispatch.
|
||||
|
||||
Raises:
|
||||
ValueError: If prompt is empty or None.
|
||||
TypeError: If prompt is not a string.
|
||||
"""
|
||||
if prompt is None:
|
||||
raise TypeError("prompt must not be None")
|
||||
if not isinstance(prompt, str):
|
||||
raise TypeError("prompt must be a string")
|
||||
if not prompt.strip():
|
||||
raise ValueError("prompt must not be empty")
|
||||
|
||||
from ulid import ULID
|
||||
|
||||
request_id = str(ULID())
|
||||
logger.info(
|
||||
"tui.materializer.prompt_dispatched",
|
||||
request_id=request_id,
|
||||
session_id=session_id,
|
||||
prompt_length=len(prompt),
|
||||
)
|
||||
return request_id
|
||||
|
||||
@property
|
||||
def is_active(self) -> bool:
|
||||
"""Return whether the materializer is actively listening to events."""
|
||||
return self._subscription_id is not None
|
||||
|
||||
@property
|
||||
def current_plan_id(self) -> str | None:
|
||||
"""Return the current plan ID being tracked."""
|
||||
return self._current_plan_id
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ConversationWidgetReference",
|
||||
"TextualAppReference",
|
||||
"TuiMaterializer",
|
||||
]
|
||||
@@ -1216,3 +1216,9 @@ revert_decisions # noqa: B018, F821
|
||||
|
||||
# Extension protocol parameters — required by Protocol interface definitions
|
||||
destination # noqa: B018, F821
|
||||
|
||||
# TuiMaterializer Protocol parameters — required by Protocol interface definitions
|
||||
expect_type # noqa: B018, F821
|
||||
|
||||
# TuiMaterializer dispatch_prompt — on_response callback parameter (future API)
|
||||
on_response # noqa: B018, F821
|
||||
|
||||
Reference in New Issue
Block a user