Merge branch 'master' into generic-type

Author: xepozz

psalm-baseline.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<files psalm-version="6.14.3@d0b040a91f280f071c1abcb1b77ce3822058725a">
+<files psalm-version="6.16.0@7cf3e8b988edd75e0766963b0b9e671b220f5785">
   <file src="src/Activity/ActivityInterface.php">
     <DeprecatedClass>
       <code><![CDATA[NamedArgumentConstructor]]></code>
@@ -134,15 +134,9 @@
       <code><![CDATA[object]]></code>
       <code><![CDATA[object]]></code>
     </InvalidReturnType>
-    <LessSpecificReturnStatement>
-      <code><![CDATA[new self($serviceClient, $options, $converter, $interceptorProvider)]]></code>
-    </LessSpecificReturnStatement>
     <MissingParamType>
       <code><![CDATA[$workflow]]></code>
     </MissingParamType>
-    <MoreSpecificReturnType>
-      <code><![CDATA[static]]></code>
-    </MoreSpecificReturnType>
     <RedundantFunctionCall>
       <code><![CDATA[\sprintf]]></code>
       <code><![CDATA[\sprintf]]></code>
@@ -1356,6 +1350,8 @@
             $converter ?? DataConverter::createDefault(),
             $rpc ?? Goridge::create(),
             $credentials,
+            $pluginRegistry,
+            $client,
         )]]></code>
     </UnsafeInstantiation>
   </file>
@@ -1410,6 +1406,12 @@
       <code><![CDATA[$response->toArray()['assets']]]></code>
     </PossiblyUndefinedStringArrayOffset>
   </file>
+  <file src="testing/src/Environment.php">
+    <PossiblyNullReference>
+      <code><![CDATA[stop]]></code>
+      <code><![CDATA[stop]]></code>
+    </PossiblyNullReference>
+  </file>
   <file src="testing/src/Replay/WorkflowReplayer.php">
     <UndefinedMethod>
       <code><![CDATA[getWorkflowType]]></code>

src/Client/ScheduleClient.php

@@ -32,6 +32,11 @@
 use Temporal\DataConverter\DataConverter;
 use Temporal\DataConverter\DataConverterInterface;
 use Temporal\Internal\Mapper\ScheduleMapper;
+use Temporal\Internal\Interceptor\Pipeline;
+use Temporal\Plugin\ConnectionPluginInterface;
+use Temporal\Plugin\PluginRegistry;
+use Temporal\Plugin\ScheduleClientPluginContext;
+use Temporal\Plugin\ScheduleClientPluginInterface;
 use Temporal\Internal\Marshaller\Mapper\AttributeMapperFactory;
 use Temporal\Internal\Marshaller\Marshaller;
 use Temporal\Internal\Marshaller\MarshallerInterface;
@@ -45,14 +50,39 @@ final class ScheduleClient implements ScheduleClientInterface
     private DataConverterInterface $converter;
     private MarshallerInterface $marshaller;
     private ProtoToArrayConverter $protoConverter;
+    private PluginRegistry $pluginRegistry;
 
     public function __construct(
         ServiceClientInterface $serviceClient,
         ?ClientOptions $options = null,
         ?DataConverterInterface $converter = null,
+        ?PluginRegistry $pluginRegistry = null,
     ) {
         $this->clientOptions = $options ?? new ClientOptions();
         $this->converter = $converter ?? DataConverter::createDefault();
+        $this->pluginRegistry = $pluginRegistry ?? new PluginRegistry();
+
+        // Apply connection plugins (before client-level configuration)
+        $connectionPlugins = $this->pluginRegistry->getPlugins(ConnectionPluginInterface::class);
+        $serviceClient = Pipeline::prepare($connectionPlugins)
+            /** @see ConnectionPluginInterface::configureServiceClient() */
+            ->with(static fn(ServiceClientInterface $serviceClient) => $serviceClient, 'configureServiceClient')($serviceClient);
+
+        $pluginContext = new ScheduleClientPluginContext(
+            clientOptions: $this->clientOptions,
+            dataConverter: $this->converter,
+        );
+        $schedulePlugins = $this->pluginRegistry->getPlugins(ScheduleClientPluginInterface::class);
+        /** @see ScheduleClientPluginInterface::configureScheduleClient() */
+        Pipeline::prepare($schedulePlugins)
+            ->with(static fn() => null, 'configureScheduleClient')($pluginContext);
+
+        $this->clientOptions = $pluginContext->getClientOptions();
+        $pluginConverter = $pluginContext->getDataConverter();
+        if ($pluginConverter !== null) {
+            $this->converter = $pluginConverter;
+        }
+
         $this->marshaller = new Marshaller(
             new AttributeMapperFactory(new AttributeReader()),
         );
@@ -71,8 +101,9 @@ public static function create(
         ServiceClientInterface $serviceClient,
         ?ClientOptions $options = null,
         ?DataConverterInterface $converter = null,
+        ?PluginRegistry $pluginRegistry = null,
     ): ScheduleClientInterface {
-        return new self($serviceClient, $options, $converter);
+        return new self($serviceClient, $options, $converter, $pluginRegistry);
     }
 
     public function createSchedule(

src/Client/WorkflowClient.php

@@ -38,6 +38,13 @@
 use Temporal\Interceptor\WorkflowClientCallsInterceptor;
 use Temporal\Internal\Client\ActivityCompletionClient;
 use Temporal\Internal\Client\WorkflowProxy;
+use Temporal\Plugin\ClientPluginContext;
+use Temporal\Plugin\ClientPluginInterface;
+use Temporal\Plugin\CompositePipelineProvider;
+use Temporal\Plugin\ConnectionPluginInterface;
+use Temporal\Plugin\PluginRegistry;
+use Temporal\Plugin\ScheduleClientPluginInterface;
+use Temporal\Plugin\WorkerPluginInterface;
 use Temporal\Internal\Client\WorkflowRun;
 use Temporal\Internal\Client\WorkflowStarter;
 use Temporal\Internal\Client\WorkflowStub;
@@ -63,6 +70,7 @@ class WorkflowClient implements WorkflowClientInterface
     private DataConverterInterface $converter;
     private ?WorkflowStarter $starter = null;
     private WorkflowReader $reader;
+    private PluginRegistry $pluginRegistry;
 
     /** @var Pipeline<WorkflowClientCallsInterceptor, mixed> */
     private Pipeline $interceptorPipeline;
@@ -72,11 +80,40 @@ public function __construct(
         ?ClientOptions $options = null,
         ?DataConverterInterface $converter = null,
         ?PipelineProvider $interceptorProvider = null,
+        ?PluginRegistry $pluginRegistry = null,
     ) {
-        $this->interceptorPipeline = ($interceptorProvider ?? new SimplePipelineProvider())
-            ->getPipeline(WorkflowClientCallsInterceptor::class);
+        $this->pluginRegistry = $pluginRegistry ?? new PluginRegistry();
         $this->clientOptions = $options ?? new ClientOptions();
         $this->converter = $converter ?? DataConverter::createDefault();
+
+        // Apply connection plugins (before client-level configuration)
+        $connectionPlugins = $this->pluginRegistry->getPlugins(ConnectionPluginInterface::class);
+        $serviceClient = Pipeline::prepare($connectionPlugins)
+            /** @see ConnectionPluginInterface::configureServiceClient() */
+            ->with(static fn(ServiceClientInterface $serviceClient) => $serviceClient, 'configureServiceClient')($serviceClient);
+
+        $pluginContext = new ClientPluginContext(
+            clientOptions: $this->clientOptions,
+            dataConverter: $this->converter,
+        );
+        $clientPlugins = $this->pluginRegistry->getPlugins(ClientPluginInterface::class);
+        Pipeline::prepare($clientPlugins)
+            /** @see ClientPluginInterface::configureClient() */
+            ->with(static fn(ClientPluginContext $pluginContext) => $pluginContext, 'configureClient')($pluginContext);
+
+        $this->clientOptions = $pluginContext->getClientOptions();
+        $pluginConverter = $pluginContext->getDataConverter();
+        if ($pluginConverter !== null) {
+            $this->converter = $pluginConverter;
+        }
+
+        // Build interceptor pipeline: merge plugin-contributed interceptors with user-provided ones
+        $provider = new CompositePipelineProvider(
+            $pluginContext->getInterceptors(),
+            $interceptorProvider ?? new SimplePipelineProvider(),
+        );
+
+        $this->interceptorPipeline = $provider->getPipeline(WorkflowClientCallsInterceptor::class);
         $this->reader = new WorkflowReader($this->createReader());
 
         // Set Temporal-Namespace metadata
@@ -88,16 +125,34 @@ public function __construct(
         );
     }
 
-    /**
-     * @return static
-     */
     public static function create(
         ServiceClientInterface $serviceClient,
         ?ClientOptions $options = null,
         ?DataConverterInterface $converter = null,
         ?PipelineProvider $interceptorProvider = null,
+        ?PluginRegistry $pluginRegistry = null,
     ): self {
-        return new self($serviceClient, $options, $converter, $interceptorProvider);
+        return new self($serviceClient, $options, $converter, $interceptorProvider, $pluginRegistry);
+    }
+
+    /**
+     * Get plugins that also implement WorkerPluginInterface for propagation to workers.
+     *
+     * @return list<WorkerPluginInterface>
+     */
+    public function getWorkerPlugins(): array
+    {
+        return $this->pluginRegistry->getPlugins(WorkerPluginInterface::class);
+    }
+
+    /**
+     * Get plugins that also implement ScheduleClientPluginInterface for propagation to schedule clients.
+     *
+     * @return list<ScheduleClientPluginInterface>
+     */
+    public function getScheduleClientPlugins(): array
+    {
+        return $this->pluginRegistry->getPlugins(ScheduleClientPluginInterface::class);
     }
 
     public function getServiceClient(): ServiceClientInterface

src/Interceptor/SimplePipelineProvider.php

@@ -22,14 +22,28 @@ class SimplePipelineProvider implements PipelineProvider
      * @param array<array-key, Interceptor> $interceptors
      */
     public function __construct(
-        private iterable $interceptors = [],
+        private readonly iterable $interceptors = [],
     ) {}
 
+    /**
+     * Create a new provider with additional interceptors prepended.
+     *
+     * @param list<Interceptor> $interceptors Interceptors to prepend before existing ones.
+     */
+    public function withPrependedInterceptors(array $interceptors): self
+    {
+        if ($interceptors === []) {
+            return $this;
+        }
+
+        return new self(\array_merge($interceptors, [...$this->interceptors]));
+    }
+
     public function getPipeline(string $interceptorClass): Pipeline
     {
         return $this->cache[$interceptorClass] ??= Pipeline::prepare(
             \array_filter(
-                $this->interceptors,
+                [...$this->interceptors],
                 static fn(Interceptor $i): bool => $i instanceof $interceptorClass,
             ),
         );

src/Internal/Interceptor/Pipeline.php

@@ -53,7 +53,7 @@ private function __construct(
     /**
      * Make sure that interceptors implement the same interface.
      *
-     * @template T of Interceptor
+     * @template T of object
      *
      * @param iterable<T> $interceptors
      *
@@ -64,6 +64,21 @@ public static function prepare(iterable $interceptors): self
         return new self($interceptors);
     }
 
+    /**
+     * Merge two pipelines into one. Interceptors from the first pipeline run before the second.
+     *
+     * @template T of object
+     *
+     * @param self<T, mixed> $first
+     * @param self<T, mixed> $second
+     *
+     * @return self<T, mixed>
+     */
+    public static function merge(self $first, self $second): self
+    {
+        return new self([...$first->interceptors, ...$second->interceptors]);
+    }
+
     /**
      * @param non-empty-string $method Method name of the all interceptors.
      *

src/Internal/Transport/Router/GetWorkerInfo.php

@@ -18,6 +18,8 @@
 use Temporal\Internal\Declaration\Prototype\WorkflowPrototype;
 use Temporal\Internal\Marshaller\MarshallerInterface;
 use Temporal\Internal\Repository\RepositoryInterface;
+use Temporal\Plugin\PluginInterface;
+use Temporal\Plugin\PluginRegistry;
 use Temporal\Worker\ServiceCredentials;
 use Temporal\Worker\Transport\Command\ServerRequestInterface;
 use Temporal\Worker\WorkerInterface;
@@ -28,6 +30,7 @@ public function __construct(
         private readonly RepositoryInterface $queues,
         private readonly MarshallerInterface $marshaller,
         private readonly ServiceCredentials $credentials,
+        private readonly PluginRegistry $pluginRegistry,
     ) {}
 
     public function handle(ServerRequestInterface $request, array $headers, Deferred $resolver): void
@@ -54,6 +57,10 @@ private function workerToArray(WorkerInterface $worker): array
             'Name' => $activity->getID(),
         ];
 
+        $map = $this->map($this->pluginRegistry->getPlugins(PluginInterface::class), static fn(PluginInterface $plugin): array => [
+            'Name' => $plugin->getName(),
+            'Version' => null,
+        ]);
         return [
             'TaskQueue'  => $worker->getID(),
             'Options'    => $this->marshaller->marshal($worker->getOptions()),
@@ -62,6 +69,7 @@ private function workerToArray(WorkerInterface $worker): array
             // ActivityInfo[]
             'Activities' => $this->map($worker->getActivities(), $activityMap),
             'PhpSdkVersion' => SdkVersion::getSdkVersion(),
+            'Plugins' => $map,
             'Flags' => (object) $this->prepareFlags(),
         ];
     }

src/Internal/Transport/Server.php

@@ -13,8 +13,11 @@
 
 use React\Promise\PromiseInterface;
 use Temporal\Internal\Exception\UndefinedRequestException;
+use Temporal\Internal\Interceptor\Pipeline;
 use Temporal\Internal\Queue\QueueInterface;
 use Temporal\Internal\Transport\Request\UndefinedResponse;
+use Temporal\Plugin\PluginRegistry;
+use Temporal\Plugin\WorkerPluginInterface;
 use Temporal\Worker\Transport\Command\Client\FailedClientResponse;
 use Temporal\Worker\Transport\Command\Client\SuccessClientResponse;
 use Temporal\Worker\Transport\Command\FailureResponseInterface;
@@ -34,14 +37,18 @@ final class Server implements ServerInterface
 
     private \Closure $onMessage;
     private QueueInterface $queue;
+    private Pipeline $interceptors;
 
     /**
      * @psalm-param OnMessageHandler $onMessage
      */
-    public function __construct(QueueInterface $queue, callable $onMessage)
+    public function __construct(QueueInterface $queue, callable $onMessage, PluginRegistry $pluginRegistry)
     {
         $this->queue = $queue;
 
+        $plugins = $pluginRegistry->getPlugins(WorkerPluginInterface::class);
+        $this->interceptors = Pipeline::prepare($plugins);
+
         $this->onMessage($onMessage);
     }
 
@@ -56,7 +63,11 @@ public function onMessage(callable $then): void
     public function dispatch(ServerRequestInterface $request, array $headers): void
     {
         try {
-            $result = ($this->onMessage)($request, $headers);
+            $result = $this->interceptors->with(
+                static fn(callable $handler, ServerRequestInterface $request, array $headers): PromiseInterface => $handler($request, $headers),
+                /** @see WorkerPluginInterface::runWorker() */
+                'runWorker',
+            )($this->onMessage, $request, $headers);
         } catch (\Throwable $e) {
             $this->queue->push(new FailedClientResponse($request->getID(), $e));
 

src/Plugin/AbstractPlugin.php

@@ -0,0 +1,34 @@
+<?php
+
+/**
+ * This file is part of Temporal package.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Temporal\Plugin;
+
+/**
+ * Abstract base class providing no-op defaults for all plugin methods.
+ *
+ * Plugin authors can extend this and override only what they need.
+ */
+abstract class AbstractPlugin implements TemporalPluginInterface
+{
+    use ConnectionPluginTrait;
+    use ClientPluginTrait;
+    use ScheduleClientPluginTrait;
+    use WorkerPluginTrait;
+
+    public function __construct(
+        private readonly string $name,
+    ) {}
+
+    public function getName(): string
+    {
+        return $this->name;
+    }
+}