diff --git a/docs/portal/build/workers.md b/docs/portal/build/workers.md
index 41e9e64..ca88fbc 100644
--- a/docs/portal/build/workers.md
+++ b/docs/portal/build/workers.md
@@ -99,6 +99,14 @@ An activity option such as `['queue' => 'payments-v2']` selects the queue of a n
 
 ### Poll timeouts and early disconnects
 
+When the Server advertises `task_delivery`, the managed worker negotiates
+offer/accept delivery. Receiving an offer does not reserve a task or start an
+activity attempt. The SDK accepts the selected offer before running a handler.
+If that response is lost, it retries the same packet and poll identity until
+the Server confirms the claim, expiry, or loss of registration authority.
+An older node returning an error cannot switch that poll to legacy delivery.
+Servers without this capability continue using the existing poll protocol.
+
 With the default cURL transport, supporting Servers send progress during idle
 worker polls. Progress keeps an idle connection active while the complete
 request remains bounded to the requested wait plus five seconds. Ordinary polls
diff --git a/docs/protocol.md b/docs/protocol.md
index 6ce822c..7814f0c 100644
--- a/docs/protocol.md
+++ b/docs/protocol.md
@@ -198,6 +198,16 @@ may have committed a claim before the connection disappeared, so `task=null`
 does not establish that no lease exists. This does not make arbitrary database
 errors or unrelated structured HTTP 500/503 responses retryable.
 
+With negotiated `task_delivery`, worker polls use `/offers`. An offer alone
+grants no execution authority. Positive acceptance includes the original
+registration token and selected offer packet; its request body remains fixed
+through uncertain replies. Managed workers retain this state across client
+clones and honor the normal retry backoff. If using the low-level
+`poll*TaskResponse()` methods, catch `TaskDeliveryUncertain` and retry with its
+`details['poll_request_id']` on the same Client. An ordinary empty response to
+an acceptance does not prove that no claim committed. A confirmed offer expiry
+permits a new poll; lost registration authority stops that worker.
+
 Before replaying or completing a workflow task, the managed worker requires a
 successful lease-renewal response whose task ID, attempt, and lease owner match
 the current claim. A typed `renewed=false`, `retryable=true` response is retried
diff --git a/src/Client.php b/src/Client.php
index d4d3aa4..cf5ba4a 100644
--- a/src/Client.php
+++ b/src/Client.php
@@ -62,6 +62,9 @@
 use DurableWorkflow\Worker\WorkerSessionOptions;
 use DurableWorkflow\Worker\ActivityLeaseBinding;
 use DurableWorkflow\Worker\ActivityLeaseProcess;
+use DurableWorkflow\Worker\TaskDelivery;
+use DurableWorkflow\Worker\TaskDeliveryAttempt;
+use DurableWorkflow\Transport\PollTransport;
 use InvalidArgumentException;
 
 /** Synchronous control-plane and worker-plane client for the standalone server. */
@@ -81,6 +84,7 @@ final class Client implements WorkflowClientInterface
     private FailureReportLimiter $failureReports;
     private bool $boundedWorkerRequests = false;
     private ActivityLeaseBinding $activityLeaseBinding;
+    private TaskDelivery $taskDelivery;
     private ?string $activityLeaseBootstrap = null;
     private ?ExecutionTelemetry $executionTelemetry = null;
     private bool $observabilityContextSupported = false;
@@ -143,6 +147,7 @@ public function __construct(
         $this->payloadUploads = new RuntimePayloadUploads($this->transport, $this->baseUri);
         $this->failureReports = new FailureReportLimiter(new FailureReportLimits(), $this->transport);
         $this->activityLeaseBinding = new ActivityLeaseBinding();
+        $this->taskDelivery = new TaskDelivery();
     }
 
     /** Narrow the encoded failure-report request budget for an application transport. */
@@ -1576,6 +1581,7 @@ public function registerWorker(
             'memo_upserts',
             'typed_search_attributes',
             'durable_selection',
+            TaskDelivery::CAPABILITY,
         ],
         int $maxConcurrentWorkflowTasks = 1,
         int $maxConcurrentActivityTasks = 1,
@@ -1585,6 +1591,10 @@ public function registerWorker(
         int $maxConcurrentWorkerSessions = 10,
         ?array $workflowDefinitionFingerprints = null,
     ): array {
+        if ($this->workerProtocolVersion !== '1.20') {
+            $capabilities = array_values(array_filter($capabilities,
+                static fn (string $capability): bool => $capability !== TaskDelivery::CAPABILITY));
+        }
         if (in_array('message_streams', $capabilities, true) && !Version::supportsMessageStreams()) {
             throw new InvalidArgumentException('Message streams require worker protocol 1.15 or newer.');
         }
@@ -1606,7 +1616,7 @@ public function registerWorker(
             }
         }
 
-        return $this->worker('POST', '/worker/register', $this->withoutNulls([
+        $response = $this->worker('POST', '/worker/register', $this->withoutNulls([
             'worker_id' => $workerId,
             'task_queue' => $taskQueue,
             'runtime' => 'php',
@@ -1621,7 +1631,11 @@ public function registerWorker(
             'max_concurrent_activity_tasks' => $maxConcurrentActivityTasks,
             'max_concurrent_worker_sessions' => $maxConcurrentWorkerSessions,
             'build_id' => $buildId,
+            'process_metrics' => ['process_id' => getmypid(), 'process_uptime_seconds' => 0],
         ]));
+        $this->taskDelivery->registered($workerId, $taskQueue, $this->workerProtocolVersion, $capabilities, $response);
+
+        return $response;
     }
 
     /** @return array<string, mixed> */
@@ -1678,6 +1692,7 @@ public function heartbeatWorker(string $workerId, array $taskSlots = [], ?Reques
     public function deregisterWorker(string $workerId): void
     {
         $this->control('DELETE', '/workers/'.$this->segment($workerId));
+        $this->taskDelivery->forget($workerId);
     }
 
     /** @return array{worker_id: string, outcome: string, recovered_workflow_task_count: int} */
@@ -1702,6 +1717,8 @@ public function deregisterWorkerRegistration(string $workerId, ?string $registra
                 'invalid_worker_deregistration_receipt', $response);
         }
 
+        $this->taskDelivery->forget($workerId);
+
         /** @var array{worker_id: string, outcome: string, recovered_workflow_task_count: int} $response */
         return $response;
     }
@@ -2592,7 +2609,12 @@ private function pollTaskResponse(string $path, array $body): array
         $retryTransientFailure = true;
         while (true) {
             try {
-                return $this->worker('POST', $path, $body);
+                return $this->taskDelivery->poll($path, $body,
+                    fn (string $wirePath, array $wireBody, ?TaskDeliveryAttempt $selected): array =>
+                        $this->request('POST', $wirePath, true, $wireBody,
+                            responseValidator: static fn (array $response) =>
+                                TaskDelivery::assertResponse($response, $wireBody, $wirePath, $selected)))
+                    ?? $this->worker('POST', $path, $body);
             } catch (ServerException $exception) {
                 if ($retryTransientFailure && $exception->isTransientConnectionFailure()) {
                     $retryTransientFailure = false;
@@ -2659,7 +2681,7 @@ private function sendTransportRequest(string $method, string $path, bool $worker
     }
 
     /** @return array<string, string> */
-    private function requestHeaders(bool $worker): array
+    private function requestHeaders(bool $worker, string $path = ''): array
     {
         $headers = [
             'Accept' => 'application/json',
@@ -2671,6 +2693,9 @@ private function requestHeaders(bool $worker): array
         if ($this->authentication !== null) {
             $headers = array_merge($headers, $this->authentication->headers($worker));
         }
+        if ($worker && PollTransport::taskKind('POST', $path) !== null && str_ends_with($path, '/offers')) {
+            $headers[TaskDelivery::HEADER] = TaskDelivery::VERSION;
+        }
 
         return $headers;
     }
@@ -2678,6 +2703,7 @@ private function requestHeaders(bool $worker): array
     /**
      * @param array<string, mixed>|null $body
      * @param (\Closure(array<string, mixed>): list<list<int|string>>)|null $payloadPaths
+     * @param (\Closure(array<string, mixed>): void)|null $responseValidator
      * @return array<string, mixed>
      */
     private function request(
@@ -2689,11 +2715,13 @@ private function request(
         ?RequestBudget $budget = null,
         ?\Closure $payloadPaths = null,
         ?string $requestId = null,
+        ?\Closure $responseValidator = null,
     ): array {
-        $headers = $this->requestHeaders($worker);
+        $headers = $this->requestHeaders($worker, $path);
+        $logicalPath = PollTransport::logicalPath($path);
         try {
             $callerBudget = $budget;
-            $poll = str_ends_with($path, '/poll') && is_int($body['timeout_seconds'] ?? null);
+            $poll = PollTransport::taskKind($method, $path) !== null && is_int($body['timeout_seconds'] ?? null);
             $pollSeconds = $poll ? max(0, min(60, $body['timeout_seconds'])) : 0;
             $budget ??= $worker && $this->boundedWorkerRequests ? new RequestBudget($pollSeconds + 5) : null;
             if ($body !== null) {
@@ -2703,7 +2731,7 @@ private function request(
                 $body = $this->failureReports->request($path, $body);
             }
             $activityLease = $worker ? $this->activityLeaseBinding->current() : null;
-            if ($activityLease !== null && $activityLease->governs($path)) {
+            if ($activityLease !== null && $activityLease->governs($logicalPath)) {
                 // Uploading an external result is not publication authority.
                 // Check again before its final Worker POST and bound the send
                 // by the latest acknowledged grant, not an old cached lease.
@@ -2713,21 +2741,27 @@ private function request(
             $response = $this->sendTransportRequest($method, $path, $worker, $headers, $body, $budget);
 
             $budget?->remainingSeconds();
+            if ($responseValidator !== null) {
+                if (!is_array($response) || array_is_list($response)) {
+                    throw new ServerException('The task-delivery response is invalid.', 200, 'invalid_task_delivery_response');
+                }
+                $responseValidator($response);
+            }
             // Control I/O and supervised payload I/O have separate bounds.
             // Explicit caller/cleanup budgets still cover their entire request.
             $responseBudget = $budget !== null && $poll && $callerBudget === null ? new RequestBudget(5) : $budget;
-            $canStreamBounded = $worker && ($this->boundedWorkerRequests || $path === '/worker/query-tasks/poll')
+            $canStreamBounded = $worker && ($this->boundedWorkerRequests || $logicalPath === '/worker/query-tasks/poll')
                 && $this->transport instanceof \DurableWorkflow\Transport\BoundedProgressPayloadTransport
                 && $this->transport instanceof BoundedTransport && $this->transport->supportsBoundedRequests();
             $observerBudget = $canStreamBounded && $callerBudget === null ? null : $responseBudget;
             $progress = $worker && $this->workerPayloadObserver !== null && is_array($response) && !array_is_list($response)
-                ? ($this->workerPayloadObserver)($path, $body, $response, $observerBudget) : null;
+                ? ($this->workerPayloadObserver)($logicalPath, $body, $response, $observerBudget) : null;
             $streamBounded = $canStreamBounded && $progress !== null;
             if ($streamBounded && $callerBudget === null) {
                 $responseBudget = null;
             }
             $result = is_array($response) && !array_is_list($response)
-                ? (new RuntimePayloads($this->transport, $this->baseUri, $headers, $this->maxExternalPayloadBytes, $responseBudget, $progress, $streamBounded))->response($response, $path, $worker, $payloadPaths?->__invoke($response))
+                ? (new RuntimePayloads($this->transport, $this->baseUri, $headers, $this->maxExternalPayloadBytes, $responseBudget, $progress, $streamBounded))->response($response, $logicalPath, $worker, $payloadPaths?->__invoke($response))
                 : [];
             $responseBudget?->remainingSeconds();
 
diff --git a/src/Exception/TaskDeliveryUncertain.php b/src/Exception/TaskDeliveryUncertain.php
new file mode 100644
index 0000000..035f351
--- /dev/null
+++ b/src/Exception/TaskDeliveryUncertain.php
@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Exception;
+
+use Throwable;
+
+/** Retry only the original logical poll; this exception grants no execution. */
+final class TaskDeliveryUncertain extends ServerException
+{
+    public function __construct(string $pollRequestId, int $status, ?Throwable $previous = null)
+    {
+        parent::__construct('Task delivery is not confirmed. Retry the same poll and selected offer.',
+            $status, 'task_delivery_outcome_unknown', [
+                'task' => null, 'poll_request_id' => $pollRequestId,
+                'retry_same_poll_request_id' => true,
+                ...($previous instanceof ServerException && is_int($previous->details['retry_after_seconds'] ?? null)
+                    && $previous->details['retry_after_seconds'] > 0
+                    ? ['retry_after_seconds' => $previous->details['retry_after_seconds']] : []),
+            ], $previous);
+    }
+}
diff --git a/src/Transport/PollTransport.php b/src/Transport/PollTransport.php
index 4aaf208..3f71944 100644
--- a/src/Transport/PollTransport.php
+++ b/src/Transport/PollTransport.php
@@ -19,7 +19,8 @@ public static function taskKind(string $method, string $uri): ?string
         $path = parse_url($uri, PHP_URL_PATH);
         if (strtoupper($method) === 'POST' && is_string($path)) {
             foreach (self::KINDS as $kind) {
-                if (str_ends_with($path, '/worker/'.$kind.'-tasks/poll')) {
+                if (str_ends_with($path, '/worker/'.$kind.'-tasks/poll')
+                    || str_ends_with($path, '/worker/'.$kind.'-tasks/offers')) {
                     return $kind;
                 }
             }
@@ -28,6 +29,13 @@ public static function taskKind(string $method, string $uri): ?string
         return null;
     }
 
+    /** Payload observers and claim supervision use the original logical poll. */
+    public static function logicalPath(string $path): string
+    {
+        return self::taskKind('POST', $path) !== null && str_ends_with($path, '/offers')
+            ? substr($path, 0, -7).'/poll' : $path;
+    }
+
     /** @return array{status: int, headers: array<string, string>, body: array<string, mixed>} */
     public static function decode(string $rawBody, int $outerStatus): array
     {
diff --git a/src/Worker.php b/src/Worker.php
index 57192c9..9299526 100644
--- a/src/Worker.php
+++ b/src/Worker.php
@@ -656,6 +656,7 @@ private function registerWithRetry(): ?array
                         'worker_sessions',
                         'sticky_execution',
                         'cross_kind_poll_wake',
+                        ...($this->client->workerProtocolVersion === '1.20' ? [\DurableWorkflow\Worker\TaskDelivery::CAPABILITY] : []),
                         ...($this->enableCooperativeCancellation ? ['cooperative_cancellation', 'nexus_cooperative_cancellation'] : []),
                         ...($this->enablePreparedLocalActivities ? ['prepared_local_activities'] : []),
                         ...($this->preparedLocalActivityGroupsSupported ? ['prepared_local_activity_groups'] : []),
@@ -929,6 +930,7 @@ private function pollWithRetry(string $taskKind, \Closure $poll, int $requestedT
                 }
 
                 if (!$exception->isTransientConnectionFailure() && !$exception->isTransientUpstreamFailure()
+                    && !$exception instanceof \DurableWorkflow\Exception\TaskDeliveryUncertain
                     && !PollResponse::isTransientPayloadFailure($exception)
                     && !$databaseUnavailable && !$storageAdmission) {
                     $requestId = 'php-'.$taskKind.'-poll-'.bin2hex(random_bytes(16));
diff --git a/src/Worker/PollResponse.php b/src/Worker/PollResponse.php
index e09477d..62a4c26 100644
--- a/src/Worker/PollResponse.php
+++ b/src/Worker/PollResponse.php
@@ -6,6 +6,7 @@
 
 use DurableWorkflow\Exception\ExternalPayloadException;
 use DurableWorkflow\Exception\ServerException;
+use DurableWorkflow\Exception\TaskDeliveryUncertain;
 
 /** Classifies full worker poll response envelopes using protocol fields. */
 final class PollResponse
@@ -23,6 +24,7 @@ final class PollResponse
         'worker_heartbeat_stale',
         'worker_draining',
         'worker_stopped',
+        'worker_registration_lost_authority',
     ];
 
     /** @param array<string, mixed> $response */
@@ -42,7 +44,8 @@ public static function isTerminal(array $response): bool
      */
     public static function isTransientFailure(ServerException $exception): bool
     {
-        if ($exception->isTransientConnectionFailure() || $exception->isTransientUpstreamFailure()
+        if ($exception instanceof TaskDeliveryUncertain
+            || $exception->isTransientConnectionFailure() || $exception->isTransientUpstreamFailure()
             || self::isTransientPayloadFailure($exception)) {
             return true;
         }
diff --git a/src/Worker/TaskDelivery.php b/src/Worker/TaskDelivery.php
new file mode 100644
index 0000000..2228716
--- /dev/null
+++ b/src/Worker/TaskDelivery.php
@@ -0,0 +1,187 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Worker;
+
+use Closure;
+use DurableWorkflow\Exception\ServerException;
+use DurableWorkflow\Exception\TaskDeliveryUncertain;
+use DurableWorkflow\Transport\PollTransport;
+
+/** Per-Client registration and unresolved delivery state, shared by its clones. */
+final class TaskDelivery
+{
+    public const CAPABILITY = 'task_delivery_offers';
+    public const SCHEMA = 'durable-workflow.task-delivery.v1';
+    public const HEADER = 'X-Durable-Workflow-Task-Delivery';
+    public const VERSION = 'offer-accept.v1';
+
+    /** @var array<string, array{queue: string, token: string}> */
+    private array $registrations = [];
+    /** @var array<string, TaskDeliveryAttempt> */
+    private array $pending = [];
+
+    /**
+     * @param list<string> $capabilities
+     * @param array<string, mixed> $response
+     */
+    public function registered(string $workerId, string $queue, string $protocol, array $capabilities, array $response): void
+    {
+        $server = $response['server_capabilities'] ?? [];
+        $feature = is_array($server) ? ($server['task_delivery'] ?? null) : null;
+        if ($protocol !== '1.20' || !in_array(self::CAPABILITY, $capabilities, true)
+            || $feature === null || (is_array($feature) && ($feature['supported'] ?? null) === false)) {
+            foreach ($this->pending as $attempt) {
+                if ($attempt->workerId === $workerId) {
+                    throw new ServerException('An unresolved delivery cannot change transport modes.', 201, 'invalid_task_delivery_capability');
+                }
+            }
+            $this->forget($workerId);
+            return;
+        }
+        $token = $response['registration_token'] ?? null;
+        if (!is_array($feature) || array_is_list($feature) || ($feature['supported'] ?? null) !== true
+            || ($feature['schema'] ?? null) !== self::SCHEMA || ($feature['version'] ?? null) !== self::VERSION
+            || ($feature['header'] ?? null) !== self::HEADER || ($feature['worker_capability'] ?? null) !== self::CAPABILITY
+            || ($feature['minimum_protocol_version'] ?? null) !== '1.20' || ($feature['endpoint_suffix'] ?? null) !== '/offers'
+            || ($feature['unaccepted_offer_reserves_task'] ?? null) !== false
+            || !is_int($feature['acceptance_timeout_seconds'] ?? null)
+            || $feature['acceptance_timeout_seconds'] < 1 || $feature['acceptance_timeout_seconds'] > 10
+            || !is_string($token) || preg_match('/^[a-f0-9]{32}$/D', $token) !== 1) {
+            throw new ServerException('The task-delivery registration contract is invalid.', 201, 'invalid_task_delivery_capability');
+        }
+        if (($this->registrations[$workerId] ?? null) !== ['queue' => $queue, 'token' => $token]) {
+            $this->forget($workerId);
+        }
+        $this->registrations[$workerId] = ['queue' => $queue, 'token' => $token];
+    }
+
+    /**
+     * @param array<string, mixed> $body
+     * @param Closure(string, array<string, mixed>, ?TaskDeliveryAttempt): array<string, mixed> $send
+     * @return array<string, mixed>|null Null means negotiation never selected this mode.
+     */
+    public function poll(string $path, array $body, Closure $send): ?array
+    {
+        $workerId = (string) $body['worker_id'];
+        $registration = $this->registrations[$workerId] ?? null;
+        if ($registration === null || $registration['queue'] !== ($body['task_queue'] ?? null)) {
+            return null;
+        }
+        $key = hash('sha256', json_encode([$path, $workerId, $body['task_queue'], $body['poll_request_id']], JSON_THROW_ON_ERROR));
+        $selected = $this->pending[$key] ?? null;
+        try {
+            if ($selected === null) {
+                $wirePath = substr($path, 0, -5).'/offers';
+                $body['registration_token'] = $registration['token'];
+                $response = $send($wirePath, $body, null);
+                if (!array_key_exists('task_offer', $response)) {
+                    return $response;
+                }
+                $candidate = TaskDeliveryAttempt::fromResponse($wirePath, $body, $response,
+                    (string) PollTransport::taskKind('POST', $path));
+                // Concurrent callers for one logical poll select only its first packet.
+                $selected = $this->pending[$key] ??= $candidate;
+            }
+            $response = $send($selected->path, $selected->requestBody(), $selected);
+            if ($response['task'] === null && !PollResponse::isTerminal($response)) {
+                throw new TaskDeliveryUncertain((string) $body['poll_request_id'], 200);
+            }
+        } catch (ServerException $exception) {
+            $response = $this->refusedResponse($exception, $workerId);
+            if ($response === null) {
+                if ($this->isUncertain($exception, $selected)) {
+                    throw new TaskDeliveryUncertain((string) $body['poll_request_id'], $exception->status, $exception);
+                }
+                throw $exception;
+            }
+        }
+        $this->resolved($key, $selected);
+
+        return $response;
+    }
+
+    /**
+     * Validate before payload hydration can initialize a claim supervisor.
+     * @param array<string, mixed> $response
+     * @param array<string, mixed> $body
+     */
+    public static function assertResponse(array $response, array $body, string $path, ?TaskDeliveryAttempt $selected): void
+    {
+        $task = $response['task'] ?? null;
+        $kind = PollTransport::taskKind('POST', $path);
+        $idField = match ($kind) {
+            'query' => 'query_task_id', 'update-validation' => 'update_validation_task_id', default => 'task_id',
+        };
+        if (!array_key_exists('task', $response)
+            || (($selected !== null || $task !== null) && array_key_exists('task_offer', $response))
+            || (($response['poll_status'] ?? null) === 'offered' && !array_key_exists('task_offer', $response))
+            || ($task !== null && (!is_array($task) || array_is_list($task)
+                || !is_string($task[$idField] ?? null) || $task[$idField] === ''
+                || ($task['lease_owner'] ?? null) !== $body['worker_id']
+                || ($selected !== null && $task[$idField] !== $selected->taskId)
+                || (isset($task['task_kind']) && $task['task_kind'] !== str_replace('-', '_', (string) $kind))))) {
+            throw new ServerException('The task-delivery response is invalid.', 200, 'invalid_task_delivery_response');
+        }
+    }
+
+    public function forget(string $workerId): void
+    {
+        unset($this->registrations[$workerId]);
+        foreach ($this->pending as $key => $attempt) {
+            if ($attempt->workerId === $workerId) {
+                unset($this->pending[$key]);
+            }
+        }
+    }
+
+    /** @return array<string, int> */
+    public function __debugInfo(): array
+    {
+        return ['registered_workers' => count($this->registrations), 'unresolved_deliveries' => count($this->pending)];
+    }
+
+    private function definitiveRefusal(ServerException $exception): bool
+    {
+        $response = $exception->details;
+
+        return $exception->status === 409 && $response !== null && !array_is_list($response)
+            && array_key_exists('task', $response) && $response['task'] === null
+            && ($response['reason'] ?? null) === $exception->reason
+            && (in_array($exception->reason, ['task_offer_expired', 'worker_registration_lost_authority'], true)
+                || PollResponse::isTerminal($response));
+    }
+
+    /** @return array<string, mixed>|null */
+    private function refusedResponse(ServerException $exception, string $workerId): ?array
+    {
+        if (!$this->definitiveRefusal($exception)) {
+            return null;
+        }
+        $response = $exception->details ?? [];
+        if ($exception->reason === 'task_offer_expired') {
+            $response['poll_status'] = 'empty';
+        } else {
+            $this->forget($workerId);
+        }
+
+        return $response;
+    }
+
+    private function isUncertain(ServerException $exception, ?TaskDeliveryAttempt $selected): bool
+    {
+        return !$exception instanceof TaskDeliveryUncertain && !PollResponse::isTransientPayloadFailure($exception)
+            && (($selected !== null && ($exception->status >= 500 || $exception->status === 409 || $exception->status === 429
+                    || $exception->reason === 'invalid_task_delivery_response'
+                    || in_array($exception->status, [0, 404, 405], true)))
+                || ($selected === null && in_array($exception->status, [404, 405], true)));
+    }
+
+    private function resolved(string $key, ?TaskDeliveryAttempt $selected): void
+    {
+        if (($this->pending[$key] ?? null) === $selected) {
+            unset($this->pending[$key]);
+        }
+    }
+}
diff --git a/src/Worker/TaskDeliveryAttempt.php b/src/Worker/TaskDeliveryAttempt.php
new file mode 100644
index 0000000..64cff2c
--- /dev/null
+++ b/src/Worker/TaskDeliveryAttempt.php
@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Worker;
+
+use DurableWorkflow\Exception\ServerException;
+
+/** The first selected packet remains immutable through uncertain replies. */
+final class TaskDeliveryAttempt
+{
+    /** @param array<string, mixed> $body */
+    private function __construct(
+        public readonly string $workerId,
+        public readonly string $path,
+        public readonly string $taskId,
+        private readonly array $body,
+    ) {
+    }
+
+    /**
+     * @param array<string, mixed> $body
+     * @param array<string, mixed> $response
+     */
+    public static function fromResponse(string $path, array $body, array $response, string $kind): self
+    {
+        $offer = $response['task_offer'] ?? null;
+        if (!array_key_exists('task', $response) || $response['task'] !== null || ($response['poll_status'] ?? null) !== 'offered'
+            || !is_array($offer) || array_is_list($offer)
+            || ($offer['schema'] ?? null) !== TaskDelivery::SCHEMA
+            || ($offer['task_kind'] ?? null) !== str_replace('-', '_', $kind)
+            || !is_string($offer['task_id'] ?? null) || $offer['task_id'] === ''
+            || !is_string($offer['token'] ?? null) || $offer['token'] === '' || strlen($offer['token']) > 8192
+            || !is_int($offer['expires_at_ms'] ?? null) || $offer['expires_at_ms'] < 1) {
+            throw new ServerException('The server returned an invalid task offer.', 200, 'invalid_task_delivery_offer');
+        }
+
+        return new self((string) $body['worker_id'], $path, $offer['task_id'],
+            array_replace($body, ['task_offer_token' => $offer['token'], 'timeout_seconds' => 0]));
+    }
+
+    /** @return array<string, mixed> */
+    public function requestBody(): array
+    {
+        return $this->body;
+    }
+
+    /** @return array<string, string> */
+    public function __debugInfo(): array
+    {
+        return ['worker_id' => $this->workerId, 'path' => $this->path, 'task_id' => $this->taskId];
+    }
+}
diff --git a/tests/ClientContractTest.php b/tests/ClientContractTest.php
index 439a9c9..eef3ec1 100644
--- a/tests/ClientContractTest.php
+++ b/tests/ClientContractTest.php
@@ -119,6 +119,7 @@ public function testDefaultWorkerRegistrationAdvertisesTheCompletePhpCapabilityS
                     'memo_upserts',
                     'typed_search_attributes',
                     'durable_selection',
+                    'task_delivery_offers',
                 ],
                 'capability_manifest' => [
                     'local_activities' => [
@@ -140,6 +141,7 @@ public function testDefaultWorkerRegistrationAdvertisesTheCompletePhpCapabilityS
                 'max_concurrent_workflow_tasks' => 1,
                 'max_concurrent_activity_tasks' => 1,
                 'max_concurrent_worker_sessions' => 10,
+                'process_metrics' => ['process_id' => getmypid(), 'process_uptime_seconds' => 0],
             ],
         ], $transport->requests[0]);
     }
diff --git a/tests/WorkerAuthoringTest.php b/tests/WorkerAuthoringTest.php
index b8918bf..930c2af 100644
--- a/tests/WorkerAuthoringTest.php
+++ b/tests/WorkerAuthoringTest.php
@@ -182,6 +182,7 @@ public function testHighLevelWorkerAdvertisesTheCompleteManagedCapabilitySet():
                     'worker_sessions',
                     'sticky_execution',
                     'cross_kind_poll_wake',
+                    'task_delivery_offers',
                 ],
                 'capability_manifest' => [
                     'local_activities' => [
@@ -204,6 +205,7 @@ public function testHighLevelWorkerAdvertisesTheCompleteManagedCapabilitySet():
                 'max_concurrent_activity_tasks' => 1,
                 'max_concurrent_worker_sessions' => 10,
                 'build_id' => 'release-a',
+                'process_metrics' => ['process_id' => getmypid(), 'process_uptime_seconds' => 0],
             ],
         ], $transport->requests[0]);
     }
diff --git a/tests/WorkerPollTransportTest.php b/tests/WorkerPollTransportTest.php
index eda12b1..2ecf2df 100644
--- a/tests/WorkerPollTransportTest.php
+++ b/tests/WorkerPollTransportTest.php
@@ -51,6 +51,27 @@ public static function pollKinds(): array
         return [['workflow'], ['activity'], ['query'], ['update-validation']];
     }
 
+    #[DataProvider('pollKinds')]
+    public function testOfferAndAcceptanceKeepProgressNegotiationAndTheirAbsoluteBudgets(string $kind): void
+    {
+        $calls = [];
+        $http = new GuzzleClient(['handler' => static function (RequestInterface $request, array $options) use (&$calls) {
+            $calls[] = [$request, $options];
+            return Create::promiseFor(new Response(200, ['Content-Type' => self::MEDIA_TYPE], self::frame(200)));
+        }]);
+        $transport = new Psr18Transport($http);
+        $uri = 'https://server.example/api/worker/'.$kind.'-tasks/offers';
+        $transport->send('POST', $uri, [], ['timeout_seconds' => 4, 'poll_request_id' => 'same-poll']);
+        $transport->send('POST', $uri, [], ['timeout_seconds' => 0, 'poll_request_id' => 'same-poll', 'task_offer_token' => 'same-packet']);
+        self::assertSame(9, $calls[0][1]['timeout']);
+        self::assertSame(5, $calls[1][1]['timeout']);
+        foreach ($calls as [$request, $options]) {
+            self::assertSame(self::MEDIA_TYPE.', application/json;q=0.9', $request->getHeaderLine('Accept'));
+            self::assertFalse($options['stream']);
+        }
+        self::assertSame('same-packet', json_decode((string) $calls[1][0]->getBody(), true)['task_offer_token']);
+    }
+
     public function testAShorterInjectedGuzzleDeadlineIsPreserved(): void
     {
         $observed = null;
diff --git a/tests/WorkerTaskDeliveryTest.php b/tests/WorkerTaskDeliveryTest.php
new file mode 100644
index 0000000..5ab31ba
--- /dev/null
+++ b/tests/WorkerTaskDeliveryTest.php
@@ -0,0 +1,271 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Tests;
+
+use DurableWorkflow\Client;
+use DurableWorkflow\Exception\ServerException;
+use DurableWorkflow\Exception\TaskDeliveryUncertain;
+use DurableWorkflow\Exception\TransportException;
+use DurableWorkflow\Tests\Support\FakeTransport;
+use DurableWorkflow\Transport\PollTransport;
+use DurableWorkflow\Worker;
+use DurableWorkflow\Worker\PollResponse;
+use DurableWorkflow\Worker\TaskDelivery;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+
+final class WorkerTaskDeliveryTest extends TestCase
+{
+    #[DataProvider('kinds')]
+    public function testOnlyPositiveAcceptanceReturnsAnExecutableTask(string $kind): void
+    {
+        $transport = new FakeTransport([self::registration(), self::offer($kind), self::claimed($kind)]);
+        $observed = [];
+        $client = (new Client('https://server.example', transport: $transport))->withWorkerPayloadObserver(
+            static function (string $path, ?array $body, array $response) use (&$observed): ?\Closure {
+                if ($response['task'] ?? null) {
+                    $observed[] = $path;
+                }
+                return null;
+            });
+        $client->registerWorker('worker', 'orders', [], []);
+        self::assertSame(self::claimed($kind), self::poll($client, $kind));
+        self::assertCount(3, $transport->requests);
+        self::assertSame(['/worker/'.$kind.'-tasks/poll'], $observed);
+        $initial = $transport->requests[1];
+        $accept = $transport->requests[2];
+        self::assertStringEndsWith('/worker/'.$kind.'-tasks/offers', $initial['uri']);
+        self::assertSame($initial['uri'], $accept['uri']);
+        self::assertSame(TaskDelivery::VERSION, $initial['headers'][TaskDelivery::HEADER]);
+        self::assertSame($initial['headers'], $accept['headers']);
+        self::assertSame(4, $initial['body']['timeout_seconds']);
+        self::assertSame(array_replace($initial['body'], ['timeout_seconds' => 0, 'task_offer_token' => 'opaque-first-packet']), $accept['body']);
+        self::assertSame(str_repeat('a', 32), $initial['body']['registration_token']);
+        self::assertSame(getmypid(), $transport->requests[0]['body']['process_metrics']['process_id']);
+        self::assertArrayNotHasKey(TaskDelivery::HEADER, $transport->requests[0]['headers']);
+    }
+
+    #[DataProvider('uncertainOutcomes')]
+    public function testUncertainRepliesRetryOnlyTheFirstSelectedPacket(TransportException|array|null $outcome): void
+    {
+        $transport = new FakeTransport([self::registration(), self::offer('workflow'), $outcome, self::claimed('workflow')]);
+        $client = new Client('https://server.example', transport: $transport);
+        $client->registerWorker('worker', 'orders', [], []);
+        try {
+            self::poll($client, 'workflow');
+            self::fail('The uncertain acceptance must not authorize execution or start another poll.');
+        } catch (TaskDeliveryUncertain $error) {
+            self::assertTrue(PollResponse::isTransientFailure($error));
+            self::assertSame('fixed-poll', $error->details['poll_request_id']);
+            self::assertArrayNotHasKey('task_offer_token', $error->details);
+        }
+        // A clone and changed wait/page inputs cannot replace an unresolved packet.
+        $clone = $client->withWorkerPollCutoffObserver(static function (): void {});
+        self::assertSame(self::claimed('workflow'), $clone->pollWorkflowTaskResponse('worker', 'orders', 19, 'fixed-poll', 30));
+        self::assertCount(4, $transport->requests);
+        self::assertSame($transport->requests[2], $transport->requests[3]);
+        self::assertSame('opaque-first-packet', $transport->requests[3]['body']['task_offer_token']);
+    }
+
+    public static function uncertainOutcomes(): iterable
+    {
+        yield 'lost response' => [new TransportException('Connection reset.', transientConnectionFailure: true)];
+        yield 'gateway timeout' => [new TransportException('Gateway timeout.', 504)];
+        yield 'older node' => [new TransportException('Not found.', 404)];
+        yield 'structured backend failure' => [new TransportException('Backend unavailable.', 503,
+            ['task' => null, 'reason' => 'backend_unavailable', 'retryable' => true])];
+        yield 'lock pressure' => [new TransportException('Lock pressure.', 503,
+            ['task' => null, 'reason' => 'backend_lock_pressure', 'poll_status' => 'backend_lock_pressure', 'retry_after_seconds' => 1])];
+        yield 'packet unavailable on another node' => [new TransportException('Unavailable.', 409,
+            ['task' => null, 'reason' => 'task_offer_unavailable'])];
+        yield 'empty is not proof of no claim' => [['task' => null, 'poll_status' => 'empty']];
+        yield 'malformed response' => [null];
+        yield 'wrong task' => [['task' => ['task_id' => 'other', 'lease_owner' => 'worker']]];
+        yield 'wrong owner' => [['task' => ['task_id' => 'task', 'lease_owner' => 'other']]];
+    }
+
+    public function testAnOlderNodeCannotDowngradeTheNegotiatedEndpoint(): void
+    {
+        $transport = new FakeTransport([self::registration(), new TransportException('Not found.', 404), ['task' => null]]);
+        $client = new Client('https://server.example', transport: $transport);
+        $client->registerWorker('worker', 'orders', [], []);
+        try {
+            self::poll($client, 'activity');
+            self::fail('Negotiated delivery cannot fall back to a legacy immediate claim.');
+        } catch (TaskDeliveryUncertain) {
+            self::assertSame(['task' => null], self::poll($client, 'activity'));
+        }
+        self::assertSame($transport->requests[1], $transport->requests[2]);
+        self::assertStringEndsWith('/activity-tasks/offers', $transport->requests[2]['uri']);
+    }
+
+    public function testAConfirmedExpiryReleasesTheSelectedPacket(): void
+    {
+        $transport = new FakeTransport([self::registration(), self::offer('workflow'),
+            new TransportException('Expired.', 409, ['task' => null, 'poll_status' => 'task_offer_expired', 'reason' => 'task_offer_expired']),
+            self::offer('workflow', 'second-packet'), self::claimed('workflow')]);
+        $client = new Client('https://server.example', transport: $transport);
+        $client->registerWorker('worker', 'orders', [], []);
+        self::assertSame('empty', self::poll($client, 'workflow')['poll_status']);
+        self::assertSame(self::claimed('workflow'), self::poll($client, 'workflow'));
+        self::assertArrayNotHasKey('task_offer_token', $transport->requests[3]['body']);
+        self::assertSame('second-packet', $transport->requests[4]['body']['task_offer_token']);
+    }
+
+    public function testRegistrationCannotDowngradeAnUnresolvedAcceptance(): void
+    {
+        $transport = new FakeTransport([self::registration(), self::offer('workflow'),
+            new TransportException('Gateway timeout.', 504), ['registered' => true], self::claimed('workflow')]);
+        $client = new Client('https://server.example', transport: $transport);
+        $client->registerWorker('worker', 'orders', [], []);
+        try { self::poll($client, 'workflow'); self::fail('Expected an uncertain acceptance.'); }
+        catch (TaskDeliveryUncertain) {}
+        try { $client->registerWorker('worker', 'orders', [], []); self::fail('Expected a rejected downgrade.'); }
+        catch (ServerException $error) { self::assertSame('invalid_task_delivery_capability', $error->reason); }
+        self::assertSame(self::claimed('workflow'), self::poll($client, 'workflow'));
+        self::assertSame($transport->requests[2], $transport->requests[4]);
+    }
+
+    public function testAnAuthoritativeWorkerStopIsNotRetriedAsAnUncertainAcceptance(): void
+    {
+        $terminal = ['task' => null, 'poll_status' => 'draining', 'reason' => 'worker_draining'];
+        $transport = new FakeTransport([self::registration(), self::offer('workflow'),
+            new TransportException('Worker is draining.', 409, $terminal)]);
+        $client = new Client('https://server.example', transport: $transport);
+        $client->registerWorker('worker', 'orders', [], []);
+        self::assertSame($terminal, self::poll($client, 'workflow'));
+        self::assertTrue(PollResponse::isTerminal($terminal));
+        self::assertCount(3, $transport->requests);
+    }
+
+    public function testReplacementRegistrationDiscardsOnlyTheOldIncarnationsPacket(): void
+    {
+        $replacement = self::registration();
+        $replacement['registration_token'] = str_repeat('b', 32);
+        $transport = new FakeTransport([self::registration(), self::offer('activity'),
+            new TransportException('Gateway timeout.', 504), $replacement, self::offer('activity', 'replacement-packet'), self::claimed('activity')]);
+        $client = new Client('https://server.example', transport: $transport);
+        $client->registerWorker('worker', 'orders', [], []);
+        try { self::poll($client, 'activity'); self::fail('Expected an uncertain acceptance.'); }
+        catch (TaskDeliveryUncertain) {}
+        $client->registerWorker('worker', 'orders', [], []);
+        self::assertSame(self::claimed('activity'), self::poll($client, 'activity'));
+        self::assertSame(str_repeat('b', 32), $transport->requests[4]['body']['registration_token']);
+        self::assertArrayNotHasKey('task_offer_token', $transport->requests[4]['body']);
+        self::assertSame('replacement-packet', $transport->requests[5]['body']['task_offer_token']);
+    }
+
+    #[DataProvider('legacyRegistrations')]
+    public function testLegacyDeliveryIsChosenOnlyAtRegistration(array $registration, string $protocol, ?array $capabilities): void
+    {
+        $transport = new FakeTransport([$registration, ['task' => null]]);
+        $client = new Client('https://server.example', transport: $transport, workerProtocolVersion: $protocol);
+        if ($capabilities === null) {
+            $client->registerWorker('worker', 'orders', [], []);
+        } else {
+            $client->registerWorker('worker', 'orders', [], [], $capabilities);
+        }
+        self::poll($client, 'workflow');
+        self::assertStringEndsWith('/workflow-tasks/poll', $transport->requests[1]['uri']);
+        self::assertArrayNotHasKey(TaskDelivery::HEADER, $transport->requests[1]['headers']);
+        if ($protocol === '1.19') {
+            self::assertNotContains(TaskDelivery::CAPABILITY, $transport->requests[0]['body']['capabilities']);
+        }
+    }
+
+    public static function legacyRegistrations(): iterable
+    {
+        yield 'no server feature' => [['registered' => true], '1.20', null];
+        yield 'explicit worker opt-out' => [self::registration(), '1.20', ['query_tasks']];
+        yield 'older protocol' => [self::registration(), '1.19', null];
+    }
+
+    public function testAContradictoryOfferCannotReachPayloadOrExecutionObservers(): void
+    {
+        $contradictory = self::claimed('workflow') + ['task_offer' => self::offer('workflow')['task_offer']];
+        $transport = new FakeTransport([self::registration(), $contradictory]);
+        $claims = 0;
+        $client = (new Client('https://server.example', transport: $transport))->withWorkerPayloadObserver(
+            static function (string $path, ?array $body, array $response) use (&$claims): ?\Closure {
+                if ($response['task'] ?? null) { ++$claims; }
+                return null;
+            });
+        $client->registerWorker('worker', 'orders', [], []);
+        try {
+            self::poll($client, 'workflow');
+            self::fail('A contradictory offer cannot authorize a task.');
+        } catch (ServerException $error) {
+            self::assertSame('invalid_task_delivery_response', $error->reason);
+        }
+        self::assertSame(0, $claims);
+        self::assertCount(2, $transport->requests);
+    }
+
+    public function testTheManagedWorkerRetainsItsLogicalPollAfterAnUncertainAcceptance(): void
+    {
+        $now = 0.0;
+        $retries = [];
+        $transport = new FakeTransport([self::registration(), self::offer('workflow'),
+            new TransportException('Not found.', 404),
+            ['task' => null, 'poll_status' => 'stopped', 'reason' => 'worker_stopped']]);
+        $worker = new Worker(new Client('https://server.example', transport: $transport), 'orders', workerId: 'worker',
+            clock: static function () use (&$now): float { return $now; },
+            sleeper: static function (int $microseconds) use (&$now): void { $now += $microseconds / 1_000_000; },
+            transientPollRetryObserver: static function (string $kind, int $attempt, float $delay, ServerException $error) use (&$retries): void {
+                $retries[] = [$kind, $attempt, $error::class];
+            });
+        $worker->run(4);
+        self::assertSame([['workflow', 1, TaskDeliveryUncertain::class]], $retries);
+        self::assertSame($transport->requests[2], $transport->requests[3]);
+        self::assertContains(TaskDelivery::CAPABILITY, $transport->requests[0]['body']['capabilities']);
+        self::assertCount(4, $transport->requests);
+    }
+
+    public function testOfferTransportUsesTheSameProgressAndLogicalPathForAllKinds(): void
+    {
+        foreach (['workflow', 'activity', 'query', 'update-validation'] as $kind) {
+            $path = '/worker/'.$kind.'-tasks/offers';
+            self::assertSame($kind, PollTransport::taskKind('POST', $path));
+            self::assertSame('/worker/'.$kind.'-tasks/poll', PollTransport::logicalPath($path));
+            self::assertNull(PollTransport::taskKind('GET', $path));
+        }
+    }
+
+    public static function kinds(): array { return [['workflow'], ['activity'], ['query']]; }
+
+    private static function poll(Client $client, string $kind): array
+    {
+        return match ($kind) {
+            'workflow' => $client->pollWorkflowTaskResponse('worker', 'orders', 4, 'fixed-poll'),
+            'activity' => $client->pollActivityTaskResponse('worker', 'orders', 4, 'fixed-poll'),
+            'query' => $client->pollQueryTaskResponse('worker', 'orders', 4, 'fixed-poll'),
+        };
+    }
+
+    private static function offer(string $kind, string $token = 'opaque-first-packet'): array
+    {
+        return ['task' => null, 'poll_status' => 'offered', 'task_offer' => [
+            'schema' => TaskDelivery::SCHEMA, 'task_kind' => $kind, 'task_id' => 'task',
+            'token' => $token, 'expires_at_ms' => 1000,
+        ]];
+    }
+
+    private static function claimed(string $kind): array
+    {
+        return ['task' => [($kind === 'query' ? 'query_task_id' : 'task_id') => 'task', 'lease_owner' => 'worker', 'task_kind' => $kind]];
+    }
+
+    private static function registration(): array
+    {
+        return ['registered' => true, 'registration_token' => str_repeat('a', 32), 'heartbeat_interval_seconds' => 10,
+            'server_capabilities' => ['task_delivery' => [
+                'schema' => TaskDelivery::SCHEMA, 'supported' => true, 'minimum_protocol_version' => '1.20',
+                'worker_capability' => TaskDelivery::CAPABILITY, 'header' => TaskDelivery::HEADER,
+                'version' => TaskDelivery::VERSION, 'endpoint_suffix' => '/offers',
+                'acceptance_timeout_seconds' => 2, 'unaccepted_offer_reserves_task' => false,
+            ], 'worker_deregistration_fencing' => ['schema' => 'durable-workflow.v2.worker-deregistration.v1',
+                'supported' => true, 'endpoint' => '/worker/registrations/{workerId}/deregister', 'receipt_retention_seconds' => 600]]];
+    }
+}