feat(tui): implement TuiMaterializer bridging A2A event queue to conversation view with ThoughtBlockWidget #11179
@@ -5,6 +5,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
- **`feat(tui): TuiMaterializer bridges A2A event queue to conversation view with ThoughtBlockWidget` (**#5326**):** Implemented `TuiMaterializer` class in `src/cleveragents/tui/materializers.py` that subscribes to an `A2aEventQueue`, translates inbound `TaskStatusUpdateEvent` and `TaskArtifactUpdateEvent` instances into ``ThoughtBlockWidget``-style entries rendered in the TUI conversation stream. Status events produce collapsed thought blocks with status icons (checkmark, cross, warning) showing phase/state timestamps; artifact events produce expanded thought blocks rendering file modifications, diff URLs, and structured metadata. Includes headless buffer for test-inspectable output, subscription lifecycle (`start`/`stop`), and graceful degradation when Textual is unavailable. BDD test coverage in `features/tui_materializer.feature`. Part of Epic (v3.7.0 milestone — TUI Implementation).
|
||||
|
||||
- **`task-implementor` posts work-started notification comments** (#11031): Both
|
||||
the `issue_impl` and `pr_fix` procedures now post an informational "work
|
||||
started" comment to the Forgejo issue/PR before beginning implementation.
|
||||
|
||||
@@ -43,3 +43,5 @@ Below are some of the specific details of various contributions.
|
||||
* HAL 9000 has contributed database resource types (PostgreSQL, SQLite) with transaction-based sandbox strategy: implemented ``DatabaseResourceHandler`` providing full CRUD operations (`read`, `write`, `delete`, `list_children`) and connection validation with automatic credential masking for PostgreSQL and SQLite backends. Includes ``TransactionSandbox`` infrastructure wired into ``SandboxFactory``, BDD test coverage in ``features/database_resources.feature``, and Robot Framework integration tests in ``robot/database_resources.robot`` (PR #10591 / issue #8608, Epic #8568).
|
||||
* HAL 9000 has contributed the agents plan rollback command (PR #8674 / issue #8557): implemented checkpoint-based plan state restoration with the `agents plan rollback <plan-id> [<checkpoint-id>]` CLI command as part of Epic #8493, enabling plans to be restored to previous checkpoints, discarding post-checkpoint decisions, and resuming execution from the rolled-back state. Supported by `--yes/-y`, `--to-checkpoint`, and `--format/-f` flags. Includes comprehensive BDD test coverage (>= 97%) for rollback, decision discarding, and plan resume functionality.
|
||||
* HAL 9000 has contributed the PyYAML security upgrade (PR #11012 / issue #9055): added `pyyaml>=6.0.3` dependency constraint to address known YAML parsing vulnerabilities.
|
||||
|
||||
* HAL 9000 has contributed the TuiMaterializer bridge between A2A event queue and TUI conversation view via thought blocks (PR #11179 / issue #5326): implemented `TuiMaterializer` class that subscribes to `A2aEventQueue`, translates `TaskStatusUpdateEvent` and `TaskArtifactUpdateEvent` events into ``ThoughtBlockWidget``-style entries with status icons, phase metadata, artifact file listings, and diff URLs. Includes subscription lifecycle management, headless buffer for testing, graceful Textual-degradation fallback, and comprehensive BDD test coverage (17 scenarios). Part of v3.7.0 milestone (Epic — TUI Implementation).
|
||||
|
||||
@@ -0,0 +1,279 @@
|
||||
"""Step definitions for TUI Materializer — A2A event queue bridge."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
|
||||
from cleveragents.a2a.events import A2aEventQueue
|
||||
from cleveragents.a2a.models import A2aEvent
|
||||
|
||||
|
||||
# ── Background helpers (shared by scenarios) ───────────────────────────
|
||||
|
||||
|
||||
def _make_queue() -> A2aEventQueue:
|
||||
"""Create a fresh in-memory event queue for testing."""
|
||||
return A2aEventQueue()
|
||||
|
||||
|
||||
def _captured_view() -> tuple[list[str], list[bool], Any]:
|
||||
"""Return (texts, flags, callback) for capturing conversation_view."""
|
||||
texts: list[str] = []
|
||||
flags: list[bool] = []
|
||||
|
||||
def _view(text: str, expanded: bool) -> None:
|
||||
texts.append(text)
|
||||
flags.append(expanded)
|
||||
|
||||
return texts, flags, _view
|
||||
|
||||
|
||||
# ── Given steps ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@given("a fresh A2aEventQueue")
|
||||
def step_queue(context: Any) -> None:
|
||||
context._a2a_queue = _make_queue()
|
||||
|
||||
|
||||
@given("a captured conversation_view callback")
|
||||
def step_captured_view(context: Any) -> None:
|
||||
texts, flags, cb = _captured_view()
|
||||
context._conv_texts = texts
|
||||
context._conv_flags = flags
|
||||
context._conv_callback = cb
|
||||
|
||||
|
||||
# ── When steps ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@when("I start the TuiMaterializer with the event queue")
|
||||
def step_start(context: Any) -> None:
|
||||
from cleveragents.tui.materializers import TuiMaterializer
|
||||
|
||||
cb = context._conv_callback
|
||||
|
||||
tm = TuiMaterializer(context._a2a_queue, cb)
|
||||
tm.start()
|
||||
context._tui_mat = tm
|
||||
|
||||
|
||||
@when("I stop the TuiMaterializer")
|
||||
def step_stop(context: Any) -> None:
|
||||
context._tui_mat.stop()
|
||||
|
||||
|
||||
@when("I start the TuiMaterializer again")
|
||||
def step_start_again(context: Any) -> None:
|
||||
context._tui_mat.start() # should be no-op
|
||||
|
||||
|
||||
@given('I publish a TaskStatusUpdateEvent with status "{status}" and phase "{phase}"')
|
||||
def step_publish_status(context: Any, status: str, phase: str | None) -> None:
|
||||
"""Publish a TaskStatusUpdateEvent (phase=None → null).
|
||||
|
||||
Behave passes unquoted ``null`` as the literal string ``"null"``, so we
|
||||
normalise that to Python :class:`None` to match the intended semantics.
|
||||
"""
|
||||
_phase = None if phase == "null" else phase
|
||||
event = A2aEvent(
|
||||
event_type="TaskStatusUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"status": status, "phase": _phase if _phase else "n/a"},
|
||||
)
|
||||
context._a2a_queue.publish(event)
|
||||
|
||||
|
||||
@given('I publish a TaskStatusUpdateEvent with status "{status}" and no phase')
|
||||
def step_publish_status_no_phase(context: Any, status: str) -> None:
|
||||
event = A2aEvent(
|
||||
event_type="TaskStatusUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"status": status},
|
||||
)
|
||||
context._a2a_queue.publish(event)
|
||||
|
||||
|
||||
@given("I publish a TaskStatusUpdateEvent with artifact metadata")
|
||||
def step_publish_artifact(context: Any) -> None:
|
||||
event = A2aEvent(
|
||||
event_type="TaskArtifactUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={
|
||||
"status": "complete",
|
||||
"files": ["src/example.py"],
|
||||
"diff_url": "https://example.com/diff",
|
||||
},
|
||||
)
|
||||
context._a2a_queue.publish(event)
|
||||
|
||||
|
||||
@given("I publish a TaskArtifactUpdateEvent with three modified files")
|
||||
def step_publish_artifact_three_files(context: Any) -> None:
|
||||
files = [
|
||||
{"path": "src/main.py", "type": "modified"},
|
||||
{"path": "tests/test_main.py", "type": "modified"},
|
||||
{"path": "docs/readme.md", "type": "added"},
|
||||
]
|
||||
event = A2aEvent(
|
||||
event_type="TaskArtifactUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"files": files},
|
||||
)
|
||||
context._a2a_queue.publish(event)
|
||||
|
||||
|
||||
@given("I publish a TaskArtifactUpdateEvent with a single file path artifact")
|
||||
def step_publish_artifact_single_file(context: Any) -> None:
|
||||
event = A2aEvent(
|
||||
event_type="TaskArtifactUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"files": "src/only.py"},
|
||||
)
|
||||
context._a2a_queue.publish(event)
|
||||
|
||||
|
||||
@given("I publish a TaskArtifactUpdateEvent with a diff URL artifact")
|
||||
def step_publish_artifact_diff_url(context: Any) -> None:
|
||||
event = A2aEvent(
|
||||
event_type="TaskArtifactUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"diff_url": "https://example.org/diff/abc123"},
|
||||
)
|
||||
context._a2a_queue.publish(event)
|
||||
|
||||
|
||||
@given('I publish an A2aEvent of type "{event_type}"')
|
||||
def step_publish_generic(context: Any, event_type: str) -> None:
|
||||
event = A2aEvent(
|
||||
event_type=event_type,
|
||||
plan_id="test-plan-001",
|
||||
data={"note": "custom"},
|
||||
)
|
||||
context._a2a_queue.publish(event)
|
||||
|
||||
|
||||
@when("the TuiMaterializer processes it")
|
||||
def step_process(context: Any) -> None:
|
||||
# Events are published synchronously above; the subscription callback
|
||||
# fires inline. Nothing else needed here.
|
||||
pass
|
||||
|
||||
|
||||
@when("I publish two events with the materializer running")
|
||||
def step_publish_two_events(context: Any) -> None:
|
||||
event1 = A2aEvent(
|
||||
event_type="TaskStatusUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"status": "running", "phase": "strategize"},
|
||||
)
|
||||
event2 = A2aEvent(
|
||||
event_type="TaskArtifactUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"files": ["src/main.py"]},
|
||||
)
|
||||
context._a2a_queue.publish(event1)
|
||||
context._a2a_queue.publish(event2)
|
||||
|
||||
|
||||
@when("I publish two events and wait")
|
||||
def step_publish_two_and_wait(context: Any) -> None:
|
||||
event1 = A2aEvent(
|
||||
event_type="TaskStatusUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"status": "running"},
|
||||
)
|
||||
event2 = A2aEvent(
|
||||
event_type="TaskArtifactUpdateEvent",
|
||||
plan_id="test-plan-001",
|
||||
data={"files": ["src/test.py"]},
|
||||
)
|
||||
context._a2a_queue.publish(event1)
|
||||
context._a2a_queue.publish(event2)
|
||||
|
||||
|
||||
@when("I call clear_buffer on the TuiMaterializer")
|
||||
def step_clear_buffer(context: Any) -> None:
|
||||
context._tui_mat.clear_buffer()
|
||||
|
||||
|
||||
# ── Then steps ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@then("the materializer should be started")
|
||||
def step_mat_started(context: Any) -> None:
|
||||
assert context._tui_mat.is_started is True
|
||||
|
||||
|
||||
@then("the materializer should have an active subscription")
|
||||
def step_mat_has_subscription(context: Any) -> None:
|
||||
assert context._tui_mat._subscription_id is not None
|
||||
|
||||
|
||||
@then("the materializer should not be started")
|
||||
def step_mat_not_started(context: Any) -> None:
|
||||
assert context._tui_mat.is_started is False
|
||||
|
||||
|
||||
@then("the materializer should have no active subscription")
|
||||
def step_mat_no_subscription(context: Any) -> None:
|
||||
assert context._tui_mat._subscription_id is None
|
||||
|
||||
|
||||
@then("the conversation view should contain {expected}")
|
||||
def step_conv_contains(context: Any, expected: str) -> None:
|
||||
found = any(expected in t for t in context._conv_texts)
|
||||
assert found, (
|
||||
f"Expected '{expected}' in conversation texts; got: {context._conv_texts}"
|
||||
)
|
||||
|
||||
|
||||
@then("the event entry should be in collapsed state")
|
||||
def step_collapsed(context: Any) -> None:
|
||||
last_flag = context._conv_flags[-1] if context._conv_flags else False
|
||||
assert last_flag is False, f"Expected collapsed (False), got {last_flag}"
|
||||
|
||||
|
||||
@then("the event entry should be in expanded state")
|
||||
def step_expanded(context: Any) -> None:
|
||||
last_flag = context._conv_flags[-1] if context._conv_flags else False
|
||||
assert last_flag is True, f"Expected expanded (True), got {last_flag}"
|
||||
|
||||
|
||||
@then("the widget buffer should have exactly {n:d} entries")
|
||||
def step_buffer_count(context: Any, n: int) -> None:
|
||||
buf = context._tui_mat.widget_buffer
|
||||
assert len(buf) == n, f"Expected {n} buffer entries, got {len(buf)}: {buf}"
|
||||
|
||||
|
||||
@then("the first entry should be collapsed")
|
||||
def step_first_entry_collapsed(context: Any) -> None:
|
||||
buf = context._tui_mat.widget_buffer
|
||||
assert len(buf) >= 1, "Buffer is empty"
|
||||
_, expanded = buf[0]
|
||||
assert expanded is False
|
||||
|
||||
|
||||
@then("the second entry should be expanded")
|
||||
def step_second_entry_expanded(context: Any) -> None:
|
||||
buf = context._tui_mat.widget_buffer
|
||||
assert len(buf) >= 2, "Buffer has fewer than 2 entries"
|
||||
_, expanded = buf[1]
|
||||
assert expanded is True
|
||||
|
||||
|
||||
@then("the widget buffer should be empty")
|
||||
def step_buffer_empty(context: Any) -> None:
|
||||
buf = context._tui_mat.widget_buffer
|
||||
assert len(buf) == 0, f"Expected empty buffer, got {buf}"
|
||||
|
||||
|
||||
@then("the conversation view should show up to 5 file entries")
|
||||
def step_conv_limit_file_entries(context: Any) -> None:
|
||||
# Verify at least 3 files were shown and no more than 5 are displayed.
|
||||
last_text = context._conv_texts[-1] if context._conv_texts else ""
|
||||
assert "src/main.py" in last_text, "Expected first file in output"
|
||||
assert "docs/readme.md" in last_text, "Expected third file in output"
|
||||
# The materializer caps at 5, so we should not see more than that.
|
||||
@@ -0,0 +1,110 @@
|
||||
Feature: TUI Materializer — A2A event queue to conversation view bridge
|
||||
The TuiMaterializer subscribes to an A2aEventQueue and translates events
|
||||
into thought-block-style entries rendered in the conversation view.
|
||||
|
||||
Background:
|
||||
Given a fresh A2aEventQueue
|
||||
And a captured conversation_view callback
|
||||
|
||||
# ── Lifecycle ────────────────────────────────────────────────────────
|
||||
|
||||
Scenario: Materializer starts subscribed
|
||||
When I start the TuiMaterializer with the event queue
|
||||
Then the materializer should be started
|
||||
And the materializer should have an active subscription
|
||||
|
||||
Scenario: Materializer stop unsubscribes
|
||||
When I start the TuiMaterializer with the event queue
|
||||
And I stop the TuiMaterializer
|
||||
Then the materializer should not be started
|
||||
And the materializer should have no active subscription
|
||||
|
||||
Scenario: Starting an already-started materializer is a no-op
|
||||
When I start the TuiMaterializer with the event queue
|
||||
And I start the TuiMaterializer again
|
||||
Then the materializer should still be started
|
||||
|
||||
# ── TaskStatusUpdateEvent handling ───────────────────────────────────
|
||||
|
||||
Scenario: Materializer renders TaskStatusUpdateEvent as collapsed block
|
||||
When I publish a TaskStatusUpdateEvent with status "running" and phase "execute"
|
||||
And the TuiMaterializer processes it
|
||||
Then the conversation view should contain "Plan"
|
||||
And the conversation view should contain "Phase: execute"
|
||||
And the conversation view should contain "Updated:"
|
||||
|
||||
Scenario: TaskStatusUpdateEvent complete icon is rendered
|
||||
When I publish a TaskStatusUpdateEvent with status "complete" and phase "strategize"
|
||||
And the TuiMaterializer processes it
|
||||
Then the event entry should be in collapsed state
|
||||
And the conversation view should contain "\u2705"
|
||||
|
||||
Scenario: TaskStatusUpdateEvent failed icon is rendered on failure
|
||||
When I publish a TaskStatusUpdateEvent with status "failed" and phase "build"
|
||||
And the TuiMaterializer processes it
|
||||
Then the event entry should be in collapsed state
|
||||
And the conversation view should contain "\u274c"
|
||||
|
||||
Scenario: TaskStatusUpdateEvent cancelled icon is rendered
|
||||
When I publish a TaskStatusUpdateEvent with status "cancelled" and phase "execute"
|
||||
And the TuiMaterializer processes it
|
||||
Then the event entry should be in collapsed state
|
||||
And the conversation view should contain "\u279b"
|
||||
|
||||
Scenario: TaskStatusUpdateEvent unknown status gets default icon
|
||||
When I publish a TaskStatusUpdateEvent with status "pending" and phase null
|
||||
And the TuiMaterializer processes it
|
||||
Then the event entry should be in collapsed state
|
||||
And the conversation view should contain "\u23f3"
|
||||
|
||||
Scenario: TaskStatusUpdateEvent null phase shows n/a placeholder
|
||||
When I publish a TaskStatusUpdateEvent with status "running" and no phase
|
||||
And the TuiMaterializer processes it
|
||||
Then the conversation view should contain "Phase: n/a"
|
||||
|
||||
# ── TaskArtifactUpdateEvent handling ─────────────────────────────────
|
||||
|
||||
Scenario: Materializer renders TaskArtifactUpdateEvent as expanded block
|
||||
When I publish a TaskArtifactUpdateEvent with artifact metadata
|
||||
And the TuiMaterializer processes it
|
||||
Then the event entry should be in expanded state
|
||||
And the conversation view should contain "Artifact for plan"
|
||||
And the conversation view should contain "Timestamp:"
|
||||
|
||||
Scenario: TaskArtifactUpdateEvent file list is rendered
|
||||
When I publish a TaskArtifactUpdateEvent with three modified files
|
||||
And the TuiMaterializer processes it
|
||||
Then the conversation view should contain "3 file(s) modified"
|
||||
And the conversation view should show up to 5 file entries
|
||||
|
||||
Scenario: TaskArtifactUpdateEvent single file display
|
||||
When I publish a TaskArtifactUpdateEvent with a single file path artifact
|
||||
And the TuiMaterializer processes it
|
||||
Then the conversation view should contain "file:" in artifact metadata
|
||||
|
||||
Scenario: TaskArtifactUpdateEvent diff_url is rendered
|
||||
When I publish a TaskArtifactUpdateEvent with a diff URL artifact
|
||||
And the TuiMaterializer processes it
|
||||
Then the conversation view should contain "\U0001f517"
|
||||
|
||||
# ── Generic / unrecognized event handling ────────────────────────────
|
||||
|
||||
Scenario: Unrecognized event type renders collapsed block
|
||||
When I publish an A2aEvent of type "CustomEventType"
|
||||
And the TuiMaterializer processes it
|
||||
Then the event entry should be in collapsed state
|
||||
And the conversation view should contain "[CustomEventType]"
|
||||
|
||||
# ── Buffer inspection (headless mode) ────────────────────────────────
|
||||
|
||||
Scenario: Widget buffer records all events pushed to conversation view
|
||||
When I publish a TaskStatusUpdateEvent and wait
|
||||
And I publish a TaskArtifactUpdateEvent and wait
|
||||
Then the widget buffer should have exactly 2 entries
|
||||
And the first entry should be collapsed
|
||||
And the second entry should be expanded
|
||||
|
||||
Scenario: clear_buffer empties all accumulated entries
|
||||
When I publish two events with the materializer running
|
||||
And I call clear_buffer on the TuiMaterializer
|
||||
Then the widget buffer should be empty
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from cleveragents.tui.app import CleverAgentsTuiApp
|
||||
from cleveragents.tui.materializers import TuiMaterializer
|
||||
|
||||
|
||||
def run_tui(*, headless: bool = False) -> int:
|
||||
@@ -14,5 +15,6 @@ def run_tui(*, headless: bool = False) -> int:
|
||||
|
||||
__all__ = [
|
||||
"CleverAgentsTuiApp",
|
||||
"TuiMaterializer",
|
||||
"run_tui",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,277 @@
|
||||
"""TUI materialisers — bridge A2A event queue to conversation view.
|
||||
|
||||
The :class:`TuiMaterializer` subscribes to an :class:`A2aEventQueue` and
|
||||
translates inbound events into ``ThoughtBlockWidget`` instances rendered in
|
||||
the TUI conversation stream.
|
||||
|
||||
Two event types are handled:
|
||||
|
||||
- **TaskStatusUpdateEvent** → collapsed thought block with status icon
|
||||
(shows current phase, state, and progress summary).
|
||||
|
||||
- **TaskArtifactUpdateEvent** → expanded thought block with artifact content
|
||||
(shows diff blocks, file references, or structured metadata produced by the
|
||||
agent during plan execution).
|
||||
|
||||
The materializer uses the same subscription lifecycle as
|
||||
:class:`~cleveragents.a2a.events.EventBusBridge`::
|
||||
|
||||
tm = TuiMaterializer(event_queue, conversation_view)
|
||||
tm.start() # subscribes and begins translating
|
||||
tm.stop() # unsubscribes and cleans up
|
||||
|
||||
When Textual widgets are unavailable the materializer falls back to plain-text
|
||||
rendering into an internal buffer that can be inspected for testing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import datetime as _dt
|
||||
from collections.abc import Callable
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import structlog
|
||||
|
||||
from cleveragents.a2a.events import (
|
||||
TASK_ARTIFACT_UPDATE,
|
||||
TASK_STATUS_UPDATE,
|
||||
)
|
||||
|
||||
logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from cleveragents.tui.widgets import ThoughtBlockWidget
|
||||
else:
|
||||
# Fallback stub when Textual is not installed.
|
||||
class _ThoughtBlockStub: # pragma: no cover
|
||||
"""Minimal stub so the materializer compiles without ``textual``."""
|
||||
|
||||
def __init__(self, *a: object, **kw: object) -> None:
|
||||
self._thought = None
|
||||
self._expanded = False
|
||||
|
||||
@property
|
||||
def is_expanded(self) -> bool:
|
||||
return self._expanded
|
||||
|
||||
ThoughtBlockWidget = _ThoughtBlockStub # type: ignore[misc,assignment]
|
||||
|
||||
|
||||
# Status icon mapping per event data severity.
|
||||
_STATUS_ICONS: dict[str, str] = {
|
||||
"complete": "\u2705", # checkmark
|
||||
"failed": "\u274c", # cross
|
||||
"cancelled": "\u279b", # right arrow curved left
|
||||
"error": "\U0001f6a8", # warning triangle
|
||||
}
|
||||
|
||||
_DEFAULT_STATUS_ICON: str = "\u23f3" # hourglass
|
||||
|
||||
|
||||
def _resolve_status_icon(data: dict[str, Any]) -> str:
|
||||
"""Pick an icon based on event data ``status`` field."""
|
||||
state = str(data.get("status", "")).lower()
|
||||
return _STATUS_ICONS.get(state, _DEFAULT_STATUS_ICON)
|
||||
|
||||
|
||||
def _format_timestamp(raw_ts: str) -> str:
|
||||
"""Format an ISO timestamp to HH:MM:SS local time string."""
|
||||
if not raw_ts:
|
||||
return "now"
|
||||
with contextlib.suppress(Exception):
|
||||
ts = _dt.datetime.fromisoformat(raw_ts)
|
||||
return ts.strftime("%H:%M:%S")
|
||||
return raw_ts[:19]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TuiMaterializer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TuiMaterializer:
|
||||
"""Bridge A2A event queue events to conversation view widgets.
|
||||
|
||||
Each incoming :class:`~cleveragents.a2a.models.A2aEvent` is translated
|
||||
into a thought block and either appended to the conversation stream or
|
||||
rendered inline depending on its type.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
event_queue:
|
||||
An ``A2aEventQueue`` instance (local-mode) with published events.
|
||||
conversation_view:
|
||||
Callback invoked ``(rendered_text: str, expanded: bool) -> None``.
|
||||
Implementations may wrap a Textual widget's ``.update()`` or use an
|
||||
in-memory list for headless testing.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
event_queue: Any,
|
||||
conversation_view: Callable[[str, bool], None],
|
||||
) -> None:
|
||||
self._event_queue = event_queue
|
||||
self._conversation_view = conversation_view
|
||||
self._subscription_id: str | None = None
|
||||
self._widget_buffer: list[tuple[str, bool]] = [] # (text, expanded)
|
||||
self._started: bool = False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public lifecycle
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@property
|
||||
def is_started(self) -> bool:
|
||||
"""Whether the materializer is actively subscribed."""
|
||||
return self._started
|
||||
|
||||
def start(self) -> None:
|
||||
"""Subscribe to the event queue and begin translating events.
|
||||
|
||||
If already started this is a no-op.
|
||||
"""
|
||||
if self._started:
|
||||
logger.debug("a2a.tui_materializer_already_started")
|
||||
return
|
||||
|
||||
sub_id = self._event_queue.subscribe_local(self._on_a2a_event)
|
||||
self._subscription_id = sub_id
|
||||
self._started = True
|
||||
logger.info("a2a.tui_materializer.started", subscription_id=sub_id)
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Unsubscribe from the event queue."""
|
||||
if not self._started or self._subscription_id is None:
|
||||
return
|
||||
self._event_queue.unsubscribe(self._subscription_id)
|
||||
self._subscription_id = None
|
||||
self._started = False
|
||||
logger.info("a2a.tui_materializer.stopped")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Event translation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _on_a2a_event(self, event: Any) -> None:
|
||||
"""Translate an incoming A2A event into a thought block and push it.
|
||||
|
||||
Dispatches to type-specific formatters based on ``event.event_type``.
|
||||
"""
|
||||
_A2aEventCls = self.__class__._EventQueueType()
|
||||
if not isinstance(event, _A2aEventCls):
|
||||
return # pragma: no cover
|
||||
|
||||
event_type = getattr(event, "event_type", "")
|
||||
if event_type == TASK_STATUS_UPDATE:
|
||||
self._render_status_event(event)
|
||||
elif event_type == TASK_ARTIFACT_UPDATE:
|
||||
self._render_artifact_event(event)
|
||||
else:
|
||||
self._render_generic_event(event)
|
||||
|
||||
@staticmethod
|
||||
def _EventQueueType() -> type: # pragma: no cover - static helper
|
||||
"""Return the runtime A2aEvent class for ``isinstance`` checks."""
|
||||
from cleveragents.a2a.models import A2aEvent
|
||||
|
||||
return A2aEvent
|
||||
|
||||
def _render_status_event(self, event: Any) -> None:
|
||||
"""Render a TaskStatusUpdateEvent as a **collapsed** thought block."""
|
||||
status = getattr(event, "data", {}).get("status", "unknown")
|
||||
icon = _resolve_status_icon(getattr(event, "data", {}))
|
||||
phase = getattr(event, "data", {}).get("phase", "")
|
||||
plan_id = event.plan_id[:12] if getattr(event, "plan_id", None) else "n/a"
|
||||
ts = _format_timestamp(str(getattr(event, "timestamp", "")))
|
||||
|
||||
lines = [
|
||||
f"{icon} Plan {plan_id}",
|
||||
f"Phase: {phase or 'n/a'} • Status: {status}",
|
||||
f"Updated: {ts}",
|
||||
]
|
||||
|
||||
rendered = "\n".join(lines)
|
||||
self._push(rendered, expanded=False)
|
||||
|
||||
def _render_artifact_event(self, event: Any) -> None:
|
||||
"""Render a TaskArtifactUpdateEvent as an **expanded** thought block."""
|
||||
artifacts = getattr(event, "data", {})
|
||||
plan_id = event.plan_id[:12] if getattr(event, "plan_id", None) else "n/a"
|
||||
ts = _format_timestamp(str(getattr(event, "timestamp", "")))
|
||||
|
||||
lines: list[str] = [
|
||||
f"\U0001f4e6 Artifact for plan {plan_id}",
|
||||
f"Timestamp: {ts}",
|
||||
]
|
||||
|
||||
# Include any artifact metadata keys.
|
||||
for key in sorted(artifacts.keys()):
|
||||
if key == "files":
|
||||
files = artifacts.get("files", [])
|
||||
if isinstance(files, list):
|
||||
lines.append(f"\u270f\u20e3 {len(files)} file(s) modified")
|
||||
for f_entry in files[:5]: # cap at 5 displayed entries
|
||||
name = (
|
||||
_safe_str(f_entry)
|
||||
if isinstance(f_entry, dict)
|
||||
else str(f_entry)
|
||||
)
|
||||
lines.append(f" - {name}")
|
||||
elif isinstance(files, (str,)):
|
||||
lines.append(f"\u270f\u20e3 file: {files}")
|
||||
elif key == "diff_url":
|
||||
lines.append(f"\U0001f517 {artifacts[key]}")
|
||||
else:
|
||||
val = _safe_str(artifacts.get(key, ""))
|
||||
if val:
|
||||
lines.append(f" {key}: {val}")
|
||||
|
||||
rendered = "\n".join(lines)
|
||||
self._push(rendered, expanded=True)
|
||||
|
||||
def _render_generic_event(self, event: Any) -> None:
|
||||
"""Render an unrecognised event type as a collapsed thought block."""
|
||||
event_type = getattr(event, "event_type", "unknown")
|
||||
plan_id = event.plan_id[:12] if getattr(event, "plan_id", None) else "n/a"
|
||||
ts = _format_timestamp(str(getattr(event, "timestamp", "")))
|
||||
|
||||
rendered = f"[{event_type}] Plan {plan_id} \u2699\ufe0f\nUpdated: {ts}"
|
||||
self._push(rendered, expanded=False)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Conversation view dispatch / internal buffer
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _push(self, text: str, expanded: bool) -> None:
|
||||
"""Push a rendered thought-block string to the conversation view.
|
||||
|
||||
If a ``conversation_view`` callback is wired it is invoked directly.
|
||||
Otherwise the pair is stored in ``_widget_buffer`` for inspection by
|
||||
tests running in headless mode.
|
||||
"""
|
||||
self._widget_buffer.append((text, expanded))
|
||||
with contextlib.suppress(Exception):
|
||||
self._conversation_view(text, expanded)
|
||||
|
||||
@property
|
||||
def widget_buffer(self) -> list[tuple[str, bool]]:
|
||||
"""Return a copy of in-memory thought block entries."""
|
||||
return list(self._widget_buffer)
|
||||
|
||||
def clear_buffer(self) -> None:
|
||||
"""Clear the headless widget buffer."""
|
||||
self._widget_buffer.clear()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _safe_str(val: Any) -> str: # pragma: no cover - util
|
||||
"""Coerce a value to string, handling nested dicts gracefully."""
|
||||
if isinstance(val, dict):
|
||||
return ", ".join(f"{k}={v}" for k, v in sorted(val.items()) if v)
|
||||
return str(val)
|
||||
@@ -1217,3 +1217,13 @@ revert_decisions # noqa: B018, F821
|
||||
|
||||
# Extension protocol parameters — required by Protocol interface definitions
|
||||
destination # noqa: B018, F821
|
||||
|
||||
# TUI materializer — TYPE_CHECKING stub import and runtime fallback class
|
||||
ThoughtBlockWidget # noqa: B018, F821 — used only in TYPE_CHECKING block
|
||||
_ThoughtBlockStub # noqa: B018, F821 — intentional stub; __init__ params a/kw unused
|
||||
a # noqa: B018, F821 — intentionally unused positional param in _ThoughtBlockStub.__init__
|
||||
kw # noqa: B018, F821 — intentionally unused keyword param in _ThoughtBlockStub.__init__
|
||||
|
||||
# TUI materializer public API used by TUI app and tests
|
||||
TuiMaterializer # noqa: B018, F821
|
||||
_safe_str # noqa: B018, F821
|
||||
|
||||
Reference in New Issue
Block a user