Ken
11b3bc2bbf
Deploy Research Workbench / deploy (push) Failing after 14m1s
- bundle/bundle.md: Add provider-anthropic with source URL and orchestrator config override to disable extended_thinking (foundation enables it by default, causes empty text responses for simple queries) - Dockerfile: Add amplifier-module-provider-anthropic pip install for container - backend/agui_adapter.py: Fix thinking block handling and empty text block rendering - Thinking blocks no longer fall through to text handler - Text blocks only emit START when actual content arrives - Handle amplifierd wire format (block_index/block_type, complete text in content_block:end) - frontend/src/hooks/useChat.ts: Reset assistantId on TEXT_MESSAGE_END to prevent stale message IDs across turns - tests/test_agui_adapter.py: Add extended thinking test coverage Chat functionality now works end-to-end. Verified: login -> create session -> send message -> receive AI response visible in chat UI. Generated with Amplifier
329 lines
12 KiB
Python
329 lines
12 KiB
Python
"""Tests for agui_adapter module — AmplifierdEventTranslator."""
|
|
|
|
import sys
|
|
import os
|
|
|
|
# Add backend to sys.path so we can import agui_adapter
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
|
|
|
|
from agui_adapter import AmplifierdEventTranslator
|
|
|
|
|
|
class TestTextBlocks:
|
|
"""Tests for text content_block events → AG-UI text message events."""
|
|
|
|
def setup_method(self):
|
|
self.translator = AmplifierdEventTranslator(
|
|
run_id="run_1", thread_id="thread_1"
|
|
)
|
|
|
|
def test_text_block_start(self):
|
|
"""text_block_start → no events emitted (lazy: start is deferred to first delta)."""
|
|
event = {
|
|
"event_type": "content_block:start",
|
|
"data": {"index": 0, "type": "text"},
|
|
}
|
|
result = self.translator.translate(event)
|
|
assert result == []
|
|
|
|
def test_text_delta(self):
|
|
"""First text_delta → TEXT_MESSAGE_START then TEXT_MESSAGE_CONTENT (lazy start)."""
|
|
# Start the text block — no event emitted yet
|
|
start_event = {
|
|
"event_type": "content_block:start",
|
|
"data": {"index": 0, "type": "text"},
|
|
}
|
|
self.translator.translate(start_event)
|
|
|
|
delta_event = {
|
|
"event_type": "content_block:delta",
|
|
"data": {
|
|
"index": 0,
|
|
"delta": {"type": "text_delta", "text": "Hello "},
|
|
},
|
|
}
|
|
result = self.translator.translate(delta_event)
|
|
# Lazy start: first delta emits [TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT]
|
|
assert len(result) == 2
|
|
start_ev, content_ev = result
|
|
assert start_ev.type.value == "TEXT_MESSAGE_START"
|
|
assert start_ev.role == "assistant"
|
|
assert start_ev.message_id
|
|
assert content_ev.type.value == "TEXT_MESSAGE_CONTENT"
|
|
assert content_ev.delta == "Hello "
|
|
assert content_ev.message_id == start_ev.message_id
|
|
|
|
def test_text_block_end(self):
|
|
"""text_block_end after content → TEXT_MESSAGE_END with matching messageId."""
|
|
# Start block, send a delta (which lazily opens it), then close it
|
|
self.translator.translate({
|
|
"event_type": "content_block:start",
|
|
"data": {"index": 0, "type": "text"},
|
|
})
|
|
self.translator.translate({
|
|
"event_type": "content_block:delta",
|
|
"data": {"index": 0, "delta": {"type": "text_delta", "text": "Hi"}},
|
|
})
|
|
|
|
end_event = {
|
|
"event_type": "content_block:end",
|
|
"data": {"index": 0},
|
|
}
|
|
result = self.translator.translate(end_event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
assert ev.type.value == "TEXT_MESSAGE_END"
|
|
|
|
|
|
class TestToolBlocks:
|
|
"""Tests for tool_use content_block events and tool results → AG-UI tool call events."""
|
|
|
|
def setup_method(self):
|
|
self.translator = AmplifierdEventTranslator(
|
|
run_id="run_1", thread_id="thread_1"
|
|
)
|
|
|
|
def test_tool_use_start(self):
|
|
"""tool_use_start → TOOL_CALL_START with toolCallName='bash', toolCallId='tool_123'."""
|
|
event = {
|
|
"event_type": "content_block:start",
|
|
"data": {
|
|
"index": 0,
|
|
"type": "tool_use",
|
|
"id": "tool_123",
|
|
"name": "bash",
|
|
},
|
|
}
|
|
result = self.translator.translate(event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
assert ev.type.value == "TOOL_CALL_START"
|
|
assert ev.tool_call_name == "bash"
|
|
assert ev.tool_call_id == "tool_123"
|
|
|
|
def test_tool_use_delta(self):
|
|
"""tool_use_delta → TOOL_CALL_ARGS with delta='{"command": "ls"}'."""
|
|
# Start the tool block first
|
|
start_event = {
|
|
"event_type": "content_block:start",
|
|
"data": {
|
|
"index": 0,
|
|
"type": "tool_use",
|
|
"id": "tool_123",
|
|
"name": "bash",
|
|
},
|
|
}
|
|
self.translator.translate(start_event)
|
|
|
|
delta_event = {
|
|
"event_type": "content_block:delta",
|
|
"data": {
|
|
"index": 0,
|
|
"delta": {
|
|
"type": "input_json_delta",
|
|
"partial_json": '{"command": "ls"}',
|
|
},
|
|
},
|
|
}
|
|
result = self.translator.translate(delta_event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
assert ev.type.value == "TOOL_CALL_ARGS"
|
|
assert ev.delta == '{"command": "ls"}'
|
|
|
|
def test_tool_use_end(self):
|
|
"""tool_use_end → TOOL_CALL_END."""
|
|
# Start first
|
|
start_event = {
|
|
"event_type": "content_block:start",
|
|
"data": {
|
|
"index": 0,
|
|
"type": "tool_use",
|
|
"id": "tool_123",
|
|
"name": "bash",
|
|
},
|
|
}
|
|
self.translator.translate(start_event)
|
|
|
|
end_event = {
|
|
"event_type": "content_block:end",
|
|
"data": {"index": 0},
|
|
}
|
|
result = self.translator.translate(end_event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
assert ev.type.value == "TOOL_CALL_END"
|
|
|
|
def test_tool_result(self):
|
|
"""tool_result → TOOL_CALL_RESULT with toolCallId."""
|
|
event = {
|
|
"event_type": "tool:result",
|
|
"data": {"tool_use_id": "tool_123", "content": "output"},
|
|
}
|
|
result = self.translator.translate(event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
assert ev.type.value == "TOOL_CALL_RESULT"
|
|
assert ev.tool_call_id == "tool_123"
|
|
|
|
def test_tool_result_with_mcp_app_metadata(self):
|
|
"""tool_result_with_mcp_app_metadata → TOOL_CALL_RESULT preserves _meta.ui.resourceUri='app://chart' and structuredContent.chartType='bar'."""
|
|
event = {
|
|
"event_type": "tool:result",
|
|
"data": {
|
|
"tool_use_id": "tool_123",
|
|
"content": "output",
|
|
"_meta": {"ui": {"resourceUri": "app://chart"}},
|
|
"structuredContent": {"chartType": "bar"},
|
|
},
|
|
}
|
|
result = self.translator.translate(event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
assert ev.type.value == "TOOL_CALL_RESULT"
|
|
d = ev.model_dump()
|
|
assert d.get("_meta", {}).get("ui", {}).get("resourceUri") == "app://chart"
|
|
assert d.get("structuredContent", {}).get("chartType") == "bar"
|
|
|
|
def test_tool_result_without_mcp_app_metadata(self):
|
|
"""tool_result_without_mcp_app_metadata → no _meta or structuredContent keys."""
|
|
event = {
|
|
"event_type": "tool:result",
|
|
"data": {"tool_use_id": "tool_123", "content": "output"},
|
|
}
|
|
result = self.translator.translate(event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
d = ev.model_dump()
|
|
assert "_meta" not in d
|
|
assert "structuredContent" not in d
|
|
|
|
|
|
class TestExtendedThinking:
|
|
"""Tests for extended thinking blocks — must produce zero AG-UI events."""
|
|
|
|
def setup_method(self):
|
|
self.translator = AmplifierdEventTranslator(
|
|
run_id="run_1", thread_id="thread_1"
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _thinking_start(self, index: int = 0) -> dict:
|
|
return {
|
|
"event_type": "content_block:start",
|
|
"data": {"index": index, "type": "thinking"},
|
|
}
|
|
|
|
def _thinking_delta(self, text: str, index: int = 0) -> dict:
|
|
return {
|
|
"event_type": "content_block:delta",
|
|
"data": {"index": index, "delta": {"type": "thinking_delta", "thinking": text}},
|
|
}
|
|
|
|
def _thinking_end(self, index: int = 0) -> dict:
|
|
return {"event_type": "content_block:end", "data": {"index": index}}
|
|
|
|
def _text_start(self, index: int = 1) -> dict:
|
|
return {
|
|
"event_type": "content_block:start",
|
|
"data": {"index": index, "type": "text"},
|
|
}
|
|
|
|
def _text_delta(self, text: str, index: int = 1) -> dict:
|
|
return {
|
|
"event_type": "content_block:delta",
|
|
"data": {"index": index, "delta": {"type": "text_delta", "text": text}},
|
|
}
|
|
|
|
def _text_end(self, index: int = 1) -> dict:
|
|
return {"event_type": "content_block:end", "data": {"index": index}}
|
|
|
|
# ------------------------------------------------------------------
|
|
# tests
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_thinking_block_produces_no_events(self):
|
|
"""thinking start/delta/end → all return [] (thinking blocks are skipped)."""
|
|
assert self.translator.translate(self._thinking_start()) == []
|
|
assert self.translator.translate(self._thinking_delta("Let me reason…")) == []
|
|
assert self.translator.translate(self._thinking_end()) == []
|
|
|
|
def test_empty_text_block_produces_no_events(self):
|
|
"""text start then immediate end with no deltas → no AG-UI events emitted."""
|
|
assert self.translator.translate(self._text_start()) == []
|
|
assert self.translator.translate(self._text_end()) == []
|
|
|
|
def test_thinking_then_empty_text_produces_no_events(self):
|
|
"""Full sequence: thinking block + empty text block → zero AG-UI events total."""
|
|
events = [
|
|
self._thinking_start(index=0),
|
|
self._thinking_delta("I need to think…", index=0),
|
|
self._thinking_end(index=0),
|
|
self._text_start(index=1),
|
|
self._text_end(index=1),
|
|
]
|
|
all_results = []
|
|
for ev in events:
|
|
all_results.extend(self.translator.translate(ev))
|
|
assert all_results == []
|
|
|
|
def test_thinking_then_nonempty_text_works(self):
|
|
"""Thinking block (skip) then text block with actual content → START + CONTENT + END."""
|
|
# Thinking block — all silent
|
|
self.translator.translate(self._thinking_start(index=0))
|
|
self.translator.translate(self._thinking_delta("Reasoning…", index=0))
|
|
self.translator.translate(self._thinking_end(index=0))
|
|
|
|
# Text block — lazy open on first delta
|
|
self.translator.translate(self._text_start(index=1))
|
|
|
|
first_delta_result = self.translator.translate(self._text_delta("Hello", index=1))
|
|
assert len(first_delta_result) == 2
|
|
start_ev, content_ev = first_delta_result
|
|
assert start_ev.type.value == "TEXT_MESSAGE_START"
|
|
assert start_ev.role == "assistant"
|
|
message_id = start_ev.message_id
|
|
assert content_ev.type.value == "TEXT_MESSAGE_CONTENT"
|
|
assert content_ev.delta == "Hello"
|
|
assert content_ev.message_id == message_id
|
|
|
|
# Subsequent delta — only CONTENT (no second START)
|
|
second_delta_result = self.translator.translate(self._text_delta(" world", index=1))
|
|
assert len(second_delta_result) == 1
|
|
assert second_delta_result[0].type.value == "TEXT_MESSAGE_CONTENT"
|
|
assert second_delta_result[0].delta == " world"
|
|
|
|
# End — TEXT_MESSAGE_END because content was started
|
|
end_result = self.translator.translate(self._text_end(index=1))
|
|
assert len(end_result) == 1
|
|
assert end_result[0].type.value == "TEXT_MESSAGE_END"
|
|
assert end_result[0].message_id == message_id
|
|
|
|
|
|
class TestLifecycle:
|
|
"""Tests for lifecycle events → AG-UI run lifecycle events."""
|
|
|
|
def setup_method(self):
|
|
self.translator = AmplifierdEventTranslator(
|
|
run_id="run_1", thread_id="thread_1"
|
|
)
|
|
|
|
def test_orchestrator_complete(self):
|
|
"""orchestrator_complete → RUN_FINISHED with threadId and runId."""
|
|
event = {"event_type": "orchestrator:complete", "data": {}}
|
|
result = self.translator.translate(event)
|
|
assert len(result) == 1
|
|
ev = result[0]
|
|
assert ev.type.value == "RUN_FINISHED"
|
|
assert ev.thread_id == "thread_1"
|
|
assert ev.run_id == "run_1"
|
|
|
|
def test_unknown_event(self):
|
|
"""unknown event → empty list."""
|
|
event = {"event_type": "some:unknown_event", "data": {}}
|
|
result = self.translator.translate(event)
|
|
assert result == []
|