[Bug] Annotated constraints and validators on workflow/activity/handler parameters are ignored by the Pydantic data converter

What are you really trying to do?

Validate workflow and activity arguments with Pydantic constraints without wrapping every argument in a model, for example:

@workflow.run
async def run(self, qty: PositiveInt, sku: Annotated[str, AfterValidator(strip_upper)]) -> str: ...

with pydantic_data_converter (the README says it supports "all types supported by Pydantic").

Describe the bug

Annotated[...] metadata on the parameters and return type of workflows, activities, signals, queries and updates never reaches the payload converter. temporalio.common._type_hints_from_func calls typing.get_type_hints(func) without include_extras=True, so every Annotated[T, ...] is reduced to T before it is stored as arg_types/ret_type. With the Pydantic converter this means:

• qty: PositiveInt (or Annotated[int, Field(gt=0)]) accepts -5
• Annotated[str, AfterValidator(...)] never runs its validator
• the same applies to activity parameters, signal/query/update parameters, and to return types (decoded by the caller)

The identical annotations on a Pydantic model field are enforced, so def run(self, order: Order) rejects a bad quantity while def run(self, qty: PositiveInt) accepts it.

The Annotated-in-sandbox problem fixed in #1914 (#1897) had the same symptom (constraints silently dropped) but a different cause; this one happens outside the sandbox path too.

Minimal Reproduction

import asyncio
import uuid
from datetime import timedelta
from typing import Annotated

from pydantic import AfterValidator, BaseModel, PositiveInt

from temporalio import activity, workflow
from temporalio.client import Client, WorkflowFailureError
from temporalio.contrib.pydantic import pydantic_data_converter
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

Sku = Annotated[str, AfterValidator(lambda s: s.strip().upper())]


class Order(BaseModel):
    qty: PositiveInt
    sku: Sku


@activity.defn
async def reserve(qty: PositiveInt, sku: Sku) -> str:
    return f"activity got qty={qty!r} sku={sku!r}"


@workflow.defn
class ParamWorkflow:
    @workflow.run
    async def run(self, qty: PositiveInt, sku: Sku) -> str:
        res = await workflow.execute_activity(
            reserve,
            args=[qty, "  " + sku.lower() + " "],
            start_to_close_timeout=timedelta(seconds=10),
        )
        return f"workflow got qty={qty!r} sku={sku!r}; {res}"


@workflow.defn
class ModelWorkflow:
    @workflow.run
    async def run(self, order: Order) -> str:
        return f"workflow got {order!r}"


async def task_failure(handle) -> str | None:
    async for ev in handle.fetch_history_events():
        if ev.HasField("workflow_task_failed_event_attributes"):
            return ev.workflow_task_failed_event_attributes.failure.message
    return None


async def main():
    async with await WorkflowEnvironment.start_local(
        data_converter=pydantic_data_converter
    ) as env:
        client: Client = env.client
        async with Worker(
            client,
            task_queue="tq",
            workflows=[ParamWorkflow, ModelWorkflow],
            activities=[reserve],
        ):
            print(
                "valid args:",
                await client.execute_workflow(
                    ParamWorkflow.run,
                    args=[3, "  abc-1 "],
                    id=str(uuid.uuid4()),
                    task_queue="tq",
                ),
            )
            for name, run, kwargs in [
                ("params", ParamWorkflow.run, {"args": [-5, "abc"]}),
                ("model field", ModelWorkflow.run, {"arg": {"qty": -5, "sku": "abc"}}),
            ]:
                handle = await client.start_workflow(
                    run,
                    id=str(uuid.uuid4()),
                    task_queue="tq",
                    task_timeout=timedelta(seconds=2),
                    execution_timeout=timedelta(seconds=5),
                    **kwargs,
                )
                try:
                    print(f"qty=-5 as {name}: result:", await handle.result())
                except WorkflowFailureError as e:
                    print(
                        f"qty=-5 as {name}: not accepted ({type(e.cause).__name__}),",
                        "task failure:",
                        (await task_failure(handle) or "none")[:40],
                    )


if __name__ == "__main__":
    asyncio.run(main())

Output on current main (646d70f5):

valid args: workflow got qty=3 sku='  abc-1 '; activity got qty=3 sku='    abc-1  '
qty=-5 as params: result: workflow got qty=-5 sku='abc'; activity got qty=-5 sku='  abc '
qty=-5 as model field: not accepted (TimeoutError), task failure: Failed decoding arguments

Expected: the first line shows sku='ABC-1' in both places, and the second line is rejected like the third (workflow task fails with Failed decoding arguments, caused by 1 validation error for constrained-int ... Input should be greater than 0).

The hints the converter receives, from _type_hints_from_func(reserve), are ([<class 'int'>, <class 'str'>], <class 'str'>).

Environment/Versions

• OS and processor: macOS (Darwin 25.3.0), arm64
• Temporal Version: temporalio main at 646d70f5 (pyproject version 1.34.0), with the 1.34.0 wheel's bridge extension; Python 3.12.14, pydantic 2.14.1 / pydantic-core 2.50.1; local dev server v1.8.3 (server 1.32.0) via WorkflowEnvironment.start_local()
• Are you using Docker or Kubernetes or building Temporal from source? No

Additional context

Cause: temporalio/common.py _type_hints_from_func (hints = get_type_hints(func)), consumed by activity.py, workflow/_definition.py and workflow/_handlers.py (run, signal, query, update, and dynamic variants).

I have a working change on a branch and would like to agree on the behaviour before opening a PR, because it is user-visible:

1. Keep Annotated in the stored hints (get_type_hints(..., include_extras=True)), let the Pydantic converter hand the full hint to its TypeAdapter (it already does), and make the default converter ignore Annotated metadata (unwrap to the underlying type, as it effectively does today). Only the Pydantic converter changes behaviour. This is what I have implemented. Cost: custom converters that inspect type_hint themselves would now see Annotated[...] for parameters that use it (previously they saw the stripped type), and a Pydantic user whose arguments violated an Annotated constraint that was silently ignored will now get failed workflow tasks (the same as they would with the constraint on a model field).
2. Same as 1 but opt-in for Pydantic, e.g. a flag on PydanticPayloadConverter/PydanticJSONPlainPayloadConverter (default off to keep today's behaviour). Avoids the behaviour change but means the documented "all types supported by Pydantic" needs an extra switch.
3. Keep stripping hints for everyone and document that Annotated metadata on top-level parameters is not honoured (wrap values in a model). No code change.

Related: default-converter value_to_type also fails on Annotated nested inside generics today (e.g. list[Annotated[int, "x"]] raises Unserializable type); option 1 fixes that as a side effect.

Which of these would you prefer? Happy to send the PR for option 1 (or 2).