Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d4059e5bfe | |||
| 80b02051bc | |||
| e712c9e7e4 |
@@ -0,0 +1,83 @@
|
||||
Feature: LspRuntime workspace path containment
|
||||
As a security-conscious platform
|
||||
I need LspRuntime._read_file to enforce workspace path containment
|
||||
So that path traversal attacks cannot read files outside the workspace
|
||||
|
||||
# ── _read_file static method containment ──────────────────────────
|
||||
|
||||
Scenario: read_file allows a file inside the workspace
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file inside the workspace with content "safe content"
|
||||
When lspc I call read_file with the workspace path
|
||||
Then lspc the file content should be "safe content"
|
||||
And lspc no error should be raised
|
||||
|
||||
Scenario: read_file blocks a file outside the workspace
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file outside the workspace
|
||||
When lspc I call read_file with the workspace path
|
||||
Then lspc an LspError should be raised with message containing "outside workspace"
|
||||
|
||||
Scenario: read_file blocks path traversal using dot-dot segments
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file outside the workspace
|
||||
When lspc I call read_file with a traversal path and the workspace path
|
||||
Then lspc an LspError should be raised with message containing "outside workspace"
|
||||
|
||||
Scenario: read_file without workspace path has no containment check
|
||||
Given lspc I have a file outside the workspace
|
||||
When lspc I call read_file without a workspace path
|
||||
Then lspc no error should be raised
|
||||
|
||||
# ── get_diagnostics containment ────────────────────────────────────
|
||||
|
||||
Scenario: get_diagnostics blocks file outside workspace
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file outside the workspace
|
||||
And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace
|
||||
When lspc I try to get diagnostics for "local/pyright" on the outside file
|
||||
Then lspc an LspError should be raised with message containing "outside workspace"
|
||||
|
||||
Scenario: get_diagnostics allows file inside workspace
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file inside the workspace with content "x = 1"
|
||||
And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace
|
||||
When lspc I get diagnostics for "local/pyright" on the inside file
|
||||
Then lspc diagnostics should be returned as a list
|
||||
And lspc no error should be raised
|
||||
|
||||
# ── get_completions containment ────────────────────────────────────
|
||||
|
||||
Scenario: get_completions blocks file outside workspace
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file outside the workspace
|
||||
And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace
|
||||
When lspc I try to get completions for "local/pyright" on the outside file at line 1 column 1
|
||||
Then lspc an LspError should be raised with message containing "outside workspace"
|
||||
|
||||
# ── get_hover containment ──────────────────────────────────────────
|
||||
|
||||
Scenario: get_hover blocks file outside workspace
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file outside the workspace
|
||||
And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace
|
||||
When lspc I try to get hover for "local/pyright" on the outside file at line 1 column 1
|
||||
Then lspc an LspError should be raised with message containing "outside workspace"
|
||||
|
||||
# ── get_definitions containment ────────────────────────────────────
|
||||
|
||||
Scenario: get_definitions blocks file outside workspace
|
||||
Given lspc I have a temp workspace directory
|
||||
And lspc I have a file outside the workspace
|
||||
And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace
|
||||
When lspc I try to get definitions for "local/pyright" on the outside file at line 1 column 1
|
||||
Then lspc an LspError should be raised with message containing "outside workspace"
|
||||
|
||||
# ── workspace path not registered ─────────────────────────────────
|
||||
|
||||
Scenario: get_diagnostics without registered workspace has no containment check
|
||||
Given lspc I have a file outside the workspace
|
||||
And lspc I create an LspRuntime with a healthy mock server "local/pyright" without workspace
|
||||
When lspc I get diagnostics for "local/pyright" on the outside file
|
||||
Then lspc diagnostics should be returned as a list
|
||||
And lspc no error should be raised
|
||||
@@ -0,0 +1,310 @@
|
||||
"""Step definitions for lsp_path_containment.feature.
|
||||
|
||||
Tests workspace path containment in LspRuntime._read_file to prevent
|
||||
path traversal attacks. Uses the ``lspc`` step prefix to avoid
|
||||
Behave AmbiguousStep errors.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.lsp.errors import LspError
|
||||
from cleveragents.lsp.lifecycle import LspLifecycleManager
|
||||
from cleveragents.lsp.runtime import LspRuntime
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_mock_client() -> MagicMock:
|
||||
"""Create a mock LSP client with the methods runtime calls."""
|
||||
client = MagicMock(name="mock_lsp_client")
|
||||
client.did_open = MagicMock()
|
||||
client.did_close = MagicMock()
|
||||
client.get_diagnostics = MagicMock(return_value=[])
|
||||
client.get_completions = MagicMock(return_value=[])
|
||||
client.get_hover = MagicMock(return_value=None)
|
||||
client.get_definitions = MagicMock(return_value=[])
|
||||
return client
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("lspc I have a temp workspace directory")
|
||||
def step_lspc_create_workspace(context: Context) -> None:
|
||||
workspace = tempfile.mkdtemp(prefix="lspc_workspace_")
|
||||
context.lspc_workspace = workspace
|
||||
context.lspc_error = None
|
||||
|
||||
def cleanup() -> None:
|
||||
import shutil
|
||||
|
||||
if os.path.exists(workspace):
|
||||
shutil.rmtree(workspace, ignore_errors=True)
|
||||
|
||||
context.add_cleanup(cleanup)
|
||||
|
||||
|
||||
@given('lspc I have a file inside the workspace with content "{content}"')
|
||||
def step_lspc_create_inside_file(context: Context, content: str) -> None:
|
||||
fd, path = tempfile.mkstemp(
|
||||
suffix=".py",
|
||||
dir=context.lspc_workspace,
|
||||
prefix="inside_",
|
||||
)
|
||||
with os.fdopen(fd, "w") as f:
|
||||
f.write(content)
|
||||
context.lspc_inside_file = path
|
||||
|
||||
def cleanup() -> None:
|
||||
if os.path.exists(path):
|
||||
os.unlink(path)
|
||||
|
||||
context.add_cleanup(cleanup)
|
||||
|
||||
|
||||
@given("lspc I have a file outside the workspace")
|
||||
def step_lspc_create_outside_file(context: Context) -> None:
|
||||
fd, path = tempfile.mkstemp(suffix=".py", prefix="outside_")
|
||||
with os.fdopen(fd, "w") as f:
|
||||
f.write("outside content")
|
||||
context.lspc_outside_file = path
|
||||
context.lspc_error = None
|
||||
|
||||
def cleanup() -> None:
|
||||
if os.path.exists(path):
|
||||
os.unlink(path)
|
||||
|
||||
context.add_cleanup(cleanup)
|
||||
|
||||
|
||||
@given('lspc I create an LspRuntime with a healthy mock server "{name}" and workspace')
|
||||
def step_lspc_create_runtime_with_workspace(context: Context, name: str) -> None:
|
||||
mock_client = _make_mock_client()
|
||||
|
||||
mock_lifecycle = MagicMock(spec=LspLifecycleManager)
|
||||
mock_lifecycle.health_check = MagicMock(return_value=True)
|
||||
mock_lifecycle.get_client = MagicMock(return_value=mock_client)
|
||||
mock_lifecycle.start_server = MagicMock()
|
||||
|
||||
runtime = LspRuntime(lifecycle_manager=mock_lifecycle)
|
||||
# Register the workspace path by calling start_server
|
||||
# We need to mock the registry lookup too
|
||||
from cleveragents.lsp.models import LspServerConfig
|
||||
from cleveragents.lsp.registry import LspRegistry
|
||||
|
||||
registry = LspRegistry()
|
||||
config = LspServerConfig(name=name, command="echo", languages=["python"])
|
||||
registry.register(config)
|
||||
|
||||
runtime = LspRuntime(registry=registry, lifecycle_manager=mock_lifecycle)
|
||||
runtime.start_server(name, context.lspc_workspace)
|
||||
|
||||
context.lspc_runtime = runtime
|
||||
context.lspc_mock_client = mock_client
|
||||
context.lspc_error = None
|
||||
|
||||
|
||||
@given(
|
||||
'lspc I create an LspRuntime with a healthy mock server "{name}" without workspace'
|
||||
)
|
||||
def step_lspc_create_runtime_without_workspace(context: Context, name: str) -> None:
|
||||
mock_client = _make_mock_client()
|
||||
|
||||
mock_lifecycle = MagicMock(spec=LspLifecycleManager)
|
||||
mock_lifecycle.health_check = MagicMock(return_value=True)
|
||||
mock_lifecycle.get_client = MagicMock(return_value=mock_client)
|
||||
|
||||
runtime = LspRuntime(lifecycle_manager=mock_lifecycle)
|
||||
# Do NOT call start_server — no workspace path registered
|
||||
|
||||
context.lspc_runtime = runtime
|
||||
context.lspc_mock_client = mock_client
|
||||
context.lspc_error = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("lspc I call read_file with the workspace path")
|
||||
def step_lspc_read_file_with_workspace(context: Context) -> None:
|
||||
# Determine which file to use: inside or outside
|
||||
file_path = getattr(context, "lspc_inside_file", None) or getattr(
|
||||
context, "lspc_outside_file", None
|
||||
)
|
||||
try:
|
||||
context.lspc_file_content = LspRuntime._read_file(
|
||||
file_path, context.lspc_workspace
|
||||
)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_file_content = None
|
||||
|
||||
|
||||
@when("lspc I call read_file with a traversal path and the workspace path")
|
||||
def step_lspc_read_file_traversal(context: Context) -> None:
|
||||
# Build a traversal path: workspace/subdir/../../outside_file
|
||||
outside_file = context.lspc_outside_file
|
||||
workspace = context.lspc_workspace
|
||||
# Construct a path that starts inside the workspace but traverses out
|
||||
traversal_path = os.path.join(
|
||||
workspace, "subdir", "..", "..", outside_file.lstrip("/")
|
||||
)
|
||||
try:
|
||||
context.lspc_file_content = LspRuntime._read_file(traversal_path, workspace)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_file_content = None
|
||||
|
||||
|
||||
@when("lspc I call read_file without a workspace path")
|
||||
def step_lspc_read_file_no_workspace(context: Context) -> None:
|
||||
file_path = context.lspc_outside_file
|
||||
try:
|
||||
context.lspc_file_content = LspRuntime._read_file(file_path)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_file_content = None
|
||||
|
||||
|
||||
@when('lspc I try to get diagnostics for "{name}" on the outside file')
|
||||
def step_lspc_get_diagnostics_outside(context: Context, name: str) -> None:
|
||||
try:
|
||||
context.lspc_result = context.lspc_runtime.get_diagnostics(
|
||||
name, context.lspc_outside_file
|
||||
)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_result = None
|
||||
|
||||
|
||||
@when('lspc I get diagnostics for "{name}" on the inside file')
|
||||
def step_lspc_get_diagnostics_inside(context: Context, name: str) -> None:
|
||||
try:
|
||||
context.lspc_result = context.lspc_runtime.get_diagnostics(
|
||||
name, context.lspc_inside_file
|
||||
)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_result = None
|
||||
|
||||
|
||||
@when('lspc I get diagnostics for "{name}" on the outside file')
|
||||
def step_lspc_get_diagnostics_outside_no_ws(context: Context, name: str) -> None:
|
||||
try:
|
||||
context.lspc_result = context.lspc_runtime.get_diagnostics(
|
||||
name, context.lspc_outside_file
|
||||
)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_result = None
|
||||
|
||||
|
||||
@when(
|
||||
'lspc I try to get completions for "{name}" on the outside file'
|
||||
" at line {line:d} column {col:d}"
|
||||
)
|
||||
def step_lspc_get_completions_outside(
|
||||
context: Context, name: str, line: int, col: int
|
||||
) -> None:
|
||||
try:
|
||||
context.lspc_result = context.lspc_runtime.get_completions(
|
||||
name, context.lspc_outside_file, line, col
|
||||
)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_result = None
|
||||
|
||||
|
||||
@when(
|
||||
'lspc I try to get hover for "{name}" on the outside file'
|
||||
" at line {line:d} column {col:d}"
|
||||
)
|
||||
def step_lspc_get_hover_outside(
|
||||
context: Context, name: str, line: int, col: int
|
||||
) -> None:
|
||||
try:
|
||||
context.lspc_result = context.lspc_runtime.get_hover(
|
||||
name, context.lspc_outside_file, line, col
|
||||
)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_result = None
|
||||
|
||||
|
||||
@when(
|
||||
'lspc I try to get definitions for "{name}" on the outside file'
|
||||
" at line {line:d} column {col:d}"
|
||||
)
|
||||
def step_lspc_get_definitions_outside(
|
||||
context: Context, name: str, line: int, col: int
|
||||
) -> None:
|
||||
try:
|
||||
context.lspc_result = context.lspc_runtime.get_definitions(
|
||||
name, context.lspc_outside_file, line, col
|
||||
)
|
||||
context.lspc_error = None
|
||||
except Exception as exc:
|
||||
context.lspc_error = exc
|
||||
context.lspc_result = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('lspc the file content should be "{expected}"')
|
||||
def step_lspc_file_content(context: Context, expected: str) -> None:
|
||||
assert context.lspc_error is None, f"Expected no error, got {context.lspc_error}"
|
||||
assert context.lspc_file_content == expected, (
|
||||
f"Expected '{expected}', got '{context.lspc_file_content}'"
|
||||
)
|
||||
|
||||
|
||||
@then("lspc no error should be raised")
|
||||
def step_lspc_no_error(context: Context) -> None:
|
||||
assert context.lspc_error is None, (
|
||||
f"Expected no error, got {type(context.lspc_error).__name__}: "
|
||||
f"{context.lspc_error}"
|
||||
)
|
||||
|
||||
|
||||
@then('lspc an LspError should be raised with message containing "{msg}"')
|
||||
def step_lspc_lsp_error_msg(context: Context, msg: str) -> None:
|
||||
assert context.lspc_error is not None, "Expected an LspError but no error occurred"
|
||||
assert isinstance(context.lspc_error, LspError), (
|
||||
f"Expected LspError, got {type(context.lspc_error).__name__}: "
|
||||
f"{context.lspc_error}"
|
||||
)
|
||||
assert msg in str(context.lspc_error), (
|
||||
f"Expected '{msg}' in error message, got: {context.lspc_error}"
|
||||
)
|
||||
|
||||
|
||||
@then("lspc diagnostics should be returned as a list")
|
||||
def step_lspc_diagnostics_is_list(context: Context) -> None:
|
||||
assert context.lspc_error is None, f"Expected no error, got {context.lspc_error}"
|
||||
assert isinstance(context.lspc_result, list), (
|
||||
f"Expected list, got {type(context.lspc_result)}"
|
||||
)
|
||||
@@ -0,0 +1,154 @@
|
||||
"""Step definitions for TDD Issue #10490 — LspRuntime._read_file path containment.
|
||||
|
||||
This test captures bug #10490: ``LspRuntime._read_file()`` in
|
||||
``src/cleveragents/lsp/runtime.py`` resolves symlinks and ``..`` components
|
||||
via ``os.path.realpath()`` but does NOT verify that the resolved path is
|
||||
contained within the workspace directory. An attacker who can control
|
||||
``file_path`` (e.g. via a malicious LLM tool call) could read any file on
|
||||
the filesystem by supplying a traversal path such as
|
||||
``<workspace>/../../../etc/passwd``.
|
||||
|
||||
The test uses the ``@tdd_expected_fail`` tag until the fix in #10490 is
|
||||
merged. The tag inversion mechanism causes CI to report the scenario as
|
||||
passed while the bug is still unfixed.
|
||||
See CONTRIBUTING.md > Bug Fix Workflow > TDD Issue Test Tags.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.lsp.errors import LspError
|
||||
from cleveragents.lsp.runtime import LspRuntime
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("lsp_pc a workspace directory containing a safe file")
|
||||
def step_lsp_pc_workspace_with_safe_file(context: Context) -> None:
|
||||
"""Create a temporary workspace directory with a safe file inside it."""
|
||||
context.lsp_pc_workspace = tempfile.mkdtemp(prefix="lsp_pc_workspace_")
|
||||
safe_file_path = os.path.join(context.lsp_pc_workspace, "safe.py")
|
||||
with open(safe_file_path, "w", encoding="utf-8") as f:
|
||||
f.write("# safe content\n")
|
||||
context.lsp_pc_safe_file = safe_file_path
|
||||
|
||||
def _cleanup() -> None:
|
||||
import shutil
|
||||
|
||||
shutil.rmtree(context.lsp_pc_workspace, ignore_errors=True)
|
||||
|
||||
context.add_cleanup(_cleanup)
|
||||
|
||||
|
||||
@given("lsp_pc a sensitive file exists outside the workspace")
|
||||
def step_lsp_pc_sensitive_file_outside_workspace(context: Context) -> None:
|
||||
"""Create a temporary file OUTSIDE the workspace directory."""
|
||||
fd, outside_path = tempfile.mkstemp(
|
||||
prefix="lsp_pc_sensitive_",
|
||||
suffix=".txt",
|
||||
)
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||||
f.write("sensitive content — should not be readable via workspace traversal\n")
|
||||
context.lsp_pc_outside_file = outside_path
|
||||
|
||||
def _cleanup() -> None:
|
||||
if os.path.exists(outside_path):
|
||||
os.unlink(outside_path)
|
||||
|
||||
context.add_cleanup(_cleanup)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("lsp_pc I call _read_file with a path that traverses outside the workspace")
|
||||
def step_lsp_pc_call_read_file_traversal(context: Context) -> None:
|
||||
"""Attempt to read the outside file via a path traversal from the workspace.
|
||||
|
||||
Constructs a path like ``<workspace>/../<basename_of_outside_file>``
|
||||
which resolves to the sensitive file outside the workspace. The current
|
||||
implementation of ``_read_file`` does NOT check workspace containment,
|
||||
so it will successfully read the file instead of raising ``LspError``.
|
||||
"""
|
||||
outside_basename = os.path.basename(context.lsp_pc_outside_file)
|
||||
# Build a traversal path: workspace + "/../<outside_file_basename>"
|
||||
# os.path.realpath will resolve this to the actual outside path.
|
||||
traversal_path = os.path.join(
|
||||
context.lsp_pc_workspace,
|
||||
"..",
|
||||
outside_basename,
|
||||
)
|
||||
context.lsp_pc_traversal_path = traversal_path
|
||||
context.lsp_pc_error = None
|
||||
try:
|
||||
context.lsp_pc_result = LspRuntime._read_file(traversal_path)
|
||||
except LspError as exc:
|
||||
context.lsp_pc_error = exc
|
||||
except Exception as exc:
|
||||
context.lsp_pc_error = exc
|
||||
|
||||
|
||||
@when("lsp_pc I call _read_file with the safe file path")
|
||||
def step_lsp_pc_call_read_file_safe(context: Context) -> None:
|
||||
"""Call _read_file with a path to the safe file inside the workspace."""
|
||||
context.lsp_pc_error = None
|
||||
try:
|
||||
context.lsp_pc_result = LspRuntime._read_file(context.lsp_pc_safe_file)
|
||||
except Exception as exc:
|
||||
context.lsp_pc_error = exc
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('lsp_pc an LspError should be raised with message containing "{expected_msg}"')
|
||||
def step_lsp_pc_lsp_error_raised(context: Context, expected_msg: str) -> None:
|
||||
"""Assert that _read_file raised LspError with the expected message.
|
||||
|
||||
Bug #10490: This assertion FAILS while the bug exists because
|
||||
``_read_file`` does NOT raise ``LspError`` for paths outside the
|
||||
workspace — it successfully reads the sensitive file instead.
|
||||
|
||||
The ``@tdd_expected_fail`` tag on the scenario inverts this failure
|
||||
so CI reports the scenario as passed until the fix is merged.
|
||||
"""
|
||||
assert context.lsp_pc_error is not None, (
|
||||
"Bug #10490: LspRuntime._read_file() did NOT raise any error when "
|
||||
f"given a path traversal to a file outside the workspace.\n"
|
||||
f" Traversal path : {context.lsp_pc_traversal_path!r}\n"
|
||||
f" Resolved path : {os.path.realpath(context.lsp_pc_traversal_path)!r}\n"
|
||||
f" Workspace : {context.lsp_pc_workspace!r}\n"
|
||||
f" File was read successfully — this proves the path containment "
|
||||
f"check is missing."
|
||||
)
|
||||
assert isinstance(context.lsp_pc_error, LspError), (
|
||||
f"Expected LspError but got {type(context.lsp_pc_error).__name__}: "
|
||||
f"{context.lsp_pc_error}"
|
||||
)
|
||||
assert expected_msg in str(context.lsp_pc_error), (
|
||||
f"Expected error message to contain {expected_msg!r}, "
|
||||
f"but got: {context.lsp_pc_error}"
|
||||
)
|
||||
|
||||
|
||||
@then("lsp_pc the file content should be returned successfully")
|
||||
def step_lsp_pc_file_content_returned(context: Context) -> None:
|
||||
"""Assert that _read_file successfully returned the safe file content."""
|
||||
assert context.lsp_pc_error is None, (
|
||||
f"Expected _read_file to succeed for a file inside the workspace, "
|
||||
f"but got error: {context.lsp_pc_error}"
|
||||
)
|
||||
assert context.lsp_pc_result == "# safe content\n", (
|
||||
f"Expected file content '# safe content\\n', but got: {context.lsp_pc_result!r}"
|
||||
)
|
||||
@@ -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,37 @@
|
||||
@tdd_issue @tdd_issue_10490
|
||||
Feature: TDD Issue #10490 — LspRuntime._read_file has no workspace path containment check
|
||||
As a security-conscious developer
|
||||
I want LspRuntime._read_file() to restrict file access to the workspace directory
|
||||
So that a malicious or misconfigured LLM tool call cannot read arbitrary files
|
||||
|
||||
This test captures bug #10490. The ``_read_file`` static method in
|
||||
``src/cleveragents/lsp/runtime.py`` calls ``os.path.realpath()`` to
|
||||
resolve symlinks and ``..`` components, but does NOT verify that the
|
||||
resolved path is contained within the workspace directory. An attacker
|
||||
who can control ``file_path`` (e.g. via a malicious LLM tool call) could
|
||||
read any file on the filesystem by supplying a path such as
|
||||
``<workspace>/../../../etc/passwd``.
|
||||
|
||||
The method is a ``@staticmethod`` and has no access to the workspace path,
|
||||
so it cannot perform containment checks without a design change.
|
||||
|
||||
The scenario below uses ``@tdd_expected_fail`` because the assertion
|
||||
FAILS while the bug exists — ``_read_file`` does NOT raise ``LspError``
|
||||
when given a path that resolves outside the workspace. The tag inversion
|
||||
mechanism causes CI to report the scenario as passed until the fix lands.
|
||||
Once the fix for #10490 is merged, the ``@tdd_expected_fail`` tag must be
|
||||
removed so the scenario runs normally.
|
||||
|
||||
See CONTRIBUTING.md > Bug Fix Workflow > TDD Issue Test Tags.
|
||||
|
||||
@tdd_expected_fail
|
||||
Scenario: _read_file should reject paths outside workspace but currently does not
|
||||
Given lsp_pc a workspace directory containing a safe file
|
||||
And lsp_pc a sensitive file exists outside the workspace
|
||||
When lsp_pc I call _read_file with a path that traverses outside the workspace
|
||||
Then lsp_pc an LspError should be raised with message containing "outside workspace"
|
||||
|
||||
Scenario: _read_file can read a file inside the workspace
|
||||
Given lsp_pc a workspace directory containing a safe file
|
||||
When lsp_pc I call _read_file with the safe file path
|
||||
Then lsp_pc the file content should be returned successfully
|
||||
@@ -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
|
||||
@@ -55,6 +55,8 @@ class LspRuntime:
|
||||
) -> None:
|
||||
self._registry = registry or LspRegistry()
|
||||
self._lifecycle = lifecycle_manager or LspLifecycleManager()
|
||||
# Maps server name to resolved workspace root for path containment checks.
|
||||
self._workspace_paths: dict[str, str] = {}
|
||||
|
||||
@property
|
||||
def registry(self) -> LspRegistry:
|
||||
@@ -98,6 +100,7 @@ class LspRuntime:
|
||||
)
|
||||
|
||||
self._lifecycle.start_server(config, workspace_path)
|
||||
self._workspace_paths[name] = os.path.realpath(workspace_path)
|
||||
|
||||
def stop_server(self, name: str) -> None:
|
||||
"""Stop the LSP server identified by *name*.
|
||||
@@ -117,6 +120,7 @@ class LspRuntime:
|
||||
|
||||
logger.info("lsp.runtime.stopping_server", server=name)
|
||||
self._lifecycle.stop_server(name)
|
||||
self._workspace_paths.pop(name, None)
|
||||
|
||||
def get_diagnostics(self, name: str, file_path: str) -> list[Any]:
|
||||
"""Retrieve diagnostics for *file_path* from the named server.
|
||||
@@ -135,7 +139,8 @@ class LspRuntime:
|
||||
Raises:
|
||||
ValueError: If *name* or *file_path* is empty.
|
||||
LspServerNotFoundError: If the server is not running.
|
||||
LspError: If the server has crashed.
|
||||
LspError: If the server has crashed or the file is outside
|
||||
the workspace.
|
||||
"""
|
||||
if not name:
|
||||
raise ValueError("name must be a non-empty string")
|
||||
@@ -146,8 +151,9 @@ class LspRuntime:
|
||||
uri = self._path_to_uri(file_path)
|
||||
|
||||
# Open the file so the server analyses it
|
||||
workspace_path = self._workspace_paths.get(name)
|
||||
try:
|
||||
text = self._read_file(file_path)
|
||||
text = self._read_file(file_path, workspace_path)
|
||||
except OSError as exc:
|
||||
raise LspError(
|
||||
f"Cannot read file for diagnostics: {file_path}",
|
||||
@@ -192,7 +198,8 @@ class LspRuntime:
|
||||
Raises:
|
||||
ValueError: If inputs are invalid.
|
||||
LspServerNotFoundError: If the server is not running.
|
||||
LspError: If the server has crashed.
|
||||
LspError: If the server has crashed or the file is outside
|
||||
the workspace.
|
||||
"""
|
||||
if not name:
|
||||
raise ValueError("name must be a non-empty string")
|
||||
@@ -207,8 +214,9 @@ class LspRuntime:
|
||||
uri = self._path_to_uri(file_path)
|
||||
|
||||
# Open the file
|
||||
workspace_path = self._workspace_paths.get(name)
|
||||
try:
|
||||
text = self._read_file(file_path)
|
||||
text = self._read_file(file_path, workspace_path)
|
||||
except OSError as exc:
|
||||
raise LspError(
|
||||
f"Cannot read file for completions: {file_path}",
|
||||
@@ -249,6 +257,9 @@ class LspRuntime:
|
||||
|
||||
Returns:
|
||||
Hover result dict or ``None`` if no info is available.
|
||||
|
||||
Raises:
|
||||
LspError: If the file is outside the workspace.
|
||||
"""
|
||||
if not name:
|
||||
raise ValueError("name must be a non-empty string")
|
||||
@@ -262,8 +273,9 @@ class LspRuntime:
|
||||
client = self._get_healthy_client(name)
|
||||
uri = self._path_to_uri(file_path)
|
||||
|
||||
workspace_path = self._workspace_paths.get(name)
|
||||
try:
|
||||
text = self._read_file(file_path)
|
||||
text = self._read_file(file_path, workspace_path)
|
||||
except OSError as exc:
|
||||
raise LspError(
|
||||
f"Cannot read file for hover: {file_path}",
|
||||
@@ -303,6 +315,9 @@ class LspRuntime:
|
||||
|
||||
Returns:
|
||||
List of Location dicts.
|
||||
|
||||
Raises:
|
||||
LspError: If the file is outside the workspace.
|
||||
"""
|
||||
if not name:
|
||||
raise ValueError("name must be a non-empty string")
|
||||
@@ -316,8 +331,9 @@ class LspRuntime:
|
||||
client = self._get_healthy_client(name)
|
||||
uri = self._path_to_uri(file_path)
|
||||
|
||||
workspace_path = self._workspace_paths.get(name)
|
||||
try:
|
||||
text = self._read_file(file_path)
|
||||
text = self._read_file(file_path, workspace_path)
|
||||
except OSError as exc:
|
||||
raise LspError(
|
||||
f"Cannot read file for definitions: {file_path}",
|
||||
@@ -370,13 +386,44 @@ class LspRuntime:
|
||||
return f"file://{file_path}"
|
||||
|
||||
@staticmethod
|
||||
def _read_file(file_path: str) -> str:
|
||||
def _read_file(file_path: str, workspace_path: str | None = None) -> str:
|
||||
"""Read file contents as UTF-8 text.
|
||||
|
||||
When *workspace_path* is provided the resolved file path must be
|
||||
contained within the workspace directory. This prevents path
|
||||
traversal attacks where a caller supplies a path such as
|
||||
``../../etc/passwd`` to escape the workspace root.
|
||||
|
||||
Args:
|
||||
file_path: Path to the file to read.
|
||||
workspace_path: Optional workspace root. When supplied, the
|
||||
resolved *file_path* must start with the resolved
|
||||
*workspace_path* or an :class:`LspError` is raised.
|
||||
|
||||
Raises:
|
||||
LspError: If the resolved path is a directory or device.
|
||||
LspError: If the resolved path is a directory, device, or
|
||||
(when *workspace_path* is given) outside the workspace.
|
||||
"""
|
||||
resolved = os.path.realpath(file_path)
|
||||
if workspace_path is not None:
|
||||
resolved_workspace = os.path.realpath(workspace_path)
|
||||
# Ensure the file is strictly inside the workspace directory.
|
||||
# We append os.sep so that a workspace of "/tmp/ws" does not
|
||||
# accidentally match "/tmp/ws2/file.py".
|
||||
ws_prefix = (
|
||||
resolved_workspace
|
||||
if resolved_workspace.endswith(os.sep)
|
||||
else resolved_workspace + os.sep
|
||||
)
|
||||
if resolved != resolved_workspace and not resolved.startswith(ws_prefix):
|
||||
raise LspError(
|
||||
f"Path traversal detected: '{file_path}' is outside workspace",
|
||||
details={
|
||||
"file": file_path,
|
||||
"resolved": resolved,
|
||||
"workspace": workspace_path,
|
||||
},
|
||||
)
|
||||
if not os.path.isfile(resolved):
|
||||
raise LspError(
|
||||
f"Not a regular file: {file_path}",
|
||||
@@ -448,6 +495,7 @@ class LspRuntime:
|
||||
def stop_all(self) -> None:
|
||||
"""Shut down all running LSP servers."""
|
||||
self._lifecycle.stop_all()
|
||||
self._workspace_paths.clear()
|
||||
|
||||
|
||||
__all__ = [
|
||||
|
||||
@@ -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