refactor(nexus): remove NexusEnvironment; update task handling and client integration

Author: xepozz

src/Internal/Nexus/NexusContext.php

@@ -11,6 +11,7 @@
 
 namespace Temporal\Internal\Nexus;
 
+use Temporal\Client\WorkflowClientInterface;
 use Temporal\Internal\Interceptor\Pipeline;
 use Temporal\Nexus\Handler\OperationCancelDetails;
 use Temporal\Nexus\Handler\OperationContext;
@@ -22,22 +23,23 @@
  *
  * {@see $current} (the handler-side {@see OperationContext} with links/headers/deadline) is the
  * one invariant of a dispatch — it is always present. The remaining fields are conditional:
- * {@see $operation}/{@see $environment} are null until the worker environment is bound;
+ * {@see $operation} is null until the worker environment is bound; {@see $workflowClient} is null
+ * unless a WorkflowClient was supplied to the WorkerFactory;
  * {@see $startDetails} and {@see $cancelDetails} are mutually exclusive (a dispatch is a start
  * XOR a cancel); {@see $outboundPipeline} is null when no interceptor pipeline applies. The
- * {@see $environment} (WorkflowClient) and {@see $outboundPipeline} are internal plumbing and
+ * {@see $workflowClient} and {@see $outboundPipeline} are internal plumbing and
  * should not be consumed by user code.
  */
 final class NexusContext
 {
     /**
      * @param null|Pipeline<\Temporal\Interceptor\NexusOperationOutboundCallsInterceptor, mixed> $outboundPipeline
-     * @internal $environment and $outboundPipeline are framework plumbing.
+     * @internal $workflowClient and $outboundPipeline are framework plumbing.
      */
     public function __construct(
         public readonly OperationContext $current,
         public readonly ?NexusOperationContext $operation = null,
-        public readonly ?NexusEnvironment $environment = null,
+        public readonly ?WorkflowClientInterface $workflowClient = null,
         public readonly ?OperationStartDetails $startDetails = null,
         public readonly ?OperationCancelDetails $cancelDetails = null,
         public readonly ?Pipeline $outboundPipeline = null,

src/Internal/Nexus/NexusEnvironment.php

@@ -22,6 +22,7 @@
 use Temporal\Nexus\Handler\SyncOperationStartResult;
 use Temporal\Nexus\Header as NexusHeader;
 use Temporal\Nexus\LinkParser;
+use Temporal\Nexus\NexusOperationContext;
 use Temporal\Api\Common\V1\Payload;
 use Temporal\Api\Common\V1\Payloads;
 use Temporal\Api\Nexus\V1\CancelOperationRequest;
@@ -30,6 +31,7 @@
 use Temporal\Api\Nexus\V1\Response;
 use Temporal\Api\Nexus\V1\StartOperationRequest;
 use Temporal\Api\Nexus\V1\StartOperationResponse;
+use Temporal\Client\WorkflowClientInterface;
 use Temporal\DataConverter\DataConverterInterface;
 use Temporal\Interceptor\PipelineProvider;
 use Temporal\Interceptor\SimplePipelineProvider;
@@ -57,7 +59,7 @@ public function __construct(
         private readonly DataConverterInterface $dataConverter,
         private readonly bool $includeTracebackInFailure = true,
         private readonly PipelineProvider $interceptorProvider = new SimplePipelineProvider(),
-        private readonly ?NexusEnvironment $environment = null,
+        private readonly ?WorkflowClientInterface $workflowClient = null,
     ) {}
 
     /**
@@ -85,8 +87,11 @@ public static function deadlineFromHeaders(array $headers): ?\DateTimeImmutable
         }
     }
 
-    public function handleStartOperation(Request $request, ?MethodCanceller $methodCanceller = null): Response
-    {
+    public function handleStartOperation(
+        Request $request,
+        ?MethodCanceller $methodCanceller = null,
+        NexusOperationContext $operationContext = new NexusOperationContext(),
+    ): Response {
         $startRequest = $request->getStartOperation();
         \assert($startRequest instanceof StartOperationRequest);
 
@@ -125,7 +130,8 @@ public function handleStartOperation(Request $request, ?MethodCanceller $methodC
                 $context,
                 $details,
                 $input,
-                $this->environment,
+                $this->workflowClient,
+                $operationContext,
             );
 
             $startResponse = new StartOperationResponse();
@@ -174,8 +180,10 @@ public function handleStartOperation(Request $request, ?MethodCanceller $methodC
         }
     }
 
-    public function handleCancelOperation(Request $request): Response
-    {
+    public function handleCancelOperation(
+        Request $request,
+        NexusOperationContext $operationContext = new NexusOperationContext(),
+    ): Response {
         $cancelRequest = $request->getCancelOperation();
         \assert($cancelRequest instanceof CancelOperationRequest);
 
@@ -202,7 +210,8 @@ public function handleCancelOperation(Request $request): Response
             $this->getServiceHandler()->cancelOperation(
                 $context,
                 $details,
-                $this->environment,
+                $this->workflowClient,
+                $operationContext,
             );
 
             $response = new Response();

src/Internal/Nexus/NexusTaskHandler.php

@@ -13,6 +13,7 @@
 
 use Psr\Log\LoggerInterface;
 use Spiral\Attributes\ReaderInterface;
+use Temporal\Client\WorkflowClientInterface;
 use Temporal\DataConverter\DataConverterInterface;
 use Temporal\Exception\ExceptionInterceptorInterface;
 use Temporal\Interceptor\PipelineProvider;
@@ -23,7 +24,6 @@
 use Temporal\Internal\Declaration\Reader\ActivityReader;
 use Temporal\Internal\Declaration\Reader\NexusServiceReader;
 use Temporal\Internal\Declaration\Reader\WorkflowReader;
-use Temporal\Internal\Nexus\NexusEnvironment;
 use Temporal\Internal\Nexus\NexusInvocationRegistry;
 use Temporal\Internal\Nexus\NexusTaskHandler;
 use Temporal\Internal\Marshaller\MarshallerInterface;
@@ -63,7 +63,7 @@ public function __construct(
         public readonly ExceptionInterceptorInterface $exceptionInterceptor,
         public readonly PipelineProvider $interceptorProvider,
         public readonly LoggerInterface $logger,
-        public readonly ?NexusEnvironment $nexusEnvironment = null,
+        public readonly ?WorkflowClientInterface $workflowClient = null,
     ) {
         $this->workflows = new WorkflowCollection();
         $this->activities = new ActivityCollection();
@@ -73,7 +73,7 @@ public function __construct(
             $this->nexusServices,
             $this->dataConverter,
             interceptorProvider: $this->interceptorProvider,
-            environment: $nexusEnvironment,
+            workflowClient: $workflowClient,
         );
         $this->running = new ProcessCollection();
         $this->workflowsReader = new WorkflowReader($this->reader);
@@ -86,7 +86,7 @@ public static function fromWorkerFactory(
         ExceptionInterceptorInterface $exceptionInterceptor,
         PipelineProvider $interceptorProvider,
         LoggerInterface $logger,
-        ?NexusEnvironment $nexusEnvironment = null,
+        ?WorkflowClientInterface $workflowClient = null,
     ): self {
         return new self(
             $worker,
@@ -99,7 +99,7 @@ public static function fromWorkerFactory(
             $exceptionInterceptor,
             $interceptorProvider,
             $logger,
-            $nexusEnvironment,
+            $workflowClient,
         );
     }
 }

src/Internal/ServiceContainer.php

@@ -15,8 +15,10 @@
 use Temporal\Api\Nexus\V1\CancelOperationRequest;
 use Temporal\Api\Nexus\V1\Request;
 use Temporal\DataConverter\EncodedValues;
+use Temporal\Internal\Marshaller\MarshallerInterface;
 use Temporal\Internal\Nexus\NexusHandlerErrorException;
 use Temporal\Internal\Nexus\NexusTaskHandler;
+use Temporal\Nexus\NexusOperationContext;
 use Temporal\Worker\Transport\Command\ServerRequestInterface;
 
 /**
@@ -41,13 +43,19 @@ final class CancelNexusOperation extends Route
 {
     public function __construct(
         private readonly NexusTaskHandler $taskHandler,
+        private readonly MarshallerInterface $marshaller,
     ) {}
 
     public function handle(ServerRequestInterface $request, array $headers, Deferred $resolver): void
     {
         try {
-            $protoRequest = self::buildProtoRequest($request->getOptions());
-            $this->taskHandler->handleCancelOperation($protoRequest);
+            $options = $request->getOptions();
+            $operationContext = $this->marshaller->unmarshal($options, new NexusOperationContext());
+            $protoRequest = self::buildProtoRequest($options);
+            $this->taskHandler->handleCancelOperation(
+                $protoRequest,
+                $operationContext,
+            );
             $resolver->resolve(EncodedValues::fromValues([]));
         } catch (NexusHandlerErrorException $e) {
             // Unwrap the proto-shaped wrapper produced by NexusTaskHandler — the

src/Internal/Transport/Router/CancelNexusOperation.php

@@ -19,12 +19,14 @@
 use Temporal\DataConverter\DataConverterInterface;
 use Temporal\DataConverter\EncodedValues;
 use Temporal\DataConverter\ValuesInterface;
+use Temporal\Internal\Marshaller\MarshallerInterface;
 use Temporal\Internal\Nexus\NexusHandlerErrorException;
 use Temporal\Internal\Nexus\NexusInvocationRegistry;
 use Temporal\Internal\Nexus\NexusLinkConverter;
 use Temporal\Internal\Nexus\NexusTaskHandler;
 use Temporal\Nexus\Handler\MethodCanceller;
 use Temporal\Nexus\LinkParser;
+use Temporal\Nexus\NexusOperationContext;
 use Temporal\Worker\Transport\Command\Client\NexusOperationStarted;
 use Temporal\Worker\Transport\Command\ServerRequestInterface;
 
@@ -34,12 +36,14 @@ public function __construct(
         private readonly NexusTaskHandler $taskHandler,
         private readonly NexusInvocationRegistry $invocations,
         private readonly DataConverterInterface $dataConverter,
+        private readonly MarshallerInterface $marshaller,
     ) {}
 
     public function handle(ServerRequestInterface $request, array $headers, Deferred $resolver): void
     {
         $options = $request->getOptions();
         $invocationId = (int) ($options['invocationId'] ?? 0);
+        $operationContext = $this->marshaller->unmarshal($options, new NexusOperationContext());
 
         $canceller = null;
         if ($invocationId !== 0) {
@@ -49,7 +53,11 @@ public function handle(ServerRequestInterface $request, array $headers, Deferred
 
         try {
             $protoRequest = self::buildProtoRequest($options, $request->getPayloads());
-            $response = $this->taskHandler->handleStartOperation($protoRequest, $canceller);
+            $response = $this->taskHandler->handleStartOperation(
+                $protoRequest,
+                $canceller,
+                $operationContext,
+            );
 
             $startResponse = $response->getStartOperation();
             \assert($startResponse !== null);

src/Internal/Transport/Router/InvokeNexusOperation.php

@@ -11,14 +11,15 @@
 
 namespace Temporal\Nexus\Handler\Internal;
 
+use Temporal\Client\WorkflowClientInterface;
 use Temporal\DataConverter\ValuesInterface;
-use Temporal\Internal\Nexus\NexusEnvironment;
 use Temporal\Nexus\Exception\HandlerException;
 use Temporal\Nexus\Exception\OperationException;
 use Temporal\Nexus\Handler\OperationCancelDetails;
 use Temporal\Nexus\Handler\OperationContext;
 use Temporal\Nexus\Handler\OperationStartDetails;
 use Temporal\Nexus\Handler\OperationStartResult;
+use Temporal\Nexus\NexusOperationContext;
 
 /**
  * @internal
@@ -37,7 +38,8 @@ public function startOperation(
         OperationContext $context,
         OperationStartDetails $details,
         ValuesInterface $input,
-        ?NexusEnvironment $environment = null,
+        ?WorkflowClientInterface $workflowClient = null,
+        NexusOperationContext $operationContext = new NexusOperationContext(),
     ): OperationStartResult;
 
     /**
@@ -55,6 +57,7 @@ public function startOperation(
     public function cancelOperation(
         OperationContext $context,
         OperationCancelDetails $details,
-        ?NexusEnvironment $environment = null,
+        ?WorkflowClientInterface $workflowClient = null,
+        NexusOperationContext $operationContext = new NexusOperationContext(),
     ): void;
 }

src/Nexus/Handler/Internal/HandlerInterface.php

@@ -12,6 +12,7 @@
 namespace Temporal\Nexus\Handler\Internal;
 
 use Temporal\Api\Common\V1\Payloads;
+use Temporal\Client\WorkflowClientInterface;
 use Temporal\DataConverter\DataConverterInterface;
 use Temporal\DataConverter\EncodedValues;
 use Temporal\DataConverter\ValuesInterface;
@@ -23,7 +24,6 @@
 use Temporal\Interceptor\SimplePipelineProvider;
 use Temporal\Internal\Declaration\NexusServiceInstance;
 use Temporal\Internal\Nexus\NexusContext;
-use Temporal\Internal\Nexus\NexusEnvironment;
 use Temporal\Nexus\Exception\ErrorType;
 use Temporal\Nexus\Exception\HandlerException;
 use Temporal\Nexus\Exception\InvalidArgumentException;
@@ -92,7 +92,8 @@ public function startOperation(
         OperationContext $context,
         OperationStartDetails $details,
         ValuesInterface $input,
-        ?NexusEnvironment $environment = null,
+        ?WorkflowClientInterface $workflowClient = null,
+        NexusOperationContext $operationContext = new NexusOperationContext(),
     ): OperationStartResult {
         [$instance, $handler] = $this->resolveHandler($context);
 
@@ -119,8 +120,8 @@ public function startOperation(
         }
 
         Nexus::setCurrentContext(new NexusContext(
-            operation: self::buildPublicOperationContext($environment),
-            environment: $environment,
+            operation: self::publicOperationContext($operationContext),
+            workflowClient: $workflowClient,
             current: $contextWithServiceDefinition,
             startDetails: $details,
             outboundPipeline: $this->interceptorProvider
@@ -160,7 +161,8 @@ public function startOperation(
     public function cancelOperation(
         OperationContext $context,
         OperationCancelDetails $details,
-        ?NexusEnvironment $environment = null,
+        ?WorkflowClientInterface $workflowClient = null,
+        NexusOperationContext $operationContext = new NexusOperationContext(),
     ): void {
         [$instance, $handler] = $this->resolveHandler($context);
 
@@ -179,8 +181,8 @@ public function cancelOperation(
         $contextWithServiceDefinition = $context->withServiceDefinition($instance->prototype);
 
         Nexus::setCurrentContext(new NexusContext(
-            operation: self::buildPublicOperationContext($environment),
-            environment: $environment,
+            operation: self::publicOperationContext($operationContext),
+            workflowClient: $workflowClient,
             current: $contextWithServiceDefinition,
             cancelDetails: $details,
             outboundPipeline: $this->interceptorProvider
@@ -201,12 +203,12 @@ static function (CancelOperationInput $input) use ($handler): void {
         }
     }
 
-    private static function buildPublicOperationContext(?NexusEnvironment $environment): ?NexusOperationContext
+    private static function publicOperationContext(NexusOperationContext $operationContext): ?NexusOperationContext
     {
-        if ($environment === null) {
+        if ($operationContext->namespace === '' || $operationContext->taskQueue === '') {
             return null;
         }
-        return new NexusOperationContext($environment->namespace, $environment->taskQueue);
+        return $operationContext;
     }
 
     /**