Negotiate task offers and serialize acceptance for each logical poll

docs/portal/build/workers.md

@@ -99,6 +99,17 @@ 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.
+During acceptance, stop/drain hints and malformed replies retain that packet.
+Only the protocol's two closed HTTP `409` refusal shapes can release it or end
+the original registration; a later replacement registration is preserved.
+
 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

docs/protocol.md

@@ -198,6 +198,32 @@ 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. Definitive refusal requires
+HTTP `409` and exactly `task`, `poll_status`, and `reason`, with literal
+`task: null`. The only accepted status/reason pairs are
+`task_offer_expired` / `task_offer_expired` and
+`stale_worker_registration` / `worker_registration_lost_authority`.
+Expiry releases only the selected packet; authority loss ends only its original
+registration incarnation. Other pending requests and replacement registrations
+remain separate. The Server must establish these outcomes from receipt-first
+reconciliation and authoritative claim/registration state.
+
+The default transport checks the complete original UTF-8 JSON response for
+duplicate decoded member names before associative decoding. This also applies
+inside a progress envelope. Invalid bytes, trailing data, extra fields, mixed
+pairs, conflicting exception fields and successful terminal hints retain the
+selected packet without running a handler. Discovery before selection keeps its
+existing poll classification. A custom transport must provide original-byte
+validation through `Transport\TaskDeliveryRefusal::fromJson()` and attach that
+evidence to `TransportException`; a decoded-only refusal remains uncertain.
+
 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

src/Client.php

@@ -68,6 +68,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 DurableWorkflow\Exception\PaginationException;
 use InvalidArgumentException;
 
@@ -88,6 +91,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;
@@ -162,6 +166,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();
         $this->metricsObserver = new MetricsObserver($metrics ?? new NoopMetricsRecorder());
     }
 
@@ -1752,6 +1757,7 @@ public function registerWorker(
             'memo_upserts',
             'typed_search_attributes',
             'durable_selection',
+            TaskDelivery::CAPABILITY,
         ],
         int $maxConcurrentWorkflowTasks = 1,
         int $maxConcurrentActivityTasks = 1,
@@ -1761,6 +1767,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.');
         }
@@ -1797,7 +1807,9 @@ 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, $registration);
         if (array_key_exists('server_capabilities', $registration)) {
             $this->rememberClaimedTaskReadyAge($registration['server_capabilities']);
         }
@@ -1859,6 +1871,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} */
@@ -1883,6 +1896,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;
     }
@@ -2790,7 +2805,14 @@ private function performPollTaskResponse(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: function (array $response) use ($wireBody, $wirePath, $selected): void {
+                                $this->taskDelivery->assertCurrentRegistration($wireBody);
+                                TaskDelivery::assertResponse($response, $wireBody, $wirePath, $selected);
+                            }))
+                    ?? $this->worker('POST', $path, $body);
             } catch (ServerException $exception) {
                 if ($retryTransientFailure && $exception->isTransientConnectionFailure()) {
                     $retryTransientFailure = false;
@@ -2857,7 +2879,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',
@@ -2869,13 +2891,17 @@ 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;
     }
 
     /**
      * @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(
@@ -2887,14 +2913,16 @@ private function request(
         ?RequestBudget $budget = null,
         ?\Closure $payloadPaths = null,
         ?string $requestId = null,
+        ?\Closure $responseValidator = null,
     ): array {
         if ($this->metricsRequest !== null || !$this->metricsObserver->enabled()) {
-            return $this->performRequest($method, $path, $worker, $body, $operation, $budget, $payloadPaths, $requestId, $this->metricsRequest);
+            return $this->performRequest($method, $path, $worker, $body, $operation, $budget, $payloadPaths, $requestId,
+                $this->metricsRequest, $responseValidator);
         }
 
         return $this->observeRequest($method, $path, $worker,
             fn (ClientRequestObservation $observation): array => $this->performRequest(
-                $method, $path, $worker, $body, $operation, $budget, $payloadPaths, $requestId, $observation));
+                $method, $path, $worker, $body, $operation, $budget, $payloadPaths, $requestId, $observation, $responseValidator));
     }
 
     /** @param \Closure(ClientRequestObservation): array<string, mixed> $operation
@@ -2928,14 +2956,16 @@ private function performRequest(
         ?\Closure $payloadPaths,
         ?string $requestId,
         ?ClientRequestObservation $observation,
+        ?\Closure $responseValidator,
     ): array {
         if ($observation !== null) {
             $observation->status = null;
         }
-        $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) {
@@ -2945,7 +2975,7 @@ private function performRequest(
                 $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.
@@ -2959,21 +2989,27 @@ private function performRequest(
             }
 
             $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();
 
@@ -2983,7 +3019,8 @@ private function performRequest(
                 throw $exception;
             }
             $details = $exception->response;
-            $reason = is_array($details) && isset($details['reason']) ? (string) $details['reason'] : null;
+            $reason = is_array($details) && isset($details['reason'])
+                && ($responseValidator === null || is_string($details['reason'])) ? (string) $details['reason'] : null;
             $status = $exception->status ?? 0;
             $message = $exception->getMessage();
             if ($operation === 'query') {

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);
+    }
+}

src/Exception/TransportException.php

@@ -5,6 +5,7 @@
 namespace DurableWorkflow\Exception;
 
 use Throwable;
+use DurableWorkflow\Transport\TaskDeliveryRefusal;
 
 class TransportException extends DurableWorkflowException
 {
@@ -17,6 +18,7 @@ public function __construct(
         public readonly bool $transientConnectionFailure = false,
         /** @var array<string, string> */
         public readonly array $headers = [],
+        public readonly ?TaskDeliveryRefusal $taskDeliveryRefusal = null,
     ) {
         parent::__construct($message, $status ?? 0, $previous);
     }

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
     {