""" Claude Message Handler Platform-agnostic Claude interaction logic. Handles the core workflow of processing user messages via Claude CLI. Uses tree-based queuing for message ordering. """ import asyncio from loguru import logger from core.anthropic import format_user_error_preview, get_user_facing_error_message from core.trace import trace_event from .cli_event_constants import STATUS_MESSAGE_PREFIXES from .command_dispatcher import ( dispatch_command, message_kind_for_command, parse_command_base, ) from .event_parser import parse_cli_event from .models import IncomingMessage from .node_event_pipeline import handle_session_info_event, process_parsed_cli_event from .platforms.base import MessagingPlatform, SessionManagerInterface from .rendering.profiles import build_rendering_profile from .safe_diagnostics import format_exception_for_log from .session import SessionStore from .transcript import RenderCtx, TranscriptBuffer from .trees.queue_manager import ( MessageNode, MessageState, MessageTree, TreeQueueManager, ) from .ui_updates import ThrottledTranscriptEditor class ClaudeMessageHandler: """ Platform-agnostic handler for Claude interactions. Uses a tree-based message queue where: - New messages create a tree root - Replies become children of the message being replied to - Each node has state: PENDING, IN_PROGRESS, COMPLETED, ERROR - Per-tree queue ensures ordered processing """ def __init__( self, platform: MessagingPlatform, cli_manager: SessionManagerInterface, session_store: SessionStore, *, debug_platform_edits: bool = False, debug_subagent_stack: bool = False, log_raw_messaging_content: bool = False, log_raw_cli_diagnostics: bool = False, log_messaging_error_details: bool = False, ): self.platform = platform self.cli_manager = cli_manager self.session_store = session_store self._debug_platform_edits = debug_platform_edits self._debug_subagent_stack = debug_subagent_stack self._log_raw_messaging_content = log_raw_messaging_content self._log_raw_cli_diagnostics = log_raw_cli_diagnostics self._log_messaging_error_details = log_messaging_error_details self._tree_queue = TreeQueueManager( queue_update_callback=self.update_queue_positions, node_started_callback=self.mark_node_processing, ) self._rendering_profile = build_rendering_profile(platform.name) def format_status(self, emoji: str, label: str, suffix: str | None = None) -> str: return self._rendering_profile.format_status(emoji, label, suffix) def _parse_mode(self) -> str | None: return self._rendering_profile.parse_mode def get_render_ctx(self) -> RenderCtx: return self._rendering_profile.render_ctx def _get_limit_chars(self) -> int: return self._rendering_profile.limit_chars @property def tree_queue(self) -> TreeQueueManager: """Accessor for the current tree queue manager.""" return self._tree_queue def replace_tree_queue(self, tree_queue: TreeQueueManager) -> None: """Replace tree queue manager via explicit API.""" self._tree_queue = tree_queue self._tree_queue.set_queue_update_callback(self.update_queue_positions) self._tree_queue.set_node_started_callback(self.mark_node_processing) async def handle_message(self, incoming: IncomingMessage) -> None: """ Main entry point for handling an incoming message. Determines if this is a new conversation or reply, creates/extends the message tree, and queues for processing. """ platform_name = getattr(self.platform, "name", "messaging") trace_event( stage="ingress", event="turn.received", source=platform_name, chat_id=incoming.chat_id, platform_message_id=incoming.message_id, reply_to_message_id=incoming.reply_to_message_id, thread_id=getattr(incoming, "message_thread_id", None), message_text=incoming.text or "", ) with logger.contextualize( chat_id=incoming.chat_id, node_id=incoming.message_id ): await self._handle_message_impl(incoming) async def _handle_message_impl(self, incoming: IncomingMessage) -> None: """Implementation of handle_message with context bound.""" cmd_base = parse_command_base(incoming.text) # Record incoming message ID for best-effort UI clearing (/clear), even if # we later ignore this message (status/command/etc). try: if incoming.message_id is not None: self.session_store.record_message_id( incoming.platform, incoming.chat_id, str(incoming.message_id), direction="in", kind=message_kind_for_command(cmd_base), ) except Exception as e: logger.debug( "Failed to record incoming message_id: {}", format_exception_for_log( e, log_full_message=self._log_messaging_error_details ), ) if await dispatch_command(self, incoming, cmd_base): return # Filter out status messages (our own messages) text = incoming.text or "" if any(text.startswith(p) for p in STATUS_MESSAGE_PREFIXES): return # Check if this is a reply to an existing node in a tree parent_node_id = None tree = None if incoming.is_reply() and incoming.reply_to_message_id: # Look up if the replied-to message is in any tree (could be a node or status message) reply_id = incoming.reply_to_message_id tree = self.tree_queue.get_tree_for_node(reply_id) if tree: # Resolve to actual node ID (handles status message replies) parent_node_id = self.tree_queue.resolve_parent_node_id(reply_id) if parent_node_id: logger.info(f"Found tree for reply, parent node: {parent_node_id}") else: logger.warning( f"Reply to {incoming.reply_to_message_id} found tree but no valid parent node" ) tree = None # Treat as new conversation # Generate node ID node_id = incoming.message_id # Use pre-sent status (e.g. voice note) or send new status_text = self._get_initial_status(tree, parent_node_id) if incoming.status_message_id: status_msg_id = incoming.status_message_id await self.platform.queue_edit_message( incoming.chat_id, status_msg_id, status_text, parse_mode=self._parse_mode(), fire_and_forget=False, ) else: status_msg_id = await self.platform.queue_send_message( incoming.chat_id, status_text, reply_to=incoming.message_id, fire_and_forget=False, message_thread_id=incoming.message_thread_id, ) self.record_outgoing_message( incoming.platform, incoming.chat_id, status_msg_id, "status" ) # Create or extend tree if parent_node_id and tree and status_msg_id: # Reply to existing node - add as child tree, _node = await self.tree_queue.add_to_tree( parent_node_id=parent_node_id, node_id=node_id, incoming=incoming, status_message_id=status_msg_id, ) # Register status message as a node too for reply chains self.tree_queue.register_node(status_msg_id, tree.root_id) self.session_store.register_node(status_msg_id, tree.root_id) self.session_store.register_node(node_id, tree.root_id) elif status_msg_id: # New conversation - create new tree tree = await self.tree_queue.create_tree( node_id=node_id, incoming=incoming, status_message_id=status_msg_id, ) # Register status message self.tree_queue.register_node(status_msg_id, tree.root_id) self.session_store.register_node(node_id, tree.root_id) self.session_store.register_node(status_msg_id, tree.root_id) # Persist tree if tree: self.session_store.save_tree(tree.root_id, tree.to_dict()) # Enqueue for processing was_queued = await self.tree_queue.enqueue( node_id=node_id, processor=self._process_node, ) if was_queued and status_msg_id: queue_size = self.tree_queue.get_queue_size(node_id) trace_event( stage="routing", event="turn.queued", source=getattr(self.platform, "name", "messaging"), chat_id=incoming.chat_id, platform_message_id=node_id, status_message_id=status_msg_id, queue_size=queue_size, ) await self.platform.queue_edit_message( incoming.chat_id, status_msg_id, self.format_status( "📋", "Queued", f"(position {queue_size}) - waiting..." ), parse_mode=self._parse_mode(), ) async def update_queue_positions(self, tree: MessageTree) -> None: """Refresh queued status messages after a dequeue.""" try: queued_ids = await tree.get_queue_snapshot() except Exception as e: logger.warning( "Failed to read queue snapshot: {}", format_exception_for_log( e, log_full_message=self._log_messaging_error_details ), ) return if not queued_ids: return position = 0 for node_id in queued_ids: node = tree.get_node(node_id) if not node or node.state != MessageState.PENDING: continue position += 1 self.platform.fire_and_forget( self.platform.queue_edit_message( node.incoming.chat_id, node.status_message_id, self.format_status( "📋", "Queued", f"(position {position}) - waiting..." ), parse_mode=self._parse_mode(), ) ) async def mark_node_processing(self, tree: MessageTree, node_id: str) -> None: """Update the dequeued node's status to processing immediately.""" node = tree.get_node(node_id) if not node or node.state == MessageState.ERROR: return self.platform.fire_and_forget( self.platform.queue_edit_message( node.incoming.chat_id, node.status_message_id, self.format_status("🔄", "Processing..."), parse_mode=self._parse_mode(), ) ) def _create_transcript_and_render_ctx( self, ) -> tuple[TranscriptBuffer, RenderCtx]: """Create transcript buffer and render context for node processing.""" transcript = TranscriptBuffer( show_tool_results=False, debug_subagent_stack=self._debug_subagent_stack, ) return transcript, self.get_render_ctx() async def _process_node( self, node_id: str, node: MessageNode, ) -> None: """Core task processor - handles a single Claude CLI interaction.""" incoming = node.incoming status_msg_id = node.status_message_id chat_id = incoming.chat_id with logger.contextualize(node_id=node_id, chat_id=chat_id): await self._process_node_impl(node_id, node, chat_id, status_msg_id) async def _process_node_impl( self, node_id: str, node: MessageNode, chat_id: str, status_msg_id: str, ) -> None: """Internal implementation of _process_node with context bound.""" incoming = node.incoming tree = self.tree_queue.get_tree_for_node(node_id) if tree: await tree.update_state(node_id, MessageState.IN_PROGRESS) transcript, render_ctx = self._create_transcript_and_render_ctx() had_transcript_events = False captured_session_id = None temp_session_id = None last_status: str | None = None parent_session_id = None platform_nm = getattr(self.platform, "name", "messaging") if tree and node.parent_id: parent_session_id = tree.get_parent_session_id(node_id) if parent_session_id: trace_event( stage="claude_cli", event="claude_cli.fork.from_parent_session", source=platform_nm, chat_id=chat_id, node_id=node_id, parent_session_id=parent_session_id, ) editor = ThrottledTranscriptEditor( platform=self.platform, parse_mode=self._parse_mode(), get_limit_chars=self._get_limit_chars, transcript=transcript, render_ctx=render_ctx, node_id=node_id, chat_id=chat_id, status_msg_id=status_msg_id, debug_platform_edits=self._debug_platform_edits, log_messaging_error_details=self._log_messaging_error_details, ) async def update_ui(status: str | None = None, force: bool = False) -> None: await editor.update(status, force=force) try: try: ( cli_session, session_or_temp_id, is_new, ) = await self.cli_manager.get_or_create_session( session_id=parent_session_id ) if is_new: temp_session_id = session_or_temp_id else: captured_session_id = session_or_temp_id sess_evt = ( "claude_cli.session.pending_created" if is_new else "claude_cli.session.reused" ) trace_event( stage="claude_cli", event=sess_evt, source=platform_nm, chat_id=chat_id, node_id=node_id, status_message_id=status_msg_id, session_handle=str(session_or_temp_id), parent_resume_session_id=parent_session_id, fork_requested=bool(parent_session_id), ) trace_event( stage="claude_cli", event="claude_cli.request.sent", source=platform_nm, chat_id=chat_id, node_id=node_id, prompt=incoming.text, fork_session_arg=bool(parent_session_id), resume_session_arg=parent_session_id, ) except RuntimeError as e: error_message = get_user_facing_error_message(e) transcript.apply({"type": "error", "message": error_message}) await update_ui( self.format_status("⏳", "Session limit reached"), force=True, ) if tree: await tree.update_state( node_id, MessageState.ERROR, error_message=error_message, ) trace_event( stage="claude_cli", event="claude_cli.session.limit_reached", source=platform_nm, chat_id=chat_id, node_id=node_id, ) return async for event_data in cli_session.start_task( incoming.text, session_id=parent_session_id, fork_session=bool(parent_session_id), ): if not isinstance(event_data, dict): logger.warning( f"HANDLER: Non-dict event received: {type(event_data)}" ) continue ( captured_session_id, temp_session_id, ) = await handle_session_info_event( event_data, tree, node_id, captured_session_id, temp_session_id, cli_manager=self.cli_manager, session_store=self.session_store, ) if event_data.get("type") == "session_info": continue parsed_list = parse_cli_event( event_data, log_raw_cli=self._log_raw_cli_diagnostics ) for parsed in parsed_list: ( last_status, had_transcript_events, ) = await process_parsed_cli_event( parsed, transcript, update_ui, last_status, had_transcript_events, tree, node_id, captured_session_id, session_store=self.session_store, format_status=self.format_status, propagate_error_to_children=self._propagate_error_to_children, log_messaging_error_details=self._log_messaging_error_details, ) except asyncio.CancelledError: trace_event( stage="claude_cli", event="turn.processor.cancelled", source=platform_nm, chat_id=chat_id, node_id=node_id, ) logger.warning(f"HANDLER: Task cancelled for node {node_id}") cancel_reason = None if isinstance(node.context, dict): cancel_reason = node.context.get("cancel_reason") if cancel_reason == "stop": await update_ui(self.format_status("⏹", "Stopped."), force=True) else: transcript.apply({"type": "error", "message": "Task was cancelled"}) await update_ui(self.format_status("❌", "Cancelled"), force=True) # Do not propagate cancellation to children; a reply-scoped "/stop" # should only stop the targeted task. if tree: await tree.update_state( node_id, MessageState.ERROR, error_message="Cancelled by user" ) except Exception as e: trace_event( stage="claude_cli", event="turn.processor.exception", source=platform_nm, chat_id=chat_id, node_id=node_id, exc_type=type(e).__name__, ) logger.error( "HANDLER: Task failed with exception: {}", format_exception_for_log( e, log_full_message=self._log_messaging_error_details ), ) error_msg = format_user_error_preview(e) transcript.apply({"type": "error", "message": error_msg}) await update_ui(self.format_status("💥", "Task Failed"), force=True) if tree: await self._propagate_error_to_children( node_id, error_msg, "Parent task failed" ) finally: trace_event( stage="routing", event="turn.processor.finished", source=platform_nm, chat_id=chat_id, node_id=node_id, claude_session_id=captured_session_id or temp_session_id, ) # Free the session-manager slot. Session IDs are persisted in the tree and # can be resumed later by ID; we don't need to keep a CLISession instance # around after this node completes. try: if captured_session_id: await self.cli_manager.remove_session(captured_session_id) elif temp_session_id: await self.cli_manager.remove_session(temp_session_id) except Exception as e: logger.debug( "Failed to remove session for node {}: {}", node_id, format_exception_for_log( e, log_full_message=self._log_messaging_error_details ), ) async def _propagate_error_to_children( self, node_id: str, error_msg: str, child_status_text: str, ) -> None: """Mark node as error and propagate to pending children with UI updates.""" affected = await self.tree_queue.mark_node_error( node_id, error_msg, propagate_to_children=True ) # Update status messages for all affected children (skip first = current node) for child in affected[1:]: self.platform.fire_and_forget( self.platform.queue_edit_message( child.incoming.chat_id, child.status_message_id, self.format_status("❌", "Cancelled:", child_status_text), parse_mode=self._parse_mode(), ) ) def _get_initial_status( self, tree: object | None, parent_node_id: str | None, ) -> str: """Get initial status message text.""" if tree and parent_node_id: # Reply to existing tree if self.tree_queue.is_node_tree_busy(parent_node_id): queue_size = self.tree_queue.get_queue_size(parent_node_id) + 1 return self.format_status( "📋", "Queued", f"(position {queue_size}) - waiting..." ) return self.format_status("🔄", "Continuing conversation...") # New conversation return self.format_status("⏳", "Launching new Claude CLI instance...") async def stop_all_tasks(self) -> int: """ Stop all pending and in-progress tasks. Order of operations: 1. Cancel tree queue tasks (uses internal locking) 2. Stop CLI sessions 3. Update UI for all affected nodes """ # 1. Cancel tree queue tasks using the public async method logger.info("Cancelling tree queue tasks...") cancelled_nodes = await self.tree_queue.cancel_all() logger.info(f"Cancelled {len(cancelled_nodes)} nodes") # 2. Stop CLI sessions - this kills subprocesses and ensures everything is dead logger.info("Stopping all CLI sessions...") await self.cli_manager.stop_all() # 3. Update UI and persist state for all cancelled nodes self.update_cancelled_nodes_ui(cancelled_nodes) return len(cancelled_nodes) async def stop_task(self, node_id: str) -> int: """ Stop a single queued or in-progress task node. Used when the user replies "/stop" to a specific status/user message. """ tree = self.tree_queue.get_tree_for_node(node_id) if tree: node = tree.get_node(node_id) if node and node.state not in (MessageState.COMPLETED, MessageState.ERROR): # Used by _process_node cancellation path to render "Stopped." node.set_context({"cancel_reason": "stop"}) cancelled_nodes = await self.tree_queue.cancel_node(node_id) self.update_cancelled_nodes_ui(cancelled_nodes) return len(cancelled_nodes) def record_outgoing_message( self, platform: str, chat_id: str, msg_id: str | None, kind: str, ) -> None: """Record outgoing message ID for /clear. Best-effort, never raises.""" if not msg_id: return try: self.session_store.record_message_id( platform, chat_id, str(msg_id), direction="out", kind=kind ) except Exception as e: logger.debug( "Failed to record message_id: {}", format_exception_for_log( e, log_full_message=self._log_messaging_error_details ), ) def update_cancelled_nodes_ui(self, nodes: list[MessageNode]) -> None: """Update status messages and persist tree state for cancelled nodes.""" trees_to_save: dict[str, MessageTree] = {} for node in nodes: self.platform.fire_and_forget( self.platform.queue_edit_message( node.incoming.chat_id, node.status_message_id, self.format_status("⏹", "Stopped."), parse_mode=self._parse_mode(), ) ) tree = self.tree_queue.get_tree_for_node(node.node_id) if tree: trees_to_save[tree.root_id] = tree for root_id, tree in trees_to_save.items(): self.session_store.save_tree(root_id, tree.to_dict())