Add Temporal Operation Handler

[VegetarianOrc]

What was changed

Added support for Temporal-backed Nexus operation handlers.

This introduces @temporalio.nexus.temporal_operation, TemporalNexusClient, and TemporalOperationResult, allowing Nexus operation handlers to either return a synchronous result or start a Temporal workflow as the async backing operation. The branch also adds token parsing support for generic operation tokens, cancellation support for Temporal-backed workflow operations, Nexus workflow client overloads for the new operation shape, and focused runtime/type coverage.

Why?

This gives Nexus operation handlers a first-class way to interact with Temporal while preserving Nexus async operation semantics, workflow linking/callback behavior, cancellation handling, and typed workflow-start ergonomics.

Checklist

1. How was this tested:

   • Existing tests
   • Added type tests
   • Added tests for temporal operation

2. Any docs updates needed?

Yes. The Nexus docs should cover @temporalio.nexus.temporal_operation, TemporalNexusClient.start_workflow,
TemporalOperationResult.sync, and async workflow-backed operation behavior.

[Evanthx]

Found a few nitpicks!

[Quinn-With-Two-Ns]

Bit surprised to see a change to the workflow code here, what is this needed for?

[VegetarianOrc]

In the Python SDK, you can pass use Service Handlers directly instead of just the service contract. Without the overloads here type checking fails when you pass a @temporal_operation to a workflow.NexusClient.

@service_handler()
class EchoServiceHandler:
    @nexus.temporal_operation
    async def echo(
        self,
        _ctx: StartOperationContext,
        client: nexus.TemporalNexusClient,
        input: Input,
    ) -> nexus.TemporalOperationResult[str]:
        return await client.start_workflow(
            EchoWorkflow.run, input, id=f"echo-{input.value}"
        )

@workflow.defn
class EchoWorkflowCaller:
    @workflow.run
    async def run(self, input: Input) -> str:
        client = workflow.create_nexus_client(
            service=EchoServiceHandler,
            endpoint=make_nexus_endpoint_name(input.task_queue),
        )
        
        # Without the additions this fails to match an overload for temporal_operation due to the addition of TemporalNexusClient to the arg list.
        return await client.execute_operation(EchoServiceHandler.echo, input)

This PR updates the type tests to include this.

[VegetarianOrc]

I've swapped expected_param_type and param_type in this check here. Parameters should be contravariant (e.g. you could provide a function that accepts nexusrpc.StartOperationContext) but the previous direction was covariance (e.g. you could provide a function that accepts a child of WorkflowRunOperationContext).

In the covariant case, the type provided at runtime will never fulfill the decorated function's requirement of the more specific type. Type checkers already prevent this, but bug was that the warning below would not trigger at runtime so users that don't type check would not receive the warning.

In the contravariant case, the type provided at runtime will fulfill the decorated function's requirement of the less specific type.

The new tests in test_handler_operation_definitions.py demonstrate both paths and assert that the warning is or is not delivered as appropriate.

[VegetarianOrc]

Added this test to demonstrate how users can customize cancellation.

[VegetarianOrc]

Base class is ABC to allow customization of cancellation and require implementors to override start_operation. The decorator uses the concrete internal implementation _TemporalNexusOperationHandler to invoke the decorated method.

[Evanthx]

Nothing major, just some nitpicks!

[Evanthx]

If namespace is an empty string, this is going to be a weird error message - it'll say "expected namespace to be non-empty string, got string".

There is also a comment above in token.py saying "Allow empty string for ns" - is it allowed then? Seems like that comment or this check should change? (I noticed that this time after a prior comment from the last code review!)

[VegetarianOrc]

Good call on the error message, I didn't quite realize that! Also good call on the conflicting behavior. I got overzealous with validation here. I've changed the namespace logic to reflect what was previously supported so that should clear up both issues!

[Evanthx]

Just out of pure paranoia ... should you add a check to make sure no one calls TemporalOperationResult(token=None)? (or async_token with a None value?) It looks like that will work, then will fail with a runtime error below in _to_nexus_result if I read this right. Either way, seems worth validating the inputs in the constructor.

Also you have a constructor and then it looks like the desire is that the two class methods call the constructor? In Python, can you hide the constructor from the user to avoid confusion? If we don't want them to call it, then it's nice if it's not there to be called. But that is very much just a nitpick.

[VegetarianOrc]

You can't truly hide a constructor in Python. You can do some kinda gross things to do it, but IMO it's not worth it.

I have added a __post_init__ implementation to validate that it's been constructed in an expected state to help out with that. Now regardless of how it is constructed, it will assert that it has exactly one of value & token, and if it has a token that it's a non-empty string.