# Copyright (c) Microsoft. All rights reserved.

from dataclasses import dataclass
from typing import Any, TypeVar

import pytest
from typing_extensions import Never

from agent_framework import (
    FunctionExecutor,
    WorkflowBuilder,
    WorkflowContext,
    WorkflowMessage,
    executor,
)


# Module-level types for string forward reference tests
@dataclass
class FuncExecForwardRefMessage:
    content: str


@dataclass
class FuncExecForwardRefTypeA:
    value: str


@dataclass
class FuncExecForwardRefTypeB:
    value: int


@dataclass
class FuncExecForwardRefResponse:
    result: str


class TestFunctionExecutor:
    """Test suite for FunctionExecutor and @executor decorator."""

    def test_function_executor_basic(self):
        """Test basic FunctionExecutor creation and validation."""

        async def process_string(text: str, ctx: WorkflowContext[str]) -> None:
            await ctx.send_message(text.upper())

        func_exec = FunctionExecutor(process_string)

        # Check that handler was registered
        assert len(func_exec._handlers) == 1  # pyright: ignore[reportPrivateUsage]
        assert str in func_exec._handlers  # pyright: ignore[reportPrivateUsage]

        # Check handler spec was created
        assert len(func_exec._handler_specs) == 1  # pyright: ignore[reportPrivateUsage]
        spec = func_exec._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["name"] == "process_string"
        assert spec["message_type"] is str
        assert spec["output_types"] == [str]

    def test_executor_decorator(self):
        """Test @executor decorator creates proper FunctionExecutor."""

        @executor(id="test_executor")
        async def process_int(value: int, ctx: WorkflowContext[int]) -> None:
            await ctx.send_message(value * 2)

        assert isinstance(process_int, FunctionExecutor)
        assert process_int.id == "test_executor"
        assert int in process_int._handlers  # pyright: ignore[reportPrivateUsage]

        # Check spec
        spec = process_int._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["message_type"] is int
        assert spec["output_types"] == [int]

    def test_executor_decorator_without_id(self):
        """Test @executor decorator uses function name as default ID."""

        @executor
        async def my_function(data: dict[str, Any], ctx: WorkflowContext[Any]) -> None:
            await ctx.send_message(data)

        assert my_function.id == "my_function"

    def test_executor_decorator_without_parentheses(self):
        """Test @executor decorator works without parentheses."""

        @executor
        async def no_parens_function(data: str, ctx: WorkflowContext[str]) -> None:
            await ctx.send_message(data.upper())

        assert isinstance(no_parens_function, FunctionExecutor)
        assert no_parens_function.id == "no_parens_function"
        assert str in no_parens_function._handlers  # pyright: ignore[reportPrivateUsage]

        # Also test with single parameter function
        @executor
        async def simple_no_parens(value: int):
            return value * 2

        assert isinstance(simple_no_parens, FunctionExecutor)
        assert simple_no_parens.id == "simple_no_parens"
        assert int in simple_no_parens._handlers  # pyright: ignore[reportPrivateUsage]

    def test_union_output_types(self):
        """Test that union output types are properly inferred for both messages and workflow outputs."""

        @executor
        async def multi_output(text: str, ctx: WorkflowContext[str | int]) -> None:
            if text.isdigit():
                await ctx.send_message(int(text))
            else:
                await ctx.send_message(text.upper())

        spec = multi_output._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert set(spec["output_types"]) == {str, int}
        assert spec["workflow_output_types"] == []  # No workflow outputs defined

        # Test union types for workflow outputs too
        @executor
        async def multi_workflow_output(data: str, ctx: WorkflowContext[Never, str | int | bool]) -> None:
            if data.isdigit():
                await ctx.yield_output(int(data))
            elif data.lower() in ("true", "false"):
                await ctx.yield_output(data.lower() == "true")
            else:
                await ctx.yield_output(data.upper())

        workflow_spec = multi_workflow_output._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert workflow_spec["output_types"] == []  # None means no message outputs
        assert set(workflow_spec["workflow_output_types"]) == {str, int, bool}

    def test_none_output_type(self):
        """Test WorkflowContext produces empty output types."""

        @executor
        async def no_output(data: Any, ctx: WorkflowContext) -> None:
            # This executor doesn't send any messages
            pass

        spec = no_output._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["output_types"] == []
        assert spec["workflow_output_types"] == []  # No workflow outputs defined

    def test_any_output_type(self):
        """Test WorkflowContext[Any] and WorkflowContext[Any, Any] produce Any output types."""

        @executor
        async def any_output(data: str, ctx: WorkflowContext[Any]) -> None:
            await ctx.send_message("result")

        spec = any_output._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["output_types"] == [Any]
        assert spec["workflow_output_types"] == []  # No workflow outputs defined

        # Test both parameters as Any
        @executor
        async def any_both_output(data: str, ctx: WorkflowContext[Any, Any]) -> None:
            await ctx.send_message("message")
            await ctx.yield_output("workflow_output")

        both_spec = any_both_output._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert both_spec["output_types"] == [Any]
        assert both_spec["workflow_output_types"] == [Any]

    def test_validation_errors(self):
        """Test various validation errors in function signatures."""

        # Wrong number of parameters (now accepts 1 or 2, so 0 or 3+ should fail)
        async def no_params() -> None:
            pass

        with pytest.raises(
            ValueError, match="must have \\(message: T\\) or \\(message: T, ctx: WorkflowContext\\[U\\]\\)"
        ):
            FunctionExecutor(no_params)  # type: ignore

        async def too_many_params(data: str, ctx: WorkflowContext[str], extra: int) -> None:
            pass

        with pytest.raises(
            ValueError, match="must have \\(message: T\\) or \\(message: T, ctx: WorkflowContext\\[U\\]\\)"
        ):
            FunctionExecutor(too_many_params)  # type: ignore

        # Missing message type annotation
        async def no_msg_type(data, ctx: WorkflowContext[str]) -> None:  # type: ignore
            pass

        with pytest.raises(ValueError, match="type annotation for the message"):
            FunctionExecutor(no_msg_type)  # type: ignore

        # Missing ctx annotation (only for 2-parameter functions)
        async def no_ctx_type(data: str, ctx) -> None:  # type: ignore
            pass

        with pytest.raises(ValueError, match="must have a WorkflowContext"):
            FunctionExecutor(no_ctx_type)  # type: ignore

        # Wrong ctx type
        async def wrong_ctx_type(data: str, ctx: str) -> None:  # type: ignore
            pass

        with pytest.raises(ValueError, match="must be annotated as WorkflowContext"):
            FunctionExecutor(wrong_ctx_type)  # type: ignore

        # Unparameterized WorkflowContext is now allowed
        async def unparameterized_ctx(data: str, ctx: WorkflowContext) -> None:  # type: ignore
            pass

        # This should now succeed since unparameterized WorkflowContext is allowed
        executor = FunctionExecutor(unparameterized_ctx)
        assert executor.output_types == []  # Unparameterized has no inferred types
        assert executor.workflow_output_types == []  # No workflow output types

    async def test_execution_in_workflow(self):
        """Test that FunctionExecutor works properly in a workflow."""

        @executor(id="upper")
        async def to_upper(text: str, ctx: WorkflowContext[str]) -> None:
            result = text.upper()
            await ctx.send_message(result)

        @executor(id="reverse")
        async def reverse_text(text: str, ctx: WorkflowContext[Any, str]) -> None:
            result = text[::-1]
            await ctx.yield_output(result)

        # Verify type inference for both executors
        upper_spec = to_upper._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert upper_spec["output_types"] == [str]
        assert upper_spec["workflow_output_types"] == []  # No workflow outputs

        reverse_spec = reverse_text._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert reverse_spec["output_types"] == [Any]  # First parameter is Any
        assert reverse_spec["workflow_output_types"] == [str]  # Second parameter is str

        workflow = WorkflowBuilder(start_executor=to_upper).add_edge(to_upper, reverse_text).build()

        # Run workflow
        events = await workflow.run("hello world")
        outputs = events.get_outputs()

        # Assert that we got the expected output
        assert len(outputs) == 1
        assert outputs[0] == "DLROW OLLEH"

    def test_can_handle_method(self):
        """Test that can_handle method works with instance handlers."""

        @executor
        async def string_processor(text: str, ctx: WorkflowContext[str]) -> None:
            await ctx.send_message(text)

        assert string_processor.can_handle(WorkflowMessage(data="hello", source_id="Mock"))
        assert not string_processor.can_handle(WorkflowMessage(data=123, source_id="Mock"))
        assert not string_processor.can_handle(WorkflowMessage(data=[], source_id="Mock"))

    def test_duplicate_handler_registration(self):
        """Test that registering duplicate handlers raises an error."""

        async def first_handler(text: str, ctx: WorkflowContext[str]) -> None:
            await ctx.send_message(text)

        func_exec = FunctionExecutor(first_handler)

        # Try to register another handler for the same type
        async def second_handler(message: str, ctx: WorkflowContext[str]) -> None:
            await ctx.send_message(message)

        with pytest.raises(ValueError, match="Handler for type .* already registered"):
            func_exec._register_instance_handler(  # pyright: ignore[reportPrivateUsage]
                name="second",
                func=second_handler,
                message_type=str,
                ctx_annotation=WorkflowContext[str],
                output_types=[str],
                workflow_output_types=[],
            )

    def test_complex_type_annotations(self):
        """Test with complex type annotations like List[str], Dict[str, int], etc."""

        @executor
        async def process_list(items: list[str], ctx: WorkflowContext[dict[str, int]]) -> None:
            result = {item: len(item) for item in items}
            await ctx.send_message(result)

        spec = process_list._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["message_type"] == list[str]
        assert spec["output_types"] == [dict[str, int]]

    def test_single_parameter_function(self):
        """Test FunctionExecutor with single-parameter functions."""

        @executor(id="simple_processor")
        async def process_simple(text: str):
            return text.upper()

        assert isinstance(process_simple, FunctionExecutor)
        assert process_simple.id == "simple_processor"
        assert str in process_simple._handlers  # pyright: ignore[reportPrivateUsage]

        # Check spec - single parameter functions have no output types since they can't send messages
        spec = process_simple._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["message_type"] is str
        assert spec["output_types"] == []
        assert spec["ctx_annotation"] is None

    def test_single_parameter_validation(self):
        """Test validation for single-parameter functions."""

        # Valid single-parameter function
        async def valid_single(data: int):
            return data * 2

        func_exec = FunctionExecutor(valid_single)
        assert int in func_exec._handlers  # pyright: ignore[reportPrivateUsage]

        # Single parameter with missing type annotation should still fail
        async def no_annotation(data):  # type: ignore
            pass

        with pytest.raises(ValueError, match="type annotation for the message"):
            FunctionExecutor(no_annotation)  # type: ignore

    def test_single_parameter_can_handle(self):
        """Test that single-parameter functions work with can_handle method."""

        @executor
        async def int_processor(value: int):
            return value * 2

        assert int_processor.can_handle(WorkflowMessage(data=42, source_id="mock"))
        assert not int_processor.can_handle(WorkflowMessage(data="hello", source_id="mock"))
        assert not int_processor.can_handle(WorkflowMessage(data=[], source_id="mock"))

    async def test_single_parameter_execution(self):
        """Test that single-parameter functions can be executed properly."""

        @executor(id="double")
        async def double_value(value: int):
            return value * 2

        # Since single-parameter functions can't send messages,
        # they're typically used as terminal nodes or for side effects
        WorkflowBuilder(start_executor=double_value).build()

        # For testing purposes, we can check that the handler is registered correctly
        assert double_value.can_handle(WorkflowMessage(data=5, source_id="mock"))
        assert int in double_value._handlers  # pyright: ignore[reportPrivateUsage]

    def test_sync_function_basic(self):
        """Test basic synchronous function support."""

        @executor(id="sync_processor")
        def process_sync(text: str):
            return text.upper()

        assert isinstance(process_sync, FunctionExecutor)
        assert process_sync.id == "sync_processor"
        assert str in process_sync._handlers  # pyright: ignore[reportPrivateUsage]

        # Check spec - sync single parameter functions have no output types
        spec = process_sync._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["message_type"] is str
        assert spec["output_types"] == []
        assert spec["ctx_annotation"] is None

    def test_sync_function_with_context(self):
        """Test synchronous function with WorkflowContext."""

        @executor
        def sync_with_ctx(value: int, ctx: WorkflowContext[int]):
            # Sync functions can still use context
            return value * 2

        assert isinstance(sync_with_ctx, FunctionExecutor)
        assert sync_with_ctx.id == "sync_with_ctx"
        assert int in sync_with_ctx._handlers  # pyright: ignore[reportPrivateUsage]

        # Check spec - sync functions with context can infer output types
        spec = sync_with_ctx._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["message_type"] is int
        assert spec["output_types"] == [int]

    def test_sync_function_can_handle(self):
        """Test that sync functions work with can_handle method."""

        @executor
        def string_handler(text: str):
            return text.strip()

        assert string_handler.can_handle(WorkflowMessage(data="hello", source_id="mock"))
        assert not string_handler.can_handle(WorkflowMessage(data=123, source_id="mock"))
        assert not string_handler.can_handle(WorkflowMessage(data=[], source_id="mock"))

    def test_sync_function_validation(self):
        """Test validation for synchronous functions."""

        # Valid sync function with one parameter
        def valid_sync(data: str):
            return data.upper()

        func_exec = FunctionExecutor(valid_sync)
        assert str in func_exec._handlers  # pyright: ignore[reportPrivateUsage]

        # Valid sync function with two parameters
        def valid_sync_with_ctx(data: int, ctx: WorkflowContext[str]):
            return str(data)

        func_exec2 = FunctionExecutor(valid_sync_with_ctx)
        assert int in func_exec2._handlers  # pyright: ignore[reportPrivateUsage]

        # Sync function with missing type annotation should still fail
        def no_annotation(data):  # type: ignore  # pyright: ignore[reportUnknownVariableType]
            return data  # pyright: ignore[reportUnknownVariableType]

        with pytest.raises(ValueError, match="type annotation for the message"):
            FunctionExecutor(no_annotation)  # type: ignore

    def test_mixed_sync_async_decorator(self):
        """Test that both sync and async functions work with decorator."""

        @executor
        def sync_func(data: str):
            return data.lower()

        @executor
        async def async_func(data: str):
            return data.upper()

        # Both should be FunctionExecutor instances
        assert isinstance(sync_func, FunctionExecutor)
        assert isinstance(async_func, FunctionExecutor)

        # Both should handle strings
        assert sync_func.can_handle(WorkflowMessage(data="test", source_id="mock"))
        assert async_func.can_handle(WorkflowMessage(data="test", source_id="mock"))

        # Both should be different instances
        assert sync_func is not async_func

    async def test_sync_function_in_workflow(self):
        """Test that sync functions work properly in a workflow context."""

        @executor(id="sync_upper")
        def to_upper_sync(text: str, ctx: WorkflowContext[str]):
            return text.upper()
            # Note: For the test, we'll use a sync send mechanism
            # In practice, the wrapper handles the async conversion

        @executor(id="async_reverse")
        async def reverse_async(text: str, ctx: WorkflowContext[Any, str]):
            result = text[::-1]
            await ctx.yield_output(result)

        # Verify type inference for sync and async functions
        sync_spec = to_upper_sync._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert sync_spec["output_types"] == [str]
        assert sync_spec["workflow_output_types"] == []  # No workflow outputs

        async_spec = reverse_async._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert async_spec["output_types"] == [Any]  # First parameter is Any
        assert async_spec["workflow_output_types"] == [str]  # Second parameter is str

        # Verify the executors can handle their input types
        assert to_upper_sync.can_handle(WorkflowMessage(data="hello", source_id="mock"))
        assert reverse_async.can_handle(WorkflowMessage(data="HELLO", source_id="mock"))

        # For integration testing, we mainly verify that the handlers are properly registered
        # and the functions are wrapped correctly
        assert str in to_upper_sync._handlers  # pyright: ignore[reportPrivateUsage]
        assert str in reverse_async._handlers  # pyright: ignore[reportPrivateUsage]

    async def test_sync_function_thread_execution(self):
        """Test that sync functions run in thread pool and don't block the event loop."""
        import threading
        import time

        _ = threading.get_ident()
        execution_thread_id = None

        @executor
        def blocking_function(data: str):
            nonlocal execution_thread_id
            execution_thread_id = threading.get_ident()
            # Simulate some CPU-bound work
            time.sleep(0.01)  # Small sleep to verify thread execution
            return data.upper()

        # Verify the function is wrapped and registered
        assert str in blocking_function._handlers  # pyright: ignore[reportPrivateUsage]

        # For a more complete test, we'd need to create a full workflow context,
        # but for now we can verify that the function was properly wrapped
        # and that sync functions store the correct metadata
        assert not blocking_function._is_async  # pyright: ignore[reportPrivateUsage]
        assert not blocking_function._has_context  # pyright: ignore[reportPrivateUsage]

        # The actual thread execution test would require a full workflow setup,
        # but the important thing is that asyncio.to_thread is used in the wrapper

    def test_executor_rejects_staticmethod(self):
        """Test that @executor decorator properly rejects @staticmethod with clear error."""
        with pytest.raises(ValueError) as exc_info:

            class Example:  # pyright: ignore[reportUnusedClass]
                @executor
                @staticmethod
                async def bad_handler(data: str) -> str:
                    return data.upper()

        assert "cannot be used with @staticmethod" in str(exc_info.value)
        assert "@handler on instance methods" in str(exc_info.value)

    def test_executor_rejects_classmethod(self):
        """Test that @executor decorator properly rejects @classmethod with clear error."""
        with pytest.raises(ValueError) as exc_info:

            class Example:  # pyright: ignore[reportUnusedClass]
                @executor
                @classmethod
                async def bad_handler(cls, data: str) -> str:
                    return data.upper()

        assert "cannot be used with @classmethod" in str(exc_info.value)
        assert "@handler on instance methods" in str(exc_info.value)

    async def test_async_staticmethod_detection_behavior(self):
        """Document the behavior of inspect.iscoroutinefunction with staticmethod descriptors.

        This test explains why the unwrapping is necessary when decorators are stacked.
        """
        import asyncio
        import inspect

        # When @staticmethod is applied, it creates a descriptor
        async def my_async_func():
            await asyncio.sleep(0.001)
            return "done"

        # Apply staticmethod (what happens with innermost decorator)
        static_wrapped = staticmethod(my_async_func)

        # Direct check on descriptor object fails (this is the bug)
        assert not inspect.iscoroutinefunction(static_wrapped)
        assert isinstance(static_wrapped, staticmethod)

        # But unwrapping __func__ reveals the async function
        unwrapped = static_wrapped.__func__
        assert inspect.iscoroutinefunction(unwrapped)

        # When accessed via class attribute, Python's descriptor protocol
        # automatically unwraps it, so it works:
        class C:
            async_static = static_wrapped

        assert inspect.iscoroutinefunction(C.async_static)  # Works via descriptor protocol


class TestExecutorExplicitTypes:
    """Test suite for @executor decorator with explicit input_type and output_type parameters."""

    def test_executor_with_explicit_input_type(self):
        """Test that explicit input_type takes precedence over introspection."""

        @executor(input=str)
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Handler should be registered for str (explicit)
        assert str in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert len(process._handlers) == 1  # pyright: ignore[reportPrivateUsage]

        # Can handle str messages
        assert process.can_handle(WorkflowMessage(data="hello", source_id="mock"))
        # Cannot handle int messages
        assert not process.can_handle(WorkflowMessage(data=42, source_id="mock"))

    def test_executor_with_explicit_output_type(self):
        """Test that explicit output_type takes precedence over introspection."""

        @executor(output=int)
        async def process(message: str, ctx: WorkflowContext[str]) -> None:
            pass

        # Handler spec should have int as output type (explicit), not str (introspected)
        spec = process._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["output_types"] == [int]

        # Executor output_types property should reflect explicit type
        assert int in process.output_types
        assert str not in process.output_types

    def test_executor_with_explicit_input_and_output_types(self):
        """Test that both explicit input_type and output_type work together."""

        @executor(id="explicit_both", input=dict, output=list)
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Handler should be registered for dict (explicit input type)
        assert dict in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert len(process._handlers) == 1  # pyright: ignore[reportPrivateUsage]

        # Output type should be list (explicit)
        spec = process._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["output_types"] == [list]

        # Verify can_handle
        assert process.can_handle(WorkflowMessage(data={"key": "value"}, source_id="mock"))
        assert not process.can_handle(WorkflowMessage(data="string", source_id="mock"))

    def test_executor_with_explicit_union_input_type(self):
        """Test that explicit union input_type is handled correctly."""

        @executor(input=str | int)
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Handler should be registered for the union type
        assert len(process._handlers) == 1  # pyright: ignore[reportPrivateUsage]

        # Can handle both str and int messages
        assert process.can_handle(WorkflowMessage(data="hello", source_id="mock"))
        assert process.can_handle(WorkflowMessage(data=42, source_id="mock"))
        # Cannot handle float
        assert not process.can_handle(WorkflowMessage(data=3.14, source_id="mock"))

    def test_executor_with_explicit_union_output_type(self):
        """Test that explicit union output_type is normalized to a list."""

        @executor(output=str | int | bool)
        async def process(message: Any, ctx: WorkflowContext) -> None:
            pass

        # Output types should be a list with all union members
        assert set(process.output_types) == {str, int, bool}

    def test_executor_explicit_types_precedence_over_introspection(self):
        """Test that explicit types always take precedence over introspected types."""

        # Introspection would give: input=str, output=[int]
        # Explicit gives: input=bytes, output=[float]
        @executor(input=bytes, output=float)
        async def process(message: str, ctx: WorkflowContext[int]) -> None:
            pass

        # Should use explicit input type (bytes), not introspected (str)
        assert bytes in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert str not in process._handlers  # pyright: ignore[reportPrivateUsage]

        # Should use explicit output type (float), not introspected (int)
        assert float in process.output_types
        assert int not in process.output_types

    def test_executor_fallback_to_introspection_when_no_explicit_types(self):
        """Test that introspection is used when no explicit types are provided."""

        @executor
        async def process(message: str, ctx: WorkflowContext[int]) -> None:
            pass

        # Should use introspected types
        assert str in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert int in process.output_types

    def test_executor_partial_explicit_types(self):
        """Test that partial explicit types work (only input_type or only output_type)."""

        # Only explicit input_type, introspect output_type
        @executor(input=bytes)
        async def process_input(message: str, ctx: WorkflowContext[int]) -> None:
            pass

        assert bytes in process_input._handlers  # Explicit  # pyright: ignore[reportPrivateUsage]
        assert int in process_input.output_types  # Introspected

        # Only explicit output_type, introspect input_type
        @executor(output=float)
        async def process_output(message: str, ctx: WorkflowContext[int]) -> None:
            pass

        assert str in process_output._handlers  # Introspected  # pyright: ignore[reportPrivateUsage]
        assert float in process_output.output_types  # Explicit
        assert int not in process_output.output_types  # Not introspected when explicit provided

    def test_executor_explicit_input_type_allows_no_message_annotation(self):
        """Test that explicit input_type allows function without message type annotation."""

        @executor(input=str)
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Should work with explicit input_type
        assert str in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert process.can_handle(WorkflowMessage(data="hello", source_id="mock"))

    def test_executor_explicit_types_with_id(self):
        """Test that explicit types work together with id parameter."""

        @executor(id="custom_id", input=bytes, output=int)
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        assert process.id == "custom_id"
        assert bytes in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert int in process.output_types

    def test_executor_explicit_types_with_single_param_function(self):
        """Test that explicit input_type works with single-parameter functions."""

        @executor(input=str)
        async def process(message):  # type: ignore[no-untyped-def]
            return message.upper()  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]

        # Should work with explicit input_type
        assert str in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert process.can_handle(WorkflowMessage(data="hello", source_id="mock"))
        assert not process.can_handle(WorkflowMessage(data=42, source_id="mock"))

    def test_executor_explicit_types_with_sync_function(self):
        """Test that explicit types work with synchronous functions."""

        @executor(input=int, output=str)
        def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        assert int in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert str in process.output_types

    def test_function_executor_constructor_with_explicit_types(self):
        """Test FunctionExecutor constructor with explicit input_type and output_type."""

        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        func_exec = FunctionExecutor(process, id="test", input=dict, output=list)  # pyright: ignore[reportUnknownArgumentType]

        assert dict in func_exec._handlers  # pyright: ignore[reportPrivateUsage]
        spec = func_exec._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["message_type"] is dict
        assert spec["output_types"] == [list]

    def test_executor_explicit_union_types_via_typing_union(self):
        """Test that Union[] syntax also works for explicit types."""
        from typing import Union

        @executor(input=Union[str, int], output=Union[bool, float])
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Can handle both str and int
        assert process.can_handle(WorkflowMessage(data="hello", source_id="mock"))
        assert process.can_handle(WorkflowMessage(data=42, source_id="mock"))

        # Output types should include both
        assert set(process.output_types) == {bool, float}

    def test_executor_with_string_forward_reference_input_type(self):
        """Test that string forward references work for input_type."""

        @executor(input="FuncExecForwardRefMessage")
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Should resolve the string to the actual type
        assert FuncExecForwardRefMessage in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert process.can_handle(WorkflowMessage(data=FuncExecForwardRefMessage("hello"), source_id="mock"))

    def test_executor_with_string_forward_reference_union(self):
        """Test that string forward references work with union types."""

        @executor(input="FuncExecForwardRefTypeA | FuncExecForwardRefTypeB")
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Should handle both types
        assert process.can_handle(WorkflowMessage(data=FuncExecForwardRefTypeA("hello"), source_id="mock"))
        assert process.can_handle(WorkflowMessage(data=FuncExecForwardRefTypeB(42), source_id="mock"))

    def test_executor_with_string_forward_reference_output_type(self):
        """Test that string forward references work for output_type."""

        @executor(input=str, output="FuncExecForwardRefResponse")
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Should resolve the string output type
        assert FuncExecForwardRefResponse in process.output_types

    def test_executor_with_explicit_workflow_output_type(self):
        """Test that explicit workflow_output_type takes precedence over introspection."""

        @executor(workflow_output=bool)
        async def process(message: str, ctx: WorkflowContext[int]) -> None:
            pass

        # Handler spec should have bool as workflow_output_type (explicit)
        spec = process._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
        assert spec["workflow_output_types"] == [bool]

        # Executor workflow_output_types property should reflect explicit type
        assert bool in process.workflow_output_types
        # output_types should still come from introspection (int from WorkflowContext[int])
        assert int in process.output_types

    def test_executor_with_explicit_workflow_output_type_precedence(self):
        """Test that explicit workflow_output_type overrides introspected WorkflowContext second param."""

        @executor(workflow_output=str)
        async def process(message: int, ctx: WorkflowContext[int, bool]) -> None:
            pass

        # workflow_output_types should be str (explicit), not bool (introspected from ctx)
        assert str in process.workflow_output_types
        assert bool not in process.workflow_output_types

    def test_executor_with_all_explicit_types(self):
        """Test that all three explicit type parameters work together."""
        from typing import Any

        @executor(input=str, output=int, workflow_output=bool)
        async def process(message: Any, ctx: WorkflowContext) -> None:
            pass

        # Check input type
        assert str in process._handlers  # pyright: ignore[reportPrivateUsage]
        assert process.can_handle(WorkflowMessage(data="hello", source_id="mock"))

        # Check output_type
        assert int in process.output_types

        # Check workflow_output_type
        assert bool in process.workflow_output_types

    def test_executor_with_union_workflow_output_type(self):
        """Test that union types work for workflow_output_type."""

        @executor(workflow_output=str | int)
        async def process(message: str, ctx: WorkflowContext) -> None:
            pass

        # Should include both types from union
        assert str in process.workflow_output_types
        assert int in process.workflow_output_types

    def test_executor_with_string_forward_reference_workflow_output_type(self):
        """Test that string forward references work for workflow_output_type."""

        @executor(input=str, workflow_output="FuncExecForwardRefResponse")
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Should resolve the string workflow_output_type
        assert FuncExecForwardRefResponse in process.workflow_output_types

    def test_executor_with_string_forward_reference_union_workflow_output_type(self):
        """Test that string forward reference union types work for workflow_output_type."""

        @executor(input=str, workflow_output="FuncExecForwardRefTypeA | FuncExecForwardRefTypeB")
        async def process(message, ctx: WorkflowContext) -> None:  # type: ignore[no-untyped-def]
            pass

        # Should resolve both types from string union
        assert FuncExecForwardRefTypeA in process.workflow_output_types
        assert FuncExecForwardRefTypeB in process.workflow_output_types

    def test_executor_fallback_to_introspection_for_workflow_output_type(self):
        """Test that workflow_output_type falls back to introspection when not explicitly provided."""

        @executor
        async def process(message: str, ctx: WorkflowContext[int, bool]) -> None:
            pass

        # Should use introspected types from WorkflowContext[int, bool]
        assert int in process.output_types
        assert bool in process.workflow_output_types

    def test_function_executor_constructor_with_workflow_output_type(self):
        """Test FunctionExecutor constructor accepts workflow_output_type parameter."""

        async def my_func(message: str, ctx: WorkflowContext) -> None:
            pass

        exec_instance = FunctionExecutor(
            my_func,
            id="test_constructor",
            input=str,
            output=int,
            workflow_output=bool,
        )

        assert str in exec_instance._handlers  # pyright: ignore[reportPrivateUsage]
        assert int in exec_instance.output_types
        assert bool in exec_instance.workflow_output_types


# region Tests for unresolved TypeVar rejection in function executor registration

_FT = TypeVar("_FT")


class TestFunctionExecutorTypeVarRejection:
    """Tests that FunctionExecutor rejects unresolved TypeVar in message annotations."""

    def test_function_executor_rejects_unresolved_typevar(self):
        """Test that FunctionExecutor raises ValueError for unresolved TypeVar message annotation."""

        def echo(message: _FT) -> _FT:
            return message

        with pytest.raises(ValueError, match="unresolved TypeVar"):
            FunctionExecutor(echo, id="echo")

    def test_function_executor_rejects_typevar_with_context(self):
        """Test that FunctionExecutor raises ValueError for TypeVar even with WorkflowContext."""

        async def echo(message: _FT, ctx: WorkflowContext) -> None:
            pass

        with pytest.raises(ValueError, match="unresolved TypeVar"):
            FunctionExecutor(echo, id="echo")

    def test_function_executor_explicit_input_bypasses_typevar_check(self):
        """Test that explicit input= parameter bypasses TypeVar detection."""

        async def echo(message: _FT, ctx: WorkflowContext) -> None:
            pass

        exec_instance = FunctionExecutor(echo, id="echo", input=str, output=str)
        assert str in exec_instance.input_types

    def test_function_executor_allows_concrete_types(self):
        """Test that FunctionExecutor works normally with concrete type annotations."""

        async def handle(message: str, ctx: WorkflowContext[str]) -> None:
            pass

        exec_instance = FunctionExecutor(handle, id="concrete")
        assert str in exec_instance.input_types

    def test_function_executor_error_recommends_explicit_types(self):
        """Test that error message recommends @executor(input=..., output=...)."""

        def echo(message: _FT) -> _FT:
            return message

        with pytest.raises(ValueError, match=r"@executor\(input=<concrete_type>, output=<concrete_type>\)"):
            FunctionExecutor(echo, id="echo")


# endregion: Tests for unresolved TypeVar rejection in function executor registration


_FBT = TypeVar("_FBT", bound=str)


def test_function_executor_rejects_bounded_typevar_in_message_annotation():
    """Test that FunctionExecutor raises ValueError for a bounded TypeVar in message annotation."""

    async def process(message: _FBT, ctx: WorkflowContext) -> None:
        await ctx.send_message(message)

    with pytest.raises(ValueError, match="unresolved TypeVar"):
        FunctionExecutor(process, id="bounded")
