feat(nexus): integrate NexusEnvironment and NexusTaskHandler into ServiceContainer; streamline task handling logic

Author: xepozz

src/Internal/Nexus/NexusContext.php

@@ -20,10 +20,13 @@
 /**
  * Composite per-dispatch Nexus context, returned by {@see \Temporal\Nexus\Nexus::getCurrentContext()}.
  *
- * Carries the public Temporal-side {@see $operation} info and the handler-side
- * {@see $current} ({@see OperationContext} with links/headers/deadline). The
- * {@see $environment} (WorkflowClient) and {@see $outboundPipeline} are internal
- * plumbing and should not be consumed by user code.
+ * {@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 $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
+ * should not be consumed by user code.
  */
 final class NexusContext
 {
@@ -32,36 +35,11 @@ final class NexusContext
      * @internal $environment and $outboundPipeline are framework plumbing.
      */
     public function __construct(
+        public readonly OperationContext $current,
         public readonly ?NexusOperationContext $operation = null,
         public readonly ?NexusEnvironment $environment = null,
-        public readonly ?OperationContext $current = null,
         public readonly ?OperationStartDetails $startDetails = null,
         public readonly ?OperationCancelDetails $cancelDetails = null,
         public readonly ?Pipeline $outboundPipeline = null,
     ) {}
-
-    public function withOperation(?NexusOperationContext $operation): self
-    {
-        return new self($operation, $this->environment, $this->current, $this->startDetails, $this->cancelDetails, $this->outboundPipeline);
-    }
-
-    public function withEnvironment(?NexusEnvironment $environment): self
-    {
-        return new self($this->operation, $environment, $this->current, $this->startDetails, $this->cancelDetails, $this->outboundPipeline);
-    }
-
-    public function withCurrent(?OperationContext $current): self
-    {
-        return new self($this->operation, $this->environment, $current, $this->startDetails, $this->cancelDetails, $this->outboundPipeline);
-    }
-
-    public function withStartDetails(?OperationStartDetails $details): self
-    {
-        return new self($this->operation, $this->environment, $this->current, $details, $this->cancelDetails, $this->outboundPipeline);
-    }
-
-    public function withCancelDetails(?OperationCancelDetails $details): self
-    {
-        return new self($this->operation, $this->environment, $this->current, $this->startDetails, $details, $this->outboundPipeline);
-    }
 }

src/Internal/Nexus/NexusTaskHandler.php

@@ -47,7 +47,6 @@
 final class NexusTaskHandler
 {
     private ?ServiceHandler $serviceHandler = null;
-    private ?NexusEnvironment $environment = null;
 
     /**
      * @param bool $includeTracebackInFailure Disable in cross-trust-boundary
@@ -58,6 +57,7 @@ public function __construct(
         private readonly DataConverterInterface $dataConverter,
         private readonly bool $includeTracebackInFailure = true,
         private readonly PipelineProvider $interceptorProvider = new SimplePipelineProvider(),
+        private readonly ?NexusEnvironment $environment = null,
     ) {}
 
     /**
@@ -85,15 +85,6 @@ public static function deadlineFromHeaders(array $headers): ?\DateTimeImmutable
         }
     }
 
-    /**
-     * Bind worker environment for {@see \Temporal\Nexus\Nexus::getOperationContext()}. Idempotent.
-     */
-    public function withNexusEnvironment(NexusEnvironment $environment): self
-    {
-        $this->environment = $environment;
-        return $this;
-    }
-
     public function handleStartOperation(Request $request, ?MethodCanceller $methodCanceller = null): Response
     {
         $startRequest = $request->getStartOperation();

src/Internal/ServiceContainer.php

@@ -23,7 +23,9 @@
 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;
 use Temporal\Internal\Queue\QueueInterface;
 use Temporal\Internal\Repository\RepositoryInterface;
@@ -42,6 +44,7 @@ final class ServiceContainer
     public readonly ActivityCollection $activities;
     public readonly NexusServiceCollection $nexusServices;
     public readonly NexusInvocationRegistry $nexusInvocations;
+    public readonly NexusTaskHandler $nexusTaskHandler;
     public readonly WorkflowReader $workflowsReader;
     public readonly ActivityReader $activitiesReader;
     public readonly NexusServiceReader $nexusServicesReader;
@@ -60,11 +63,18 @@ public function __construct(
         public readonly ExceptionInterceptorInterface $exceptionInterceptor,
         public readonly PipelineProvider $interceptorProvider,
         public readonly LoggerInterface $logger,
+        public readonly ?NexusEnvironment $nexusEnvironment = null,
     ) {
         $this->workflows = new WorkflowCollection();
         $this->activities = new ActivityCollection();
         $this->nexusServices = new NexusServiceCollection();
         $this->nexusInvocations = new NexusInvocationRegistry();
+        $this->nexusTaskHandler = new NexusTaskHandler(
+            $this->nexusServices,
+            $this->dataConverter,
+            interceptorProvider: $this->interceptorProvider,
+            environment: $nexusEnvironment,
+        );
         $this->running = new ProcessCollection();
         $this->workflowsReader = new WorkflowReader($this->reader);
         $this->activitiesReader = new ActivityReader($this->reader);
@@ -76,6 +86,7 @@ public static function fromWorkerFactory(
         ExceptionInterceptorInterface $exceptionInterceptor,
         PipelineProvider $interceptorProvider,
         LoggerInterface $logger,
+        ?NexusEnvironment $nexusEnvironment = null,
     ): self {
         return new self(
             $worker,
@@ -88,6 +99,7 @@ public static function fromWorkerFactory(
             $exceptionInterceptor,
             $interceptorProvider,
             $logger,
+            $nexusEnvironment,
         );
     }
 }

src/Nexus/Nexus.php

@@ -53,9 +53,7 @@ public static function getCurrentContext(): NexusContext
      */
     public static function getCurrentOperationContext(): OperationContext
     {
-        return self::getCurrentContext()->current ?? throw new \LogicException(
-            'Nexus::getCurrentOperationContext() called outside a Nexus operation dispatch.',
-        );
+        return self::getCurrentContext()->current;
     }
 
     /**

src/Worker/Worker.php

@@ -15,8 +15,6 @@
 use Temporal\Internal\Declaration\EntityNameValidator;
 use Temporal\Internal\Events\EventEmitterTrait;
 use Temporal\Internal\Events\EventListenerInterface;
-use Temporal\Internal\Nexus\NexusEnvironment;
-use Temporal\Internal\Nexus\NexusTaskHandler;
 use Temporal\Internal\Repository\RepositoryInterface;
 use Temporal\Internal\ServiceContainer;
 use Temporal\Internal\Transport\Router;
@@ -37,21 +35,18 @@ class Worker implements WorkerInterface, EventListenerInterface, DispatcherInter
     private RouterInterface $router;
     private ServiceContainer $services;
     private RPCConnectionInterface $rpc;
-    private ?NexusEnvironment $nexusEnvironment;
 
     public function __construct(
         string $taskQueue,
         WorkerOptions $options,
         ServiceContainer $serviceContainer,
         RPCConnectionInterface $rpc,
-        ?NexusEnvironment $nexusEnvironment = null,
     ) {
         EntityNameValidator::validateTaskQueue($taskQueue);
 
         $this->rpc = $rpc;
         $this->name = $taskQueue;
         $this->options = $options;
-        $this->nexusEnvironment = $nexusEnvironment;
 
         $this->services = $serviceContainer;
         $this->router = $this->createRouter();
@@ -122,6 +117,14 @@ public function getActivities(): RepositoryInterface
 
     public function registerNexusServiceImplementation(object ...$services): WorkerInterface
     {
+        if ($this->services->nexusEnvironment === null && $services !== []) {
+            throw new \LogicException(
+                'Cannot register Nexus service implementations on a worker without a WorkflowClient. ' .
+                'Pass a WorkflowClient to the WorkerFactory (e.g. WorkerFactory::create(client: $workflowClient)) ' .
+                '— Nexus operations require cluster access.',
+            );
+        }
+
         foreach ($services as $service) {
             $prototype = $this->services->nexusServicesReader->fromClass(\get_class($service));
             $this->services->nexusServices->add($prototype->withInstance($service), false);
@@ -153,27 +156,10 @@ protected function createRouter(): RouterInterface
         $router->add(new Router\StackTrace($this->services->running));
 
         // Nexus routes
-        $nexusHandler = $this->createNexusTaskHandler();
-        $invocations = $this->services->nexusInvocations;
-        $router->add(new Router\InvokeNexusOperation($nexusHandler, $invocations, $this->services->dataConverter));
-        $router->add(new Router\CancelNexusOperation($nexusHandler));
-        $router->add(new Router\CancelNexusOperationMethod($invocations));
+        $router->add(new Router\InvokeNexusOperation($this->services->nexusTaskHandler, $this->services->nexusInvocations, $this->services->dataConverter));
+        $router->add(new Router\CancelNexusOperation($this->services->nexusTaskHandler));
+        $router->add(new Router\CancelNexusOperationMethod($this->services->nexusInvocations));
 
         return $router;
     }
-
-    private function createNexusTaskHandler(): NexusTaskHandler
-    {
-        $handler = new NexusTaskHandler(
-            $this->services->nexusServices,
-            $this->services->dataConverter,
-            interceptorProvider: $this->services->interceptorProvider,
-        );
-
-        if ($this->nexusEnvironment !== null) {
-            $handler->withNexusEnvironment($this->nexusEnvironment);
-        }
-
-        return $handler;
-    }
 }

src/WorkerFactory.php

@@ -214,9 +214,9 @@ public function newWorker(
                     $options->enableLoggingInReplay,
                     $taskQueue,
                 ),
+                $nexusEnvironment,
             ),
             $this->rpc,
-            $nexusEnvironment,
         );
 
         // Call initializeWorker hooks (forward order)

testing/src/WorkerFactory.php

@@ -14,6 +14,7 @@
 use Temporal\Interceptor\PipelineProvider;
 use Temporal\Interceptor\SimplePipelineProvider;
 use Temporal\Internal\Interceptor\Pipeline;
+use Temporal\Internal\Nexus\NexusEnvironment;
 use Temporal\Internal\ServiceContainer;
 use Temporal\Internal\Workflow\Logger;
 use Temporal\Plugin\CompositePipelineProvider;
@@ -94,6 +95,15 @@ public function newWorker(
             $interceptorProvider ?? new SimplePipelineProvider(),
         );
 
+        /** @psalm-suppress ArgumentTypeCoercion */
+        $nexusEnvironment = $this->workflowClient !== null
+            ? new NexusEnvironment(
+                $this->workflowClient->getClientOptions()->namespace,
+                $taskQueue,
+                $this->workflowClient,
+            )
+            : null;
+
         $worker = new WorkerMock(
             new Worker(
                 $taskQueue,
@@ -107,6 +117,7 @@ public function newWorker(
                         $options->enableLoggingInReplay,
                         $taskQueue,
                     ),
+                    $nexusEnvironment,
                 ),
                 $this->rpc,
             ),

tests/Nexus/Unit/NexusOutboundInterceptorTest.php

@@ -18,6 +18,7 @@
 use Temporal\Interceptor\Trait\NexusOperationOutboundCallsInterceptorTrait;
 use Temporal\Internal\Interceptor\Pipeline;
 use Temporal\Internal\Nexus\NexusContext;
+use Temporal\Nexus\Handler\OperationContext;
 use Temporal\Nexus\Nexus;
 use Temporal\Nexus\NexusOperationContext;
 
@@ -56,6 +57,7 @@ public function getInfo(GetInfoInput $input, callable $next): NexusOperationCont
 
         $info = new NexusOperationContext('ns', 'tq');
         Nexus::setCurrentContext(new NexusContext(
+            current: new OperationContext(service: 'svc', operation: 'op'),
             operation: $info,
             outboundPipeline: Pipeline::prepare([$record('A'), $record('B')]),
         ));
@@ -79,6 +81,7 @@ public function getInfo(GetInfoInput $input, callable $next): NexusOperationCont
         };
 
         Nexus::setCurrentContext(new NexusContext(
+            current: new OperationContext(service: 'svc', operation: 'op'),
             operation: new NexusOperationContext('ns', 'tq'),
             outboundPipeline: Pipeline::prepare([$rewriting]),
         ));
@@ -92,7 +95,10 @@ public function getInfo(GetInfoInput $input, callable $next): NexusOperationCont
     public function testGetInfoReturnsInfoDirectlyWithoutPipeline(): void
     {
         $info = new NexusOperationContext('ns', 'tq');
-        Nexus::setCurrentContext(new NexusContext(operation: $info));
+        Nexus::setCurrentContext(new NexusContext(
+            current: new OperationContext(service: 'svc', operation: 'op'),
+            operation: $info,
+        ));
 
         self::assertSame($info, Nexus::getOperationContext());
     }

tests/Unit/Nexus/NexusContextAccessorTestCase.php

@@ -8,6 +8,7 @@
 use Temporal\Client\WorkflowClientInterface;
 use Temporal\Internal\Nexus\NexusContext;
 use Temporal\Internal\Nexus\NexusEnvironment;
+use Temporal\Nexus\Handler\OperationContext;
 use Temporal\Nexus\Nexus;
 use PHPUnit\Framework\Attributes\CoversClass;
 use Temporal\Nexus\NexusOperationContext;
@@ -41,6 +42,7 @@ public function testReturnsCurrentContext(): void
         $ctx = new NexusOperationContext('ns', 'tq');
 
         Nexus::setCurrentContext(new NexusContext(
+            current: new OperationContext(service: 'svc', operation: 'op'),
             operation: $ctx,
             environment: new NexusEnvironment('ns', 'tq', $client),
         ));
@@ -62,7 +64,10 @@ public function testPublicContextDoesNotExposeWorkflowClient(): void
 
     public function testClearingRestoresOutsideBehavior(): void
     {
-        Nexus::setCurrentContext(new NexusContext(operation: new NexusOperationContext('ns', 'tq')));
+        Nexus::setCurrentContext(new NexusContext(
+            current: new OperationContext(service: 'svc', operation: 'op'),
+            operation: new NexusOperationContext('ns', 'tq'),
+        ));
         Nexus::setCurrentContext(null);
 
         $this->expectException(\LogicException::class);