# SPDX-FileCopyrightText: 2022-present deepset GmbH # # SPDX-License-Identifier: Apache-2.0 import asyncio from typing import Any import pytest from haystack.components.agents.state.state import State from haystack.components.agents.tool_calling import ToolNotFoundException, _run_tool from haystack.dataclasses import ChatMessage, ToolCall from haystack.human_in_the_loop import ( AlwaysAskPolicy, AskOncePolicy, BlockingConfirmationStrategy, ConfirmationUIResult, NeverAskPolicy, SimpleConsoleUI, ToolExecutionDecision, ) from haystack.human_in_the_loop.strategies import ( _apply_tool_execution_decisions, _process_confirmation_strategies, _run_confirmation_strategies, _run_confirmation_strategies_async, _update_chat_history, ) from haystack.tools import Tool, create_tool_from_function def addition_tool(a: int, b: int) -> int: return a + b def multiplication_tool(a: int, b: int) -> int: return a * b @pytest.fixture def tools() -> list[Tool]: add_tool = create_tool_from_function( function=addition_tool, name="addition_tool", description="A tool that adds two integers together." ) mult_tool = create_tool_from_function( function=multiplication_tool, name="multiplication_tool", description="A tool that multiplies two integers." ) return [add_tool, mult_tool] class TestBlockingConfirmationStrategy: def test_initialization(self): strategy = BlockingConfirmationStrategy(confirmation_policy=AskOncePolicy(), confirmation_ui=SimpleConsoleUI()) assert isinstance(strategy.confirmation_policy, AskOncePolicy) assert isinstance(strategy.confirmation_ui, SimpleConsoleUI) def test_to_dict(self): strategy = BlockingConfirmationStrategy(confirmation_policy=AskOncePolicy(), confirmation_ui=SimpleConsoleUI()) strategy_dict = strategy.to_dict() assert strategy_dict == { "type": "haystack.human_in_the_loop.strategies.BlockingConfirmationStrategy", "init_parameters": { "confirmation_policy": { "type": "haystack.human_in_the_loop.policies.AskOncePolicy", "init_parameters": {}, }, "confirmation_ui": { "type": "haystack.human_in_the_loop.user_interfaces.SimpleConsoleUI", "init_parameters": {}, }, "reject_template": "Tool execution for '{tool_name}' was rejected by the user.", "modify_template": "The parameters for tool '{tool_name}' were updated by the user to:" "\n{final_tool_params}", "user_feedback_template": "With user feedback: {feedback}", }, } def test_from_dict(self): strategy_dict = { "type": "haystack.human_in_the_loop.strategies.BlockingConfirmationStrategy", "init_parameters": { "confirmation_policy": { "type": "haystack.human_in_the_loop.policies.AskOncePolicy", "init_parameters": {}, }, "confirmation_ui": { "type": "haystack.human_in_the_loop.user_interfaces.SimpleConsoleUI", "init_parameters": {}, }, }, } strategy = BlockingConfirmationStrategy.from_dict(strategy_dict) assert isinstance(strategy, BlockingConfirmationStrategy) assert isinstance(strategy.confirmation_policy, AskOncePolicy) assert isinstance(strategy.confirmation_ui, SimpleConsoleUI) def test_run_confirm(self, monkeypatch): strategy = BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ) # Mock the UI to always confirm def mock_get_user_confirmation(tool_name, tool_description, tool_params): return ConfirmationUIResult(action="confirm") monkeypatch.setattr(strategy.confirmation_ui, "get_user_confirmation", mock_get_user_confirmation) decision = strategy.run(tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"}) assert decision.tool_name == "test_tool" assert decision.execute is True assert decision.final_tool_params == {"param1": "value1"} def test_run_modify(self, monkeypatch): strategy = BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ) # Mock the UI to always modify def mock_get_user_confirmation(tool_name, tool_description, tool_params): return ConfirmationUIResult(action="modify", new_tool_params={"param1": "new_value"}) monkeypatch.setattr(strategy.confirmation_ui, "get_user_confirmation", mock_get_user_confirmation) decision = strategy.run(tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"}) assert decision.tool_name == "test_tool" assert decision.execute is True assert decision.final_tool_params == {"param1": "new_value"} assert decision.feedback == ( "The parameters for tool 'test_tool' were updated by the user to:\n{'param1': 'new_value'}" ) def test_run_reject(self, monkeypatch): strategy = BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ) # Mock the UI to always reject def mock_get_user_confirmation(tool_name, tool_description, tool_params): return ConfirmationUIResult(action="reject", feedback="Not needed") monkeypatch.setattr(strategy.confirmation_ui, "get_user_confirmation", mock_get_user_confirmation) decision = strategy.run(tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"}) assert decision.tool_name == "test_tool" assert decision.execute is False assert decision.final_tool_params is None assert decision.feedback == ( "Tool execution for 'test_tool' was rejected by the user. With user feedback: Not needed" ) class TestRunConfirmationStrategies: def test_run_confirmation_strategies_no_strategy(self, tools): teds = _run_confirmation_strategies( confirmation_strategies={}, messages_with_tool_calls=[ ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"param1": "value1"})]) ], tools=tools, ) assert teds == [ ToolExecutionDecision(tool_name=tools[0].name, execute=True, final_tool_params={"param1": "value1"}) ] def test_run_confirmation_strategies_with_strategy(self, tools): teds = _run_confirmation_strategies( confirmation_strategies={ tools[0].name: BlockingConfirmationStrategy( confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI() ) }, messages_with_tool_calls=[ ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"param1": "value1"})]) ], tools=tools, ) assert teds == [ ToolExecutionDecision(tool_name=tools[0].name, execute=True, final_tool_params={"param1": "value1"}) ] def test_run_confirmation_strategies_with_multi_tool_tuple_key(self, tools): add_tool, mult_tool = tools[0], tools[1] strategy = BlockingConfirmationStrategy(confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI()) teds = _run_confirmation_strategies( confirmation_strategies={(add_tool.name, mult_tool.name): strategy}, messages_with_tool_calls=[ ChatMessage.from_assistant( tool_calls=[ToolCall(add_tool.name, {"a": 1, "b": 2}), ToolCall(mult_tool.name, {"a": 3, "b": 4})] ) ], tools=tools, ) assert len(teds) == 2 assert teds[0].tool_name == add_tool.name assert teds[0].execute is True assert teds[1].tool_name == mult_tool.name assert teds[1].execute is True def test_run_confirmation_strategies_unknown_tool_passes_through(self, tools): # The model hallucinated a tool name that isn't in the available tools. Confirmation is skipped and the # call passes through unchanged so the tool-calling code can report it (ToolNotFoundException). teds = _run_confirmation_strategies( confirmation_strategies={ tools[0].name: BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ) }, messages_with_tool_calls=[ ChatMessage.from_assistant( tool_calls=[ToolCall(id="tc-1", tool_name="hallucinated_tool", arguments={"a": 1})] ) ], tools=tools, ) assert teds == [ ToolExecutionDecision( tool_call_id="tc-1", tool_name="hallucinated_tool", execute=True, final_tool_params={"a": 1} ) ] class TestUnknownToolEndToEnd: def test_hallucinated_tool_survives_confirmation_and_is_reported_as_not_found(self, tools): # A confirmation strategy is configured, and the model emits a tool call for a tool that doesn't exist. # The hallucinated call must pass through confirmation unchanged and then be reported by the tool-calling # code as not found, exactly as it would be without any confirmation strategy configured. state = State(schema={"messages": {"type": list[ChatMessage]}}) tool_call = ToolCall(id="tc-1", tool_name="hallucinated_tool", arguments={"a": 1}) assistant_message = ChatMessage.from_assistant(tool_calls=[tool_call]) state.set("messages", [ChatMessage.from_user("do something"), assistant_message]) new_chat_history = _process_confirmation_strategies( confirmation_strategies={ tools[0].name: BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ) }, messages_with_tool_calls=[assistant_message], tools=tools, state=state, ) # The hallucinated call survives the confirmation layer unchanged (it is not dropped), as the pending call # on the last message of the returned history. pending_messages = [new_chat_history[-1]] surviving_calls = [tc for message in pending_messages for tc in (message.tool_calls or [])] assert surviving_calls == [tool_call] # raise_on_failure=True: the tool-calling code raises ToolNotFoundException. with pytest.raises(ToolNotFoundException): _run_tool(messages=pending_messages, state=state, tools=tools) # raise_on_failure=False: it returns an error tool message instead of crashing. tool_messages, _ = _run_tool(messages=pending_messages, state=state, tools=tools, raise_on_failure=False) assert tool_messages[0].tool_call_results[0].error assert "not found" in tool_messages[0].tool_call_results[0].result class TestApplyToolExecutionDecisions: @pytest.fixture def assistant_message(self, tools): tool_call = ToolCall(tool_name=tools[0].name, arguments={"a": 1, "b": 2}, id="1") return ChatMessage.from_assistant(tool_calls=[tool_call]) def test_reject(self, tools, assistant_message): rejection_messages, new_tool_call_messages = _apply_tool_execution_decisions( tool_call_messages=[assistant_message], tool_execution_decisions=[ ToolExecutionDecision( tool_name=tools[0].name, execute=False, tool_call_id="1", feedback=( "The tool execution for 'addition_tool' was rejected by the user. With feedback: Not needed" ), ) ], ) assert rejection_messages == [ assistant_message, ChatMessage.from_tool( tool_result=( "The tool execution for 'addition_tool' was rejected by the user. With feedback: Not needed" ), origin=assistant_message.tool_call, error=True, ), ] assert new_tool_call_messages == [] def test_confirm(self, tools, assistant_message): rejection_messages, new_tool_call_messages = _apply_tool_execution_decisions( tool_call_messages=[assistant_message], tool_execution_decisions=[ ToolExecutionDecision( tool_name=tools[0].name, execute=True, tool_call_id="1", final_tool_params={"a": 1, "b": 2} ) ], ) assert rejection_messages == [] assert new_tool_call_messages == [assistant_message] def test_modify(self, tools, assistant_message): rejection_messages, new_tool_call_messages = _apply_tool_execution_decisions( tool_call_messages=[assistant_message], tool_execution_decisions=[ ToolExecutionDecision( tool_name=tools[0].name, execute=True, tool_call_id="1", final_tool_params={"a": 5, "b": 6}, feedback="The parameters for tool 'addition_tool' were updated by the user to:\n{'a': 5, 'b': 6}", ) ], ) assert rejection_messages == [] assert new_tool_call_messages == [ ChatMessage.from_user( "The parameters for tool 'addition_tool' were updated by the user to:\n{'a': 5, 'b': 6}" ), ChatMessage.from_assistant( tool_calls=[ToolCall(tool_name=tools[0].name, arguments={"a": 5, "b": 6}, id="1")] ), ] def test_two_teds_same_name_no_ids(self): message_with_tool_calls = ChatMessage.from_assistant( text="I'll extract the information about the people mentioned in the context.", # Same tool name with different params but missing IDs tool_calls=[ ToolCall(tool_name="add_database_tool", arguments={"name": "Malte"}), ToolCall(tool_name="add_database_tool", arguments={"name": "Milos"}), ], ) # This raises a ValueError because tool_call_id is missing and there are multiple tool calls with the same name # so we cannot disambiguate which TED applies to which tool call. with pytest.raises( ValueError, match="ToolExecutionDecisions are missing tool_call_id fields and there are multiple tool calls with the " "same name", ): _apply_tool_execution_decisions( tool_call_messages=[message_with_tool_calls], tool_execution_decisions=[ ToolExecutionDecision( tool_name="add_database_tool", execute=True, final_tool_params={"name": "Malte"} ), ToolExecutionDecision( tool_name="add_database_tool", execute=True, final_tool_params={"name": "Milos"} ), ], ) class TestUpdateChatHistory: @pytest.fixture def chat_history_one_tool_call(self): return [ ChatMessage.from_user("Hello"), ChatMessage.from_assistant(tool_calls=[ToolCall("tool1", {"a": 1, "b": 2}, id="1")]), ] def test_update_chat_history_rejection(self, chat_history_one_tool_call): """Test that the new history includes a tool call result message after a rejection.""" rejection_messages = [ ChatMessage.from_assistant(tool_calls=[chat_history_one_tool_call[1].tool_call]), ChatMessage.from_tool( tool_result="The tool execution for 'tool1' was rejected by the user. With feedback: Not needed", origin=chat_history_one_tool_call[1].tool_call, error=True, ), ] updated_messages = _update_chat_history( chat_history_one_tool_call, rejection_messages=rejection_messages, tool_call_and_explanation_messages=[] ) assert updated_messages == [chat_history_one_tool_call[0], *rejection_messages] def test_update_chat_history_confirm(self, chat_history_one_tool_call): """No changes should be made if the tool call was confirmed.""" tool_call_messages = [ChatMessage.from_assistant(tool_calls=[chat_history_one_tool_call[1].tool_call])] updated_messages = _update_chat_history( chat_history_one_tool_call, rejection_messages=[], tool_call_and_explanation_messages=tool_call_messages ) assert updated_messages == chat_history_one_tool_call def test_update_chat_history_modify(self, chat_history_one_tool_call): """Test that the new history includes a user message and updated tool call after a modification.""" tool_call_messages = [ ChatMessage.from_user( "The parameters for tool 'tool1' were updated by the user to:\n{'param': 'new_value'}" ), ChatMessage.from_assistant(tool_calls=[ToolCall("tool1", {"param": "new_value"}, id="1")]), ] updated_messages = _update_chat_history( chat_history_one_tool_call, rejection_messages=[], tool_call_and_explanation_messages=tool_call_messages ) assert updated_messages == [chat_history_one_tool_call[0], *tool_call_messages] def test_update_chat_history_modify_two_tool_calls(self): tool_call_message = ChatMessage.from_assistant( tool_calls=[ToolCall("tool1", {"a": 1, "b": 2}, id="1"), ToolCall("tool2", {"a": 3, "b": 4}, id="2")] ) chat_history = [ChatMessage.from_user("What is 1 + 2? and 3 + 4?"), tool_call_message] rejection_messages, modified_tool_call_messages = _apply_tool_execution_decisions( tool_call_messages=[tool_call_message], tool_execution_decisions=[ ToolExecutionDecision( tool_name="tool1", execute=True, tool_call_id="1", final_tool_params={"a": 5, "b": 6}, feedback="The parameters for tool 'tool1' were updated by the user to:\n{'a': 5, 'b': 6}", ), ToolExecutionDecision( tool_name="tool2", execute=True, tool_call_id="2", final_tool_params={"a": 7, "b": 8}, feedback="The parameters for tool 'tool2' were updated by the user to:\n{'a': 7, 'b': 8}'", ), ], ) updated_messages = _update_chat_history( chat_history, rejection_messages=rejection_messages, tool_call_and_explanation_messages=modified_tool_call_messages, ) assert updated_messages == [ chat_history[0], ChatMessage.from_user("The parameters for tool 'tool1' were updated by the user to:\n{'a': 5, 'b': 6}"), ChatMessage.from_user("The parameters for tool 'tool2' were updated by the user to:\n{'a': 7, 'b': 8}'"), ChatMessage.from_assistant( tool_calls=[ ToolCall(tool_name="tool1", arguments={"a": 5, "b": 6}, id="1", extra=None), ToolCall(tool_name="tool2", arguments={"a": 7, "b": 8}, id="2", extra=None), ] ), ] def test_update_chat_history_two_tool_calls_modify_and_reject(self): tool_call_message = ChatMessage.from_assistant( tool_calls=[ToolCall("tool1", {"a": 1, "b": 2}, id="1"), ToolCall("tool2", {"a": 3, "b": 4}, id="2")] ) chat_history = [ChatMessage.from_user("What is 1 + 2? and 3 + 4?"), tool_call_message] rejection_messages, modified_tool_call_messages = _apply_tool_execution_decisions( tool_call_messages=[tool_call_message], tool_execution_decisions=[ ToolExecutionDecision( tool_name="tool1", execute=True, tool_call_id="1", final_tool_params={"a": 5, "b": 6}, feedback="The parameters for tool 'tool1' were updated by the user to:\n{'a': 5, 'b': 6}", ), ToolExecutionDecision( tool_name="tool2", execute=False, tool_call_id="2", feedback="The tool execution for 'tool2' was rejected by the user. With feedback: Not needed", ), ], ) updated_messages = _update_chat_history( chat_history, rejection_messages=rejection_messages, tool_call_and_explanation_messages=modified_tool_call_messages, ) assert updated_messages == [ chat_history[0], ChatMessage.from_assistant( tool_calls=[ToolCall(tool_name="tool2", arguments={"a": 3, "b": 4}, id="2", extra=None)] ), ChatMessage.from_tool( tool_result="The tool execution for 'tool2' was rejected by the user. With feedback: Not needed", origin=ToolCall(tool_name="tool2", arguments={"a": 3, "b": 4}, id="2", extra=None), error=True, ), ChatMessage.from_user("The parameters for tool 'tool1' were updated by the user to:\n{'a': 5, 'b': 6}"), ChatMessage.from_assistant( tool_calls=[ToolCall(tool_name="tool1", arguments={"a": 5, "b": 6}, id="1", extra=None)] ), ] class ConfirmationStrategyContextCapturingStrategy: def __init__(self): self.captured_confirmation_strategy_context: dict[str, Any] | None = None def run( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], tool_call_id: str | None = None, confirmation_strategy_context: dict[str, Any] | None = None, ) -> ToolExecutionDecision: self.captured_confirmation_strategy_context = confirmation_strategy_context return ToolExecutionDecision( tool_name=tool_name, execute=True, tool_call_id=tool_call_id, final_tool_params=tool_params ) async def run_async( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], tool_call_id: str | None = None, confirmation_strategy_context: dict[str, Any] | None = None, ) -> ToolExecutionDecision: self.captured_confirmation_strategy_context = confirmation_strategy_context return ToolExecutionDecision( tool_name=tool_name, execute=True, tool_call_id=tool_call_id, final_tool_params=tool_params ) def to_dict(self) -> dict[str, Any]: return {"type": "test.RunContextCapturingStrategy", "init_parameters": {}} @classmethod def from_dict(cls, data: dict[str, Any]) -> "ConfirmationStrategyContextCapturingStrategy": return cls() class TestRunContext: def test_confirmation_strategy_context_passed_to_strategy(self, tools): confirmation_strategy_context = {"event_queue": "mock_queue", "redis_client": "mock_redis"} capturing_strategy = ConfirmationStrategyContextCapturingStrategy() teds = _run_confirmation_strategies( confirmation_strategies={tools[0].name: capturing_strategy}, messages_with_tool_calls=[ ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"a": 1, "b": 2})]) ], tools=tools, confirmation_strategy_context=confirmation_strategy_context, ) # Verify the strategy received the confirmation_strategy_context directly assert capturing_strategy.captured_confirmation_strategy_context is not None assert capturing_strategy.captured_confirmation_strategy_context == confirmation_strategy_context assert len(teds) == 1 assert teds[0].execute is True @pytest.mark.asyncio async def test_confirmation_strategy_context_passed_to_strategy_async(self, tools): confirmation_strategy_context = {"websocket": "mock_websocket", "request_id": "12345"} capturing_strategy = ConfirmationStrategyContextCapturingStrategy() teds = await _run_confirmation_strategies_async( confirmation_strategies={tools[0].name: capturing_strategy}, messages_with_tool_calls=[ ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"a": 1, "b": 2})]) ], tools=tools, confirmation_strategy_context=confirmation_strategy_context, ) # Verify the strategy received the confirmation_strategy_context directly assert capturing_strategy.captured_confirmation_strategy_context is not None assert capturing_strategy.captured_confirmation_strategy_context == confirmation_strategy_context assert len(teds) == 1 assert teds[0].execute is True class TrueAsyncConfirmationStrategy: """ A confirmation strategy with truly async behavior for testing purposes. This strategy simulates an async operation (like waiting for a WebSocket response) by using asyncio.sleep. It demonstrates how custom strategies can implement non-blocking async confirmation flows. """ def __init__(self, delay: float = 0.01, decision: str = "confirm"): self.delay = delay self.decision = decision self.async_was_called = False self.sync_was_called = False def run( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], tool_call_id: str | None = None, confirmation_strategy_context: dict[str, Any] | None = None, ) -> ToolExecutionDecision: """Sync version - should NOT be called when run_async is available.""" self.sync_was_called = True return ToolExecutionDecision( tool_name=tool_name, execute=True, tool_call_id=tool_call_id, final_tool_params=tool_params ) async def run_async( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], tool_call_id: str | None = None, confirmation_strategy_context: dict[str, Any] | None = None, ) -> ToolExecutionDecision: """Truly async version that simulates waiting for external confirmation.""" self.async_was_called = True # Simulate async operation (e.g., waiting for WebSocket response) await asyncio.sleep(self.delay) if self.decision == "reject": return ToolExecutionDecision( tool_name=tool_name, execute=False, tool_call_id=tool_call_id, feedback=f"Tool '{tool_name}' was rejected asynchronously.", ) return ToolExecutionDecision( tool_name=tool_name, execute=True, tool_call_id=tool_call_id, final_tool_params=tool_params ) def to_dict(self) -> dict[str, Any]: return {"type": "test.TrueAsyncConfirmationStrategy", "init_parameters": {"delay": self.delay}} @classmethod def from_dict(cls, data: dict[str, Any]) -> "TrueAsyncConfirmationStrategy": return cls(**data.get("init_parameters", {})) class TestAsyncConfirmationStrategies: @pytest.mark.asyncio async def test_async_strategy_confirm(self): strategy = TrueAsyncConfirmationStrategy(delay=0.01, decision="confirm") decision = await strategy.run_async( tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"} ) assert strategy.async_was_called is True assert strategy.sync_was_called is False assert decision.tool_name == "test_tool" assert decision.execute is True assert decision.final_tool_params == {"param1": "value1"} @pytest.mark.asyncio async def test_async_strategy_reject(self): strategy = TrueAsyncConfirmationStrategy(delay=0.01, decision="reject") decision = await strategy.run_async( tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"} ) assert strategy.async_was_called is True assert decision.execute is False assert decision.feedback is not None and "rejected asynchronously" in decision.feedback @pytest.mark.asyncio async def test_async_strategy_used_by_run_confirmation_strategies_async(self, tools): strategy = TrueAsyncConfirmationStrategy(delay=0.01, decision="confirm") teds = await _run_confirmation_strategies_async( confirmation_strategies={tools[0].name: strategy}, messages_with_tool_calls=[ ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"a": 1, "b": 2})]) ], tools=tools, ) # Verify that only the async method was called assert strategy.async_was_called is True assert strategy.sync_was_called is False assert len(teds) == 1 assert teds[0].execute is True @pytest.mark.asyncio async def test_blocking_strategy_run_async(self, monkeypatch): strategy = BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ) # Mock the UI to always confirm def mock_get_user_confirmation(tool_name, tool_description, tool_params): return ConfirmationUIResult(action="confirm") monkeypatch.setattr(strategy.confirmation_ui, "get_user_confirmation", mock_get_user_confirmation) decision = await strategy.run_async( tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"} ) assert decision.tool_name == "test_tool" assert decision.execute is True assert decision.final_tool_params == {"param1": "value1"} @pytest.mark.asyncio async def test_run_confirmation_strategies_async_no_strategy(self, tools): teds = await _run_confirmation_strategies_async( confirmation_strategies={}, messages_with_tool_calls=[ ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"a": 1, "b": 2})]) ], tools=tools, ) assert len(teds) == 1 assert teds[0].tool_name == tools[0].name assert teds[0].execute is True assert teds[0].final_tool_params == {"a": 1, "b": 2} @pytest.mark.asyncio async def test_run_confirmation_strategies_async_with_strategy(self, tools): teds = await _run_confirmation_strategies_async( confirmation_strategies={ tools[0].name: BlockingConfirmationStrategy( confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI() ) }, messages_with_tool_calls=[ ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"a": 1, "b": 2})]) ], tools=tools, ) assert len(teds) == 1 assert teds[0].tool_name == tools[0].name assert teds[0].execute is True @pytest.mark.asyncio async def test_run_confirmation_strategies_async_unknown_tool_passes_through(self, tools): # The model hallucinated a tool name that isn't in the available tools. Confirmation is skipped and the # call passes through unchanged so the tool-calling code can report it (ToolNotFoundException). teds = await _run_confirmation_strategies_async( confirmation_strategies={ tools[0].name: BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ) }, messages_with_tool_calls=[ ChatMessage.from_assistant( tool_calls=[ToolCall(id="tc-1", tool_name="hallucinated_tool", arguments={"a": 1})] ) ], tools=tools, ) assert teds == [ ToolExecutionDecision( tool_call_id="tc-1", tool_name="hallucinated_tool", execute=True, final_tool_params={"a": 1} ) ]