diff --git a/pyproject.toml b/pyproject.toml index 966ac0152..f9cfa06a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [project] name = "uipath-langchain" -version = "0.18.51" +version = "0.18.52" description = "Python SDK that enables developers to build and deploy LangGraph agents to the UiPath Cloud Platform" readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" dependencies = [ - "uipath>=2.14.40, <2.15.0", + "uipath>=2.14.44, <2.15.0", "uipath-core>=0.5.33, <0.6.0", "uipath-platform>=0.2.39, <0.3.0", "uipath-runtime>=0.13.5, <0.14.0", diff --git a/src/uipath_langchain/agent/tools/a2a/conversational_agent_tool.py b/src/uipath_langchain/agent/tools/a2a/conversational_agent_tool.py index 742bcdc95..b8e23f46b 100644 --- a/src/uipath_langchain/agent/tools/a2a/conversational_agent_tool.py +++ b/src/uipath_langchain/agent/tools/a2a/conversational_agent_tool.py @@ -3,12 +3,20 @@ A conversational agent is a published Orchestrator release. It is resolved by process name in the execution folder, like a process tool, and reached through AgentHub's ``a2a/{folderKey}/{releaseId}`` endpoint. Unlike a Remote A2A agent -(``a2a_tool``) it is not looked up in the Remote A2A registry, and the model -addresses its conversations by a ``conversation_id`` it passes back. - -In the ReAct graph the conversations are kept in ``tools_storage``. Where tools -run without a wrapper (advanced mode) nothing is stored. Client lifecycle is -managed by the caller via ``ConversationalAgentA2aClient.dispose()`` or the +(``a2a_tool``) it is not looked up in the Remote A2A registry. + +The tool's ``settings.conversationMode`` picks how the model deals with +conversations. In ``single`` mode (the default) the model passes only a message +and every call continues the tool's one conversation. In ``multiple`` mode the +model addresses conversations by a ``conversation_id`` it passes back. + +In the ReAct graph the conversations are kept in ``tools_storage``. Each chat +message is a new job with empty tool state, but the earlier turns are replayed in +the messages, so a call that finds no current conversation rebuilds it from this +tool's last result there. Where tools run without a wrapper (advanced mode) +nothing is stored: the tool gets the graph's messages injected and reads the +conversation from them. Client lifecycle is managed by the caller via +``ConversationalAgentA2aClient.dispose()`` or the ``open_conversational_agent_tools`` async context manager. """ @@ -17,7 +25,15 @@ import os from contextlib import AsyncExitStack, asynccontextmanager from logging import getLogger -from typing import Any, AsyncGenerator, Awaitable, Callable, Sequence +from typing import ( + Annotated, + Any, + AsyncGenerator, + Awaitable, + Callable, + Literal, + Sequence, +) import httpx from a2a.client import Client @@ -30,8 +46,9 @@ ) from a2a.utils.errors import InvalidRequestError, TaskNotFoundError from google.protobuf.json_format import ParseDict -from langchain_core.messages import ToolCall, ToolMessage +from langchain_core.messages import AIMessage, AnyMessage, ToolCall, ToolMessage from langchain_core.tools import BaseTool +from langgraph.prebuilt import InjectedState from langgraph.types import Command from opentelemetry import trace as otel_trace from pydantic import BaseModel, Field @@ -73,6 +90,21 @@ "continue a specific conversation." ) +_SINGLE_CONVERSATION_HINT = ( + "Each call continues one ongoing conversation with this agent. " + "Make one call at a time." +) + +_NOTE_NOT_FOUND = ( + "The previous conversation was not found; a new conversation was started." +) +_NOTE_ENDED = "The previous conversation had ended; a new conversation was started." +_NOTE_NO_MEMORY = ( + "The agent has no memory of earlier messages; if your message depends on earlier " + "context, send it again with that context." +) +_HANDLE_ARGUMENTS = ("conversation_id", "new_conversation") + class A2aConversationInput(BaseModel): """Input schema for a conversational agent tool.""" @@ -88,6 +120,15 @@ class A2aConversationInput(BaseModel): ) +class A2aSingleConversationInput(BaseModel): + """Input schema for a conversational agent tool in single mode. + + Carries only the message: every call continues the tool's one conversation. + """ + + message: str = Field(description="The message to send to the agent.") + + @resource_override(resource_type="process", resource_identifier="process_name") async def _resolve_conversational_a2a_url( sdk: Any, process_name: str, folder_path: str | None @@ -308,12 +349,16 @@ def _record_conversation( *, closed: bool, make_current: bool, + keep_closed_current: bool = False, ) -> dict[str, Any]: """Return a copy of the state with the conversation recorded as just used. - ``last_used`` is a counter derived from the stored entries, so the state is - deterministic. Beyond ``_MAX_CONVERSATIONS`` the least recently used - conversation is evicted, closed ones first. + A closed conversation stops being the current one unless + ``keep_closed_current`` (single mode, which has no other conversation to + point at and starts a new one on the next call). ``last_used`` is a counter + derived from the stored entries, so the state is deterministic. Beyond + ``_MAX_CONVERSATIONS`` the least recently used conversation is evicted, + closed ones first. """ conversations = { key: dict(value) for key, value in conversation_state["conversations"].items() @@ -325,7 +370,7 @@ def _record_conversation( "last_used": last_used, } current = conversation_id if make_current else conversation_state["current"] - if closed and current == conversation_id: + if closed and not keep_closed_current and current == conversation_id: current = None while len(conversations) > _MAX_CONVERSATIONS: evicted = min( @@ -367,6 +412,167 @@ def _load_conversation_state(stored: dict[str, Any]) -> dict[str, Any]: } +def _unavailable_reason( + error: _A2aRequestError, +) -> Literal["not_found", "ended"] | None: + """Tell a conversation AgentHub no longer has (or has ended) from other errors.""" + if isinstance(error.__cause__, TaskNotFoundError): + return "not_found" + if _is_terminal_state_error(error): + return "ended" + return None + + +def _empty_conversation_state() -> dict[str, Any]: + return {"current": None, "conversations": {}} + + +def _message_text(content: str | list[str | dict[str, Any]]) -> str: + if isinstance(content, str): + return content + return "".join( + block if isinstance(block, str) else str(block.get("text", "")) + for block in content + if isinstance(block, str) or block.get("type") == "text" + ) + + +def _read_result(message: ToolMessage) -> tuple[str | None, bool]: + """Read the conversation of one of this tool's earlier results, and whether it ended. + + A result without a conversation, such as an error, a cancelled confirmation or + an interrupted call, gives ``None``. A failed request still names its + conversation, which is alive. + """ + if message.status == "error": + return None, False + try: + payload = json.loads(_message_text(message.content)) + except ValueError: + return None, False + # A call whose arguments the user edited is wrapped with the executed arguments. + if ( + isinstance(payload, dict) + and "meta" in payload + and isinstance(payload.get("result"), dict) + ): + payload = payload["result"] + if not isinstance(payload, dict): + return None, False + conversation_id = payload.get("conversation_id") + if not isinstance(conversation_id, str) or not conversation_id: + return None, False + ended = ( + payload.get("error") == "conversation_closed" + or payload.get("task_state") in _TERMINAL_TASK_STATES + ) + return conversation_id, ended + + +def _called_tool_name( + messages: Sequence[AnyMessage], before: int, tool_call_id: str +) -> str | None: + """Name of the tool that the nearest earlier assistant message called with this id.""" + for index in range(before - 1, -1, -1): + message = messages[index] + if isinstance(message, AIMessage): + for tool_call in message.tool_calls: + if tool_call["id"] == tool_call_id: + return tool_call["name"] + return None + + +def _conversation_id_from_history( + messages: Sequence[AnyMessage], tool_name: str +) -> str | None: + """Return the newest conversation in this tool's results in the history that is open. + + Walks backwards over the tool's results, skipping those without a conversation + and the conversations that a later result shows as ended. A result is this + tool's when the assistant tool call it answers names the tool; + ``ToolMessage.name`` is only the fallback, because replayed messages may not + carry it. + """ + ended: set[str] = set() + for index in range(len(messages) - 1, -1, -1): + message = messages[index] + if not isinstance(message, ToolMessage): + continue + called = ( + _called_tool_name(messages, index, message.tool_call_id) or message.name + ) + if called != tool_name: + continue + conversation_id, has_ended = _read_result(message) + if conversation_id is None: + continue + if has_ended: + ended.add(conversation_id) + elif conversation_id not in ended: + return conversation_id + return None + + +def _seed_from_history( + conversation_state: dict[str, Any], + messages: Sequence[AnyMessage], + tool_name: str, +) -> dict[str, Any]: + """Adopt the conversation of the tool's last result, if the state has none. + + Each chat message is a new job whose tool state is empty, while the replayed + messages still hold the earlier results. A conversation the state already + knows, such as one that has ended, is never reopened this way. + """ + if conversation_state["current"] is not None: + return conversation_state + conversation_id = _conversation_id_from_history(messages, tool_name) + if ( + conversation_id is None + or conversation_id in conversation_state["conversations"] + ): + return conversation_state + return _record_conversation( + conversation_state, + conversation_id, + None, + closed=False, + make_current=True, + ) + + +def _record_reply( + conversation_state: dict[str, Any], + conversation_id: str | None, + task_id: str | None, + reply: tuple[str, str, str | None, str | None], + *, + keep_closed_current: bool = False, +) -> tuple[dict[str, Any], dict[str, Any]]: + """Record the conversation a reply belongs to and return the result with the state. + + ``conversation_id`` and ``task_id`` are those the message was sent with. + """ + text, task_state, new_task_id, new_context_id = reply + response_conversation_id = new_context_id or new_task_id or conversation_id + if response_conversation_id: + conversation_state = _record_conversation( + conversation_state, + response_conversation_id, + new_task_id or task_id, + closed=task_state in _TERMINAL_TASK_STATES, + make_current=response_conversation_id != conversation_id + or conversation_state["current"] is None, + keep_closed_current=keep_closed_current, + ) + result = { + "agent_response": text, + "conversation_id": response_conversation_id, + "task_state": task_state, + } + return result, conversation_state + + async def _converse( args: A2aConversationInput, conversation_state: dict[str, Any], @@ -410,12 +616,10 @@ async def _converse( task_id = known_conversation["task_id"] if known_conversation else None try: - text, task_state, new_task_id, new_context_id = await send( - args.message, task_id, conversation_id - ) + reply = await send(args.message, task_id, conversation_id) except _A2aRequestError as e: - task_not_found = isinstance(e.__cause__, TaskNotFoundError) - if conversation_id and (task_not_found or _is_terminal_state_error(e)): + reason = _unavailable_reason(e) if conversation_id else None + if conversation_id and reason: logger.info("A2A conversation %s is unavailable: %s", conversation_id, e) conversation_state = _record_conversation( conversation_state, @@ -426,7 +630,7 @@ async def _converse( ) message = ( "This conversation is no longer available. " - if task_not_found + if reason == "not_found" else "This conversation has ended. " ) return ( @@ -443,25 +647,77 @@ async def _converse( conversation_state, ) - response_conversation_id = new_context_id or new_task_id or conversation_id - if response_conversation_id: + result, conversation_state = _record_reply( + conversation_state, conversation_id, task_id, reply + ) + if conversation_id and result["conversation_id"] != conversation_id: + result["note"] = _NOTE_NOT_FOUND + return json.dumps(result), conversation_state + + +async def _converse_single( + message: str, + conversation_state: dict[str, Any], + send: _ConversationSend, +) -> tuple[str, dict[str, Any]]: + """Send one message to the state's ``current`` conversation. + + When it has ended, or AgentHub no longer has it, a new one is started and the + result says so in ``note``: the model cannot start one itself. A message + that hit such a conversation is sent again, since AgentHub did not act on it. + """ + conversation_id = conversation_state["current"] + known_conversation = ( + conversation_state["conversations"].get(conversation_id) + if conversation_id + else None + ) + note: str | None = None + if known_conversation and known_conversation["state"] == "closed": + conversation_id, known_conversation, note = None, None, _NOTE_ENDED + task_id = known_conversation["task_id"] if known_conversation else None + + try: + reply = await send(message, task_id, conversation_id) + except _A2aRequestError as e: + reason = _unavailable_reason(e) if conversation_id else None + if conversation_id is None or reason is None: + logger.exception("A2A conversation request failed") + return ( + _conversation_error("request_failed", conversation_id, f"Error: {e}"), + conversation_state, + ) + logger.info("A2A conversation %s is unavailable: %s", conversation_id, e) conversation_state = _record_conversation( conversation_state, - response_conversation_id, - new_task_id or task_id, - closed=task_state in _TERMINAL_TASK_STATES, - make_current=response_conversation_id != conversation_id - or conversation_state["current"] is None, - ) - result = { - "agent_response": text, - "conversation_id": response_conversation_id, - "task_state": task_state, - } - if conversation_id and response_conversation_id != conversation_id: - result["note"] = ( - "The previous conversation was not found; a new conversation was started." + conversation_id, + task_id, + closed=True, + make_current=False, + keep_closed_current=True, ) + note = _NOTE_NOT_FOUND if reason == "not_found" else _NOTE_ENDED + conversation_id, task_id = None, None + try: + reply = await send(message, None, None) + except _A2aRequestError as retry_error: + logger.exception("A2A conversation request failed") + return ( + _conversation_error("request_failed", None, f"Error: {retry_error}"), + conversation_state, + ) + + result, conversation_state = _record_reply( + conversation_state, + conversation_id, + task_id, + reply, + keep_closed_current=True, + ) + if conversation_id and result["conversation_id"] != conversation_id: + note = _NOTE_NOT_FOUND + if note: + result["note"] = f"{note} {_NOTE_NO_MEMORY}" return json.dumps(result), conversation_state @@ -470,21 +726,35 @@ def _create_conversation_tool( description: str, metadata: dict[str, str], send: _ConversationSend, + mode: Literal["single", "multiple"], ) -> BaseTool: """Create the LLM-facing tool for a conversational agent. - The model addresses conversations by the ``conversation_id`` returned in each - result. In the ReAct graph the wrapper keeps them in ``tools_storage``. Where - tools run without a wrapper (advanced mode) nothing is stored: tool instances - outlive a run, so a call that passes no ``conversation_id`` starts a new - conversation instead of continuing one from an earlier run. + The ReAct wrapper keeps the conversations in ``tools_storage``. Where tools + run without a wrapper (advanced mode) nothing is stored, because tool + instances outlive a run: the graph's messages are injected instead and the + conversation is read from them, as a new job does. Called outside a graph, + every call without a ``conversation_id`` starts a new conversation. """ - async def _send_stateless( + async def _send_single( + *, + message: str, + messages: Annotated[Sequence[AnyMessage], InjectedState("messages")] = (), + ) -> str: + result, _ = await _converse_single( + message, + _seed_from_history(_empty_conversation_state(), messages, name), + send, + ) + return result + + async def _send_multiple( *, message: str, conversation_id: str | None = None, new_conversation: bool | None = False, + messages: Annotated[Sequence[AnyMessage], InjectedState("messages")] = (), ) -> str: result, _ = await _converse( A2aConversationInput( @@ -492,7 +762,7 @@ async def _send_stateless( conversation_id=conversation_id, new_conversation=new_conversation, ), - {"current": None, "conversations": {}}, + _seed_from_history(_empty_conversation_state(), messages, name), send, ) return result @@ -502,15 +772,33 @@ async def _conversation_wrapper( call: ToolCall, state: AgentGraphState, ) -> ToolWrapperReturnType: - conversation_state = _load_conversation_state( - state.inner_state.tools_storage.get(tool.name) or {} + conversation_state = _seed_from_history( + _load_conversation_state( + state.inner_state.tools_storage.get(tool.name) or {} + ), + state.messages, + tool.name, ) - result, updated_state = await _converse( - A2aConversationInput.model_validate(call["args"]), - conversation_state, - send, - ) + if mode == "single": + ignored = [arg for arg in _HANDLE_ARGUMENTS if arg in call["args"]] + if ignored: + logger.info( + "Ignoring %s: conversational agent tool '%s' is in single mode", + ", ".join(ignored), + tool.name, + ) + result, updated_state = await _converse_single( + A2aSingleConversationInput.model_validate(call["args"]).message, + conversation_state, + send, + ) + else: + result, updated_state = await _converse( + A2aConversationInput.model_validate(call["args"]), + conversation_state, + send, + ) return Command( update={ @@ -525,13 +813,22 @@ async def _conversation_wrapper( } ) - tool = A2aStructuredToolWithWrapper( - name=name, - description=f"{description} {_CONVERSATION_HINT}", - coroutine=_send_stateless, - args_schema=A2aConversationInput, - metadata=metadata, - ) + if mode == "single": + tool = A2aStructuredToolWithWrapper( + name=name, + description=f"{description} {_SINGLE_CONVERSATION_HINT}", + coroutine=_send_single, + args_schema=A2aSingleConversationInput, + metadata=metadata, + ) + else: + tool = A2aStructuredToolWithWrapper( + name=name, + description=f"{description} {_CONVERSATION_HINT}", + coroutine=_send_multiple, + args_schema=A2aConversationInput, + metadata=metadata, + ) tool.set_tool_wrappers(awrapper=_conversation_wrapper) return tool @@ -625,6 +922,7 @@ async def _invoke( _build_description(agent_card), metadata, _invoke, + config.settings.conversation_mode or "single", ) diff --git a/tests/agent/tools/test_conversational_agent_tool.py b/tests/agent/tools/test_conversational_agent_tool.py index fbb840af7..b6675e20d 100644 --- a/tests/agent/tools/test_conversational_agent_tool.py +++ b/tests/agent/tools/test_conversational_agent_tool.py @@ -7,9 +7,10 @@ import asyncio import json +import logging from collections.abc import Callable from types import SimpleNamespace -from typing import Any, cast +from typing import Any, Literal, cast import httpx import pytest @@ -25,8 +26,10 @@ TaskState, TaskStatus, ) -from a2a.utils.errors import InvalidRequestError -from langchain_core.messages import AIMessage, ToolMessage +from a2a.utils.errors import InvalidRequestError, TaskNotFoundError +from langchain_core.language_models.fake_chat_models import GenericFakeChatModel +from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage +from langchain_core.utils.function_calling import convert_to_openai_tool from langgraph.graph import END, START, StateGraph from langgraph.types import Command from opentelemetry import trace as otel_trace @@ -40,12 +43,21 @@ from uipath.agent.models.agent import ( AgentConversationalAgentToolProperties, AgentConversationalAgentToolResourceConfig, + AgentConversationalAgentToolSettings, AgentToolType, ) +from uipath.core.chat import ( + UiPathConversationContentPart, + UiPathConversationMessage, + UiPathConversationToolCall, + UiPathConversationToolCallResult, + UiPathInlineValue, +) from uipath.platform.common import GenericResourceOverwrite, ResourceOverwritesContext from uipath.platform.errors import FolderNotFoundException import uipath_langchain.agent.tools.a2a.conversational_agent_tool as conv_tool +from uipath_langchain.agent.advanced.agent import create_advanced_agent from uipath_langchain.agent.react.types import AgentGraphState from uipath_langchain.agent.tools.a2a.a2a_tool import ( A2aClient, @@ -53,6 +65,7 @@ ) from uipath_langchain.agent.tools.a2a.conversational_agent_tool import ( A2aConversationInput, + A2aSingleConversationInput, ConversationalAgentA2aClient, _send_message, create_conversational_agent_tools_and_clients, @@ -63,6 +76,7 @@ wrap_tools_with_error_handling, ) from uipath_langchain.chat.hitl import REQUIRE_CONVERSATIONAL_CONFIRMATION +from uipath_langchain.runtime.messages import UiPathChatMessagesMapper def _create_tools( @@ -161,15 +175,18 @@ def _fake_create(self: ClientFactory, agent_card: AgentCard): def _make_conversational_resource( *, + name: str = "Support Agent", folder_path: str | None = "Shared/Agents", cached_agent_card: dict[str, Any] | None = None, is_enabled: bool = True, process_name: str | None = "My Conversational Agent", require_confirmation: bool = False, + conversation_mode: Literal["single", "multiple"] | None = "multiple", ) -> AgentConversationalAgentToolResourceConfig: + """Build a resource. Most tests exercise the conversation handles of multiple mode.""" return AgentConversationalAgentToolResourceConfig( id="conv-id", - name="Support Agent", + name=name, description="Answers support questions", is_enabled=is_enabled, type=AgentToolType.CONVERSATIONAL_AGENT, @@ -184,6 +201,9 @@ def _make_conversational_resource( cached_agent_card=cached_agent_card, require_conversational_confirmation=require_confirmation, ), + settings=AgentConversationalAgentToolSettings( + conversation_mode=conversation_mode + ), ) @@ -705,15 +725,18 @@ async def test_conversational_cached_v03_card_selects_protocol_0_3( async def _run_through_tool_node( - tools: list[Any], args: dict[str, Any] + tools: list[Any], + args: dict[str, Any], + history: list[BaseMessage] | None = None, ) -> dict[str, Any]: tool_node = wrap_tools_with_error_handling(create_tool_node(tools))[tools[0].name] state = AgentGraphState( messages=[ + *(history or []), AIMessage( content="", tool_calls=[{"name": tools[0].name, "args": args, "id": "call-1"}], - ) + ), ] ) result = await tool_node.ainvoke(state) @@ -831,9 +854,13 @@ def _terminal_task( def _conversation_tool( - monkeypatch: pytest.MonkeyPatch, client: _ScriptedA2aClient + monkeypatch: pytest.MonkeyPatch, + client: _ScriptedA2aClient, + conversation_mode: Literal["single", "multiple"] | None = "multiple", ) -> tuple[A2aStructuredToolWithWrapper, Any]: - tools, clients = _create_tools([_make_conversational_resource()]) + tools, clients = _create_tools( + [_make_conversational_resource(conversation_mode=conversation_mode)] + ) async def _get(): return client @@ -848,9 +875,11 @@ async def _converse( tool: A2aStructuredToolWithWrapper, wrapper: Any, storage: dict[str, Any] | None, + *, + history: list[BaseMessage] | None = None, **args: Any, ) -> tuple[dict[str, Any], dict[str, Any]]: - state = AgentGraphState() + state = AgentGraphState(messages=history or []) if storage is not None: state.inner_state.tools_storage[tool.name] = storage call = {"name": tool.name, "args": {"message": "hi", **args}, "id": "call-1"} @@ -1548,3 +1577,1044 @@ async def test_conversational_failed_request_marks_the_span_error( (span,) = [s for s in exporter.get_finished_spans() if s.name == "Support Agent"] assert span.status.status_code == StatusCode.ERROR assert "conversation_id" not in (span.attributes or {}) + + +# --- Conversation mode: single (the default) and multiple --------------------- + + +_CANCELLED_JSON = json.dumps({"meta": "Cancelled by user"}) + + +def _request_failed_json(cid: str | None) -> str: + return json.dumps( + {"error": "request_failed", "conversation_id": cid, "message": "Error: boom"} + ) + + +def _conversation_closed_json(cid: str) -> str: + return json.dumps( + {"error": "conversation_closed", "conversation_id": cid, "message": "ended"} + ) + + +def _assert_restart_note(note: str) -> None: + assert "new conversation was started" in note + assert "The agent has no memory of earlier messages" in note + assert "send it again with that context" in note + + +def _result_json(cid: str | None, state: str = "input_required") -> str: + return json.dumps( + {"agent_response": "earlier", "conversation_id": cid, "task_state": state} + ) + + +def _turn( + tool_name: str, + content: str, + call_id: str, + *, + status: Literal["success", "error"] = "success", + name: str | None = None, +) -> list[BaseMessage]: + """One replayed turn: an assistant tool call and the tool's result. + + The result carries no ``name`` unless given, as after a replay. + """ + return [ + AIMessage( + content="", + tool_calls=[ + {"name": tool_name, "args": {"message": "earlier"}, "id": call_id} + ], + ), + ToolMessage(content=content, tool_call_id=call_id, status=status, name=name), + ] + + +def _with_history(*turns: list[BaseMessage]) -> list[BaseMessage]: + return [ + HumanMessage(content="hello"), + *[message for turn in turns for message in turn], + HumanMessage(content="and now?"), + ] + + +@pytest.mark.parametrize("mode", [None, "single"]) +def test_single_mode_is_the_default_and_exposes_only_the_message( + mode: Literal["single"] | None, +) -> None: + tools, _ = _create_tools([_make_conversational_resource(conversation_mode=mode)]) + tool = tools[0] + + assert tool.args_schema is A2aSingleConversationInput + assert set(A2aSingleConversationInput.model_fields) == {"message"} + assert set(tool.tool_call_schema.model_fields) == {"message"} + parameters = convert_to_openai_tool(tool)["function"]["parameters"] + assert set(parameters["properties"]) == {"message"} + assert parameters["required"] == ["message"] + assert tool.description.startswith("Answers support questions") + assert "Each call continues one ongoing conversation" in tool.description + assert "Make one call at a time." in tool.description + assert "conversation_id" not in tool.description + assert "new_conversation" not in tool.description + + +def test_multiple_mode_exposes_the_handle_arguments_to_the_model() -> None: + tools, _ = _create_tools( + [_make_conversational_resource(conversation_mode="multiple")] + ) + tool = tools[0] + + assert tool.args_schema is A2aConversationInput + assert set(tool.tool_call_schema.model_fields) == { + "message", + "conversation_id", + "new_conversation", + } + assert "Each call continues one ongoing conversation" not in tool.description + assert "Make one call at a time." not in tool.description + assert "new_conversation=true" in tool.description + + +async def test_single_mode_repeated_calls_continue_one_conversation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient( + [_open_task("c1", "one"), _open_task("c1", "two"), _open_task("c1", "three")] + ) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + + result, entry = await _converse(tool, wrapper, None) + assert result == { + "agent_response": "one", + "conversation_id": "c1", + "task_state": "input_required", + } + result, entry = await _converse(tool, wrapper, entry) + result, entry = await _converse(tool, wrapper, entry) + + assert [sent.message.context_id for sent in client.sent] == ["", "c1", "c1"] + assert [sent.message.task_id for sent in client.sent] == ["", "c1", "c1"] + assert result["conversation_id"] == "c1" + assert "note" not in result + assert entry["current"] == "c1" + assert set(entry["conversations"]) == {"c1"} + + +async def test_single_mode_ignores_handle_arguments_the_model_may_still_send( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + storage = _entry({"c1": ("open", 1)}, current="c1") + + result, entry = await _converse( + tool, wrapper, storage, conversation_id="other", new_conversation=True + ) + + assert result["conversation_id"] == "c1" + assert client.sent[0].message.context_id == "c1" + assert entry["current"] == "c1" + + +async def test_single_mode_terminal_reply_then_next_call_starts_a_new_conversation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_terminal_task("c1"), _open_task("c2", "fresh")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + storage = _entry({"c1": ("open", 1)}, current="c1") + + result, entry = await _converse(tool, wrapper, storage) + + assert result["task_state"] == "failed" + assert result["conversation_id"] == "c1" + assert "note" not in result + assert entry["conversations"]["c1"]["state"] == "closed" + + result, entry = await _converse(tool, wrapper, entry) + + assert "error" not in result + assert result["agent_response"] == "fresh" + assert result["conversation_id"] == "c2" + _assert_restart_note(result["note"]) + assert len(client.sent) == 2 # Nothing was sent into the ended conversation. + assert client.sent[1].message.context_id == "" + assert client.sent[1].message.task_id == "" + assert entry["current"] == "c2" + assert entry["conversations"]["c1"]["state"] == "closed" + + +@pytest.mark.parametrize( + "error", + [ + TaskNotFoundError(), + InvalidRequestError("Cannot send a message to a task in a terminal state."), + ], + ids=["task-not-found", "terminal-state"], +) +async def test_single_mode_unavailable_conversation_restarts_and_resends_the_message( + monkeypatch: pytest.MonkeyPatch, error: Exception +) -> None: + client = _ScriptedA2aClient([error, _open_task("c2", "fresh")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + storage = _entry({"c1": ("open", 1)}, current="c1") + + result, entry = await _converse(tool, wrapper, storage) + + assert "error" not in result + assert result["agent_response"] == "fresh" + assert result["conversation_id"] == "c2" + _assert_restart_note(result["note"]) + assert [sent.message.context_id for sent in client.sent] == ["c1", ""] + assert client.sent[1].message.task_id == "" + assert client.sent[1].message.parts[0].text == "hi" + assert entry["current"] == "c2" + assert entry["conversations"]["c1"]["state"] == "closed" + + +async def test_single_mode_failed_restart_is_a_request_failed_result( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient( + [TaskNotFoundError(), RuntimeError("boom"), _open_task("c3")] + ) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + storage = _entry({"c1": ("open", 1)}, current="c1") + + result, entry = await _converse(tool, wrapper, storage) + + assert result["error"] == "request_failed" + assert result["conversation_id"] is None + assert "boom" in result["message"] + assert entry["conversations"]["c1"]["state"] == "closed" + + result, entry = await _converse(tool, wrapper, entry) + + assert result["conversation_id"] == "c3" + _assert_restart_note(result["note"]) + assert [sent.message.context_id for sent in client.sent] == ["c1", "", ""] + + +async def test_single_mode_other_errors_keep_the_conversation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([RuntimeError("boom"), _open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + storage = _entry({"c1": ("open", 1)}, current="c1") + + result, entry = await _converse(tool, wrapper, storage) + + assert result["error"] == "request_failed" + assert result["conversation_id"] == "c1" + assert entry == storage + assert len(client.sent) == 1 + + result, _ = await _converse(tool, wrapper, entry) + + assert result["conversation_id"] == "c1" + assert client.sent[1].message.context_id == "c1" + + +async def test_single_mode_request_failure_on_the_first_call_has_a_null_id( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([RuntimeError("boom")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + + result, entry = await _converse(tool, wrapper, None) + + assert result["error"] == "request_failed" + assert result["conversation_id"] is None + assert entry == {"current": None, "conversations": {}} + + +async def test_single_mode_replaced_conversation_id_adds_a_note( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c9")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + storage = _entry({"c1": ("open", 1)}, current="c1") + + result, entry = await _converse(tool, wrapper, storage) + + assert result["conversation_id"] == "c9" + _assert_restart_note(result["note"]) + assert entry["current"] == "c9" + + +async def test_single_mode_continues_after_the_tool_node_and_reducer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c1"), _open_task("c1")]) + tools, clients = _create_tools( + [_make_conversational_resource(conversation_mode=None)] + ) + + async def _get(): + return client + + monkeypatch.setattr(clients[0], "get", _get) + node = create_tool_node(tools)[tools[0].name] + builder = StateGraph(AgentGraphState) + builder.add_node("first", node) + builder.add_node("second", node) + builder.add_edge(START, "first") + builder.add_edge("first", "second") + builder.add_edge("second", END) + + final = await builder.compile().ainvoke( + AgentGraphState( + messages=[ + AIMessage( + content="", + tool_calls=[ + { + "name": tools[0].name, + "args": {"message": "a"}, + "id": "call-1", + }, + { + "name": tools[0].name, + "args": {"message": "b"}, + "id": "call-2", + }, + ], + ) + ] + ) + ) + + stored = final["inner_state"].tools_storage[tools[0].name] + assert stored["current"] == "c1" + assert set(stored["conversations"]) == {"c1"} + assert [sent.message.context_id for sent in client.sent] == ["", "c1"] + + +@pytest.mark.parametrize("protocol_version", ["1.0", "0.3"]) +async def test_single_mode_missing_task_starts_a_new_conversation_over_the_wire( + monkeypatch: pytest.MonkeyPatch, protocol_version: str +) -> None: + requests = _patch_jsonrpc_script( + monkeypatch, + [ + _wire_open_task(protocol_version, "c1"), + {"error": {"code": -32001, "message": "Task not found"}}, + _wire_open_task(protocol_version, "c2"), + ], + ) + resource = _make_conversational_resource( + conversation_mode=None, + cached_agent_card=_cached_card() if protocol_version == "0.3" else None, + ) + async with open_conversational_agent_tools( + [resource], is_conversational=False + ) as tools: + tool = cast(A2aStructuredToolWithWrapper, tools[0]) + _, entry = await _converse(tool, tool.awrapper, None) + result, entry = await _converse(tool, tool.awrapper, entry) + + assert "error" not in result + assert result["conversation_id"] == "c2" + _assert_restart_note(result["note"]) + assert len(requests) == 3 + resent = json.loads(requests[-1].content)["params"]["message"] + assert not resent.get("taskId") + assert not resent.get("contextId") + assert entry["current"] == "c2" + + +async def test_single_mode_direct_invocation_outside_a_graph_has_no_history( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("run1"), _open_task("run2")]) + tool, _ = _conversation_tool(monkeypatch, client, None) + + first = json.loads(await tool.ainvoke({"message": "a"})) + second = json.loads(await tool.ainvoke({"message": "b"})) + + assert (first["conversation_id"], second["conversation_id"]) == ("run1", "run2") + assert [sent.message.context_id for sent in client.sent] == ["", ""] + + +# --- Cross-message continuity: rebuilt from the replayed history -------------- + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_fresh_job_continues_the_conversation_from_replayed_history( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c1", "again")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + result, entry = await _converse(tool, wrapper, None, history=history) + + assert result["agent_response"] == "again" + assert result["conversation_id"] == "c1" + assert "note" not in result + assert client.sent[0].message.context_id == "c1" + assert entry["current"] == "c1" + assert entry["conversations"]["c1"]["state"] == "open" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_of_a_different_tool_is_ignored( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history(_turn("Other_Agent", _result_json("c9"), "earlier-1")) + + result, entry = await _converse(tool, wrapper, None, history=history) + + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "" + assert set(entry["conversations"]) == {"c2"} + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +@pytest.mark.parametrize( + ("content", "status"), + [ + (_conversation_closed_json("c1"), "success"), + (_result_json("c1"), "error"), + ("Invalid input for tool", "error"), + (_CANCELLED_JSON, "success"), + ("", "error"), + ("not json at all", "success"), + (json.dumps(["conversation_id", "c1"]), "success"), + (json.dumps({"agent_response": "no id"}), "success"), + (_request_failed_json(None), "success"), + ], + ids=[ + "conversation-closed", + "error-status", + "plain-error", + "cancelled-confirmation", + "interrupted", + "not-json", + "not-an-object", + "no-conversation-id", + "failed-on-a-new-conversation", + ], +) +async def test_unusable_result_in_history_is_ignored( + monkeypatch: pytest.MonkeyPatch, + mode: Literal["multiple"] | None, + content: str, + status: Literal["success", "error"], +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history(_turn(tool.name, content, "earlier-1", status=status)) + + result, entry = await _converse(tool, wrapper, None, history=history) + + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "" + assert set(entry["conversations"]) == {"c2"} + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_uses_the_last_result_of_this_tool( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history( + _turn(tool.name, _result_json("c1"), "earlier-1"), + _turn(tool.name, _result_json("c2"), "earlier-2"), + _turn("Other_Agent", _result_json("c9"), "earlier-3"), + ) + + result, _ = await _converse(tool, wrapper, None, history=history) + + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "c2" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_matches_each_result_to_its_nearest_tool_call( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + """Some providers reuse tool call ids from one turn to the next.""" + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history( + _turn(tool.name, _result_json("c1"), "call_1"), + _turn("Other_Agent", _result_json("c9"), "call_1"), + ) + + result, _ = await _converse(tool, wrapper, None, history=history) + + assert result["conversation_id"] == "c1" + assert client.sent[0].message.context_id == "c1" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_falls_back_to_the_result_name_without_a_matching_call( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = [ + HumanMessage(content="hello"), + ToolMessage(content=_result_json("c1"), tool_call_id="orphan", name=tool.name), + HumanMessage(content="and now?"), + ] + + result, _ = await _converse(tool, wrapper, None, history=history) + + assert client.sent[0].message.context_id == "c1" + assert result["conversation_id"] == "c1" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_reads_a_result_wrapped_by_the_confirmation_envelope( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + envelope = json.dumps( + { + "meta": {"message": "User has modified the tool arguments"}, + "result": json.loads(_result_json("c1")), + } + ) + history = _with_history(_turn(tool.name, envelope, "earlier-1")) + + result, _ = await _converse(tool, wrapper, None, history=history) + + assert client.sent[0].message.context_id == "c1" + assert result["conversation_id"] == "c1" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_reads_a_result_replayed_as_content_blocks( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history(_turn(tool.name, "", "earlier-1")) + history[2].content = [{"type": "text", "text": _result_json("c1")}] + + await _converse(tool, wrapper, None, history=history) + + assert client.sent[0].message.context_id == "c1" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_current_conversation_in_the_state_beats_the_history( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + storage = _entry({"c2": ("open", 1)}, current="c2") + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + result, entry = await _converse(tool, wrapper, storage, history=history) + + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "c2" + assert set(entry["conversations"]) == {"c2"} + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_does_not_reopen_a_conversation_the_state_knows_is_closed( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + storage = _entry({"c1": ("closed", 1)}, current=None) + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + result, _ = await _converse(tool, wrapper, storage, history=history) + + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "" + + +async def test_multiple_mode_continues_across_jobs_unless_told_otherwise( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c1"), _open_task("c5"), _open_task("c7")]) + tool, wrapper = _conversation_tool(monkeypatch, client, "multiple") + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + result, _ = await _converse(tool, wrapper, None, history=history) + assert result["conversation_id"] == "c1" + assert client.sent[0].message.context_id == "c1" + + result, _ = await _converse( + tool, wrapper, None, history=history, new_conversation=True + ) + assert result["conversation_id"] == "c5" + assert client.sent[1].message.context_id == "" + + result, _ = await _converse( + tool, wrapper, None, history=history, conversation_id="c7" + ) + assert client.sent[2].message.context_id == "c7" + + +async def test_multiple_mode_still_reports_a_closed_conversation_to_the_model( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient( + [InvalidRequestError("Cannot send a message to a task in a terminal state.")] + ) + tool, wrapper = _conversation_tool(monkeypatch, client, "multiple") + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + result, entry = await _converse(tool, wrapper, None, history=history) + + assert result["error"] == "conversation_closed" + assert result["conversation_id"] == "c1" + assert "new_conversation=true" in result["message"] + assert entry["conversations"]["c1"]["state"] == "closed" + assert entry["current"] is None + assert len(client.sent) == 1 + + +async def test_single_mode_through_the_tool_node_continues_from_the_state_messages( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tools, clients = _create_tools( + [_make_conversational_resource(conversation_mode=None)] + ) + + async def _get(): + return client + + monkeypatch.setattr(clients[0], "get", _get) + history = _with_history(_turn(tools[0].name, _result_json("c1"), "earlier-1")) + + content = await _run_through_tool_node(tools, {"message": "next"}, history) + + assert content["conversation_id"] == "c1" + assert client.sent[0].message.context_id == "c1" + + +# --- Advanced (deepagents) mode: the history reaches the tool as injected state --- + + +class _AdvancedModel(GenericFakeChatModel): + model_name: str = "test-model" + + def _get_ls_params(self, stop: list[str] | None = None, **kwargs: Any) -> Any: + return {"ls_provider": "openai", "ls_model_name": self.model_name} + + def bind_tools(self, tools: Any, **kwargs: Any) -> "_AdvancedModel": + return self + + +def _tool_call(name: str, args: dict[str, Any], call_id: str) -> AIMessage: + return AIMessage( + content="", tool_calls=[{"name": name, "args": args, "id": call_id}] + ) + + +async def _run_advanced( + tool: Any, script: list[AIMessage], messages: list[BaseMessage] +) -> list[BaseMessage]: + """Run a real deepagents graph, which hands the tool its state.""" + model = _AdvancedModel(messages=iter([*script, *[AIMessage(content="done")] * 10])) + graph = create_advanced_agent(model=model, tools=[tool]) + result = await graph.ainvoke({"messages": messages}) + return cast(list[BaseMessage], result["messages"]) + + +def _tool_results(messages: list[BaseMessage]) -> list[dict[str, Any]]: + return [ + json.loads(str(message.content)) + for message in messages + if isinstance(message, ToolMessage) + ] + + +async def test_advanced_single_mode_continues_one_conversation_within_a_run( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c1", "one"), _open_task("c1", "two")]) + tool, _ = _conversation_tool(monkeypatch, client, None) + + messages = await _run_advanced( + tool, + [ + _tool_call(tool.name, {"message": "a"}, "call-1"), + _tool_call(tool.name, {"message": "b"}, "call-2"), + ], + [HumanMessage(content="hi")], + ) + + assert [sent.message.context_id for sent in client.sent] == ["", "c1"] + assert [r["conversation_id"] for r in _tool_results(messages)] == ["c1", "c1"] + + +async def test_advanced_single_mode_continues_from_the_replayed_history( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c1", "again")]) + tool, _ = _conversation_tool(monkeypatch, client, None) + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + messages = await _run_advanced( + tool, [_tool_call(tool.name, {"message": "next"}, "call-1")], history + ) + + assert client.sent[0].message.context_id == "c1" + assert _tool_results(messages)[-1]["agent_response"] == "again" + + +async def test_advanced_single_mode_restarts_an_ended_conversation_with_a_note( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient( + [ + InvalidRequestError("Cannot send a message to a task in a terminal state."), + _open_task("c2", "fresh"), + ] + ) + tool, _ = _conversation_tool(monkeypatch, client, None) + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + messages = await _run_advanced( + tool, [_tool_call(tool.name, {"message": "next"}, "call-1")], history + ) + + result = _tool_results(messages)[-1] + assert "error" not in result + assert result["conversation_id"] == "c2" + _assert_restart_note(result["note"]) + assert [sent.message.context_id for sent in client.sent] == ["c1", ""] + + +async def test_advanced_single_mode_ignores_the_history_of_another_tool( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, _ = _conversation_tool(monkeypatch, client, None) + history = _with_history(_turn("Other_Agent", _result_json("c9"), "earlier-1")) + + await _run_advanced( + tool, [_tool_call(tool.name, {"message": "next"}, "call-1")], history + ) + + assert client.sent[0].message.context_id == "" + + +async def test_advanced_multiple_mode_reads_the_history_and_honors_the_handles( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _ScriptedA2aClient([_open_task("c1"), _open_task("c5"), _open_task("c7")]) + tool, _ = _conversation_tool(monkeypatch, client, "multiple") + history = _with_history(_turn(tool.name, _result_json("c1"), "earlier-1")) + + await _run_advanced( + tool, + [ + _tool_call(tool.name, {"message": "a"}, "call-1"), + _tool_call(tool.name, {"message": "b", "new_conversation": True}, "call-2"), + _tool_call(tool.name, {"message": "c", "conversation_id": "c7"}, "call-3"), + ], + history, + ) + + assert [sent.message.context_id for sent in client.sent] == ["c1", "", "c7"] + + +# --- Which earlier result the history rebuild trusts --------------------------- + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +@pytest.mark.parametrize( + ("content", "status"), + [ + (_request_failed_json(None), "success"), + ("Invalid input for tool", "error"), + (_CANCELLED_JSON, "success"), + ("", "error"), + (_conversation_closed_json("c2"), "success"), + ], + ids=[ + "failed-on-a-new-conversation", + "tool-error", + "cancelled-confirmation", + "interrupted", + "closed-other-conversation", + ], +) +async def test_history_skips_an_unusable_last_result_and_continues_the_good_one( + monkeypatch: pytest.MonkeyPatch, + mode: Literal["multiple"] | None, + content: str, + status: Literal["success", "error"], +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history( + _turn(tool.name, _result_json("c1"), "earlier-1"), + _turn(tool.name, content, "earlier-2", status=status), + ) + + result, _ = await _converse(tool, wrapper, None, history=history) + + assert result["conversation_id"] == "c1" + assert client.sent[0].message.context_id == "c1" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_continues_the_conversation_of_a_failed_request( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + """A request that failed says nothing about the conversation, which lives on.""" + client = _ScriptedA2aClient([_open_task("c1"), _open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + + history = _with_history(_turn(tool.name, _request_failed_json("c1"), "earlier-1")) + result, _ = await _converse(tool, wrapper, None, history=history) + assert result["conversation_id"] == "c1" + assert client.sent[0].message.context_id == "c1" + + history = _with_history( + _turn(tool.name, _result_json("c1"), "earlier-1"), + _turn(tool.name, _request_failed_json("c2"), "earlier-2"), + ) + result, _ = await _converse(tool, wrapper, None, history=history) + assert client.sent[1].message.context_id == "c2" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +@pytest.mark.parametrize("state", ["completed", "failed", "canceled", "rejected"]) +async def test_history_does_not_reopen_a_conversation_whose_task_has_ended( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None, state: str +) -> None: + client = _ScriptedA2aClient([_open_task("c2"), _open_task("c3")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + + ended_only = _with_history(_turn(tool.name, _result_json("c1", state), "earlier-1")) + result, _ = await _converse(tool, wrapper, None, history=ended_only) + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "" + + # The earlier good result of the same conversation does not bring it back. + ended_after_good = _with_history( + _turn(tool.name, _result_json("c1"), "earlier-1"), + _turn(tool.name, _result_json("c1", state), "earlier-2"), + ) + result, _ = await _converse(tool, wrapper, None, history=ended_after_good) + assert result["conversation_id"] == "c3" + assert client.sent[1].message.context_id == "" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_does_not_reopen_a_conversation_the_agent_reported_closed( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history( + _turn(tool.name, _result_json("c1"), "earlier-1"), + _turn(tool.name, _conversation_closed_json("c1"), "earlier-2"), + ) + + result, _ = await _converse(tool, wrapper, None, history=history) + + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "" + + +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_falls_back_to_an_older_conversation_that_is_still_open( + monkeypatch: pytest.MonkeyPatch, mode: Literal["multiple"] | None +) -> None: + client = _ScriptedA2aClient([_open_task("c2")]) + tool, wrapper = _conversation_tool(monkeypatch, client, mode) + history = _with_history( + _turn(tool.name, _result_json("c2"), "earlier-1"), + _turn(tool.name, _result_json("c1"), "earlier-2"), + _turn(tool.name, _result_json("c1", "failed"), "earlier-3"), + ) + + result, _ = await _converse(tool, wrapper, None, history=history) + + assert result["conversation_id"] == "c2" + assert client.sent[0].message.context_id == "c2" + + +# --- Handle arguments the model sends in single mode --------------------------- + + +async def test_single_mode_logs_the_handle_arguments_it_ignores( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + storage = _entry({"c1": ("open", 1)}, current="c1") + + with caplog.at_level(logging.INFO, logger=conv_tool.logger.name): + await _converse( + tool, wrapper, storage, conversation_id="other", new_conversation=True + ) + + (record,) = [r for r in caplog.records if "single" in r.getMessage()] + assert record.levelno == logging.INFO + assert "conversation_id, new_conversation" in record.getMessage() + assert tool.name in record.getMessage() + + +async def test_single_mode_does_not_log_when_only_the_message_is_sent( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, None) + + with caplog.at_level(logging.INFO, logger=conv_tool.logger.name): + await _converse(tool, wrapper, None) + + assert not [r for r in caplog.records if "single" in r.getMessage()] + + +async def test_multiple_mode_does_not_log_its_handle_arguments( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + client = _ScriptedA2aClient([_open_task("c1")]) + tool, wrapper = _conversation_tool(monkeypatch, client, "multiple") + + with caplog.at_level(logging.INFO, logger=conv_tool.logger.name): + await _converse(tool, wrapper, None, conversation_id="c1") + + assert not [r for r in caplog.records if "single" in r.getMessage()] + + +# --- The history as the chat runtime really stores and replays it -------------- + + +class _MemoryStorage: + """The part of the runtime storage that the message mapper uses.""" + + def __init__(self) -> None: + self.values: dict[tuple[str, str, str], Any] = {} + + async def get_value(self, runtime_id: str, namespace: str, key: str) -> Any: + return self.values.get((runtime_id, namespace, key)) + + async def set_value( + self, runtime_id: str, namespace: str, key: str, value: Any + ) -> None: + self.values[(runtime_id, namespace, key)] = value + + +_CAS_TIME = "2025-01-15T10:30:00Z" + + +def _cas_message( + message_id: str, + role: str, + text: str = "", + tool_calls: list[UiPathConversationToolCall] | None = None, +) -> UiPathConversationMessage: + return UiPathConversationMessage( + message_id=message_id, + role=role, + created_at=_CAS_TIME, + updated_at=_CAS_TIME, + content_parts=[ + UiPathConversationContentPart( + content_part_id=f"{message_id}-part", + mime_type="text/plain", + data=UiPathInlineValue(inline=text), + citations=[], + created_at=_CAS_TIME, + updated_at=_CAS_TIME, + ) + ] + if text + else [], + tool_calls=tool_calls or [], + interrupts=[], + ) + + +@pytest.mark.parametrize( + "resource_name", + ["Support Agent", "Support (EU) / Agent v2!", " Support Agent "], + ids=["spaces", "special-characters", "padded"], +) +@pytest.mark.parametrize("mode", [None, "multiple"]) +async def test_history_survives_the_chat_message_mapper_round_trip( + monkeypatch: pytest.MonkeyPatch, + resource_name: str, + mode: Literal["multiple"] | None, +) -> None: + client = _ScriptedA2aClient([_open_task("c1", "first"), _open_task("c1", "second")]) + tools, clients = _create_tools( + [_make_conversational_resource(name=resource_name, conversation_mode=mode)] + ) + + async def _get(): + return client + + monkeypatch.setattr(clients[0], "get", _get) + tool = cast(A2aStructuredToolWithWrapper, tools[0]) + assert tool.awrapper is not None + + # Job 1: the model calls the tool, and the runtime turns both messages into + # the events that the chat service stores. + ai_message = AIMessage( + id="msg-1", + content="", + tool_calls=[{"name": tool.name, "args": {"message": "hi"}, "id": "call-1"}], + ) + wrapper: Any = tool.awrapper + command = await wrapper( + tool, ai_message.tool_calls[0], AgentGraphState(messages=[ai_message]) + ) + tool_message = command.update["messages"][0] + mapper = UiPathChatMessagesMapper("runtime-1", _MemoryStorage()) + start_events = await mapper.map_event(ai_message) + end_events = await mapper.map_event(tool_message) + start = next( + e.tool_call.start + for e in start_events or [] + if e.tool_call and e.tool_call.start + ) + end = next( + e.tool_call.end for e in end_events or [] if e.tool_call and e.tool_call.end + ) + + # Job 2: the chat service replays what it stored. + replayed = UiPathChatMessagesMapper("runtime-2", None).map_messages( + [ + _cas_message("user-1", "user", "hello"), + _cas_message( + "msg-1", + "assistant", + tool_calls=[ + UiPathConversationToolCall( + tool_call_id="call-1", + name=start.tool_name, + input={"message": "hi"}, + result=UiPathConversationToolCallResult( + output=end.output, is_error=bool(end.is_error) + ), + timestamp=_CAS_TIME, + created_at=_CAS_TIME, + updated_at=_CAS_TIME, + ) + ], + ), + _cas_message("user-2", "user", "and now?"), + ] + ) + + assert replayed[1].tool_calls[0]["name"] == tool.name + assert replayed[2].name is None + assert json.loads(replayed[2].content) == json.loads(tool_message.content) + + result, _ = await _converse(tool, tool.awrapper, None, history=replayed) + + assert result["conversation_id"] == "c1" + assert client.sent[1].message.context_id == "c1" diff --git a/uv.lock b/uv.lock index 6338805c9..1efc6862d 100644 --- a/uv.lock +++ b/uv.lock @@ -4806,7 +4806,7 @@ wheels = [ [[package]] name = "uipath" -version = "2.14.40" +version = "2.14.44" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "applicationinsights" }, @@ -4831,9 +4831,9 @@ dependencies = [ { name = "uipath-platform" }, { name = "uipath-runtime" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/99/d5/6d2351bfbba2249f26f99c45fae71f2385116ead0d4a62e993f9855efc9d/uipath-2.14.40.tar.gz", hash = "sha256:7c553a24298024b0f2ff55e26c79a05686be1fbd986f9bc2b8898af2fdcc9120", size = 4598398, upload-time = "2026-10-15T14:51:48.697Z" } +sdist = { url = "https://files.pythonhosted.org/packages/47/bb/8ecf728db4af71bc56c1acd82139093993e422f4ae7603eb9636953d8193/uipath-2.14.44.tar.gz", hash = "sha256:fc8d7e5d80def0f2b8427729dad94ce711ecfd080f68ab8b8c96b5f65c3a49a6", size = 4600379, upload-time = "2026-10-26T12:02:45.185Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/0c/4f/07f1b71bdd2d09b1ac3aa7944e3c6587f78df2dfa67efda7a5d45ae04d5c/uipath-2.14.40-py3-none-any.whl", hash = "sha256:0c8361e381f1bdba3a92d502cc86f3a2a12e964d303d45a1f654f4216e01a5c1", size = 455968, upload-time = "2026-10-15T14:51:46.925Z" }, + { url = "https://files.pythonhosted.org/packages/07/be/0749ab94b39049ac3fb8dbcdbcbda749b5cb86b3953c605800bf85821227/uipath-2.14.44-py3-none-any.whl", hash = "sha256:68aeaf0012cd8afc96c736ffede3f039d65bab929fb9ca9b61ed8dd4e567ba96", size = 456647, upload-time = "2026-10-26T12:02:42.583Z" }, ] [[package]] @@ -4861,7 +4861,7 @@ wheels = [ [[package]] name = "uipath-langchain" -version = "0.18.51" +version = "0.18.52" source = { editable = "." } dependencies = [ { name = "a2a-sdk" }, @@ -4960,7 +4960,7 @@ requires-dist = [ { name = "pydantic-settings", specifier = ">=2.6.0" }, { name = "python-dotenv", specifier = ">=1.2.2,<2.0.0" }, { name = "rdflib", specifier = ">=7.0.0,<8.0.0" }, - { name = "uipath", specifier = ">=2.14.40,<2.15.0" }, + { name = "uipath", specifier = ">=2.14.44,<2.15.0" }, { name = "uipath-core", specifier = ">=0.5.33,<0.6.0" }, { name = "uipath-langchain", extras = ["code-interpreter"], marker = "extra == 'all'" }, { name = "uipath-langchain", extras = ["video"], marker = "extra == 'all'" },