test_process_tool.py

"""Tests for process_tool.py."""

import os
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from uipath.agent.models.agent import (
    AgentProcessToolProperties,
    AgentProcessToolResourceConfig,
    AgentToolType,
)
from uipath.platform.common import WaitJob
from uipath.platform.orchestrator import Job

from uipath_langchain.agent.tools.process_tool import create_process_tool


@pytest.fixture
def process_resource():
    """Create a minimal process tool resource config."""
    return AgentProcessToolResourceConfig(
        type=AgentToolType.PROCESS,
        name="test_process",
        description="Test process description",
        input_schema={"type": "object", "properties": {}},
        output_schema={"type": "object", "properties": {}},
        properties=AgentProcessToolProperties(
            process_name="MyProcess",
            folder_path="/Shared/MyFolder",
        ),
    )


@pytest.fixture
def flow_resource():
    """Create a process tool resource config of type Flow (Maestro Flow release)."""
    return AgentProcessToolResourceConfig(
        type=AgentToolType.FLOW,
        name="test_flow",
        description="Test flow description",
        input_schema={"type": "object", "properties": {}},
        output_schema={"type": "object", "properties": {}},
        properties=AgentProcessToolProperties(
            process_name="MyFlow",
            folder_path="/Shared/Flows",
        ),
    )


@pytest.fixture
def function_resource():
    """Create a process tool resource config of type Function (coded function release)."""
    return AgentProcessToolResourceConfig(
        type=AgentToolType.FUNCTION,
        name="test_function",
        description="Test function description",
        input_schema={"type": "object", "properties": {}},
        output_schema={"type": "object", "properties": {}},
        properties=AgentProcessToolProperties(
            process_name="MyFunction",
            folder_path="/Shared/Functions",
        ),
    )


@pytest.fixture
def process_resource_with_inputs():
    """Create a process tool resource config with input arguments."""
    return AgentProcessToolResourceConfig(
        type=AgentToolType.PROCESS,
        name="data_processor",
        description="Process data with arguments",
        input_schema={
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "count": {"type": "integer"},
            },
        },
        output_schema={
            "type": "object",
            "properties": {
                "result": {"type": "string"},
            },
        },
        properties=AgentProcessToolProperties(
            process_name="DataProcessor",
            folder_path="/Shared/DataFolder",
        ),
    )


class TestProcessToolMetadata:
    """Test that process tool has correct metadata for observability."""

    def test_process_tool_has_metadata(self, process_resource):
        """Test that process tool has metadata dict."""
        tool = create_process_tool(process_resource)

        assert tool.metadata is not None
        assert isinstance(tool.metadata, dict)

    def test_process_tool_metadata_has_tool_type(self, process_resource):
        """Test that metadata contains tool_type derived from resource type."""
        tool = create_process_tool(process_resource)
        assert tool.metadata is not None
        assert tool.metadata["tool_type"] == "process"

    def test_process_tool_metadata_has_display_name(self, process_resource):
        """Test that metadata contains display_name from process_name."""
        tool = create_process_tool(process_resource)
        assert tool.metadata is not None
        assert tool.metadata["display_name"] == "MyProcess"

    @patch.dict(os.environ, {"UIPATH_FOLDER_PATH": "/Shared/TestFolder"})
    def test_process_tool_metadata_has_folder_path(self, process_resource):
        """Test that metadata contains folder_path for span attributes."""
        tool = create_process_tool(process_resource)
        assert tool.metadata is not None
        assert tool.metadata["folder_path"] == "/Shared/TestFolder"

    def test_process_tool_metadata_has_span_context(self, process_resource):
        """Test that metadata contains _span_context dict for tracing."""
        tool = create_process_tool(process_resource)
        assert tool.metadata is not None
        assert "_span_context" in tool.metadata
        assert isinstance(tool.metadata["_span_context"], dict)

    def test_process_tool_metadata_tool_type_uses_resource_type(self):
        """Test that tool_type is derived from resource.type.lower()."""
        resource = AgentProcessToolResourceConfig(
            type=AgentToolType.PROCESS,
            name="test_process",
            description="Test",
            input_schema={"type": "object", "properties": {}},
            output_schema={"type": "object", "properties": {}},
            properties=AgentProcessToolProperties(
                process_name="MyProcess",
                folder_path="/Shared",
            ),
        )
        tool = create_process_tool(resource)
        assert tool.metadata is not None
        assert tool.metadata["tool_type"] == resource.type.lower()


class TestProcessToolInvocation:
    """Test process tool invocation behavior: invoke then interrupt."""

    @pytest.mark.asyncio
    @patch.dict(os.environ, {"UIPATH_FOLDER_PATH": "/Shared/MyFolder"})
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_invoke_calls_processes_invoke_async(
        self, mock_uipath_class, mock_interrupt, process_resource
    ):
        """Test that invoking the tool calls client.processes.invoke_async."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key-123"
        mock_job.folder_key = "folder-key-123"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        await tool.ainvoke({})

        mock_client.processes.invoke_async.assert_called_once_with(
            name="MyProcess",
            input_arguments={},
            folder_path="/Shared/MyFolder",
            attachments=[],
            parent_span_id=None,
            parent_operation_id=None,
            run_as_me=None,
        )

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_invoke_interrupts_with_wait_job(
        self, mock_uipath_class, mock_interrupt, process_resource
    ):
        """Test that after invoking, the tool interrupts with WaitJobRaw."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key-456"
        mock_job.folder_key = "folder-key-456"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        await tool.ainvoke({})

        mock_interrupt.assert_called_once()
        wait_job_arg = mock_interrupt.call_args[0][0]
        assert isinstance(wait_job_arg, WaitJob)
        assert wait_job_arg.job == mock_job
        assert wait_job_arg.process_folder_key == "folder-key-456"

    @pytest.mark.asyncio
    @patch.dict(os.environ, {"UIPATH_FOLDER_PATH": "/Shared/DataFolder"})
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_invoke_passes_input_arguments(
        self, mock_uipath_class, mock_interrupt, process_resource_with_inputs
    ):
        """Test that input arguments are correctly passed to invoke_async."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource_with_inputs)
        await tool.ainvoke({"name": "test-data", "count": 42})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["input_arguments"] == {"name": "test-data", "count": 42}
        assert call_kwargs["name"] == "DataProcessor"
        assert call_kwargs["folder_path"] == "/Shared/DataFolder"

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_invoke_returns_output_from_extract(
        self, mock_uipath_class, mock_interrupt, process_resource
    ):
        """Test that the tool returns the extracted job output on success."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(
            return_value='{"output_arg": "value123"}'
        )
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        result = await tool.ainvoke({})

        assert result == {"output_arg": "value123"}

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_invoke_returns_error_message_on_faulted_job(
        self, mock_uipath_class, mock_interrupt, process_resource
    ):
        """Test that the tool returns an error message string when the job is faulted."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "faulted"
        mock_resumed_job.job_error = None
        mock_resumed_job.info = "Something went wrong in the workflow"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        result = await tool.ainvoke({})

        assert isinstance(result, str)
        assert "Something went wrong in the workflow" in result


class TestProcessToolSpanContext:
    """Test that _span_context is properly wired for tracing."""

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_span_context_parent_span_id_passed_to_invoke(
        self, mock_uipath_class, mock_interrupt, process_resource
    ):
        """Test that parent_span_id from _span_context is forwarded to invoke_async."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        assert tool.metadata is not None

        # Simulate tracing setting parent_span_id via the shared _span_context
        tool.metadata["_span_context"]["parent_span_id"] = "span-abc-123"

        await tool.ainvoke({})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["parent_span_id"] == "span-abc-123"

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_span_context_consumed_after_invoke(
        self, mock_uipath_class, mock_interrupt, process_resource
    ):
        """Test that parent_span_id is popped (consumed) from _span_context after use."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        assert tool.metadata is not None
        tool.metadata["_span_context"]["parent_span_id"] = "span-xyz"

        await tool.ainvoke({})

        # parent_span_id should be consumed (popped) after invocation
        assert "parent_span_id" not in tool.metadata["_span_context"]

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_span_context_defaults_to_none_when_empty(
        self, mock_uipath_class, mock_interrupt, process_resource
    ):
        """Test that parent_span_id defaults to None when _span_context is empty."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        # Don't set any parent_span_id
        await tool.ainvoke({})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["parent_span_id"] is None


class TestProcessToolRunAsMe:
    """Test RunAsMe propagation passed top-down from tool factory."""

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_run_as_me_true_passed_to_invoke(
        self,
        mock_uipath_class,
        mock_interrupt,
        process_resource,
    ):
        """Test RunAsMe=True is forwarded to invoke_async when set."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource, run_as_me=True)
        await tool.ainvoke({})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["run_as_me"] is True

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_run_as_me_false_sends_none(
        self,
        mock_uipath_class,
        mock_interrupt,
        process_resource,
    ):
        """Test RunAsMe=None when run_as_me=False (default)."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource, run_as_me=False)
        await tool.ainvoke({})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["run_as_me"] is None

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_run_as_me_default_sends_none(
        self,
        mock_uipath_class,
        mock_interrupt,
        process_resource,
    ):
        """Test RunAsMe=None when run_as_me not specified (default)."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource)
        await tool.ainvoke({})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["run_as_me"] is None


class TestProcessToolFlowType:
    """Test that a Flow-type resource flows through create_process_tool correctly."""

    def test_flow_tool_metadata_has_flow_tool_type(self, flow_resource):
        """A Flow resource produces a tool with metadata.tool_type == 'flow'."""
        tool = create_process_tool(flow_resource)
        assert tool.metadata is not None
        assert tool.metadata["tool_type"] == "flow"

    def test_flow_tool_metadata_has_display_name(self, flow_resource):
        """Flow tool metadata exposes the process_name as display_name."""
        tool = create_process_tool(flow_resource)
        assert tool.metadata is not None
        assert tool.metadata["display_name"] == "MyFlow"

    @pytest.mark.asyncio
    @patch.dict(os.environ, {"UIPATH_FOLDER_PATH": "/Shared/Flows"})
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_flow_tool_invokes_processes_invoke_async(
        self, mock_uipath_class, mock_interrupt, flow_resource
    ):
        """Flow tool invokes client.processes.invoke_async (same path as Process)."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "flow-job-key"
        mock_job.folder_key = "flow-folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(flow_resource)
        await tool.ainvoke({})

        mock_client.processes.invoke_async.assert_called_once_with(
            name="MyFlow",
            input_arguments={},
            folder_path="/Shared/Flows",
            attachments=[],
            parent_span_id=None,
            parent_operation_id=None,
            run_as_me=None,
        )

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_flow_tool_uses_non_agent_bts_key(
        self, mock_uipath_class, mock_interrupt, flow_resource
    ):
        """Flow tool stores the BTS tracking key under 'wait_for_job_key' (not agent variant)."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "flow-job-key"
        mock_job.folder_key = "flow-folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(flow_resource)
        assert tool.metadata is not None

        await tool.ainvoke({})

        bts_context = tool.metadata["_bts_context"]
        assert bts_context.get("wait_for_job_key") == "flow-job-key"
        assert "wait_for_agent_job_key" not in bts_context


@pytest.fixture
def process_resource_with_conversation_id():
    """Resource whose input schema declares the reserved conversation-id arg."""
    return AgentProcessToolResourceConfig(
        type=AgentToolType.PROCESS,
        name="conv_process",
        description="Process that consumes conversation id",
        input_schema={
            "type": "object",
            "properties": {
                "topic": {"type": "string"},
                "UIPATH_RESERVED_CONVERSATIONID": {"type": "string"},
            },
        },
        output_schema={"type": "object", "properties": {}},
        properties=AgentProcessToolProperties(
            process_name="ConvProcess",
            folder_path="/Shared/Conv",
        ),
    )


class TestProcessToolConversationIdInjection:
    """Auto-inject conversation id when the tool's input schema declares it."""

    @pytest.mark.asyncio
    @patch.dict(os.environ, {"UIPATH_CONVERSATION_ID": "conv-xyz"})
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_injects_conversation_id_when_schema_declares_it(
        self,
        mock_uipath_class,
        mock_interrupt,
        process_resource_with_conversation_id,
    ):
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"
        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client
        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource_with_conversation_id)
        await tool.ainvoke({"topic": "hello"})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["input_arguments"] == {
            "topic": "hello",
            "UIPATH_RESERVED_CONVERSATIONID": "conv-xyz",
        }

    @pytest.mark.asyncio
    @patch.dict(os.environ, {"UIPATH_CONVERSATION_ID": "from-runtime"})
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_runtime_value_overrides_caller_supplied_value(
        self,
        mock_uipath_class,
        mock_interrupt,
        process_resource_with_conversation_id,
    ):
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"
        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client
        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource_with_conversation_id)
        await tool.ainvoke(
            {"topic": "hi", "UIPATH_RESERVED_CONVERSATIONID": "from-llm"}
        )

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert (
            call_kwargs["input_arguments"]["UIPATH_RESERVED_CONVERSATIONID"]
            == "from-runtime"
        )

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_omits_when_conversation_id_missing(
        self,
        mock_uipath_class,
        mock_interrupt,
        process_resource_with_conversation_id,
    ):
        os.environ.pop("UIPATH_CONVERSATION_ID", None)
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"
        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client
        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource_with_conversation_id)
        await tool.ainvoke({"topic": "hi"})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert call_kwargs["input_arguments"].get("UIPATH_RESERVED_CONVERSATIONID") is None

    @pytest.mark.asyncio
    @patch.dict(os.environ, {"UIPATH_CONVERSATION_ID": "conv-xyz"})
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_skips_injection_when_schema_does_not_declare_it(
        self,
        mock_uipath_class,
        mock_interrupt,
        process_resource_with_inputs,
    ):
        mock_job = MagicMock(spec=Job)
        mock_job.key = "job-key"
        mock_job.folder_key = "folder-key"
        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client
        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(process_resource_with_inputs)
        await tool.ainvoke({"name": "x", "count": 1})

        call_kwargs = mock_client.processes.invoke_async.call_args[1]
        assert "UIPATH_RESERVED_CONVERSATIONID" not in call_kwargs["input_arguments"]


class TestProcessToolFunctionType:
    """Test that a Function-type resource flows through create_process_tool correctly."""

    def test_function_tool_metadata_has_function_tool_type(self, function_resource):
        """A Function resource produces a tool with metadata.tool_type == 'function'."""
        tool = create_process_tool(function_resource)
        assert tool.metadata is not None
        assert tool.metadata["tool_type"] == "function"

    def test_function_tool_metadata_has_display_name(self, function_resource):
        """Function tool metadata exposes the process_name as display_name."""
        tool = create_process_tool(function_resource)
        assert tool.metadata is not None
        assert tool.metadata["display_name"] == "MyFunction"

    @pytest.mark.asyncio
    @patch.dict(os.environ, {"UIPATH_FOLDER_PATH": "/Shared/Functions"})
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_function_tool_invokes_processes_invoke_async(
        self, mock_uipath_class, mock_interrupt, function_resource
    ):
        """Function tool invokes client.processes.invoke_async (same path as Process)."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "function-job-key"
        mock_job.folder_key = "function-folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(function_resource)
        await tool.ainvoke({})

        mock_client.processes.invoke_async.assert_called_once_with(
            name="MyFunction",
            input_arguments={},
            folder_path="/Shared/Functions",
            attachments=[],
            parent_span_id=None,
            parent_operation_id=None,
            run_as_me=None,
        )

    @pytest.mark.asyncio
    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
    @patch("uipath_langchain.agent.tools.process_tool.UiPath")
    async def test_function_tool_uses_non_agent_bts_key(
        self, mock_uipath_class, mock_interrupt, function_resource
    ):
        """Function tool stores the BTS tracking key under 'wait_for_job_key' (not agent variant)."""
        mock_job = MagicMock(spec=Job)
        mock_job.key = "function-job-key"
        mock_job.folder_key = "function-folder-key"

        mock_resumed_job = MagicMock(spec=Job)
        mock_resumed_job.state = "successful"

        mock_client = MagicMock()
        mock_client.processes.invoke_async = AsyncMock(return_value=mock_job)
        mock_client.jobs.extract_output_async = AsyncMock(return_value=None)
        mock_uipath_class.return_value = mock_client

        mock_interrupt.return_value = mock_resumed_job

        tool = create_process_tool(function_resource)
        assert tool.metadata is not None

        await tool.ainvoke({})

        bts_context = tool.metadata["_bts_context"]
        assert bts_context.get("wait_for_job_key") == "function-job-key"
        assert "wait_for_agent_job_key" not in bts_context