"""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 == []