diff --git a/src/Client/ClientOptions.php b/src/Client/ClientOptions.php
index bfb4d5a46..dff7ea6ee 100644
--- a/src/Client/ClientOptions.php
+++ b/src/Client/ClientOptions.php
@@ -14,6 +14,7 @@
 use JetBrains\PhpStorm\ExpectedValues;
 use JetBrains\PhpStorm\Pure;
 use Temporal\Api\Enums\V1\QueryRejectCondition;
+use Temporal\Common\PayloadLimitOptions;
 use Temporal\Internal\Assert;
 
 /**
@@ -36,6 +37,11 @@ class ClientOptions
     #[ExpectedValues(valuesFromClass: QueryRejectCondition::class)]
     public int $queryRejectionCondition = QueryRejectCondition::QUERY_REJECT_CONDITION_NONE;
 
+    /**
+     * @experimental This API is experimental and may change in the future.
+     */
+    public ?PayloadLimitOptions $payloadLimits = null;
+
     /**
      * ClientOptions constructor.
      */
@@ -44,6 +50,19 @@ public function __construct()
         $this->identity = \sprintf('%d@%s', (string) \getmypid(), (string) \gethostname());
     }
 
+    /**
+     * @experimental This API is experimental and may change in the future.
+     */
+    #[Pure]
+    public function withPayloadLimits(?PayloadLimitOptions $options): self
+    {
+        $self = clone $this;
+
+        $self->payloadLimits = $options;
+
+        return $self;
+    }
+
     /**
      * @param non-empty-string $namespace
      * @return $this
diff --git a/src/Client/GRPC/BaseClient.php b/src/Client/GRPC/BaseClient.php
index a86160945..dd36b1713 100644
--- a/src/Client/GRPC/BaseClient.php
+++ b/src/Client/GRPC/BaseClient.php
@@ -14,14 +14,17 @@
 use Carbon\CarbonInterval;
 use Grpc\BaseStub;
 use Grpc\UnaryCall;
+use Psr\Log\LoggerInterface;
 use Temporal\Client\Common\BackoffThrottler;
 use Temporal\Client\Common\RpcRetryOptions;
 use Temporal\Client\GRPC\Connection\Connection;
 use Temporal\Client\GRPC\Connection\ConnectionInterface;
+use Temporal\Common\PayloadLimitOptions;
 use Temporal\Exception\Client\CanceledException;
 use Temporal\Exception\Client\ServiceClientException;
 use Temporal\Exception\Client\TimeoutException;
 use Temporal\Interceptor\GrpcClientInterceptor;
+use Temporal\Internal\Client\PayloadSizeChecker;
 use Temporal\Internal\Interceptor\Pipeline;
 
 abstract class BaseClient implements GrpcClientInterface
@@ -38,6 +41,8 @@ abstract class BaseClient implements GrpcClientInterface
     private Connection $connection;
     private ContextInterface $context;
     private \Stringable|string $apiKey = '';
+    private ?PayloadSizeChecker $payloadSizeChecker = null;
+    private bool $payloadLimitsConfigured = false;
 
     /**
      * @param BaseStub|\Closure(): BaseStub $serviceClient Service Client or its factory
@@ -163,6 +168,34 @@ public function close(): void
         $this->connection->disconnect();
     }
 
+    /**
+     * @experimental This API is experimental and may change in the future.
+     */
+    final public function withPayloadLimits(PayloadLimitOptions $options, LoggerInterface $logger): static
+    {
+        $clone = clone $this;
+        $clone->payloadSizeChecker = $options->isEnabled()
+            ? new PayloadSizeChecker($options, $logger)
+            : null;
+        $clone->payloadLimitsConfigured = true;
+        return $clone;
+    }
+
+    /**
+     * @internal
+     */
+    final public function withDefaultPayloadLimits(PayloadLimitOptions $options, LoggerInterface $logger): static
+    {
+        if ($this->payloadLimitsConfigured) {
+            return $this;
+        }
+
+        $clone = $this->withPayloadLimits($options, $logger);
+        $clone->payloadLimitsConfigured = false;
+
+        return $clone;
+    }
+
     /**
      * @param null|Pipeline<GrpcClientInterceptor, object> $pipeline
      */
@@ -208,6 +241,8 @@ protected function invoke(string $method, object $arg, ?ContextInterface $ctx =
             ] + $ctx->getMetadata());
         }
 
+        $this->payloadSizeChecker?->check($method, $arg);
+
         return $this->invokePipeline !== null
             ? ($this->invokePipeline)($method, $arg, $ctx)
             : $this->call($method, $arg, $ctx);
diff --git a/src/Client/ScheduleClient.php b/src/Client/ScheduleClient.php
index d7a0ca545..b21b5fbcd 100644
--- a/src/Client/ScheduleClient.php
+++ b/src/Client/ScheduleClient.php
@@ -22,6 +22,9 @@
 use Temporal\Api\Workflowservice\V1\ListSchedulesRequest;
 use Temporal\Client\Common\ClientContextTrait;
 use Temporal\Client\Common\Paginator;
+use Temporal\Client\GRPC\BaseClient;
+use Temporal\Worker\Logger\StderrLogger;
+use Temporal\Common\PayloadLimitOptions;
 use Temporal\Client\GRPC\ServiceClientInterface;
 use Temporal\Client\Schedule\BackfillPeriod;
 use Temporal\Client\Schedule\Info\ScheduleListEntry;
@@ -29,6 +32,7 @@
 use Temporal\Client\Schedule\ScheduleHandle;
 use Temporal\Client\Schedule\ScheduleOptions;
 use Temporal\Common\Uuid;
+use Psr\Log\LoggerInterface;
 use Temporal\DataConverter\DataConverter;
 use Temporal\DataConverter\DataConverterInterface;
 use Temporal\Internal\Mapper\ScheduleMapper;
@@ -57,6 +61,7 @@ public function __construct(
         ?ClientOptions $options = null,
         ?DataConverterInterface $converter = null,
         ?PluginRegistry $pluginRegistry = null,
+        ?LoggerInterface $logger = null,
     ) {
         $this->clientOptions = $options ?? new ClientOptions();
         $this->converter = $converter ?? DataConverter::createDefault();
@@ -88,6 +93,13 @@ public function __construct(
         );
         $this->protoConverter = new ProtoToArrayConverter($this->converter);
 
+        if ($serviceClient instanceof BaseClient) {
+            $serviceClient = $serviceClient->withDefaultPayloadLimits(
+                $this->clientOptions->payloadLimits ?? PayloadLimitOptions::new(),
+                $logger ?? new StderrLogger(),
+            );
+        }
+
         // Set Temporal-Namespace metadata
         $context = $serviceClient->getContext();
         $this->client = $serviceClient->withContext(
@@ -102,8 +114,9 @@ public static function create(
         ?ClientOptions $options = null,
         ?DataConverterInterface $converter = null,
         ?PluginRegistry $pluginRegistry = null,
+        ?LoggerInterface $logger = null,
     ): ScheduleClientInterface {
-        return new self($serviceClient, $options, $converter, $pluginRegistry);
+        return new self($serviceClient, $options, $converter, $pluginRegistry, $logger);
     }
 
     public function createSchedule(
diff --git a/src/Client/WorkflowClient.php b/src/Client/WorkflowClient.php
index b73409dcf..1b1f475dd 100644
--- a/src/Client/WorkflowClient.php
+++ b/src/Client/WorkflowClient.php
@@ -12,6 +12,7 @@
 namespace Temporal\Client;
 
 use Doctrine\Common\Annotations\Reader;
+use Psr\Log\LoggerInterface;
 use JetBrains\PhpStorm\Deprecated;
 use Spiral\Attributes\AnnotationReader;
 use Spiral\Attributes\AttributeReader;
@@ -23,6 +24,9 @@
 use Temporal\Api\Workflowservice\V1\ListWorkflowExecutionsRequest;
 use Temporal\Client\Common\ClientContextTrait;
 use Temporal\Client\Common\Paginator;
+use Temporal\Client\GRPC\BaseClient;
+use Temporal\Worker\Logger\StderrLogger;
+use Temporal\Common\PayloadLimitOptions;
 use Temporal\Client\GRPC\ServiceClientInterface;
 use Temporal\Client\Update\LifecycleStage;
 use Temporal\Client\Update\UpdateHandle;
@@ -81,6 +85,7 @@ public function __construct(
         ?DataConverterInterface $converter = null,
         ?PipelineProvider $interceptorProvider = null,
         ?PluginRegistry $pluginRegistry = null,
+        ?LoggerInterface $logger = null,
     ) {
         $this->pluginRegistry = $pluginRegistry ?? new PluginRegistry();
         $this->clientOptions = $options ?? new ClientOptions();
@@ -116,6 +121,13 @@ public function __construct(
         $this->interceptorPipeline = $provider->getPipeline(WorkflowClientCallsInterceptor::class);
         $this->reader = new WorkflowReader($this->createReader());
 
+        if ($serviceClient instanceof BaseClient) {
+            $serviceClient = $serviceClient->withDefaultPayloadLimits(
+                $this->clientOptions->payloadLimits ?? PayloadLimitOptions::new(),
+                $logger ?? new StderrLogger(),
+            );
+        }
+
         // Set Temporal-Namespace metadata
         $context = $serviceClient->getContext();
         $this->client = $serviceClient->withContext(
@@ -131,8 +143,9 @@ public static function create(
         ?DataConverterInterface $converter = null,
         ?PipelineProvider $interceptorProvider = null,
         ?PluginRegistry $pluginRegistry = null,
+        ?LoggerInterface $logger = null,
     ): self {
-        return new self($serviceClient, $options, $converter, $interceptorProvider, $pluginRegistry);
+        return new self($serviceClient, $options, $converter, $interceptorProvider, $pluginRegistry, $logger);
     }
 
     /**
diff --git a/src/Common/PayloadLimitOptions.php b/src/Common/PayloadLimitOptions.php
new file mode 100644
index 000000000..63c976517
--- /dev/null
+++ b/src/Common/PayloadLimitOptions.php
@@ -0,0 +1,86 @@
+<?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\Common;
+
+/**
+ * @experimental This API is experimental and may change in the future.
+ */
+final class PayloadLimitOptions
+{
+    public const DEFAULT_PAYLOAD_SIZE_WARNING = 512 * 1024;
+    public const DEFAULT_MEMO_SIZE_WARNING = 2 * 1024;
+
+    /**
+     * @param null|positive-int $payloadSizeWarning
+     * @param null|positive-int $memoSizeWarning
+     */
+    public function __construct(
+        public readonly ?int $payloadSizeWarning = self::DEFAULT_PAYLOAD_SIZE_WARNING,
+        public readonly ?int $memoSizeWarning = self::DEFAULT_MEMO_SIZE_WARNING,
+    ) {
+        self::assertPositive($payloadSizeWarning, 'payloadSizeWarning');
+        self::assertPositive($memoSizeWarning, 'memoSizeWarning');
+    }
+
+    /**
+     * @experimental This API is experimental and may change in the future.
+     */
+    public static function new(): self
+    {
+        return new self();
+    }
+
+    /**
+     * @experimental This API is experimental and may change in the future.
+     */
+    public static function disabled(): self
+    {
+        return new self(null, null);
+    }
+
+    /**
+     * @param null|positive-int $bytes
+     *
+     * @experimental This API is experimental and may change in the future.
+     */
+    public function withPayloadSizeWarning(?int $bytes): self
+    {
+        return new self($bytes, $this->memoSizeWarning);
+    }
+
+    /**
+     * @param null|positive-int $bytes
+     *
+     * @experimental This API is experimental and may change in the future.
+     */
+    public function withMemoSizeWarning(?int $bytes): self
+    {
+        return new self($this->payloadSizeWarning, $bytes);
+    }
+
+    /**
+     * @experimental This API is experimental and may change in the future.
+     */
+    public function isEnabled(): bool
+    {
+        return $this->payloadSizeWarning !== null || $this->memoSizeWarning !== null;
+    }
+
+    private static function assertPositive(?int $value, string $name): void
+    {
+        if ($value !== null && $value <= 0) {
+            throw new \InvalidArgumentException(
+                "`$name` must be a positive number of bytes or NULL to disable the warning.",
+            );
+        }
+    }
+}
diff --git a/src/Internal/Client/PayloadSizeChecker.php b/src/Internal/Client/PayloadSizeChecker.php
new file mode 100644
index 000000000..585b1e6ab
--- /dev/null
+++ b/src/Internal/Client/PayloadSizeChecker.php
@@ -0,0 +1,199 @@
+<?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\Internal\Client;
+
+use Psr\Log\LoggerInterface;
+use Temporal\Api\Common\V1\Memo;
+use Temporal\Api\Common\V1\Payloads;
+use Temporal\Api\Failure\V1\Failure;
+use Temporal\Api\Workflowservice\V1\CreateScheduleRequest;
+use Temporal\Api\Workflowservice\V1\ExecuteMultiOperationRequest;
+use Temporal\Api\Workflowservice\V1\QueryWorkflowRequest;
+use Temporal\Api\Workflowservice\V1\RecordActivityTaskHeartbeatByIdRequest;
+use Temporal\Api\Workflowservice\V1\RecordActivityTaskHeartbeatRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCanceledByIdRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCanceledRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCompletedByIdRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCompletedRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskFailedByIdRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskFailedRequest;
+use Temporal\Api\Workflowservice\V1\ResetWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\StartBatchOperationRequest;
+use Temporal\Api\Workflowservice\V1\SignalWithStartWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\SignalWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\StartWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\TerminateWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\UpdateScheduleRequest;
+use Temporal\Api\Workflowservice\V1\UpdateWorkflowExecutionRequest;
+use Google\Protobuf\Internal\Message;
+use Temporal\Common\PayloadLimitOptions;
+
+/**
+ * @internal
+ */
+final class PayloadSizeChecker
+{
+    private const MESSAGE_CODE = 'TMPRL1103';
+    private const MAX_FAILURE_DEPTH = 20;
+
+    public function __construct(
+        private readonly PayloadLimitOptions $limits,
+        private readonly LoggerInterface $logger,
+    ) {}
+
+    /**
+     * @param non-empty-string $method
+     */
+    public function check(string $method, object $request): void
+    {
+        try {
+            $this->inspect($method, $request);
+        } catch (\Throwable) {
+        }
+    }
+
+    private static function sizeOf(?Message $message): int
+    {
+        return $message === null ? 0 : \strlen($message->serializeToString());
+    }
+
+    private function inspect(string $method, object $request): void
+    {
+        switch (true) {
+            case $request instanceof StartWorkflowExecutionRequest:
+                $this->payloads($method, $request->getInput());
+                $this->payloads($method, $request->getLastCompletionResult());
+                $this->memo($method, $request->getMemo());
+                return;
+
+            case $request instanceof SignalWithStartWorkflowExecutionRequest:
+                $this->payloads($method, $request->getInput());
+                $this->payloads($method, $request->getSignalInput());
+                $this->memo($method, $request->getMemo());
+                return;
+
+            case $request instanceof ExecuteMultiOperationRequest:
+                foreach ($request->getOperations() as $operation) {
+                    $nested = $operation->getStartWorkflow() ?? $operation->getUpdateWorkflow();
+                    if ($nested !== null) {
+                        $this->inspect($method, $nested);
+                    }
+                }
+                return;
+
+            case $request instanceof SignalWorkflowExecutionRequest:
+                $this->payloads($method, $request->getInput());
+                return;
+
+            case $request instanceof UpdateWorkflowExecutionRequest:
+                $this->payloads($method, $request->getRequest()?->getInput()?->getArgs());
+                return;
+
+            case $request instanceof QueryWorkflowRequest:
+                $this->payloads($method, $request->getQuery()?->getQueryArgs());
+                return;
+
+            case $request instanceof RespondActivityTaskCompletedRequest:
+            case $request instanceof RespondActivityTaskCompletedByIdRequest:
+                $this->payloads($method, $request->getResult());
+                return;
+
+            case $request instanceof RespondActivityTaskFailedRequest:
+            case $request instanceof RespondActivityTaskFailedByIdRequest:
+                $this->failure($method, $request->getFailure());
+                $this->payloads($method, $request->getLastHeartbeatDetails());
+                return;
+
+            case $request instanceof RespondActivityTaskCanceledRequest:
+            case $request instanceof RespondActivityTaskCanceledByIdRequest:
+            case $request instanceof RecordActivityTaskHeartbeatRequest:
+            case $request instanceof RecordActivityTaskHeartbeatByIdRequest:
+            case $request instanceof TerminateWorkflowExecutionRequest:
+                $this->payloads($method, $request->getDetails());
+                return;
+
+            case $request instanceof CreateScheduleRequest:
+                $action = $request->getSchedule()?->getAction()?->getStartWorkflow();
+                if ($action === null) {
+                    return;
+                }
+
+                $this->warn(
+                    $method,
+                    'payloads',
+                    self::sizeOf($request->getMemo()) + self::sizeOf($action->getInput()),
+                    $this->limits->payloadSizeWarning,
+                );
+                return;
+
+            case $request instanceof StartBatchOperationRequest:
+                $this->payloads($method, $request->getSignalOperation()?->getInput());
+                $this->payloads($method, $request->getTerminationOperation()?->getDetails());
+                return;
+
+            case $request instanceof ResetWorkflowExecutionRequest:
+                foreach ($request->getPostResetOperations() as $operation) {
+                    $this->payloads($method, $operation->getSignalWorkflow()?->getInput());
+                }
+                return;
+
+            case $request instanceof UpdateScheduleRequest:
+                $action = $request->getSchedule()?->getAction()?->getStartWorkflow();
+                $this->payloads($method, $action?->getInput());
+                $this->memo($method, $action?->getMemo());
+                return;
+        }
+    }
+
+    private function failure(string $method, ?Failure $failure, int $depth = 0): void
+    {
+        if ($failure === null || $depth >= self::MAX_FAILURE_DEPTH) {
+            return;
+        }
+
+        $this->payloads($method, $failure->getApplicationFailureInfo()?->getDetails());
+        $this->payloads($method, $failure->getCanceledFailureInfo()?->getDetails());
+        $this->payloads($method, $failure->getTimeoutFailureInfo()?->getLastHeartbeatDetails());
+        $this->payloads($method, $failure->getResetWorkflowFailureInfo()?->getLastHeartbeatDetails());
+
+        $this->failure($method, $failure->getCause(), $depth + 1);
+    }
+
+    private function payloads(string $method, ?Payloads $payloads): void
+    {
+        $this->warn($method, 'payloads', self::sizeOf($payloads), $this->limits->payloadSizeWarning);
+    }
+
+    private function memo(string $method, ?Memo $memo): void
+    {
+        $this->warn($method, 'memo', self::sizeOf($memo), $this->limits->memoSizeWarning);
+    }
+
+    /**
+     * @param non-empty-string $kind
+     */
+    private function warn(string $method, string $kind, int $size, ?int $limit): void
+    {
+        if ($limit === null || $size <= $limit) {
+            return;
+        }
+
+        $this->logger->warning(
+            \sprintf(
+                '[%s] Attempted to upload %s with size that exceeded the warning limit.',
+                self::MESSAGE_CODE,
+                $kind,
+            ),
+            ['method' => $method, 'size' => $size, 'limit' => $limit],
+        );
+    }
+}
diff --git a/src/Worker/Logger/StderrLogger.php b/src/Worker/Logger/StderrLogger.php
index 84bca25cb..a7457d4ba 100644
--- a/src/Worker/Logger/StderrLogger.php
+++ b/src/Worker/Logger/StderrLogger.php
@@ -13,7 +13,12 @@ final class StderrLogger implements LoggerInterface
 
     public function log($level, \Stringable|string $message, array $context = []): void
     {
-        \fwrite(\STDERR, \sprintf(
+        $stream = \defined('STDERR') ? \STDERR : \fopen('php://stderr', 'wb');
+        if ($stream === false) {
+            return;
+        }
+
+        \fwrite($stream, \sprintf(
             "[%s] %s: %s%s\n",
             (new \DateTimeImmutable())->format('Y-m-d H:i:s'),
             $level,
diff --git a/tests/Functional/Client/PayloadWarningTestCase.php b/tests/Functional/Client/PayloadWarningTestCase.php
new file mode 100644
index 000000000..9ac4633f9
--- /dev/null
+++ b/tests/Functional/Client/PayloadWarningTestCase.php
@@ -0,0 +1,84 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Functional\Client;
+
+use Psr\Log\AbstractLogger;
+use Temporal\Client\GRPC\ServiceClient;
+use Temporal\Client\WorkflowClient;
+use Temporal\Common\PayloadLimitOptions;
+use Temporal\Testing\TemporalServer;
+use Temporal\Tests\Workflow\SimpleWorkflow;
+
+/**
+ * @group client
+ * @group functional
+ */
+class PayloadWarningTestCase extends AbstractClient
+{
+    private AbstractLogger $logger;
+
+    public function testOversizedInputIsReportedAndStillAccepted(): void
+    {
+        $client = $this->createClientWithLogger();
+        $input = \str_repeat('x', PayloadLimitOptions::DEFAULT_PAYLOAD_SIZE_WARNING + 1);
+
+        $run = $client->start($client->newWorkflowStub(SimpleWorkflow::class), $input);
+
+        self::assertSame(\strtoupper($input), $run->getResult('string'));
+
+        $records = $this->warnings();
+        self::assertCount(1, $records);
+        self::assertSame('StartWorkflowExecution', $records[0]['context']['method']);
+        self::assertGreaterThan(
+            PayloadLimitOptions::DEFAULT_PAYLOAD_SIZE_WARNING,
+            $records[0]['context']['size'],
+        );
+    }
+
+    public function testInputBelowTheLimitIsNotReported(): void
+    {
+        $client = $this->createClientWithLogger();
+
+        $run = $client->start($client->newWorkflowStub(SimpleWorkflow::class), 'hello');
+
+        self::assertSame('HELLO', $run->getResult('string'));
+        self::assertSame([], $this->warnings());
+    }
+
+    protected function setUp(): void
+    {
+        $this->logger = new class extends AbstractLogger {
+            public array $records = [];
+
+            public function log($level, \Stringable|string $message, array $context = []): void
+            {
+                $this->records[] = ['message' => (string) $message, 'context' => $context];
+            }
+        };
+
+        parent::setUp();
+    }
+
+    private function createClientWithLogger(): WorkflowClient
+    {
+        return new WorkflowClient(
+            ServiceClient::create(TemporalServer::address()),
+            logger: $this->logger,
+        );
+    }
+
+    /**
+     * @return list<array{message: string, context: array}>
+     */
+    private function warnings(): array
+    {
+        return \array_values(
+            \array_filter(
+                $this->logger->records,
+                static fn(array $record): bool => \str_contains($record['message'], '[TMPRL1103]'),
+            ),
+        );
+    }
+}
diff --git a/tests/Unit/Client/GRPC/PayloadLimitsTestCase.php b/tests/Unit/Client/GRPC/PayloadLimitsTestCase.php
new file mode 100644
index 000000000..7e01fd458
--- /dev/null
+++ b/tests/Unit/Client/GRPC/PayloadLimitsTestCase.php
@@ -0,0 +1,329 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\Client\GRPC;
+
+use PHPUnit\Framework\TestCase;
+use Psr\Log\AbstractLogger;
+use Temporal\Api\Common\V1\Payload;
+use Temporal\Api\Common\V1\Payloads;
+use Temporal\Api\Workflowservice\V1\CreateScheduleRequest;
+use Temporal\Api\Workflowservice\V1\CreateScheduleResponse;
+use Temporal\Api\Workflowservice\V1\StartWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\WorkflowServiceClient;
+use Temporal\Client\ClientOptions;
+use Temporal\Client\Schedule\Action\StartWorkflowAction;
+use Temporal\Client\Schedule\Schedule;
+use Temporal\Client\ScheduleClient;
+use Temporal\Client\GRPC\Connection\ConnectionState;
+use Temporal\Client\GRPC\ContextInterface;
+use Temporal\Client\GRPC\ServiceClient;
+use Temporal\Client\GRPC\StatusCode;
+use Temporal\Common\PayloadLimitOptions;
+use Temporal\Client\WorkflowClient;
+use Temporal\Interceptor\GrpcClientInterceptor;
+use Temporal\Internal\Interceptor\Pipeline;
+
+final class PayloadLimitsTestCase extends TestCase
+{
+    /** @var array<array-key, array{string, array}> */
+    private array $records = [];
+
+    public function testWarningIsLoggedOnRpcCall(): void
+    {
+        $client = $this->createClient()->withPayloadLimits(
+            new PayloadLimitOptions(1024, 1024),
+            $this->createLogger(),
+        );
+
+        $client->testCall($this->request(2000));
+
+        self::assertCount(1, $this->records);
+        self::assertStringContainsString('[TMPRL1103]', $this->records[0][0]);
+        self::assertSame('testCall', $this->records[0][1]['method']);
+    }
+
+    public function testNoWarningWithoutLimits(): void
+    {
+        $client = $this->createClient();
+
+        $client->testCall($this->request(2000));
+
+        self::assertSame([], $this->records);
+    }
+
+    public function testLimitsCanBeDisabled(): void
+    {
+        $client = $this->createClient()->withPayloadLimits(
+            PayloadLimitOptions::disabled(),
+            $this->createLogger(),
+        );
+
+        $client->testCall($this->request(2000));
+
+        self::assertSame([], $this->records);
+    }
+
+    public function testWarningsCanBeTurnedOff(): void
+    {
+        $client = $this->createClient()
+            ->withPayloadLimits(new PayloadLimitOptions(1024, 1024), $this->createLogger())
+            ->withPayloadLimits(PayloadLimitOptions::disabled(), $this->createLogger());
+
+        $client->testCall($this->request(2000));
+
+        self::assertSame([], $this->records);
+    }
+
+    public function testWarnsWhenTheInterceptorPipelineIsInstalledFirst(): void
+    {
+        $client = $this->createClient()
+            ->withInterceptorPipeline(Pipeline::prepare([$this->passThroughInterceptor()]))
+            ->withPayloadLimits(new PayloadLimitOptions(1024, 1024), $this->createLogger());
+
+        $client->testCall($this->request(2000));
+
+        self::assertCount(1, $this->records);
+    }
+
+    public function testWarnsWhenTheLimitsAreInstalledFirst(): void
+    {
+        $client = $this->createClient()
+            ->withPayloadLimits(new PayloadLimitOptions(1024, 1024), $this->createLogger())
+            ->withInterceptorPipeline(Pipeline::prepare([$this->passThroughInterceptor()]));
+
+        $client->testCall($this->request(2000));
+
+        self::assertCount(1, $this->records);
+    }
+
+    public function testWorkflowClientEnablesWarnings(): void
+    {
+        $client = new WorkflowClient(
+            $this->createClient(),
+            (new ClientOptions())->withPayloadLimits(new PayloadLimitOptions(1024, 1024)),
+            logger: $this->createLogger(),
+        );
+
+        $serviceClient = $this->serviceClientOf($client);
+        $serviceClient->testCall($this->request(2000));
+
+        self::assertCount(1, $this->records);
+        self::assertStringContainsString('[TMPRL1103]', $this->records[0][0]);
+    }
+
+    public function testWorkflowClientRespectsDisabledLimits(): void
+    {
+        $client = new WorkflowClient(
+            $this->createClient(),
+            (new ClientOptions())->withPayloadLimits(PayloadLimitOptions::disabled()),
+            logger: $this->createLogger(),
+        );
+
+        $serviceClient = $this->serviceClientOf($client);
+        $serviceClient->testCall($this->request(1024 * 1024));
+
+        self::assertSame([], $this->records);
+    }
+
+    public function testWorkflowClientWarnsWithTheDefaultLimits(): void
+    {
+        $client = new WorkflowClient($this->createClient(), logger: $this->createLogger());
+
+        $serviceClient = $this->serviceClientOf($client);
+        $serviceClient->testCall($this->request(PayloadLimitOptions::DEFAULT_PAYLOAD_SIZE_WARNING + 1));
+
+        self::assertCount(1, $this->records);
+        self::assertStringContainsString('[TMPRL1103]', $this->records[0][0]);
+    }
+
+    public function testWorkflowClientKeepsSilentBelowTheDefaultLimits(): void
+    {
+        $client = new WorkflowClient($this->createClient(), logger: $this->createLogger());
+
+        $serviceClient = $this->serviceClientOf($client);
+        $serviceClient->testCall($this->request(1024));
+
+        self::assertSame([], $this->records);
+    }
+
+    public function testRetriedCallIsMeasuredOnce(): void
+    {
+        $client = $this->createClient(failures: 2)->withPayloadLimits(
+            new PayloadLimitOptions(1024, 1024),
+            $this->createLogger(),
+        );
+
+        $client->testCall($this->request(2000));
+
+        self::assertCount(1, $this->records);
+    }
+
+    public function testExplicitLimitsOfTheServiceClientSurviveTheClient(): void
+    {
+        $client = new WorkflowClient(
+            $this->createClient()->withPayloadLimits(PayloadLimitOptions::disabled(), $this->createLogger()),
+        );
+
+        $serviceClient = $this->serviceClientOf($client);
+        $serviceClient->testCall($this->request(1024 * 1024));
+
+        self::assertSame([], $this->records);
+    }
+
+    public function testExplicitLoggerOfTheServiceClientSurvivesTheClient(): void
+    {
+        $serviceClient = $this->createClient()
+            ->withPayloadLimits(new PayloadLimitOptions(1024, 1024), $this->createLogger());
+
+        $client = new WorkflowClient($serviceClient);
+
+        $serviceClient = $this->serviceClientOf($client);
+        $serviceClient->testCall($this->request(2000));
+
+        self::assertCount(1, $this->records);
+    }
+
+    public function testTheSecondClientOnAServiceClientKeepsItsOwnLimits(): void
+    {
+        $first = new WorkflowClient($this->createClient(), logger: $this->createLogger());
+
+        $second = new WorkflowClient(
+            $this->serviceClientOf($first),
+            (new ClientOptions())->withPayloadLimits(PayloadLimitOptions::disabled()),
+            logger: $this->createLogger(),
+        );
+
+        $this->serviceClientOf($second)->testCall($this->request(1024 * 1024));
+
+        self::assertSame([], $this->records);
+    }
+
+    public function testScheduleClientMeasuresTheRequestItBuilds(): void
+    {
+        ScheduleClient::create($this->createClient(), logger: $this->createLogger())
+            ->createSchedule(
+                Schedule::new()->withAction(
+                    StartWorkflowAction::new('Foo')
+                        ->withInput([\str_repeat('x', PayloadLimitOptions::DEFAULT_PAYLOAD_SIZE_WARNING + 1)]),
+                ),
+            );
+
+        self::assertCount(1, $this->records);
+        self::assertStringContainsString('[TMPRL1103]', $this->records[0][0]);
+        self::assertSame('CreateSchedule', $this->records[0][1]['method']);
+    }
+
+    public function testClientIsImmutable(): void
+    {
+        $client = $this->createClient();
+
+        $result = $client->withPayloadLimits(new PayloadLimitOptions(1024, 1024), $this->createLogger());
+
+        self::assertNotSame($client, $result);
+
+        $client->testCall($this->request(2000));
+
+        self::assertSame([], $this->records, 'The original client is not affected.');
+    }
+
+    protected function setUp(): void
+    {
+        $this->records = [];
+        parent::setUp();
+    }
+
+    private function passThroughInterceptor(): GrpcClientInterceptor
+    {
+        return new class implements GrpcClientInterceptor {
+            public function interceptCall(
+                string $method,
+                object $arg,
+                ContextInterface $ctx,
+                callable $next,
+            ): object {
+                return $next($method, $arg, $ctx);
+            }
+        };
+    }
+
+    /**
+     * @return ServiceClient&object{testCall: callable}
+     */
+    private function serviceClientOf(WorkflowClient $client): object
+    {
+        $serviceClient = $client->getServiceClient();
+
+        self::assertInstanceOf($this->createClient()::class, $serviceClient);
+
+        return $serviceClient;
+    }
+
+    private function request(int $size): StartWorkflowExecutionRequest
+    {
+        return (new StartWorkflowExecutionRequest())->setInput(
+            new Payloads(['payloads' => [(new Payload())->setData(\str_repeat('x', $size))]]),
+        );
+    }
+
+    private function createLogger(): AbstractLogger
+    {
+        return new class($this->records) extends AbstractLogger {
+            public function __construct(private array &$records) {}
+
+            public function log($level, \Stringable|string $message, array $context = []): void
+            {
+                $this->records[] = [(string) $message, $context];
+            }
+        };
+    }
+
+    /**
+     * @param int<0, max> $failures Number of retryable failures before the call succeeds.
+     */
+    private function createClient(int $failures = 0): ServiceClient
+    {
+        $stub = static fn() => new class($failures) extends WorkflowServiceClient {
+            public function __construct(private int $failures = 0) {}
+
+            public function getConnectivityState($try_to_connect = false): int
+            {
+                return ConnectionState::Ready->value;
+            }
+
+            public function CreateSchedule(CreateScheduleRequest $argument, $metadata = [], $options = [])
+            {
+                return $this->unaryCall(new CreateScheduleResponse());
+            }
+
+            public function testCall(object $arg, array $metadata = [], array $options = []): object
+            {
+                return $this->unaryCall((object) ['result' => true]);
+            }
+
+            private function unaryCall(object $result): object
+            {
+                $code = $this->failures-- > 0 ? StatusCode::UNAVAILABLE : 0;
+
+                return new class($code, $result) {
+                    public function __construct(private int $code, private object $result) {}
+
+                    public function wait(): array
+                    {
+                        return [$this->result, (object) ['code' => $this->code, 'details' => '']];
+                    }
+                };
+            }
+
+            public function close(): void {}
+        };
+
+        return new class($stub) extends ServiceClient {
+            public function testCall(object $request): mixed
+            {
+                return $this->invoke('testCall', $request, null);
+            }
+        };
+    }
+}
diff --git a/tests/Unit/Client/PayloadSizeCheckerTestCase.php b/tests/Unit/Client/PayloadSizeCheckerTestCase.php
new file mode 100644
index 000000000..4474e21cf
--- /dev/null
+++ b/tests/Unit/Client/PayloadSizeCheckerTestCase.php
@@ -0,0 +1,471 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\Client;
+
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\AbstractLogger;
+use Psr\Log\LogLevel;
+use Temporal\Api\Batch\V1\BatchOperationSignal;
+use Temporal\Api\Batch\V1\BatchOperationTermination;
+use Temporal\Api\Common\V1\Memo;
+use Temporal\Api\Common\V1\Payload;
+use Temporal\Api\Common\V1\Payloads;
+use Temporal\Api\Common\V1\SearchAttributes;
+use Temporal\Api\Failure\V1\ApplicationFailureInfo;
+use Temporal\Api\Failure\V1\CanceledFailureInfo;
+use Temporal\Api\Failure\V1\Failure;
+use Temporal\Api\Failure\V1\ResetWorkflowFailureInfo;
+use Temporal\Api\Failure\V1\TimeoutFailureInfo;
+use Temporal\Api\Query\V1\WorkflowQuery;
+use Temporal\Api\Schedule\V1\Schedule;
+use Temporal\Api\Schedule\V1\ScheduleAction;
+use Temporal\Api\Update\V1\Input;
+use Temporal\Api\Update\V1\Request as UpdateRequest;
+use Temporal\Api\Workflow\V1\NewWorkflowExecutionInfo;
+use Temporal\Api\Workflow\V1\PostResetOperation;
+use Temporal\Api\Workflow\V1\PostResetOperation\SignalWorkflow;
+use Temporal\Api\Workflowservice\V1\CreateScheduleRequest;
+use Temporal\Api\Workflowservice\V1\ExecuteMultiOperationRequest;
+use Temporal\Api\Workflowservice\V1\ExecuteMultiOperationRequest\Operation;
+use Temporal\Api\Workflowservice\V1\QueryWorkflowRequest;
+use Temporal\Api\Workflowservice\V1\RecordActivityTaskHeartbeatByIdRequest;
+use Temporal\Api\Workflowservice\V1\RecordActivityTaskHeartbeatRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCanceledByIdRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCanceledRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCompletedByIdRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCompletedRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskFailedByIdRequest;
+use Temporal\Api\Workflowservice\V1\ResetWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskFailedRequest;
+use Temporal\Api\Workflowservice\V1\StartBatchOperationRequest;
+use Temporal\Api\Workflowservice\V1\SignalWithStartWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\SignalWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\StartWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\TerminateWorkflowExecutionRequest;
+use Temporal\Api\Workflowservice\V1\UpdateScheduleRequest;
+use Temporal\Api\Workflowservice\V1\UpdateWorkflowExecutionRequest;
+use Temporal\Common\PayloadLimitOptions;
+use Temporal\Internal\Client\PayloadSizeChecker;
+use Temporal\Tests\Unit\Client\Stub\LoggerSpy;
+
+final class PayloadSizeCheckerTestCase extends TestCase
+{
+    private LoggerSpy $logger;
+
+    /**
+     * @return iterable<string, array{\Closure(): object, non-empty-string, list<string>}>
+     */
+    public static function oversizedRequests(): iterable
+    {
+        yield 'StartWorkflowExecution: input' => [
+            static fn() => (new StartWorkflowExecutionRequest())->setInput(self::payloads(2000)),
+            'StartWorkflowExecution',
+            ['payloads'],
+        ];
+
+        yield 'StartWorkflowExecution: last completion result' => [
+            static fn() => (new StartWorkflowExecutionRequest())->setLastCompletionResult(self::payloads(2000)),
+            'StartWorkflowExecution',
+            ['payloads'],
+        ];
+
+        yield 'StartWorkflowExecution: memo' => [
+            static fn() => (new StartWorkflowExecutionRequest())->setMemo(self::memo(2000)),
+            'StartWorkflowExecution',
+            ['memo'],
+        ];
+
+        yield 'SignalWithStartWorkflowExecution: input and signal input' => [
+            static fn() => (new SignalWithStartWorkflowExecutionRequest())
+                ->setInput(self::payloads(2000))
+                ->setSignalInput(self::payloads(3000)),
+            'SignalWithStartWorkflowExecution',
+            ['payloads', 'payloads'],
+        ];
+
+        yield 'ExecuteMultiOperation: nested start and update' => [
+            static fn() => (new ExecuteMultiOperationRequest())->setOperations([
+                (new Operation())->setStartWorkflow(
+                    (new StartWorkflowExecutionRequest())->setInput(self::payloads(2000)),
+                ),
+                (new Operation())->setUpdateWorkflow(self::updateRequest(3000)),
+            ]),
+            'ExecuteMultiOperation',
+            ['payloads', 'payloads'],
+        ];
+
+        yield 'SignalWorkflowExecution' => [
+            static fn() => (new SignalWorkflowExecutionRequest())->setInput(self::payloads(2000)),
+            'SignalWorkflowExecution',
+            ['payloads'],
+        ];
+
+        yield 'UpdateWorkflowExecution' => [
+            static fn() => self::updateRequest(2000),
+            'UpdateWorkflowExecution',
+            ['payloads'],
+        ];
+
+        yield 'QueryWorkflow' => [
+            static fn() => (new QueryWorkflowRequest())->setQuery(
+                (new WorkflowQuery())->setQueryArgs(self::payloads(2000)),
+            ),
+            'QueryWorkflow',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskCompleted' => [
+            static fn() => (new RespondActivityTaskCompletedRequest())->setResult(self::payloads(2000)),
+            'RespondActivityTaskCompleted',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskCompletedById' => [
+            static fn() => (new RespondActivityTaskCompletedByIdRequest())->setResult(self::payloads(2000)),
+            'RespondActivityTaskCompletedById',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskFailed: application failure details' => [
+            static fn() => (new RespondActivityTaskFailedRequest())->setFailure(
+                (new Failure())->setApplicationFailureInfo(
+                    (new ApplicationFailureInfo())->setDetails(self::payloads(2000)),
+                ),
+            ),
+            'RespondActivityTaskFailed',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskFailed: canceled failure details' => [
+            static fn() => (new RespondActivityTaskFailedRequest())->setFailure(
+                (new Failure())->setCanceledFailureInfo(
+                    (new CanceledFailureInfo())->setDetails(self::payloads(2000)),
+                ),
+            ),
+            'RespondActivityTaskFailed',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskFailed: timeout heartbeat details' => [
+            static fn() => (new RespondActivityTaskFailedRequest())->setFailure(
+                (new Failure())->setTimeoutFailureInfo(
+                    (new TimeoutFailureInfo())->setLastHeartbeatDetails(self::payloads(2000)),
+                ),
+            ),
+            'RespondActivityTaskFailed',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskFailed: reset workflow heartbeat details' => [
+            static fn() => (new RespondActivityTaskFailedRequest())->setFailure(
+                (new Failure())->setResetWorkflowFailureInfo(
+                    (new ResetWorkflowFailureInfo())->setLastHeartbeatDetails(self::payloads(2000)),
+                ),
+            ),
+            'RespondActivityTaskFailed',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskFailedById: last heartbeat details' => [
+            static fn() => (new RespondActivityTaskFailedByIdRequest())
+                ->setLastHeartbeatDetails(self::payloads(2000)),
+            'RespondActivityTaskFailedById',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskCanceled' => [
+            static fn() => (new RespondActivityTaskCanceledRequest())->setDetails(self::payloads(2000)),
+            'RespondActivityTaskCanceled',
+            ['payloads'],
+        ];
+
+        yield 'RespondActivityTaskCanceledById' => [
+            static fn() => (new RespondActivityTaskCanceledByIdRequest())->setDetails(self::payloads(2000)),
+            'RespondActivityTaskCanceledById',
+            ['payloads'],
+        ];
+
+        yield 'RecordActivityTaskHeartbeat' => [
+            static fn() => (new RecordActivityTaskHeartbeatRequest())->setDetails(self::payloads(2000)),
+            'RecordActivityTaskHeartbeat',
+            ['payloads'],
+        ];
+
+        yield 'RecordActivityTaskHeartbeatById' => [
+            static fn() => (new RecordActivityTaskHeartbeatByIdRequest())->setDetails(self::payloads(2000)),
+            'RecordActivityTaskHeartbeatById',
+            ['payloads'],
+        ];
+
+        yield 'TerminateWorkflowExecution' => [
+            static fn() => (new TerminateWorkflowExecutionRequest())->setDetails(self::payloads(2000)),
+            'TerminateWorkflowExecution',
+            ['payloads'],
+        ];
+
+        yield 'CreateSchedule: request memo and workflow input as one value' => [
+            static fn() => (new CreateScheduleRequest())
+                ->setMemo(self::memo(600))
+                ->setSchedule(self::schedule(600, 0)),
+            'CreateSchedule',
+            ['payloads'],
+        ];
+
+        yield 'CreateSchedule: the memo of the action is not measured on its own' => [
+            static fn() => (new CreateScheduleRequest())->setSchedule(self::schedule(0, 2000)),
+            'CreateSchedule',
+            [],
+        ];
+
+        yield 'StartBatchOperation: signal input' => [
+            static fn() => (new StartBatchOperationRequest())->setSignalOperation(
+                (new BatchOperationSignal())->setInput(self::payloads(2000)),
+            ),
+            'StartBatchOperation',
+            ['payloads'],
+        ];
+
+        yield 'StartBatchOperation: termination details' => [
+            static fn() => (new StartBatchOperationRequest())->setTerminationOperation(
+                (new BatchOperationTermination())->setDetails(self::payloads(2000)),
+            ),
+            'StartBatchOperation',
+            ['payloads'],
+        ];
+
+        yield 'ResetWorkflowExecution: post reset signal input' => [
+            static fn() => (new ResetWorkflowExecutionRequest())->setPostResetOperations([
+                (new PostResetOperation())->setSignalWorkflow(
+                    (new SignalWorkflow())->setInput(self::payloads(2000)),
+                ),
+                (new PostResetOperation())->setSignalWorkflow(
+                    (new SignalWorkflow())->setInput(self::payloads(3000)),
+                ),
+            ]),
+            'ResetWorkflowExecution',
+            ['payloads', 'payloads'],
+        ];
+
+        yield 'UpdateSchedule: workflow input' => [
+            static fn() => (new UpdateScheduleRequest())->setSchedule(self::schedule(2000, 0)),
+            'UpdateSchedule',
+            ['payloads'],
+        ];
+
+        yield 'UpdateSchedule: schedule memo' => [
+            static fn() => (new UpdateScheduleRequest())->setSchedule(self::schedule(0, 2000)),
+            'UpdateSchedule',
+            ['memo'],
+        ];
+    }
+
+    /**
+     * @param \Closure(): object $request
+     * @param non-empty-string $method
+     * @param list<string> $expected Kind of every expected warning, in order.
+     */
+    #[DataProvider('oversizedRequests')]
+    public function testWarnsForEveryOversizedField(\Closure $request, string $method, array $expected): void
+    {
+        $this->check($request(), $method);
+
+        self::assertSame($expected, \array_column($this->logger->records, 'kind'));
+        foreach ($this->logger->records as $record) {
+            self::assertSame(LogLevel::WARNING, $record['level']);
+            self::assertStringContainsString('[TMPRL1103]', $record['message']);
+            self::assertSame($method, $record['context']['method']);
+            self::assertSame(1024, $record['context']['limit']);
+        }
+    }
+
+    public function testExactlyTheLimitIsNotReported(): void
+    {
+        $payloads = self::payloads(100);
+        $size = \strlen($payloads->serializeToString());
+
+        $this->check(
+            (new StartWorkflowExecutionRequest())->setInput($payloads),
+            'StartWorkflowExecution',
+            new PayloadLimitOptions($size, $size),
+        );
+
+        self::assertSame([], $this->logger->records);
+    }
+
+    public function testKeepsSilentBelowTheLimit(): void
+    {
+        $this->check((new StartWorkflowExecutionRequest())->setInput(self::payloads(100)), 'StartWorkflowExecution');
+
+        self::assertSame([], $this->logger->records);
+    }
+
+    public function testMemoUsesItsOwnLimit(): void
+    {
+        $this->check(
+            (new StartWorkflowExecutionRequest())->setMemo(self::memo(2000)),
+            'StartWorkflowExecution',
+            new PayloadLimitOptions(1024 * 1024, 1024),
+        );
+
+        self::assertSame(['memo'], \array_column($this->logger->records, 'kind'));
+    }
+
+    public function testCreateScheduleDoesNotMeasureTheActionMemoOnItsOwn(): void
+    {
+        $this->check(
+            (new CreateScheduleRequest())->setMemo(self::memo(100))->setSchedule(self::schedule(10, 900)),
+            'CreateSchedule',
+            new PayloadLimitOptions(1024, 512),
+        );
+
+        self::assertSame([], \array_column($this->logger->records, 'kind'));
+    }
+
+    public function testMeasuresEveryFailureOfTheChain(): void
+    {
+        $request = (new RespondActivityTaskFailedRequest())->setFailure(
+            self::failure(2000)->setCause(self::failure(3000)),
+        );
+
+        $this->check($request, 'RespondActivityTaskFailed');
+
+        self::assertCount(2, $this->logger->records);
+    }
+
+    public function testFailureChainIsBounded(): void
+    {
+        $failure = self::failure(2000);
+        for ($i = 0; $i < 30; ++$i) {
+            $failure = self::failure(2000)->setCause($failure);
+        }
+
+        $this->check((new RespondActivityTaskFailedRequest())->setFailure($failure), 'RespondActivityTaskFailed');
+
+        self::assertCount(20, $this->logger->records);
+    }
+
+    public function testTheReportedSizeIsTheWireSize(): void
+    {
+        $payloads = self::payloads(2000);
+
+        $this->check((new StartWorkflowExecutionRequest())->setInput($payloads), 'StartWorkflowExecution');
+
+        self::assertSame(\strlen($payloads->serializeToString()), $this->logger->records[0]['context']['size']);
+    }
+
+    public function testSearchAttributesAreNotMeasured(): void
+    {
+        $request = (new StartWorkflowExecutionRequest())
+            ->setSearchAttributes(new SearchAttributes([
+                'indexed_fields' => ['attr' => self::payload(2000)],
+            ]));
+
+        $this->check($request, 'StartWorkflowExecution');
+
+        self::assertSame([], $this->logger->records);
+    }
+
+    public function testDisabledLimitsProduceNoWarning(): void
+    {
+        $request = (new StartWorkflowExecutionRequest())->setInput(self::payloads(1024 * 1024));
+
+        $this->check($request, 'StartWorkflowExecution', PayloadLimitOptions::disabled());
+
+        self::assertSame([], $this->logger->records);
+    }
+
+    public function testOnlyTheMemoWarningCanBeDisabled(): void
+    {
+        $request = (new StartWorkflowExecutionRequest())
+            ->setInput(self::payloads(2000))
+            ->setMemo(self::memo(2000));
+
+        $this->check($request, 'StartWorkflowExecution', new PayloadLimitOptions(1024, null));
+
+        self::assertSame(['payloads'], \array_column($this->logger->records, 'kind'));
+    }
+
+    public function testFailureOfTheCheckDoesNotBreakTheCall(): void
+    {
+        $logger = new class extends AbstractLogger {
+            public function log($level, \Stringable|string $message, array $context = []): void
+            {
+                throw new \RuntimeException('Broken logger');
+            }
+        };
+
+        $checker = new PayloadSizeChecker(new PayloadLimitOptions(1024, 1024), $logger);
+
+        $checker->check(
+            'StartWorkflowExecution',
+            (new StartWorkflowExecutionRequest())->setInput(self::payloads(2000)),
+        );
+
+        self::assertTrue(true);
+    }
+
+    public function testNonProtobufRequestIsIgnored(): void
+    {
+        $this->check(new \stdClass(), 'SomeCall');
+
+        self::assertSame([], $this->logger->records);
+    }
+
+    protected function setUp(): void
+    {
+        $this->logger = new LoggerSpy();
+        parent::setUp();
+    }
+
+    private static function payload(int $size): Payload
+    {
+        return (new Payload())->setData(\str_repeat('x', $size));
+    }
+
+    private static function payloads(int $size): Payloads
+    {
+        return new Payloads(['payloads' => [self::payload($size)]]);
+    }
+
+    private static function memo(int $size): Memo
+    {
+        return new Memo(['fields' => ['key' => self::payload($size)]]);
+    }
+
+    private static function failure(int $size): Failure
+    {
+        return (new Failure())->setApplicationFailureInfo(
+            (new ApplicationFailureInfo())->setDetails(self::payloads($size)),
+        );
+    }
+
+    private static function updateRequest(int $size): UpdateWorkflowExecutionRequest
+    {
+        return (new UpdateWorkflowExecutionRequest())->setRequest(
+            (new UpdateRequest())->setInput((new Input())->setArgs(self::payloads($size))),
+        );
+    }
+
+    private static function schedule(int $inputSize, int $memoSize): Schedule
+    {
+        $action = new NewWorkflowExecutionInfo();
+        if ($inputSize > 0) {
+            $action->setInput(self::payloads($inputSize));
+        }
+
+        if ($memoSize > 0) {
+            $action->setMemo(self::memo($memoSize));
+        }
+
+        return (new Schedule())->setAction((new ScheduleAction())->setStartWorkflow($action));
+    }
+
+    private function check(object $request, string $method, ?PayloadLimitOptions $options = null): void
+    {
+        $checker = new PayloadSizeChecker($options ?? new PayloadLimitOptions(1024, 1024), $this->logger);
+        $checker->check($method, $request);
+    }
+}
diff --git a/tests/Unit/Client/Stub/LoggerSpy.php b/tests/Unit/Client/Stub/LoggerSpy.php
new file mode 100644
index 000000000..35336961e
--- /dev/null
+++ b/tests/Unit/Client/Stub/LoggerSpy.php
@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\Client\Stub;
+
+use Psr\Log\AbstractLogger;
+
+/**
+ * @internal
+ */
+final class LoggerSpy extends AbstractLogger
+{
+    /** @var list<array{level: string, message: string, context: array, kind: string}> */
+    public array $records = [];
+
+    public function log($level, \Stringable|string $message, array $context = []): void
+    {
+        $message = (string) $message;
+
+        $this->records[] = [
+            'level' => (string) $level,
+            'message' => $message,
+            'context' => $context,
+            'kind' => \str_contains($message, ' memo ') ? 'memo' : 'payloads',
+        ];
+    }
+}
diff --git a/tests/Unit/Common/PayloadLimitOptionsTestCase.php b/tests/Unit/Common/PayloadLimitOptionsTestCase.php
new file mode 100644
index 000000000..00d5c63a6
--- /dev/null
+++ b/tests/Unit/Common/PayloadLimitOptionsTestCase.php
@@ -0,0 +1,113 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\Common;
+
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+use Temporal\Client\ClientOptions;
+use Temporal\Common\PayloadLimitOptions;
+
+final class PayloadLimitOptionsTestCase extends TestCase
+{
+    public function testDefaults(): void
+    {
+        $options = PayloadLimitOptions::new();
+
+        self::assertSame(512 * 1024, $options->payloadSizeWarning);
+        self::assertSame(2 * 1024, $options->memoSizeWarning);
+        self::assertTrue($options->isEnabled());
+    }
+
+    public function testUnsetLimitsMeanTheDefaultOnes(): void
+    {
+        self::assertNull((new ClientOptions())->payloadLimits);
+        self::assertNull((new ClientOptions())->withPayloadLimits(PayloadLimitOptions::new())->withPayloadLimits(null)->payloadLimits);
+    }
+
+    public function testWithersAreImmutable(): void
+    {
+        $options = PayloadLimitOptions::new();
+
+        $result = $options->withPayloadSizeWarning(1024)->withMemoSizeWarning(64);
+
+        self::assertNotSame($options, $result);
+        self::assertSame(512 * 1024, $options->payloadSizeWarning);
+        self::assertSame(1024, $result->payloadSizeWarning);
+        self::assertSame(64, $result->memoSizeWarning);
+    }
+
+    public function testNullDisablesSingleWarning(): void
+    {
+        $options = PayloadLimitOptions::new()->withPayloadSizeWarning(null);
+
+        self::assertNull($options->payloadSizeWarning);
+        self::assertTrue($options->isEnabled(), 'Memo warning is still enabled.');
+    }
+
+    public function testNullDisablesAllWarnings(): void
+    {
+        $options = PayloadLimitOptions::new()
+            ->withPayloadSizeWarning(null)
+            ->withMemoSizeWarning(null);
+
+        self::assertFalse($options->isEnabled());
+    }
+
+    public function testDisabledHasNoLimits(): void
+    {
+        $options = PayloadLimitOptions::disabled();
+
+        self::assertNull($options->payloadSizeWarning);
+        self::assertNull($options->memoSizeWarning);
+        self::assertFalse($options->isEnabled());
+    }
+
+    public function testClientOptionsWitherDoesNotMutateTheSource(): void
+    {
+        $options = new ClientOptions();
+        $limits = PayloadLimitOptions::new()->withPayloadSizeWarning(1024);
+
+        $result = $options->withPayloadLimits($limits);
+
+        self::assertNotSame($options, $result);
+        self::assertSame($limits, $result->payloadLimits);
+        self::assertNull($options->payloadLimits);
+    }
+
+
+    public function testClientOptionsDisableLimits(): void
+    {
+        $options = (new ClientOptions())->withPayloadLimits(PayloadLimitOptions::disabled());
+
+        self::assertFalse($options->payloadLimits?->isEnabled());
+    }
+
+    /**
+     * @return iterable<array-key, array{int}>
+     */
+    public static function nonPositiveValues(): iterable
+    {
+        yield [0];
+        yield [-1];
+    }
+
+    #[DataProvider('nonPositiveValues')]
+    public function testPayloadWarningRejectsNonPositive(int $value): void
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('`payloadSizeWarning` must be a positive number of bytes');
+
+        PayloadLimitOptions::new()->withPayloadSizeWarning($value);
+    }
+
+    #[DataProvider('nonPositiveValues')]
+    public function testMemoWarningRejectsNonPositive(int $value): void
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('`memoSizeWarning` must be a positive number of bytes');
+
+        PayloadLimitOptions::new()->withMemoSizeWarning($value);
+    }
+}