""" InputBar — message composition widget with validation, send pipeline, and conversation name display. """ from __future__ import annotations import time from textual.binding import Binding from textual.events import Key from textual.widgets import Label, Static, TextArea from za_tui.messages import MessageSent class SendTextArea(TextArea): """TextArea that delegates Enter key to parent InputBar for send.""" def on_key(self, event: Key) -> None: if event.key == "enter" and not event.shift: # Don't insert newline — let parent handle send event.prevent_default() event.stop() # Find parent InputBar and call send parent = self.parent while parent is not None: if isinstance(parent, InputBar): parent.on_send() return parent = parent.parent # Fallback: no InputBar parent found, let Enter do nothing def validate_message(text: str) -> tuple[str | None, str | None]: """Validate message text before sending. Returns (cleaned_text, None) on success, or (None, error_message) on failure. """ if not text or not text.strip(): return None, "Message cannot be empty" cleaned = text.strip() if len(cleaned) > 2000: return None, "Message too long (max 2000 chars)" return cleaned, None class InputHistory: """In-memory message history with up/down cycling.""" def __init__(self, max_size: int = 100) -> None: self._items: list[str] = [] self._index: int = -1 # -1 means at the bottom (no selection) self._max_size = max_size def push(self, text: str) -> None: """Add a message to history.""" self._items.append(text) if len(self._items) > self._max_size: self._items.pop(0) self._index = -1 # Reset position def up(self) -> str | None: """Go to older message. Returns None if at oldest.""" if not self._items: return None if self._index < len(self._items) - 1: self._index += 1 # index 0 = newest, index N = oldest idx = len(self._items) - 1 - self._index return self._items[idx] def down(self) -> str | None: """Go to newer message. Returns None if at newest.""" if not self._items: return None if self._index > 0: self._index -= 1 elif self._index == 0: self._index = -1 return None else: return None idx = len(self._items) - 1 - self._index return self._items[idx] class TypingThrottle: """Debounce typing indicator — allow at most one call per interval.""" def __init__(self, interval: float = 3.0) -> None: self._interval = interval self._last_sent: float = 0.0 def should_send(self) -> bool: """Return True if enough time has elapsed since last send.""" now = time.monotonic() if now - self._last_sent >= self._interval: self._last_sent = now return True return False class InputBar(Static): """Message input bar with TextArea, send method, and validation. Displays active conversation name above the input area. Supports optimistic send with callback injection from MainScreen. """ DEFAULT_CSS = """ InputBar { layout: vertical; height: auto; min-height: 3; max-height: 8; padding: 0 1; } InputBar .input-bar { height: auto; } InputBar #conv-label { height: 1; color: $text-muted; text-style: italic; } """ BINDINGS = [ Binding("enter", "send", "Send message", show=False, priority=True), ] def __init__( self, *, name: str | None = None, id: str | None = None, classes: str | None = None, ) -> None: merged_classes = f"input-bar {classes}".strip() if classes else "input-bar" super().__init__(name=name, id=id, classes=merged_classes) self._conversation_name: str | None = None self._chat_id: str = "" self._on_send_callback = None # Injected by MainScreen self._reply_quote: dict | None = None # Set by reply flow self._message_slice = None # Shared MessageSlice for optimistic display self._typing_throttle = TypingThrottle() # W37: typing indicator throttle def compose(self): yield Label("", id="conv-label") yield SendTextArea(id="message-input") def on_mount(self) -> None: """Auto-focus the TextArea when standalone (no sidebar or reply banner). In MainScreen context, Sidebar gets focus priority for j/k navigation. InputBar only auto-focuses when it's the primary interactive widget (standalone test context with no sidebar sibling). """ has_sidebar = False has_reply_banner = False if self.is_mounted: from za_tui.widgets.sidebar import Sidebar from za_tui.widgets.reply_banner import ReplyBanner has_sidebar = len(list(self.screen.query(Sidebar))) > 0 has_reply_banner = len(list(self.screen.query(ReplyBanner))) > 0 if not has_sidebar and not has_reply_banner: ta = self.query_one("#message-input", SendTextArea) ta.focus() def set_conversation_name(self, name: str) -> None: """Set the conversation name displayed above the input.""" self._conversation_name = name label = self.query_one("#conv-label", Label) label.update(name) @property def conversation_name(self) -> str | None: return self._conversation_name def on_send(self) -> None: """Send the current message text. Reads TextArea content, validates, emits MessageSent event, then clears the input. """ textarea = self.query_one("#message-input", TextArea) raw_text = textarea.text cleaned, error = validate_message(raw_text) if cleaned is None: return # Validation failed — don't send # Emit the message via callback or event if self._on_send_callback is not None: self._on_send_callback(cleaned, self._chat_id, self._reply_quote) else: # Fallback: post a MessageSent event self.post_message(MessageSent( text=cleaned, chat_id=self._chat_id, quote=self._reply_quote, )) # Optimistic display: add to slice if available if self._message_slice is not None and self._chat_id: self._message_slice.add_message(self._chat_id, { "msgId": f"opt-{id(self)}-{time.monotonic()}", "uidFrom": "self", "content": cleaned, "ts": time.time(), "status": "sending", }) else: # Try to find sibling ChatView for optimistic display from za_tui.widgets.chat_view import ChatView for cv in self.screen.query(ChatView): if cv._chat_id: cv._slice.add_message(cv._chat_id, { "msgId": f"opt-{id(self)}-{time.monotonic()}", "uidFrom": "self", "content": cleaned, "ts": time.time(), "status": "sending", }) cv.refresh() break # Track last sent for test introspection self._last_sent_message = cleaned self._send_called = True # Clear reply context self._reply_quote = None # Dismiss any ReplyBanner sibling from za_tui.widgets.reply_banner import ReplyBanner for banner in self.screen.query(ReplyBanner): banner.visible = False # Clear input textarea.clear() self._update_label_after_send() def _update_label_after_send(self) -> None: """Ensure label still shows conversation name after clearing.""" if self._conversation_name: label = self.query_one("#conv-label", Label) label.update(self._conversation_name) def on_key(self, event: Key) -> None: """Intercept keys: Enter sends, other printable keys trigger typing indicator (W37).""" if event.key == "enter" and not event.shift: event.prevent_default() self.action_send() return # W37: Send typing indicator on printable keypress if event.character and self._chat_id and self._typing_throttle.should_send(): bridge = getattr(self.app, "_bridge", None) if self.app else None if bridge: import asyncio asyncio.create_task( bridge.request("sendTyping", {"threadId": self._chat_id}) ) def action_send(self) -> None: """Action handler for sending message.""" self.on_send()