diff --git a/src/deep_code_agent/tui/bridge/agent_bridge.py b/src/deep_code_agent/tui/bridge/agent_bridge.py index 1353661..8627cfe 100644 --- a/src/deep_code_agent/tui/bridge/agent_bridge.py +++ b/src/deep_code_agent/tui/bridge/agent_bridge.py @@ -483,6 +483,11 @@ def handle_event() -> None: if widget is self._active_tool_widget: self._active_tool_widget = None + elif event.type == EventType.TODOS_UPDATE: + todos = event.data if isinstance(event.data, list) else [] + if hasattr(chat_log, "upsert_todos_card"): + chat_log.upsert_todos_card(todos) + elif event.type == EventType.HITL_INTERRUPT: status_bar.set_waiting_approval() interrupt_data = event.data diff --git a/src/deep_code_agent/tui/bridge/stream_handler.py b/src/deep_code_agent/tui/bridge/stream_handler.py index f0f2135..30d0c80 100644 --- a/src/deep_code_agent/tui/bridge/stream_handler.py +++ b/src/deep_code_agent/tui/bridge/stream_handler.py @@ -17,6 +17,7 @@ class EventType(Enum): TOOL_SUCCESS = auto() # Tool execution succeeded TOOL_ERROR = auto() # Tool execution failed TOOL_RESULT = auto() # Tool execution result (legacy) + TODOS_UPDATE = auto() # Agent todo list was created or updated HITL_INTERRUPT = auto() # Human-in-the-loop approval needed ERROR = auto() # Error occurred DONE = auto() # Stream complete @@ -49,6 +50,8 @@ class StreamHandler: handle_event(event) """ + TODO_STATUSES = {"pending", "in_progress", "completed", "failed"} + def __init__(self, agent, config: dict): """Initialize stream handler. @@ -62,6 +65,79 @@ def __init__(self, agent, config: dict): self._interrupted = False self._seen_tool_call_ids: set[str] = set() + def _normalize_todo_item(self, item: Any) -> dict[str, str] | None: + """Normalize one todo item from dict/object forms. + + LangChain's TodoListMiddleware currently emits todo objects with + ``content`` and ``status`` fields. Keep the UI tolerant of dict, + pydantic-like, and light object forms so stream chunk shape changes do + not break the TUI. + """ + if hasattr(item, "model_dump"): + try: + item = item.model_dump() + except Exception: + pass + + if isinstance(item, dict): + content = item.get("content") + status = item.get("status") + else: + content = getattr(item, "content", None) + status = getattr(item, "status", None) + + if content is None or status is None: + return None + + content_text = str(content).strip() + status_text = str(status).strip() + if not content_text or status_text not in self.TODO_STATUSES: + return None + + return {"content": content_text, "status": status_text} + + def _normalize_todos(self, todos: Any) -> list[dict[str, str]]: + """Normalize a todo list payload, skipping malformed entries.""" + if not isinstance(todos, list): + return [] + + normalized: list[dict[str, str]] = [] + for item in todos: + todo = self._normalize_todo_item(item) + if todo is not None: + normalized.append(todo) + return normalized + + def _find_todos_payload(self, chunk: Any, *, max_depth: int = 4) -> list[dict[str, str]]: + """Find and normalize the first valid ``todos`` payload in an update chunk.""" + if max_depth < 0: + return [] + + if hasattr(chunk, "model_dump"): + try: + chunk = chunk.model_dump() + except Exception: + pass + + if isinstance(chunk, dict): + if "todos" in chunk: + todos = self._normalize_todos(chunk.get("todos")) + if todos: + return todos + + for value in chunk.values(): + todos = self._find_todos_payload(value, max_depth=max_depth - 1) + if todos: + return todos + + elif isinstance(chunk, list): + for item in chunk: + todos = self._find_todos_payload(item, max_depth=max_depth - 1) + if todos: + return todos + + return [] + async def _process_stream(self, stream, include_tool_calls: bool = True) -> AsyncIterator[AgentEvent]: """Process the agent stream and yield events. @@ -182,6 +258,10 @@ async def _process_stream(self, stream, include_tool_calls: bool = True) -> Asyn yield AgentEvent(type=EventType.HITL_INTERRUPT, data=interrupt_data) return + todos = self._find_todos_payload(chunk) + if todos: + yield AgentEvent(type=EventType.TODOS_UPDATE, data=todos) + async def process(self, state: dict) -> AsyncIterator[AgentEvent]: """Process a user request and yield events. diff --git a/src/deep_code_agent/tui/widgets/__init__.py b/src/deep_code_agent/tui/widgets/__init__.py index c2953f3..020cd94 100644 --- a/src/deep_code_agent/tui/widgets/__init__.py +++ b/src/deep_code_agent/tui/widgets/__init__.py @@ -6,6 +6,7 @@ from deep_code_agent.tui.widgets.selectable_option import SelectableOption from deep_code_agent.tui.widgets.side_panel import SidePanel from deep_code_agent.tui.widgets.status_bar import StatusBar +from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard from deep_code_agent.tui.widgets.tool_call_view import ToolCallView __all__ = [ @@ -15,5 +16,6 @@ "SelectableOption", "SidePanel", "StatusBar", + "TodosProgressCard", "ToolCallView", ] diff --git a/src/deep_code_agent/tui/widgets/chat_log.py b/src/deep_code_agent/tui/widgets/chat_log.py index 6648ed0..4f4e9d7 100644 --- a/src/deep_code_agent/tui/widgets/chat_log.py +++ b/src/deep_code_agent/tui/widgets/chat_log.py @@ -4,6 +4,7 @@ from textual.reactive import reactive from deep_code_agent.tui.widgets.message_bubble import MessageBubble +from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard class ChatLog(VerticalScroll): @@ -35,11 +36,22 @@ class ChatLog(VerticalScroll): # Reactive state for tracking if auto-scroll is enabled auto_scroll = reactive(True) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._todos_card: TodosProgressCard | None = None + def compose(self): """Compose the chat log (empty initially).""" # Messages will be added dynamically return [] + def _mount_above_todos_card(self, widget) -> None: + """Mount chat content above the pinned todos card when it exists.""" + if self._todos_card is not None and self._todos_card in self.children: + self.mount(widget, before=self._todos_card) + return + self.mount(widget) + def add_user_message(self, content: str) -> MessageBubble: """Add a user message to the chat log. @@ -50,7 +62,7 @@ def add_user_message(self, content: str) -> MessageBubble: The created MessageBubble widget """ bubble = MessageBubble(content, role="user") - self.mount(bubble) + self._mount_above_todos_card(bubble) self._scroll_to_bottom() return bubble @@ -64,7 +76,7 @@ def add_agent_message(self, content: str) -> MessageBubble: The created MessageBubble widget """ bubble = MessageBubble(content, role="agent") - self.mount(bubble) + self._mount_above_todos_card(bubble) self._scroll_to_bottom() return bubble @@ -78,7 +90,7 @@ def add_system_message(self, content: str) -> MessageBubble: The created MessageBubble widget """ bubble = MessageBubble(content, role="system") - self.mount(bubble) + self._mount_above_todos_card(bubble) self._scroll_to_bottom() return bubble @@ -92,7 +104,7 @@ def add_tool_call(self, tool_name: str, args: dict) -> None: # Format the tool call nicely content = f"🔧 Tool call: {tool_name}" bubble = MessageBubble(content, role="system") - self.mount(bubble) + self._mount_above_todos_card(bubble) self._scroll_to_bottom() def add_tool_call_widget( @@ -121,15 +133,49 @@ def add_tool_call_widget( status=status, result=result ) - self.mount(widget) + self._mount_above_todos_card(widget) self._scroll_to_bottom() return widget + def upsert_todos_card(self, todos: list[dict[str, str]]) -> TodosProgressCard: + """Create or update the singleton todos progress card. + + The card is moved to the bottom on every update so the latest task + status stays visible in the chat stream. + """ + if self._todos_card is None: + self._todos_card = TodosProgressCard(todos) + self.mount(self._todos_card) + self._scroll_to_bottom() + return self._todos_card + + expanded = self._todos_card.expanded + self._todos_card.update_todos(todos) + self._todos_card.expanded = expanded + + if self.children and self.children[-1] is not self._todos_card: + try: + self.move_child(self._todos_card, after=len(self.children) - 1) + except Exception: + # Textual versions vary in move/remount behavior. If moving an + # existing widget fails, recreate the card while preserving the + # user's expanded/collapsed preference. + try: + self._todos_card.remove() + except Exception: + pass + self._todos_card = TodosProgressCard(todos, expanded=expanded) + self.mount(self._todos_card) + + self._scroll_to_bottom() + return self._todos_card + def clear_messages(self) -> None: """Clear all messages from the chat log.""" # Remove all children (MessageBubble widgets) for child in list(self.children): child.remove() + self._todos_card = None def _scroll_to_bottom(self) -> None: """Scroll to the bottom of the chat log.""" diff --git a/src/deep_code_agent/tui/widgets/message_bubble.py b/src/deep_code_agent/tui/widgets/message_bubble.py index da11bb1..d2d610b 100644 --- a/src/deep_code_agent/tui/widgets/message_bubble.py +++ b/src/deep_code_agent/tui/widgets/message_bubble.py @@ -76,5 +76,11 @@ def compose(self): def update_content(self, new_content: str) -> None: """Update the bubble content (useful for streaming).""" self.content = new_content - content_widget = self.query_one(".message-text", Static) + try: + content_widget = self.query_one(".message-text", Static) + except Exception: + # The bubble may have been mounted but not composed yet when + # streamed chunks complete in the same event-loop turn. Storing + # content is enough; compose() will render the latest value. + return content_widget.update(new_content) diff --git a/src/deep_code_agent/tui/widgets/todos_progress_card.py b/src/deep_code_agent/tui/widgets/todos_progress_card.py new file mode 100644 index 0000000..94d3993 --- /dev/null +++ b/src/deep_code_agent/tui/widgets/todos_progress_card.py @@ -0,0 +1,199 @@ +"""Todos progress card widget for displaying agent task progress.""" + +from __future__ import annotations + +from textual import events +from textual.containers import Vertical +from textual.widgets import Static + + +TodoItem = dict[str, str] + + +class TodosProgressCard(Vertical): + """A collapsible chat card showing the agent's todo progress. + + Args: + todos: Todo items with ``content`` and ``status`` fields. + expanded: Whether the card body should be visible initially. + """ + + DEFAULT_CSS = """ + TodosProgressCard { + width: 100%; + height: auto; + margin: 1 0; + padding: 1; + background: #172033; + border: solid $secondary; + display: block; + } + + TodosProgressCard .todos-header { + text-style: bold; + color: #a6e3ff; + margin-bottom: 1; + } + + TodosProgressCard .todos-body { + height: auto; + } + + TodosProgressCard .todo-row { + width: 100%; + height: auto; + text-wrap: wrap; + } + + TodosProgressCard .todo-pending { + color: $warning; + } + + TodosProgressCard .todo-in_progress { + color: $accent; + } + + TodosProgressCard .todo-completed { + color: $success; + } + + TodosProgressCard .todo-failed { + color: $error; + } + """ + + STATUS_ICONS = { + "pending": "○", + "in_progress": "◐", + "completed": "✓", + "failed": "✗", + } + + STATUS_LABELS = { + "pending": "pending", + "in_progress": "in progress", + "completed": "completed", + "failed": "failed", + } + + def __init__(self, todos: list[TodoItem] | None = None, *, expanded: bool = True, **kwargs): + super().__init__(**kwargs) + self.todos = self._coerce_todos(todos or []) + self.expanded = expanded + self._header_static: Static | None = None + self._body_container: Vertical | None = None + + def compose(self): + """Compose the todos card.""" + header = Static(self._header_text(), classes="todos-header") + self._header_static = header + yield header + + if self.expanded: + body = Vertical(classes="todos-body") + self._body_container = body + with body: + for todo in self.todos: + yield self._make_row(todo) + + def _coerce_todos(self, todos: list[TodoItem]) -> list[TodoItem]: + coerced: list[TodoItem] = [] + for todo in todos: + content = str(todo.get("content", "")).strip() + status = str(todo.get("status", "")).strip() + if content and status in self.STATUS_ICONS: + coerced.append({"content": content, "status": status}) + return coerced + + def _header_text(self) -> str: + marker = "▼" if self.expanded else "▶" + counts = {status: 0 for status in self.STATUS_ICONS} + for todo in self.todos: + counts[todo["status"]] += 1 + + summary_parts = [] + if counts["in_progress"]: + summary_parts.append(f"{counts['in_progress']} active") + if counts["completed"]: + summary_parts.append(f"{counts['completed']} done") + if counts["failed"]: + summary_parts.append(f"{counts['failed']} failed") + if counts["pending"]: + summary_parts.append(f"{counts['pending']} pending") + summary = ", ".join(summary_parts) if summary_parts else "no tasks" + + return f"{marker} 📋 Todos ({summary})" + + def _make_row(self, todo: TodoItem) -> Static: + status = todo["status"] + icon = self.STATUS_ICONS[status] + label = self.STATUS_LABELS[status] + return Static( + f"{icon} [{label}] {todo['content']}", + classes=f"todo-row todo-{status}", + ) + + def _refresh_header(self) -> None: + try: + header = self.query_one(".todos-header", Static) + header.update(self._header_text()) + self._header_static = header + except Exception: + pass + + def _remove_body(self) -> None: + if self._body_container is not None: + try: + self._body_container.remove() + except Exception: + pass + self._body_container = None + return + + try: + body = self.query_one(".todos-body", Vertical) + body.remove() + except Exception: + pass + + def _mount_body(self) -> None: + body = Vertical(classes="todos-body") + self._body_container = body + try: + self.mount(body, after=self._header_static) + except Exception: + self.mount(body) + + for todo in self.todos: + body.mount(self._make_row(todo)) + + def update_todos(self, todos: list[TodoItem]) -> None: + """Replace the rendered todo rows.""" + self.todos = self._coerce_todos(todos) + if self._header_static is None: + return + + self._refresh_header() + + if not self.expanded: + return + + self._remove_body() + self._mount_body() + + def toggle_expanded(self) -> None: + """Collapse or expand the card body.""" + self.expanded = not self.expanded + if self._header_static is None: + return + + self._refresh_header() + if self.expanded: + self._mount_body() + else: + self._remove_body() + + def on_click(self, event: events.Click) -> None: + """Toggle the card when clicked.""" + event.stop() + self.toggle_expanded() diff --git a/tests/tui/test_agent_bridge_helpers.py b/tests/tui/test_agent_bridge_helpers.py new file mode 100644 index 0000000..9c40ba3 --- /dev/null +++ b/tests/tui/test_agent_bridge_helpers.py @@ -0,0 +1,55 @@ +"""Tests for AgentBridge helper extraction methods.""" + + +def test_agent_bridge_extracts_tool_name_from_action_requests(): + from deep_code_agent.tui.bridge.agent_bridge import AgentBridge + + bridge = AgentBridge(agent=object()) + interrupt_data = { + "action_requests": [ + {"name": "write_file", "args": {"path": "x.txt"}}, + ] + } + + assert bridge._extract_tool_name_from_interrupt(interrupt_data) == "write_file" + assert bridge._extract_num_action_requests(interrupt_data) == 1 + assert bridge._extract_action_requests_from_interrupt(interrupt_data) == [ + {"name": "write_file", "args": {"path": "x.txt"}}, + ] + + +def test_agent_bridge_extracts_tool_name_from_tool_calls(): + from deep_code_agent.tui.bridge.agent_bridge import AgentBridge + + bridge = AgentBridge(agent=object()) + interrupt_data = {"tool_calls": [{"name": "read_file", "args": {"path": "x.txt"}}]} + + assert bridge._extract_tool_name_from_interrupt(interrupt_data) == "read_file" + assert bridge._extract_num_action_requests(interrupt_data) == 1 + + +def test_agent_bridge_handles_wrapped_interrupt_value(): + from deep_code_agent.tui.bridge.agent_bridge import AgentBridge + + class WrappedInterrupt: + value = {"action_requests": [{"name": "terminal", "args": {"command": "ls"}}]} + + bridge = AgentBridge(agent=object()) + + assert bridge._extract_tool_name_from_interrupt([WrappedInterrupt()]) == "terminal" + assert bridge._extract_action_requests_from_interrupt([WrappedInterrupt()]) == [ + {"name": "terminal", "args": {"command": "ls"}}, + ] + + +def test_agent_bridge_helper_fallbacks_and_config(): + from deep_code_agent.tui.bridge.agent_bridge import AgentBridge + + bridge = AgentBridge(agent=object()) + bridge.set_config({"configurable": {"thread_id": "abc"}}) + + assert bridge.config == {"configurable": {"thread_id": "abc"}} + assert bridge._extract_tool_name_from_interrupt({"unexpected": True}) == "unknown" + assert bridge._extract_action_requests_from_interrupt({"unexpected": True}) == [] + assert bridge._extract_num_action_requests({"unexpected": True}) == 1 + assert bridge.cancel_current() is None diff --git a/tests/tui/test_chat_log_todos.py b/tests/tui/test_chat_log_todos.py new file mode 100644 index 0000000..3c7ea02 --- /dev/null +++ b/tests/tui/test_chat_log_todos.py @@ -0,0 +1,93 @@ +"""Tests for ChatLog todos progress card integration.""" + +import asyncio + +from textual.app import App + + +def test_chat_log_upserts_single_todos_card(): + from deep_code_agent.tui.widgets.chat_log import ChatLog + from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard + + async def run_test(): + async with App().run_test() as pilot: + chat_log = ChatLog() + await pilot.app.mount(chat_log) + + first = chat_log.upsert_todos_card([{"content": "Plan", "status": "pending"}]) + second = chat_log.upsert_todos_card([{"content": "Verify", "status": "completed"}]) + + assert first is second + assert len([child for child in chat_log.children if isinstance(child, TodosProgressCard)]) == 1 + assert second.todos == [{"content": "Verify", "status": "completed"}] + + asyncio.run(run_test()) + + +def test_chat_log_keeps_todos_card_pinned_after_new_messages(): + from deep_code_agent.tui.widgets.chat_log import ChatLog + from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard + + async def run_test(): + async with App().run_test() as pilot: + chat_log = ChatLog() + await pilot.app.mount(chat_log) + + card = chat_log.upsert_todos_card([{"content": "Plan", "status": "pending"}]) + assert chat_log.children[-1] is card + + chat_log.add_user_message("A newer user message") + assert chat_log.children[-1] is card + + chat_log.add_agent_message("A newer message") + assert chat_log.children[-1] is card + + chat_log.add_system_message("A newer system message") + assert chat_log.children[-1] is card + + moved = chat_log.upsert_todos_card([{"content": "Plan", "status": "completed"}]) + + assert moved is card + assert isinstance(chat_log.children[-1], TodosProgressCard) + assert chat_log.children[-1] is card + + asyncio.run(run_test()) + + +def test_chat_log_preserves_card_collapsed_state_when_moved(): + from deep_code_agent.tui.widgets.chat_log import ChatLog + + async def run_test(): + async with App().run_test() as pilot: + chat_log = ChatLog() + await pilot.app.mount(chat_log) + + card = chat_log.upsert_todos_card([{"content": "Plan", "status": "pending"}]) + card.toggle_expanded() + chat_log.add_agent_message("A newer message") + moved = chat_log.upsert_todos_card([{"content": "Plan", "status": "completed"}]) + + assert moved.expanded is False + assert len(list(moved.query(".todos-body"))) == 0 + + asyncio.run(run_test()) + + +def test_chat_log_mounts_tool_widgets_above_pinned_todos_card(): + from deep_code_agent.tui.widgets.chat_log import ChatLog + + async def run_test(): + async with App().run_test() as pilot: + chat_log = ChatLog() + await pilot.app.mount(chat_log) + + card = chat_log.upsert_todos_card([{"content": "Plan", "status": "pending"}]) + + chat_log.add_tool_call("read_file", {"path": "x.txt"}) + assert chat_log.children[-1] is card + + widget = chat_log.add_tool_call_widget("write_file", {"path": "x.txt"}) + assert chat_log.children[-1] is card + assert chat_log.children[-2] is widget + + asyncio.run(run_test()) diff --git a/tests/tui/test_stream_handler.py b/tests/tui/test_stream_handler.py index 5cb82c7..c48b161 100644 --- a/tests/tui/test_stream_handler.py +++ b/tests/tui/test_stream_handler.py @@ -22,6 +22,12 @@ async def _gen(): return _gen() +class _FakeTodo: + def __init__(self, content, status): + self.content = content + self.status = status + + async def _collect(aiter): out = [] async for e in aiter: @@ -35,6 +41,7 @@ def test_event_types_include_new_tool_events(): assert hasattr(EventType, 'TOOL_START') assert hasattr(EventType, 'TOOL_SUCCESS') assert hasattr(EventType, 'TOOL_ERROR') + assert hasattr(EventType, 'TODOS_UPDATE') def test_filters_incomplete_tool_calls_without_id_or_name(): @@ -65,3 +72,73 @@ def test_dedupes_repeated_tool_calls_by_id(): tool_call_events = [e for e in events if e.type == EventType.TOOL_CALL] assert len(tool_call_events) == 1 assert tool_call_events[0].data == {"name": "write_file", "args": {}, "id": "call_123"} + + +def test_emits_todos_update_from_top_level_updates_chunk(): + from deep_code_agent.tui.bridge.stream_handler import EventType, StreamHandler + + agent = _FakeAgent(events=[ + ("updates", {"todos": [{"content": "Plan the work", "status": "pending"}]}) + ]) + handler = StreamHandler(agent, config={}) + + events = asyncio.run(_collect(handler.process({"messages": []}))) + todo_events = [e for e in events if e.type == EventType.TODOS_UPDATE] + assert len(todo_events) == 1 + assert todo_events[0].data == [{"content": "Plan the work", "status": "pending"}] + + +def test_emits_todos_update_from_nested_node_updates_chunk(): + from deep_code_agent.tui.bridge.stream_handler import EventType, StreamHandler + + agent = _FakeAgent(events=[ + ("updates", {"agent": {"todos": [{"content": "Run tests", "status": "in_progress"}]}}) + ]) + handler = StreamHandler(agent, config={}) + + events = asyncio.run(_collect(handler.process({"messages": []}))) + todo_events = [e for e in events if e.type == EventType.TODOS_UPDATE] + assert len(todo_events) == 1 + assert todo_events[0].data == [{"content": "Run tests", "status": "in_progress"}] + + +def test_normalizes_object_todo_items(): + from deep_code_agent.tui.bridge.stream_handler import EventType, StreamHandler + + agent = _FakeAgent(events=[ + ("updates", {"agent": {"todos": [_FakeTodo("Review diff", "completed")]}}) + ]) + handler = StreamHandler(agent, config={}) + + events = asyncio.run(_collect(handler.process({"messages": []}))) + todo_events = [e for e in events if e.type == EventType.TODOS_UPDATE] + assert len(todo_events) == 1 + assert todo_events[0].data == [{"content": "Review diff", "status": "completed"}] + + +def test_ignores_malformed_todos_without_crashing(): + from deep_code_agent.tui.bridge.stream_handler import EventType, StreamHandler + + agent = _FakeAgent(events=[ + ("updates", {"todos": [{"content": "Missing status"}, {"status": "pending"}, "bad"]}), + ("updates", {"agent": {"todos": "not-a-list"}}), + ]) + handler = StreamHandler(agent, config={}) + + events = asyncio.run(_collect(handler.process({"messages": []}))) + assert not any(e.type == EventType.TODOS_UPDATE for e in events) + assert events[-1].type == EventType.DONE + + +def test_supports_failed_status_in_todo_payload(): + from deep_code_agent.tui.bridge.stream_handler import EventType, StreamHandler + + agent = _FakeAgent(events=[ + ("updates", {"todos": [{"content": "Fix regression", "status": "failed"}]}) + ]) + handler = StreamHandler(agent, config={}) + + events = asyncio.run(_collect(handler.process({"messages": []}))) + todo_events = [e for e in events if e.type == EventType.TODOS_UPDATE] + assert len(todo_events) == 1 + assert todo_events[0].data == [{"content": "Fix regression", "status": "failed"}] diff --git a/tests/tui/test_todos_progress_card.py b/tests/tui/test_todos_progress_card.py new file mode 100644 index 0000000..efd26f2 --- /dev/null +++ b/tests/tui/test_todos_progress_card.py @@ -0,0 +1,98 @@ +"""Tests for TodosProgressCard widget.""" + +import asyncio + +from textual.app import App + + +def test_todos_card_defaults_expanded(): + from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard + + async def run_test(): + async with App().run_test() as pilot: + widget = TodosProgressCard([{"content": "Plan", "status": "pending"}]) + await pilot.app.mount(widget) + + assert widget.expanded is True + assert widget.todos == [{"content": "Plan", "status": "pending"}] + assert len(list(widget.query(".todo-row"))) == 1 + + asyncio.run(run_test()) + + +def test_todos_card_renders_all_statuses(): + from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard + + todos = [ + {"content": "Plan", "status": "pending"}, + {"content": "Work", "status": "in_progress"}, + {"content": "Verify", "status": "completed"}, + {"content": "Fix", "status": "failed"}, + ] + + async def run_test(): + async with App().run_test() as pilot: + widget = TodosProgressCard(todos) + await pilot.app.mount(widget) + + rows = list(widget.query(".todo-row")) + assert len(rows) == 4 + for status in ["pending", "in_progress", "completed", "failed"]: + assert len(list(widget.query(f".todo-{status}"))) == 1 + + asyncio.run(run_test()) + + +def test_todos_card_update_replaces_rows(): + from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard + + async def run_test(): + async with App().run_test() as pilot: + widget = TodosProgressCard([{"content": "Old", "status": "pending"}]) + await pilot.app.mount(widget) + + widget.update_todos([ + {"content": "New one", "status": "in_progress"}, + {"content": "New two", "status": "completed"}, + ]) + await pilot.pause() + + assert widget.todos == [ + {"content": "New one", "status": "in_progress"}, + {"content": "New two", "status": "completed"}, + ] + assert len(list(widget.query(".todo-row"))) == 2 + assert len(list(widget.query(".todo-pending"))) == 0 + + asyncio.run(run_test()) + + +def test_todos_card_update_before_compose_is_safe(): + from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard + + widget = TodosProgressCard([{"content": "Old", "status": "pending"}]) + widget.update_todos([{"content": "New", "status": "completed"}]) + + assert widget.todos == [{"content": "New", "status": "completed"}] + + +def test_todos_card_toggle_collapsed_and_expanded(): + from deep_code_agent.tui.widgets.todos_progress_card import TodosProgressCard + + async def run_test(): + async with App().run_test() as pilot: + widget = TodosProgressCard([{"content": "Plan", "status": "pending"}]) + await pilot.app.mount(widget) + + widget.toggle_expanded() + await pilot.pause() + assert widget.expanded is False + assert len(list(widget.query(".todos-body"))) == 0 + + widget.toggle_expanded() + await pilot.pause() + assert widget.expanded is True + assert len(list(widget.query(".todos-body"))) == 1 + assert len(list(widget.query(".todo-row"))) == 1 + + asyncio.run(run_test()) diff --git a/tests/tui/test_widgets.py b/tests/tui/test_widgets.py index 47b3791..d2fbdb7 100644 --- a/tests/tui/test_widgets.py +++ b/tests/tui/test_widgets.py @@ -19,3 +19,13 @@ def test_selectable_option_creation(): assert option.label == "Approve" assert option.description == "Allow execution" assert option.selected is True + + +def test_message_bubble_update_before_compose_is_safe(): + """Streaming can update a newly mounted bubble before compose runs.""" + from deep_code_agent.tui.widgets.message_bubble import MessageBubble + + bubble = MessageBubble("initial", role="agent") + bubble.update_content("latest") + + assert bubble.content == "latest"