"""Middleware for intercepting clarification requests and presenting them to the user.""" import json import logging from collections.abc import Callable from hashlib import sha256 from typing import Any, override from langchain.agents import AgentState from langchain.agents.middleware import AgentMiddleware from langchain_core.messages import ToolMessage from langgraph.graph import END from langgraph.prebuilt.tool_node import ToolCallRequest from langgraph.types import Command logger = logging.getLogger(__name__) class ClarificationMiddlewareState(AgentState): """Compatible with the `ThreadState` schema.""" pass class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]): """Intercepts clarification tool calls and interrupts execution to present questions to the user. When the model calls the `ask_clarification` tool, this middleware: 1. Intercepts the tool call before execution 2. Extracts the clarification question and metadata 3. Formats a user-friendly message 4. Returns a Command that interrupts execution and presents the question 5. Waits for user response before continuing This replaces the tool-based approach where clarification continued the conversation flow. """ state_schema = ClarificationMiddlewareState def _stable_message_id(self, tool_call_id: str, formatted_message: str) -> str: """Build a deterministic message ID so retried clarification calls replace, not append.""" if tool_call_id: return f"clarification:{tool_call_id}" digest = sha256(formatted_message.encode("utf-8")).hexdigest()[:16] return f"clarification:{digest}" def _normalize_options(self, raw_options: Any) -> list[str]: """Normalize tool-provided options into displayable string values.""" options = raw_options # Some models (e.g. Qwen3-Max) serialize array parameters as JSON strings # instead of native arrays. Deserialize and normalize so `options` # is always a list for the rendering logic below. if isinstance(options, str): try: options = json.loads(options) except (json.JSONDecodeError, TypeError): options = [options] if options is None: return [] if not isinstance(options, list): options = [options] return [str(option) for option in options] def _build_human_input_payload(self, args: dict[str, Any], *, tool_call_id: str, request_id: str) -> dict[str, Any]: """Build the structured UI payload while keeping ToolMessage.content as fallback.""" options = self._normalize_options(args.get("options", [])) clarification_type = str(args.get("clarification_type", "missing_info")) payload: dict[str, Any] = { "version": 1, "kind": "human_input_request", "source": "ask_clarification", "request_id": request_id, "clarification_type": clarification_type, "question": str(args.get("question") or ""), "input_mode": "choice_with_other" if options else "free_text", } if tool_call_id: payload["tool_call_id"] = tool_call_id if "context" in args: context = args.get("context") payload["context"] = None if context is None else str(context) if options: payload["options"] = [ { "id": f"option-{index}", "label": option, "value": option, } for index, option in enumerate(options, 1) ] return payload def _is_chinese(self, text: str) -> bool: """Check if text contains Chinese characters. Args: text: Text to check Returns: True if text contains Chinese characters """ return any("\u4e00" <= char <= "\u9fff" for char in text) def _format_clarification_message(self, args: dict) -> str: """Format the clarification arguments into a user-friendly message. Args: args: The tool call arguments containing clarification details Returns: Formatted message string """ question = args.get("question", "") clarification_type = args.get("clarification_type", "missing_info") context = args.get("context") options = self._normalize_options(args.get("options", [])) # Type-specific icons type_icons = { "missing_info": "❓", "ambiguous_requirement": "🤔", "approach_choice": "🔀", "risk_confirmation": "⚠️", "suggestion": "💡", } icon = type_icons.get(clarification_type, "❓") # Build the message naturally message_parts = [] # Add icon and question together for a more natural flow if context: # If there's context, present it first as background message_parts.append(f"{icon} {context}") message_parts.append(f"\n{question}") else: # Just the question with icon message_parts.append(f"{icon} {question}") # Add options in a cleaner format if options and len(options) > 0: message_parts.append("") # blank line for spacing for i, option in enumerate(options, 1): message_parts.append(f" {i}. {option}") return "\n".join(message_parts) def _is_disabled(self, request: ToolCallRequest) -> bool: """Whether clarifications are suppressed for this run. Non-interactive channels (e.g. GitHub webhooks) set ``disable_clarification`` in the run context because a clarification would dead-end the run — the human only "replies" via a later webhook delivery, by which point the agent's turn is long over. When set, we don't interrupt; we return a ToolMessage nudging the agent to proceed with its best judgment instead. """ runtime = getattr(request, "runtime", None) context = getattr(runtime, "context", None) if not context: return False return bool(context.get("disable_clarification")) def _handle_disabled_clarification(self, request: ToolCallRequest) -> ToolMessage: """Suppress a clarification and tell the agent to proceed. Returns a plain ToolMessage (not a ``Command(goto=END)``) so the agent loop continues instead of ending — the agent receives this as the tool result and generates again, ideally acting rather than re-asking. """ tool_call_id = request.tool_call.get("id", "") logger.info("ask_clarification suppressed (disable_clarification set); instructing agent to proceed") return ToolMessage( id=self._stable_message_id(tool_call_id, "proceed-without-clarification"), content=( "Clarification is disabled in this context — the human is not present " "to answer synchronously. Do not ask for confirmation. Proceed with your " "best judgment, carry out the requested action, and state any assumptions " "you made in your final response." ), tool_call_id=tool_call_id, name="ask_clarification", ) def _handle_clarification(self, request: ToolCallRequest) -> Command: """Handle clarification request and return command to interrupt execution. Args: request: Tool call request Returns: Command that interrupts execution with the formatted clarification message """ # Extract clarification arguments args = request.tool_call.get("args", {}) question = args.get("question", "") logger.info("Intercepted clarification request") logger.debug("Clarification question: %s", question) # Format the clarification message formatted_message = self._format_clarification_message(args) # Get the tool call ID tool_call_id = request.tool_call.get("id", "") request_id = self._stable_message_id(tool_call_id, formatted_message) human_input_payload = self._build_human_input_payload(args, tool_call_id=tool_call_id, request_id=request_id) # Create a ToolMessage with the formatted question # This will be added to the message history tool_message = ToolMessage( id=request_id, content=formatted_message, tool_call_id=tool_call_id, name="ask_clarification", artifact={"human_input": human_input_payload}, ) # Return a Command that: # 1. Adds the formatted tool message # 2. Interrupts execution by going to __end__ # Note: We don't add an extra AIMessage here - the frontend will detect # and display ask_clarification tool messages directly return Command( update={"messages": [tool_message]}, goto=END, ) @override def wrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: """Intercept ask_clarification tool calls and interrupt execution (sync version). Args: request: Tool call request handler: Original tool execution handler Returns: Command that interrupts execution with the formatted clarification message """ # Check if this is an ask_clarification tool call if request.tool_call.get("name") != "ask_clarification": # Not a clarification call, execute normally return handler(request) if self._is_disabled(request): return self._handle_disabled_clarification(request) return self._handle_clarification(request) @override async def awrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: """Intercept ask_clarification tool calls and interrupt execution (async version). Args: request: Tool call request handler: Original tool execution handler (async) Returns: Command that interrupts execution with the formatted clarification message """ # Check if this is an ask_clarification tool call if request.tool_call.get("name") != "ask_clarification": # Not a clarification call, execute normally return await handler(request) if self._is_disabled(request): return self._handle_disabled_clarification(request) return self._handle_clarification(request)