diff --git a/app/Http/Controllers/Api/ActivityTaskController.php b/app/Http/Controllers/Api/ActivityTaskController.php
index 1a53b0c89..4b668046c 100644
--- a/app/Http/Controllers/Api/ActivityTaskController.php
+++ b/app/Http/Controllers/Api/ActivityTaskController.php
@@ -23,6 +23,7 @@
 use App\Support\WorkerProtocol;
 use App\Support\WorkerProtocolMutationRetrier;
 use App\Support\WorkerSessionRegistry;
+use App\Support\WorkerTaskDelivery;
 use App\Support\WorkflowObservabilityPoll;
 use Carbon\CarbonImmutable;
 use Carbon\CarbonInterface;
@@ -133,6 +134,7 @@ public function poll(Request $request): JsonResponse
                 ),
                 timeoutSeconds: $timeoutSeconds,
                 protocolVersion: WorkerProtocol::requestVersion($request),
+                delivery: WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (\Throwable $exception) {
             if ($exception instanceof InvalidArgumentException
@@ -164,6 +166,9 @@ public function poll(Request $request): JsonResponse
 
             throw $exception;
         }
+        if (is_array($poll['task']['delivery_offer'] ?? null)) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
         $claim = $poll['task'] ?? null;
 
         if (is_array($claim)) {
diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php
index d6082e44f..da49f716b 100644
--- a/app/Http/Controllers/Api/WorkerController.php
+++ b/app/Http/Controllers/Api/WorkerController.php
@@ -42,6 +42,7 @@
 use App\Support\WorkerPollFence;
 use App\Support\WorkerProtocol;
 use App\Support\WorkerProtocolMutationRetrier;
+use App\Support\WorkerTaskDelivery;
 use App\Support\WorkerTerminalEventAttribution;
 use App\Support\WorkflowHistoryPageToken;
 use App\Support\WorkflowMetadataCapabilityPolicy;
@@ -1391,6 +1392,7 @@ public function pollWorkflowTasks(Request $request): JsonResponse
                 ),
                 timeoutSeconds: $timeoutSeconds,
                 taskKinds: $taskKinds,
+                delivery: WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (\Throwable $exception) {
             if ($exception instanceof \InvalidArgumentException
@@ -1448,6 +1450,9 @@ public function pollWorkflowTasks(Request $request): JsonResponse
             throw $exception;
         }
 
+        if (is_array($poll['task']['delivery_offer'] ?? null)) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
         $task = $this->formatTaskHistoryPagination($poll['task'] ?? null);
 
         // PHP SDKs before 2.2.0 reject the valid StartAccepted prefix when
@@ -4223,6 +4228,7 @@ public function pollQueryTasks(Request $request): JsonResponse
                 $worker,
                 $validated['poll_request_id'] ?? null,
                 $timeoutSeconds,
+                WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (QueryTaskQueueUnavailableException $exception) {
             return WorkerProtocol::json([
@@ -4255,6 +4261,10 @@ public function pollQueryTasks(Request $request): JsonResponse
             throw $exception;
         }
 
+        if (isset($poll['task']['delivery_offer'])) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
+
         $pollContext = [
             'namespace' => $namespace,
             'worker_id' => $worker->worker_id,
@@ -4264,7 +4274,17 @@ public function pollQueryTasks(Request $request): JsonResponse
             'query_task_id' => $poll['task']['query_task_id'] ?? null,
         ];
 
-        if ($poll['poll_status'] === 'empty') {
+        $this->logQueryPollResponse($pollContext);
+
+        return WorkerProtocol::json([
+            'task' => WorkflowObservabilityPoll::filterOptionalDelivered($poll['task']),
+            'poll_status' => $poll['poll_status'],
+        ]);
+    }
+
+    private function logQueryPollResponse(array $pollContext): void
+    {
+        if ($pollContext['poll_status'] === 'empty') {
             if (config('app.debug')) {
                 Log::debug('worker_query_poll_response_ready', $pollContext);
             }
@@ -4272,10 +4292,6 @@ public function pollQueryTasks(Request $request): JsonResponse
             Log::info('worker_query_poll_response_ready', $pollContext);
         }
 
-        return WorkerProtocol::json([
-            'task' => WorkflowObservabilityPoll::filterOptionalDelivered($poll['task']),
-            'poll_status' => $poll['poll_status'],
-        ]);
     }
 
     public function queryTaskStatus(Request $request, string $queryTaskId): JsonResponse
@@ -4477,6 +4493,7 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse
         $validated = $request->validate([
             'worker_id' => ['required', 'string'],
             'task_queue' => ['required', 'string'],
+            'poll_request_id' => ['nullable', 'string', 'max:255'],
             'timeout_seconds' => [
                 'nullable',
                 'integer',
@@ -4512,6 +4529,7 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse
                 $namespace,
                 $worker,
                 isset($validated['timeout_seconds']) ? (int) $validated['timeout_seconds'] : null,
+                WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (LongPollCapacityExhaustedException $exception) {
             return WorkerPollBackpressure::response(
@@ -4532,6 +4550,10 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse
             );
         }
 
+        if (isset($poll['task']['delivery_offer'])) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
+
         return WorkerProtocol::json($poll);
     }
 
diff --git a/app/Http/Middleware/WorkerPollResponseProgress.php b/app/Http/Middleware/WorkerPollResponseProgress.php
index 5a8c06ac2..ff68f32b0 100644
--- a/app/Http/Middleware/WorkerPollResponseProgress.php
+++ b/app/Http/Middleware/WorkerPollResponseProgress.php
@@ -15,6 +15,10 @@ final class WorkerPollResponseProgress
         'api/worker/activity-tasks/poll',
         'api/worker/query-tasks/poll',
         'api/worker/update-validation-tasks/poll',
+        'api/worker/workflow-tasks/offers',
+        'api/worker/activity-tasks/offers',
+        'api/worker/update-validation-tasks/offers',
+        'api/worker/query-tasks/offers',
     ];
 
     public function handle(Request $request, Closure $next): Response
diff --git a/app/Models/WorkflowUpdateValidationTask.php b/app/Models/WorkflowUpdateValidationTask.php
index 1440d5fc4..98a304b98 100644
--- a/app/Models/WorkflowUpdateValidationTask.php
+++ b/app/Models/WorkflowUpdateValidationTask.php
@@ -37,6 +37,7 @@ protected function casts(): array
             'attempt_count' => 'integer',
             'command_context' => 'array',
             'validation_errors' => 'array',
+            'delivery_receipt' => 'array',
             'lease_expires_at' => 'datetime',
             'approved_at' => 'datetime',
             'rejected_at' => 'datetime',
diff --git a/app/Support/ActivityTaskPoller.php b/app/Support/ActivityTaskPoller.php
index 907888631..3a3563bf9 100644
--- a/app/Support/ActivityTaskPoller.php
+++ b/app/Support/ActivityTaskPoller.php
@@ -47,6 +47,7 @@ public function poll(
         bool $workerSessionsAvailable = true,
         ?int $timeoutSeconds = null,
         ?string $protocolVersion = null,
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         $pollRequestId = $this->nonEmptyString($pollRequestId);
 
@@ -57,6 +58,21 @@ public function poll(
             ];
         }
 
+        if ($delivery !== null) {
+            $result = $this->performPoll(
+                namespace: $namespace, taskQueue: $taskQueue, leaseOwner: $leaseOwner,
+                buildId: $buildId, worker: $worker, pollRequestId: $pollRequestId,
+                supportedActivityTypes: $supportedActivityTypes, workerSessionsAvailable: $workerSessionsAvailable,
+                timeoutSeconds: $delivery->accepting() ? 0 : $timeoutSeconds,
+                protocolVersion: $protocolVersion, delivery: $delivery,
+            );
+            if (is_array($result['task'] ?? null) && ! isset($result['task']['delivery_offer'])) {
+                $delivery->assertReplay($result['task']['task_id']);
+            }
+
+            return $result;
+        }
+
         if ($pollRequestId !== null) {
             $task = $this->activeLeasedTaskForWorker(
                 namespace: $namespace,
@@ -440,6 +456,7 @@ private function performPoll(
         bool $workerSessionsAvailable = true,
         ?int $timeoutSeconds = null,
         ?string $protocolVersion = null,
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         $limit = max(10, max(1, (int) config('server.polling.max_tasks_per_poll', 1)) * 10);
         $nextProbeAt = null;
@@ -464,12 +481,14 @@ function () use (
                 $workerSessionsAvailable,
                 $limit,
                 $workerPollFence,
+                $delivery,
                 &$nextProbeAt,
                 &$resolvedResult,
             ): ?array {
                 app(StoragePressure::class)->requireNewWork();
 
-                if (! WorkerPollFence::isCurrent($workerPollFence)) {
+                $delivery?->lockWorker();
+                if ($delivery === null && ! WorkerPollFence::isCurrent($workerPollFence)) {
                     $resolvedResult = [
                         'task' => null,
                         'poll_status' => 'stale_worker_registration',
@@ -490,6 +509,7 @@ function () use (
                     $supportedActivityTypes,
                     $workerSessionsAvailable,
                     $workerPollFence,
+                    $delivery,
                 );
                 $nextProbeAt = $resolvedResult['next_probe_at'] ?? null;
 
@@ -694,6 +714,7 @@ private function nextTask(
         array $supportedActivityTypes = [],
         bool $workerSessionsAvailable = true,
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         return $this->withWorkerCompatibility(
             $namespace,
@@ -709,6 +730,7 @@ function () use (
                 $supportedActivityTypes,
                 $workerSessionsAvailable,
                 $workerPollFence,
+                $delivery,
             ): array {
                 if ($pollRequestId !== null) {
                     $task = $this->activeLeasedTaskForWorker(
@@ -751,6 +773,7 @@ function () use (
                                 $supportedActivityTypes,
                                 $workerSessionsAvailable,
                                 $workerPollFence,
+                                $delivery,
                             ),
                         ),
                     );
@@ -821,8 +844,9 @@ private function claimReadyTask(
         array $supportedActivityTypes = [],
         bool $workerSessionsAvailable = true,
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): ?array {
-        $readyTasks = $this->bridge->poll(
+        $readyTasks = $delivery?->accepting() ? $delivery->candidates('activity') : $this->bridge->poll(
             connection: null,
             queue: $taskQueue,
             limit: $limit,
@@ -881,6 +905,15 @@ private function claimReadyTask(
                 continue;
             }
 
+            if ($delivery !== null && ! $delivery->accepting()) {
+                $offer = $delivery->offerTask($readyTask, 'activity');
+                if ($offer !== null) {
+                    return $offer;
+                }
+
+                continue;
+            }
+
             try {
                 $claim = DB::transaction(function () use (
                     $namespace,
@@ -892,8 +925,10 @@ private function claimReadyTask(
                     $pollRequestId,
                     $taskQueue,
                     $buildId,
+                    $delivery,
                 ): ?array {
-                    if ($workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
+                    $delivery?->lockWorker();
+                    if ($delivery === null && $workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
                         throw new ActivityTaskClaimRolledBack;
                     }
 
@@ -906,6 +941,7 @@ private function claimReadyTask(
                         $pollRequestId,
                     );
 
+                    $delivery?->assertClaimable($taskId, 'activity');
                     $claim = $this->claimStatus($taskId, $leaseOwner);
 
                     if (($claim['claimed'] ?? false) !== true) {
@@ -942,6 +978,7 @@ private function claimReadyTask(
                         $pollRequestId,
                         $workerPollFence,
                     );
+                    $delivery?->bindAccepted($taskId);
 
                     return $claim;
                 });
diff --git a/app/Support/BackendUnavailable.php b/app/Support/BackendUnavailable.php
index 4b3bb86b1..58d5cdd62 100644
--- a/app/Support/BackendUnavailable.php
+++ b/app/Support/BackendUnavailable.php
@@ -72,10 +72,10 @@ public static function workerResponse(Request $request): ?JsonResponse
         }
 
         $operation = match (true) {
-            $request->is('api/worker/workflow-tasks/poll') => 'poll_workflow_task',
-            $request->is('api/worker/activity-tasks/poll') => 'poll_activity_task',
-            $request->is('api/worker/query-tasks/poll') => 'poll_query_task',
-            $request->is('api/worker/update-validation-tasks/poll') => 'poll_update_validation_task',
+            $request->is('api/worker/workflow-tasks/poll', 'api/worker/workflow-tasks/offers') => 'poll_workflow_task',
+            $request->is('api/worker/activity-tasks/poll', 'api/worker/activity-tasks/offers') => 'poll_activity_task',
+            $request->is('api/worker/query-tasks/poll', 'api/worker/query-tasks/offers') => 'poll_query_task',
+            $request->is('api/worker/update-validation-tasks/poll', 'api/worker/update-validation-tasks/offers') => 'poll_update_validation_task',
             $request->is('api/worker/register') => 'register_worker',
             $request->isMethod('POST') && $request->is('api/worker/registrations/*/deregister') => 'deregister_worker',
             $request->is('api/worker/heartbeat') => 'heartbeat_worker',
diff --git a/app/Support/TaskDeliveryOffer.php b/app/Support/TaskDeliveryOffer.php
new file mode 100644
index 000000000..6187b94a2
--- /dev/null
+++ b/app/Support/TaskDeliveryOffer.php
@@ -0,0 +1,61 @@
+<?php
+
+namespace App\Support;
+
+use Illuminate\Contracts\Encryption\DecryptException;
+use Illuminate\Support\Facades\Crypt;
+use JsonException;
+
+/** Bounded, sealed offer packets; these packets never grant execution. */
+final class TaskDeliveryOffer
+{
+    public static function encode(array $data): string
+    {
+        $token = Crypt::encryptString(json_encode($data, JSON_THROW_ON_ERROR));
+        if (strlen($token) > 8192) {
+            throw new TaskDeliveryOfferRejected('task_offer_too_large');
+        }
+
+        return $token;
+    }
+
+    public static function decode(mixed $token): array
+    {
+        try {
+            $data = is_string($token) && strlen($token) <= 8192
+                ? json_decode(Crypt::decryptString($token), true, flags: JSON_THROW_ON_ERROR) : null;
+        } catch (DecryptException|JsonException) {
+            $data = null;
+        }
+        if (! is_array($data) || ($data['schema'] ?? null) !== WorkerTaskDelivery::SCHEMA
+            || ! is_string($data['task_id'] ?? null) || ! is_string($data['task_kind'] ?? null)
+            || ! is_array($data['worker'] ?? null) || ! is_array($data['candidate'] ?? null)
+            || ! is_int($data['offered_attempt'] ?? null) || ! is_int($data['expires_at_ms'] ?? null)
+            || ! is_string($data['fingerprint'] ?? null) || ! is_string($data['delivery_id'] ?? null)) {
+            throw new TaskDeliveryOfferRejected('task_offer_invalid');
+        }
+
+        return $data;
+    }
+
+    public static function identifier(array $data): string
+    {
+        return hash('sha256', json_encode([
+            $data['schema'], $data['namespace'], $data['queue'], $data['route'], $data['poll_request_id'],
+            $data['task_kinds'], $data['worker']['registration_id'], $data['worker']['registration_incarnation'],
+            $data['task_id'], $data['task_kind'], $data['offered_attempt'], $data['fingerprint'],
+        ], JSON_THROW_ON_ERROR));
+    }
+
+    public static function assertUnexpired(array $data): void
+    {
+        if (now()->getTimestampMs() >= $data['expires_at_ms']) {
+            throw new TaskDeliveryOfferRejected('task_offer_expired');
+        }
+    }
+
+    public static function acceptanceMilliseconds(): int
+    {
+        return max(1000, min(10000, (int) config('server.polling.task_offer_accept_seconds', 2) * 1000));
+    }
+}
diff --git a/app/Support/TaskDeliveryOfferRejected.php b/app/Support/TaskDeliveryOfferRejected.php
new file mode 100644
index 000000000..d69b02932
--- /dev/null
+++ b/app/Support/TaskDeliveryOfferRejected.php
@@ -0,0 +1,13 @@
+<?php
+
+namespace App\Support;
+
+use RuntimeException;
+
+final class TaskDeliveryOfferRejected extends RuntimeException
+{
+    public function __construct(public readonly string $reason)
+    {
+        parent::__construct('The worker task offer could not be accepted.');
+    }
+}
diff --git a/app/Support/WorkerProtocol.php b/app/Support/WorkerProtocol.php
index 1c4094393..6e1edd02f 100644
--- a/app/Support/WorkerProtocol.php
+++ b/app/Support/WorkerProtocol.php
@@ -510,6 +510,17 @@ public static function serverCapabilities(): array
             'query_task_poll_request_idempotency' => true,
             'query_task_authority' => true,
             'query_task_timeouts' => self::queryTaskTimeouts(),
+            'task_delivery' => [
+                'schema' => WorkerTaskDelivery::SCHEMA,
+                'supported' => true,
+                'minimum_protocol_version' => '1.20',
+                'worker_capability' => WorkerTaskDelivery::CAPABILITY,
+                'header' => WorkerTaskDelivery::HEADER,
+                'version' => WorkerTaskDelivery::VERSION,
+                'endpoint_suffix' => '/offers',
+                'acceptance_timeout_seconds' => max(1, min(10, (int) config('server.polling.task_offer_accept_seconds', 2))),
+                'unaccepted_offer_reserves_task' => false,
+            ],
             'update_validation_tasks' => true,
             'synchronous_update_validation' => [
                 'supported' => true,
diff --git a/app/Support/WorkerTaskDelivery.php b/app/Support/WorkerTaskDelivery.php
new file mode 100644
index 000000000..1f722cb37
--- /dev/null
+++ b/app/Support/WorkerTaskDelivery.php
@@ -0,0 +1,336 @@
+<?php
+
+namespace App\Support;
+
+use App\Models\WorkerRegistration;
+use App\Models\WorkflowUpdateValidationTask;
+use Illuminate\Http\Request;
+use Illuminate\Support\Carbon;
+use Illuminate\Support\Facades\DB;
+use Workflow\V2\Enums\TaskStatus;
+use Workflow\V2\Models\ActivityExecution;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Models\WorkflowTask;
+
+/** Offers convey no lease. Only an authenticated, fenced acceptance may claim. */
+final class WorkerTaskDelivery
+{
+    public const HEADER = 'X-Durable-Workflow-Task-Delivery';
+
+    public const VERSION = 'offer-accept.v1';
+
+    public const SCHEMA = 'durable-workflow.task-delivery.v1';
+
+    public const CAPABILITY = 'task_delivery_offers';
+
+    public const RECEIPT_KEY = '_server_task_delivery';
+
+    private const CANDIDATE_FIELDS = [
+        'task_id', 'workflow_run_id', 'workflow_instance_id', 'namespace',
+        'workflow_type', 'workflow_class', 'connection', 'queue', 'compatibility',
+        'sticky_worker_id', 'sticky_until', 'available_at', 'priority',
+        'fairness_key', 'fairness_weight', 'activity_execution_id', 'activity_type',
+    ];
+
+    private function __construct(
+        private readonly string $namespace,
+        private readonly string $queue,
+        private readonly string $pollRequestId,
+        private readonly string $route,
+        private readonly array $taskKinds,
+        private readonly array $worker,
+        private readonly ?array $offer,
+    ) {}
+
+    public static function forRequest(Request $request, WorkerRegistration $worker, string $namespace, string $queue, ?string $pollRequestId): ?self
+    {
+        $version = $request->header(self::HEADER);
+        $token = $request->input('task_offer_token');
+        $offerEndpoint = str_ends_with($request->path(), '-tasks/offers');
+        if (! $offerEndpoint && $version === null && $token === null) {
+            return null;
+        }
+        if (! $offerEndpoint) {
+            throw new TaskDeliveryOfferRejected('task_delivery_offer_endpoint_required');
+        }
+        if ($version !== self::VERSION || version_compare(WorkerProtocol::requestVersion($request) ?? '0', '1.20', '<')) {
+            throw new TaskDeliveryOfferRejected('task_delivery_protocol_required');
+        }
+        if (! in_array(self::CAPABILITY, $worker->capabilities ?? [], true)
+            || ! is_string($pollRequestId) || $pollRequestId === '') {
+            throw new TaskDeliveryOfferRejected('task_delivery_registration_required');
+        }
+        $registrationToken = $request->input('registration_token');
+        if (! is_string($registrationToken) || ! is_string($worker->registration_token)
+            || ! hash_equals($worker->registration_token, $registrationToken)) {
+            throw new TaskDeliveryOfferRejected('worker_registration_lost_authority');
+        }
+        $snapshot = CooperativeCancellationPolicy::workerSnapshot($worker, WorkerProtocol::requestVersion($request));
+        $route = $request->path();
+        $offer = $token === null ? null : TaskDeliveryOffer::decode($token);
+        $taskKinds = str_contains($route, '/workflow-tasks/')
+            ? WorkflowTaskPollRequestStore::normalizeTaskKinds($request->input('task_kinds', ['workflow']) ?? ['workflow'])
+            : [];
+        $context = new self($namespace, $queue, $pollRequestId, $route, $taskKinds, $snapshot, $offer);
+        if ($offer !== null) {
+            $context->assertContext($offer);
+            $context->assertExpiredOfferOutcome();
+        }
+
+        return $context;
+    }
+
+    public function accepting(): bool
+    {
+        return $this->offer !== null;
+    }
+
+    /** @return list<array<string, mixed>> */
+    public function candidates(string $kind): array
+    {
+        return ($this->offer['task_kind'] ?? null) === $kind ? [$this->offer['candidate']] : [];
+    }
+
+    /** @param array<string, mixed> $candidate */
+    public function offerTask(array $candidate, string $kind): ?array
+    {
+        $task = NamespaceWorkflowScope::task($this->namespace, (string) ($candidate['task_id'] ?? ''));
+        if (! $task instanceof WorkflowTask || $task->status !== TaskStatus::Ready
+            || $task->task_type->value !== $kind || $task->queue !== $this->queue
+            || $task->available_at?->isFuture()) {
+            return null;
+        }
+
+        return $this->issue($task->id, $kind, (int) $task->attempt_count, $this->fingerprint($task),
+            array_intersect_key($candidate, array_flip(self::CANDIDATE_FIELDS)));
+    }
+
+    /** Seal an observation of the existing task authority without reserving it. */
+    public function issue(string $taskId, string $kind, int $attempt, string $fingerprint, array $candidate = []): array
+    {
+        $data = [
+            'schema' => self::SCHEMA, 'namespace' => $this->namespace, 'queue' => $this->queue,
+            'poll_request_id' => $this->pollRequestId, 'route' => $this->route, 'task_kinds' => $this->taskKinds,
+            'worker' => $this->worker, 'task_id' => $taskId, 'task_kind' => $kind,
+            'offered_attempt' => $attempt, 'fingerprint' => $fingerprint, 'candidate' => $candidate,
+            'expires_at_ms' => now()->getTimestampMs() + TaskDeliveryOffer::acceptanceMilliseconds(),
+        ];
+        $data['delivery_id'] = TaskDeliveryOffer::identifier($data);
+
+        return ['delivery_offer' => [
+            'schema' => self::SCHEMA, 'task_id' => $taskId, 'task_kind' => $kind,
+            'token' => TaskDeliveryOffer::encode($data),
+            'expires_at_ms' => $data['expires_at_ms'],
+        ]];
+    }
+
+    /** Called inside the existing claim transaction, after worker/run locking. */
+    public function assertClaimable(string $taskId, string $kind): void
+    {
+        $data = $this->offer;
+        if ($data === null || $data['task_id'] !== $taskId || $data['task_kind'] !== $kind) {
+            throw new TaskDeliveryOfferRejected('task_offer_mismatch');
+        }
+        $this->assertWorkerCurrent($data);
+        $task = NamespaceWorkflowScope::lockTaskForMutation($this->namespace, $taskId);
+        if (! $task instanceof WorkflowTask || $task->status !== TaskStatus::Ready
+            || $task->queue !== $this->queue || $task->task_type->value !== $kind
+            || (int) $task->attempt_count !== $data['offered_attempt']
+            || $task->available_at?->isFuture()) {
+            throw new TaskDeliveryOfferRejected('task_offer_unavailable');
+        }
+        if ($kind === 'activity') {
+            // Acquire the remaining domain lock before checking the short
+            // acceptance deadline. The bridge will reuse these locks.
+            ActivityExecution::query()->whereKey($task->payload['activity_execution_id'] ?? null)
+                ->lockForUpdate()->first();
+        }
+        $this->assertAcceptance($taskId, $kind, (int) $task->attempt_count, $this->fingerprint($task));
+        $this->assertSqlPollUnbound();
+    }
+
+    public function offeredTaskId(string $kind): ?string
+    {
+        return ($this->offer['task_kind'] ?? null) === $kind ? $this->offer['task_id'] : null;
+    }
+
+    /** Call under the domain's normal claim lock before changing its authority. */
+    public function assertAcceptance(string $taskId, string $kind, int $attempt, string $fingerprint): void
+    {
+        $data = $this->offer;
+        if ($data === null || $data['task_id'] !== $taskId || $data['task_kind'] !== $kind
+            || $data['offered_attempt'] !== $attempt) {
+            throw new TaskDeliveryOfferRejected('task_offer_unavailable');
+        }
+        $this->lockWorker();
+        if (! hash_equals($data['fingerprint'], $fingerprint)) {
+            throw new TaskDeliveryOfferRejected('task_offer_changed');
+        }
+        TaskDeliveryOffer::assertUnexpired($data);
+    }
+
+    public function lockWorker(): void
+    {
+        $this->assertWorkerCurrent($this->offer ?? ['worker' => $this->worker]);
+    }
+
+    /** One SQL claim per logical poll, including multiplexed validation tasks. */
+    public function assertSqlPollUnbound(): void
+    {
+        $this->lockWorker();
+        if (WorkflowTask::query()->where('delivery_poll_key', $this->pollKey())->lockForUpdate()->first(['id']) !== null
+            || WorkflowUpdateValidationTask::query()->where('delivery_poll_key', $this->pollKey())->lockForUpdate()->first(['id']) !== null) {
+            throw new TaskDeliveryOfferRejected('task_poll_already_accepted');
+        }
+    }
+
+    public function pollKey(): string
+    {
+        return hash('sha256', json_encode([
+            self::SCHEMA, $this->namespace, $this->queue, $this->route, $this->pollRequestId,
+            $this->worker['registration_id'], $this->worker['registration_incarnation'],
+        ], JSON_THROW_ON_ERROR));
+    }
+
+    public function acceptedReceipt(int $attempt): array
+    {
+        if ($this->offer === null) {
+            throw new TaskDeliveryOfferRejected('task_offer_mismatch');
+        }
+        TaskDeliveryOffer::assertUnexpired($this->offer);
+
+        return [
+            'schema' => self::SCHEMA, 'delivery_id' => $this->offer['delivery_id'],
+            'incarnation' => $this->worker['registration_incarnation'],
+            'route' => $this->route, 'poll_request_id' => $this->pollRequestId,
+            'task_kinds' => $this->taskKinds, 'accepted_attempt' => $attempt,
+        ];
+    }
+
+    public function assertReceipt(mixed $receipt, string $taskId, int $attempt): void
+    {
+        $this->lockWorker();
+        if (! is_array($receipt) || ($receipt['schema'] ?? null) !== self::SCHEMA
+            || ($receipt['incarnation'] ?? null) !== $this->worker['registration_incarnation']
+            || ($receipt['route'] ?? null) !== $this->route || ($receipt['poll_request_id'] ?? null) !== $this->pollRequestId
+            || ($receipt['task_kinds'] ?? null) !== $this->taskKinds
+            || ($receipt['accepted_attempt'] ?? null) !== $attempt
+            || ($this->offer !== null && ($this->offer['task_id'] !== $taskId
+                || ($receipt['delivery_id'] ?? null) !== $this->offer['delivery_id']))) {
+            throw new TaskDeliveryOfferRejected('task_offer_lost_authority');
+        }
+    }
+
+    /** Persist with the normal claim. Late failure rolls back the whole claim. */
+    public function bindAccepted(string $taskId): void
+    {
+        $data = $this->offer;
+        if ($data === null) {
+            throw new TaskDeliveryOfferRejected('task_offer_mismatch');
+        }
+        TaskDeliveryOffer::assertUnexpired($data);
+        $task = NamespaceWorkflowScope::taskQuery($this->namespace)->whereKey($taskId)->lockForUpdate()->first();
+        if (! $task instanceof WorkflowTask || $task->id !== $data['task_id']
+            || $task->status !== TaskStatus::Leased || $task->lease_owner !== $this->worker['worker_id']
+            || (int) $task->attempt_count !== $data['offered_attempt'] + 1) {
+            throw new TaskDeliveryOfferRejected('task_offer_lost_authority');
+        }
+        $payload = $task->payload ?? [];
+        $payload[self::RECEIPT_KEY] = $this->acceptedReceipt((int) $task->attempt_count);
+        $task->forceFill(['payload' => $payload, 'delivery_poll_key' => $this->pollKey()])->save();
+        $this->acceptedReceipt((int) $task->attempt_count);
+    }
+
+    /** Expired offer packets may still replay their already-accepted live claim. */
+    public function assertReplay(string $taskId): void
+    {
+        $task = NamespaceWorkflowScope::task($this->namespace, $taskId);
+        $receipt = $task?->payload[self::RECEIPT_KEY] ?? null;
+        if (! $task instanceof WorkflowTask || ! is_array($receipt)
+            || $task->status !== TaskStatus::Leased || $task->lease_owner !== $this->worker['worker_id']
+            || $task->queue !== $this->queue || ! $task->lease_expires_at?->isFuture()
+        ) {
+            throw new TaskDeliveryOfferRejected('task_offer_lost_authority');
+        }
+        $this->assertReceipt($receipt, $taskId, (int) $task->attempt_count);
+    }
+
+    private function assertContext(array $data): void
+    {
+        foreach (['namespace' => $this->namespace, 'queue' => $this->queue, 'poll_request_id' => $this->pollRequestId, 'route' => $this->route, 'task_kinds' => $this->taskKinds] as $key => $expected) {
+            if (($data[$key] ?? null) !== $expected) {
+                throw new TaskDeliveryOfferRejected('task_offer_mismatch');
+            }
+        }
+        if (($data['worker']['registration_incarnation'] ?? null) !== $this->worker['registration_incarnation']
+            || ($data['worker']['registration_id'] ?? null) !== $this->worker['registration_id']
+            || ! hash_equals($data['delivery_id'], TaskDeliveryOffer::identifier($data))) {
+            throw new TaskDeliveryOfferRejected('worker_registration_lost_authority');
+        }
+    }
+
+    /** Serialize against any in-flight acceptance before declaring no live claim. */
+    private function assertExpiredOfferOutcome(): void
+    {
+        if ($this->offer === null || now()->getTimestampMs() < $this->offer['expires_at_ms']) {
+            return;
+        }
+        DB::transaction(function (): void {
+            $this->lockWorker();
+            $id = $this->offer['task_id'];
+            $kind = $this->offer['task_kind'];
+            $task = match ($kind) {
+                'workflow', 'activity' => NamespaceWorkflowScope::task($this->namespace, $id)?->toArray(),
+                'update_validation' => WorkflowUpdateValidationTask::query()->where('namespace', $this->namespace)->find($id)?->toArray(),
+                'query' => app(WorkflowQueryTaskBroker::class)->task($id),
+                default => null,
+            };
+            $receipt = match ($kind) {
+                'workflow', 'activity' => $task['payload'][self::RECEIPT_KEY] ?? null,
+                'update_validation' => $task['delivery_receipt'] ?? null,
+                default => $task[self::RECEIPT_KEY] ?? null,
+            };
+            if (! is_array($task) || ($task['namespace'] ?? null) !== $this->namespace
+                || ($task['queue'] ?? $task['task_queue'] ?? null) !== $this->queue
+                || ($task['status'] ?? null) !== 'leased' || ($task['lease_owner'] ?? null) !== $this->worker['worker_id']
+                || ! is_string($task['lease_expires_at'] ?? null) || ! Carbon::parse($task['lease_expires_at'])->isFuture()
+                || ! is_array($receipt) || ($receipt['delivery_id'] ?? null) !== $this->offer['delivery_id']) {
+                throw new TaskDeliveryOfferRejected('task_offer_expired');
+            }
+            $this->assertReceipt($receipt, $id, (int) ($task['attempt_count'] ?? 0));
+        });
+    }
+
+    private function assertWorkerCurrent(array $data): void
+    {
+        $current = WorkerRegistration::query()->where('namespace', $this->namespace)
+            ->whereKey($this->worker['registration_id'])->lockForUpdate()->first();
+        if (! $current instanceof WorkerRegistration || ! is_string($current->registration_token)
+            || hash('sha256', $current->registration_token) !== $this->worker['registration_incarnation']) {
+            throw new TaskDeliveryOfferRejected('worker_registration_lost_authority');
+        }
+        // This contract requires the exact registration credential, unlike a
+        // legacy poll. Routine heartbeat timestamps do not replace that
+        // incarnation. Preserve the original routing/process identity checks.
+        $currentFence = WorkerPollFence::snapshot($current);
+        $expected = $data['worker'];
+        $expected['updated_at'] = $currentFence['updated_at'];
+        $expected['last_heartbeat_at'] = $currentFence['last_heartbeat_at'];
+        if (! WorkerPollFence::isCurrentForUpdate($expected)) {
+            throw new TaskDeliveryOfferRejected('worker_registration_lost_authority');
+        }
+    }
+
+    private function fingerprint(WorkflowTask $task): string
+    {
+        $run = WorkflowRun::query()->where('namespace', $this->namespace)->find($task->workflow_run_id);
+        $executionId = $task->task_type->value === 'activity' ? ($task->payload['activity_execution_id'] ?? null) : null;
+        $activityType = $executionId === null ? null : ActivityExecution::query()->whereKey($executionId)->value('activity_type');
+
+        return hash('sha256', json_encode([
+            $task->id, $task->namespace, $task->workflow_run_id, $task->task_type->value,
+            $task->queue, $task->compatibility, $task->available_at?->toJSON(),
+            $run?->workflow_type, $executionId, $activityType,
+        ], JSON_THROW_ON_ERROR));
+    }
+}
diff --git a/app/Support/WorkflowQueryTaskBroker.php b/app/Support/WorkflowQueryTaskBroker.php
index a374270d4..2626983df 100644
--- a/app/Support/WorkflowQueryTaskBroker.php
+++ b/app/Support/WorkflowQueryTaskBroker.php
@@ -544,6 +544,7 @@ public function pollResult(
         WorkerRegistration $worker,
         ?string $pollRequestId = null,
         ?int $timeoutSeconds = null,
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         $pollRequestId = $this->stringValue($pollRequestId);
         $taskQueue = (string) $worker->task_queue;
@@ -565,7 +566,7 @@ public function pollResult(
 
         $this->rememberQueryTaskPollingWorker($namespace, $taskQueue, $worker->worker_id, $timeoutSeconds);
 
-        if ($pollRequestId !== null) {
+        if ($delivery === null && $pollRequestId !== null) {
             $this->pollRequests->markCurrentIfFresh($namespace, $taskQueue, $buildId, $worker->worker_id, $pollRequestId);
 
             return $this->coordinatedPollResult(
@@ -589,9 +590,11 @@ public function pollResult(
             $supportedWorkflowTypes,
             $workflowDefinitionFingerprints,
             $buildId,
-            timeoutSeconds: $this->queryPollTimeoutSeconds($timeoutSeconds),
+            pollRequestId: $pollRequestId,
+            timeoutSeconds: $delivery?->accepting() ? 0 : $this->queryPollTimeoutSeconds($timeoutSeconds),
             workerPollFence: $workerPollFence,
             crossKindWake: $crossKindWake,
+            delivery: $delivery,
         ));
     }
 
@@ -745,6 +748,9 @@ private function pollResultFromCached(array $cached): array
      */
     private function pollResultFromValue(?array $task): array
     {
+        if (isset($task['delivery_offer'])) {
+            return ['task' => $task, 'poll_status' => 'offered'];
+        }
         if (is_array($task) && ! array_key_exists('query_task_id', $task)) {
             return [
                 'task' => null,
@@ -872,67 +878,42 @@ private function performPoll(
         ?int $timeoutSeconds = null,
         array $workerPollFence = [],
         bool $crossKindWake = false,
+        ?WorkerTaskDelivery $delivery = null,
     ): ?array {
         $workflowBlockStatus = null;
 
         $result = $this->longPoller->until(
-            function () use ($namespace, $taskQueue, $leaseOwner, $supportedWorkflowTypes, $workflowDefinitionFingerprints, $buildId, $pollRequestId, $workerPollFence, &$workflowBlockStatus): ?array {
+            function () use ($namespace, $taskQueue, $leaseOwner, $supportedWorkflowTypes, $workflowDefinitionFingerprints, $buildId, $pollRequestId, $workerPollFence, $delivery, &$workflowBlockStatus): ?array {
                 app(StoragePressure::class)->requireNewWork();
 
-                if ($workerPollFence !== [] && ! WorkerPollFence::isCurrent($workerPollFence)) {
-                    return ['poll_status' => 'stale_worker_registration'];
-                }
-
-                if (
-                    $pollRequestId !== null
-                    && ! $this->pollRequests->isCurrent($namespace, $taskQueue, $buildId, $leaseOwner, $pollRequestId)
-                ) {
-                    return ['poll_status' => 'superseded'];
+                $refusal = $this->queryPollRefusal($namespace, $taskQueue, $leaseOwner, $buildId, $pollRequestId, $workerPollFence, $delivery);
+                if ($refusal !== null) {
+                    return $refusal;
                 }
-
                 if ($pollRequestId !== null) {
-                    $activeLease = $this->activeLeasedTaskForPoller(
-                        $namespace,
-                        $taskQueue,
-                        $leaseOwner,
-                        $supportedWorkflowTypes,
-                        $workflowDefinitionFingerprints,
-                        $buildId,
-                        $pollRequestId,
+                    $offeredId = $delivery?->offeredTaskId('query');
+                    $offeredTask = $offeredId === null ? null : $this->task($offeredId);
+                    $activeLease = ($offeredTask['status'] ?? null) === 'leased' ? $offeredTask : $this->activeLeasedTaskForPoller(
+                        $namespace, $taskQueue, $leaseOwner, $supportedWorkflowTypes, $workflowDefinitionFingerprints, $buildId, $pollRequestId,
                     );
-
                     if (is_array($activeLease)) {
+                        if ($delivery !== null) {
+                            $this->assertQueryDelivery($delivery, $activeLease, $namespace, $taskQueue, $leaseOwner);
+                        }
+
                         return $this->queryTaskPayload($activeLease);
                     }
                 }
-
                 $workflowBlockStatus = $this->pendingTaskWorkflowBlockStatus(
-                    $namespace,
-                    $taskQueue,
-                    $leaseOwner,
-                    $supportedWorkflowTypes,
-                    $workflowDefinitionFingerprints,
-                    $buildId,
+                    $namespace, $taskQueue, $leaseOwner, $supportedWorkflowTypes, $workflowDefinitionFingerprints, $buildId,
                 );
 
-                if ($workflowBlockStatus === 'workflow_task_pending') {
-                    return ['poll_status' => $workflowBlockStatus];
-                }
-
-                if ($workflowBlockStatus === 'workflow_task_leased') {
-                    return null;
-                }
-
-                return $this->claimNext(
-                    $namespace,
-                    $taskQueue,
-                    $leaseOwner,
-                    $supportedWorkflowTypes,
-                    $workflowDefinitionFingerprints,
-                    $buildId,
-                    $pollRequestId,
-                    $workerPollFence,
-                );
+                return match ($workflowBlockStatus) {
+                    'workflow_task_pending' => ['poll_status' => $workflowBlockStatus],
+                    'workflow_task_leased' => null,
+                    default => $this->claimNext($namespace, $taskQueue, $leaseOwner, $supportedWorkflowTypes,
+                        $workflowDefinitionFingerprints, $buildId, $pollRequestId, $workerPollFence, $delivery),
+                };
             },
             static fn (?array $task): bool => $task !== null,
             timeoutSeconds: $timeoutSeconds,
@@ -959,6 +940,22 @@ function () use ($namespace, $taskQueue, $leaseOwner, $supportedWorkflowTypes, $
         ) ? null : $result;
     }
 
+    private function queryPollRefusal(
+        string $namespace, string $taskQueue, string $leaseOwner, ?string $buildId,
+        ?string $pollRequestId, array $workerPollFence, ?WorkerTaskDelivery $delivery,
+    ): ?array {
+        $delivery?->lockWorker();
+        if ($delivery === null && $workerPollFence !== [] && ! WorkerPollFence::isCurrent($workerPollFence)) {
+            return ['poll_status' => 'stale_worker_registration'];
+        }
+        if ($delivery === null && $pollRequestId !== null
+            && ! $this->pollRequests->isCurrent($namespace, $taskQueue, $buildId, $leaseOwner, $pollRequestId)) {
+            return ['poll_status' => 'superseded'];
+        }
+
+        return null;
+    }
+
     /**
      * @param  array{codec: string, blob: string}|null  $resultEnvelope
      * @return array<string, mixed>
@@ -1689,6 +1686,7 @@ private function claimNext(
         ?string $buildId = null,
         ?string $pollRequestId = null,
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): ?array {
         $task = $this->withQueueLock(
             $namespace,
@@ -1703,10 +1701,11 @@ private function claimNext(
                 $pollRequestId,
                 $workerPollFence,
                 $lock,
+                $delivery,
             )),
         );
 
-        if (in_array($task['poll_status'] ?? null, ['superseded', 'stale_worker_registration'], true)) {
+        if (isset($task['delivery_offer']) || in_array($task['poll_status'] ?? null, ['superseded', 'stale_worker_registration'], true)) {
             return $task;
         }
 
@@ -1793,11 +1792,30 @@ private function claimNextPendingTask(
         ?string $pollRequestId = null,
         array $workerPollFence = [],
         ?LockContract $queueLock = null,
+        ?WorkerTaskDelivery $delivery = null,
     ): ?array {
+        if ($delivery !== null) {
+            $delivery->lockWorker();
+            $offeredId = $delivery->offeredTaskId('query');
+            $offeredTask = $offeredId === null ? null : $this->task($offeredId);
+            $accepted = ($offeredTask['status'] ?? null) === 'leased' ? $offeredTask : $this->activeLeasedTaskForPoller(
+                $namespace, $taskQueue, $leaseOwner, $supportedWorkflowTypes, $workflowDefinitionFingerprints, $buildId, $pollRequestId,
+            );
+            if (is_array($accepted)) {
+                $this->assertQueryDelivery($delivery, $accepted, $namespace, $taskQueue, $leaseOwner);
+
+                return $accepted;
+            }
+        }
         $ids = $this->pendingTaskIds($namespace, $taskQueue);
         $remaining = [];
 
         foreach ($ids as $queryTaskId) {
+            if ($delivery?->accepting() && $delivery->offeredTaskId('query') !== $queryTaskId) {
+                $remaining[] = $queryTaskId;
+
+                continue;
+            }
             $task = $this->task($queryTaskId);
 
             if (! is_array($task) || ($task['status'] ?? null) !== 'pending') {
@@ -1821,13 +1839,13 @@ private function claimNextPendingTask(
             }
 
             if (
-                $pollRequestId !== null
+                $delivery === null && $pollRequestId !== null
                 && ! $this->pollRequests->isCurrent($namespace, $taskQueue, $buildId, $leaseOwner, $pollRequestId)
             ) {
                 return ['poll_status' => 'superseded'];
             }
 
-            if ($workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
+            if ($delivery === null && $workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
                 return ['poll_status' => 'stale_worker_registration'];
             }
 
@@ -1844,8 +1862,20 @@ private function claimNextPendingTask(
 
             $deadline = $this->originalDeadline($task);
             $this->assertQueueLockOwned($namespace, $taskQueue, $queueLock);
+            if ($delivery !== null) {
+                $fingerprint = $this->deliveryFingerprint($task);
+                if (! $delivery->accepting()) {
+                    return $delivery->issue($queryTaskId, 'query', (int) ($task['attempt_count'] ?? 0), $fingerprint);
+                }
+                $delivery->assertAcceptance($queryTaskId, 'query', (int) ($task['attempt_count'] ?? 0), $fingerprint);
+            }
             $leaseExpiresAt = now()->addSeconds($this->leaseTtlSeconds())->min($deadline);
-            if (! $this->store()->add($this->leaseKey($queryTaskId), $leaseOwner, $leaseExpiresAt)) {
+            // The queue lock and task record govern positive acceptance.
+            // A stale legacy hint cannot reserve an authoritative pending
+            // task, and this mode does not create or delete such hints.
+            $reserved = $delivery !== null
+                || $this->store()->add($this->leaseKey($queryTaskId), $leaseOwner, $leaseExpiresAt);
+            if (! $reserved) {
                 $remaining[] = $queryTaskId;
 
                 continue;
@@ -1860,18 +1890,7 @@ private function claimNextPendingTask(
             $task['leased_at'] = now()->toJSON();
             $task['poll_request_id'] = $pollRequestId;
 
-            $this->putTask($task);
-            $this->appendLeasedTask($namespace, $taskQueue, $leaseOwner, $queryTaskId);
-            $this->storePendingTaskIds(
-                $namespace,
-                $taskQueue,
-                array_values(array_filter(
-                    $ids,
-                    static fn (string $id): bool => $id !== $queryTaskId,
-                )),
-            );
-
-            return $task;
+            return $this->commitQueryClaim($task, $ids, $delivery, $queueLock);
         }
 
         $this->storePendingTaskIds($namespace, $taskQueue, $remaining);
@@ -1879,6 +1898,74 @@ private function claimNextPendingTask(
         return null;
     }
 
+    private function commitQueryClaim(array $task, array $pendingIds, ?WorkerTaskDelivery $delivery, ?LockContract $queueLock): array
+    {
+        $namespace = $task['namespace'];
+        $taskQueue = $task['task_queue'];
+        $queryTaskId = $task['query_task_id'];
+        $leaseOwner = $task['lease_owner'];
+        try {
+            if ($delivery !== null) {
+                // Prepare discovery before publishing the authority record.
+                // A crash may leave a harmless pending ID in this index, but
+                // must not leave an accepted task undiscoverable on recovery.
+                $this->appendLeasedTask($namespace, $taskQueue, $leaseOwner, $queryTaskId);
+                if (! in_array($queryTaskId, $this->leasedTaskIds($namespace, $taskQueue, $leaseOwner), true)) {
+                    throw new QueryTaskQueueUnavailableException($namespace, $taskQueue, 'The task acceptance index was not committed.');
+                }
+                $task[WorkerTaskDelivery::RECEIPT_KEY] = $delivery->acceptedReceipt($task['attempt_count']);
+            }
+            $this->assertQueueLockOwned($namespace, $taskQueue, $queueLock);
+            $this->putTask($task);
+            if ($delivery !== null) {
+                $confirmed = $this->task($queryTaskId);
+                if (! is_array($confirmed) || ($confirmed['status'] ?? null) !== 'leased') {
+                    throw new QueryTaskQueueUnavailableException($namespace, $taskQueue, 'The task acceptance receipt was not committed.');
+                }
+                $this->assertQueryDelivery($delivery, $confirmed, $namespace, $taskQueue, $leaseOwner);
+                $task = $confirmed;
+            }
+            if ($delivery === null) {
+                $this->appendLeasedTask($namespace, $taskQueue, $leaseOwner, $queryTaskId);
+            }
+            $this->storePendingTaskIds($namespace, $taskQueue, array_values(array_filter(
+                $pendingIds, static fn (string $id): bool => $id !== $queryTaskId,
+            )));
+        } catch (\Throwable $exception) {
+            if ($delivery === null || $exception instanceof TaskDeliveryOfferRejected) {
+                throw $exception;
+            }
+            // An uncertain cache write may already contain an accepted live
+            // receipt. Preserve it and let the identical packet replay it.
+            throw new QueryTaskQueueUnavailableException($namespace, $taskQueue, 'The task acceptance could not be confirmed.', $exception);
+        }
+
+        return $task;
+    }
+
+    private function assertQueryDelivery(WorkerTaskDelivery $delivery, array $task, string $namespace, string $taskQueue, string $leaseOwner): void
+    {
+        if (($task['namespace'] ?? null) !== $namespace || ($task['task_queue'] ?? null) !== $taskQueue
+            || ($task['lease_owner'] ?? null) !== $leaseOwner || ($task['status'] ?? null) !== 'leased'
+            || ! $this->timestamp($task['lease_expires_at'] ?? null)?->isFuture()
+            || $this->deadlineRefusal($task) !== null) {
+            throw new TaskDeliveryOfferRejected('task_offer_lost_authority');
+        }
+        $delivery->assertReceipt($task[WorkerTaskDelivery::RECEIPT_KEY] ?? null,
+            $task['query_task_id'], (int) ($task['attempt_count'] ?? 0));
+    }
+
+    private function deliveryFingerprint(array $task): string
+    {
+        foreach (['status', 'attempt_count', 'lease_owner', 'lease_expires_at', 'registration_incarnation',
+            'leased_at', 'poll_request_id', WorkerTaskDelivery::RECEIPT_KEY] as $key) {
+            unset($task[$key]);
+        }
+        ksort($task);
+
+        return hash('sha256', json_encode($task, JSON_THROW_ON_ERROR));
+    }
+
     /**
      * @param  array<string, mixed>  $task
      * @param  list<string>  $supportedWorkflowTypes
diff --git a/app/Support/WorkflowTaskKindSelector.php b/app/Support/WorkflowTaskKindSelector.php
index aabcaae9c..808214ad3 100644
--- a/app/Support/WorkflowTaskKindSelector.php
+++ b/app/Support/WorkflowTaskKindSelector.php
@@ -54,7 +54,8 @@ public function select(
             $result = $this->claimInOrder([$first, $second], $claim);
             $leasedKind = $result['task']['task_kind'] ?? null;
 
-            if (is_string($leasedKind) && in_array($leasedKind, [self::WORKFLOW, self::UPDATE_VALIDATION], true)) {
+            if (! isset($result['task']['delivery_offer'])
+                && is_string($leasedKind) && in_array($leasedKind, [self::WORKFLOW, self::UPDATE_VALIDATION], true)) {
                 DB::table(self::TABLE)
                     ->where('namespace', $namespace)
                     ->where('task_queue', $taskQueue)
diff --git a/app/Support/WorkflowTaskPoller.php b/app/Support/WorkflowTaskPoller.php
index b0c1d4ab8..0fc41b71b 100644
--- a/app/Support/WorkflowTaskPoller.php
+++ b/app/Support/WorkflowTaskPoller.php
@@ -64,6 +64,7 @@ public function poll(
         bool $acceptsQueryTasks = false,
         ?int $timeoutSeconds = null,
         array $taskKinds = ['workflow'],
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         $pollRequestId = $this->nonEmptyString($pollRequestId);
         $taskKinds = WorkflowTaskPollRequestStore::normalizeTaskKinds($taskKinds);
@@ -76,6 +77,25 @@ public function poll(
             ];
         }
 
+        if ($delivery !== null) {
+            $result = $this->performPoll(
+                request: $request, namespace: $namespace, taskQueue: $taskQueue,
+                leaseOwner: $leaseOwner, buildId: $buildId, worker: $worker,
+                pollRequestId: $pollRequestId, historyPageSize: $historyPageSize,
+                acceptHistoryEncoding: $acceptHistoryEncoding, supportedWorkflowTypes: $supportedWorkflowTypes,
+                workflowDefinitionFingerprints: $workflowDefinitionFingerprints,
+                acceptsQueryTasks: $acceptsQueryTasks,
+                timeoutSeconds: $delivery->accepting() ? 0 : $timeoutSeconds,
+                taskKinds: $taskKinds, delivery: $delivery,
+            );
+            if (is_array($result['task'] ?? null) && ! isset($result['task']['delivery_offer'])
+                && ($result['task']['task_kind'] ?? null) !== 'update_validation') {
+                $delivery->assertReplay($result['task']['task_id']);
+            }
+
+            return $result;
+        }
+
         if ($pollRequestId !== null && $this->cache->available()) {
             $taskKinds = $this->pollRequests->bindTaskKinds(
                 $namespace,
@@ -450,6 +470,7 @@ private function performPoll(
         bool $acceptsQueryTasks = false,
         ?int $timeoutSeconds = null,
         array $taskKinds = ['workflow'],
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         $limit = max(10, max(1, (int) config('server.polling.max_tasks_per_poll', 1)) * 10);
         $nextProbeAt = null;
@@ -502,12 +523,14 @@ function () use (
                 $taskKinds,
                 $limit,
                 $workerPollFence,
+                $delivery,
                 &$nextProbeAt,
                 &$resolvedResult,
             ): ?array {
                 app(StoragePressure::class)->requireNewWork();
 
-                if (! WorkerPollFence::isCurrent($workerPollFence)) {
+                $delivery?->lockWorker();
+                if ($delivery === null && ! WorkerPollFence::isCurrent($workerPollFence)) {
                     $resolvedResult = [
                         'task' => null,
                         'poll_status' => 'stale_worker_registration',
@@ -534,6 +557,7 @@ function () use (
                     acceptsQueryTasks: $acceptsQueryTasks,
                     supportsQueryTasks: $supportsQueryTasks,
                     workerPollFence: $workerPollFence,
+                    delivery: $delivery,
                 );
                 $nextProbeAt = $resolvedResult['next_probe_at'] ?? null;
 
@@ -608,6 +632,7 @@ private function nextRequestedTask(
         bool $acceptsQueryTasks = false,
         bool $supportsQueryTasks = false,
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         return $this->taskKindSelector->select(
             $namespace,
@@ -629,9 +654,10 @@ function (string $taskKind) use (
                 $acceptsQueryTasks,
                 $supportsQueryTasks,
                 $workerPollFence,
+                $delivery,
             ): array {
                 if ($taskKind === 'update_validation') {
-                    $validationTask = $this->updateValidationTasks->claimAvailable($namespace, $worker);
+                    $validationTask = $this->updateValidationTasks->claimAvailable($namespace, $worker, $delivery);
 
                     return [
                         'task' => $validationTask,
@@ -655,6 +681,7 @@ function (string $taskKind) use (
                     acceptsQueryTasks: $acceptsQueryTasks,
                     supportsQueryTasks: $supportsQueryTasks,
                     workerPollFence: $workerPollFence,
+                    delivery: $delivery,
                 );
 
                 if (is_array($result['task'] ?? null)) {
@@ -686,6 +713,7 @@ private function nextTask(
         bool $acceptsQueryTasks = false,
         bool $supportsQueryTasks = false,
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         return $this->withWorkerCompatibility(
             $namespace,
@@ -705,6 +733,7 @@ function () use (
                 $acceptsQueryTasks,
                 $supportsQueryTasks,
                 $workerPollFence,
+                $delivery,
             ): array {
                 $this->runDueServiceModeTimers($namespace, $taskQueue, $buildId);
 
@@ -722,6 +751,8 @@ function () use (
                     );
 
                     if (is_array($task)) {
+                        $delivery?->assertReplay($task['task_id']);
+
                         return [
                             'task' => $task,
                             'poll_status' => 'leased',
@@ -750,6 +781,7 @@ function () use (
                                 acceptHistoryEncoding: $acceptHistoryEncoding,
                                 supportedWorkflowTypes: $supportedWorkflowTypes,
                                 workerPollFence: $workerPollFence,
+                                delivery: $delivery,
                             ),
                         ),
                     );
@@ -798,6 +830,7 @@ function () use (
                                     acceptHistoryEncoding: $acceptHistoryEncoding,
                                     supportedWorkflowTypes: $supportedWorkflowTypes,
                                     workerPollFence: $workerPollFence,
+                                    delivery: $delivery,
                                 ),
                             ),
                         );
@@ -967,8 +1000,9 @@ private function claimReadyTask(
         ?string $acceptHistoryEncoding = null,
         array $supportedWorkflowTypes = [],
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): ?array {
-        $readyTasks = $this->pollReplayEligibleReadyTasks(
+        $readyTasks = $delivery?->accepting() ? $delivery->candidates('workflow') : $this->pollReplayEligibleReadyTasks(
             namespace: $namespace,
             taskQueue: $taskQueue,
             leaseOwner: $leaseOwner,
@@ -1056,6 +1090,15 @@ private function claimReadyTask(
                 continue;
             }
 
+            if ($delivery !== null && ! $delivery->accepting()) {
+                $offer = $delivery->offerTask($readyTask, 'workflow');
+                if ($offer !== null) {
+                    return $offer;
+                }
+
+                continue;
+            }
+
             $claim = DB::transaction(function () use (
                 $namespace,
                 $taskQueue,
@@ -1065,8 +1108,10 @@ private function claimReadyTask(
                 $pollRequestId,
                 $workerPollFence,
                 $runId,
+                $delivery,
             ): array {
-                if ($workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
+                $delivery?->lockWorker();
+                if ($delivery === null && $workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
                     return ['claimed' => false, 'reason' => 'stale_worker_registration'];
                 }
 
@@ -1099,6 +1144,7 @@ private function claimReadyTask(
                     $pollRequestId,
                 );
 
+                $delivery?->assertClaimable($taskId, 'workflow');
                 $claim = $this->bridge->claimStatus($taskId, $leaseOwner);
 
                 if (($claim['claimed'] ?? false) === true) {
@@ -1109,6 +1155,7 @@ private function claimReadyTask(
                         $pollRequestId,
                         $workerPollFence,
                     );
+                    $delivery?->bindAccepted($taskId);
                 }
 
                 return $claim;
diff --git a/app/Support/WorkflowUpdateValidationTaskBroker.php b/app/Support/WorkflowUpdateValidationTaskBroker.php
index e8e0e1c29..4e707f058 100644
--- a/app/Support/WorkflowUpdateValidationTaskBroker.php
+++ b/app/Support/WorkflowUpdateValidationTaskBroker.php
@@ -149,6 +149,7 @@ public function pollResult(
         string $namespace,
         WorkerRegistration $worker,
         ?int $timeoutSeconds = null,
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         if (! $this->workerSupportsValidation($worker)) {
             return ['task' => null, 'poll_status' => 'empty'];
@@ -157,9 +158,9 @@ public function pollResult(
         $fence = $this->workerFence($worker);
 
         return $this->longPoller->until(
-            fn (): array => $this->claimAvailableResult($namespace, $worker, $fence),
+            fn (): array => $this->claimAvailableResult($namespace, $worker, $fence, $delivery),
             static fn (array $candidate): bool => $candidate['poll_status'] !== 'empty',
-            $timeoutSeconds,
+            $delivery?->accepting() ? 0 : $timeoutSeconds,
             wakeChannels: $this->signals->updateValidationTaskPollChannels($namespace, $worker->task_queue),
             reserveWorkerWaitSlot: true,
             waitSlotPool: 'query-task',
@@ -172,16 +173,16 @@ public function pollResult(
      *
      * @return array<string, mixed>|null
      */
-    public function claimAvailable(string $namespace, WorkerRegistration $worker): ?array
+    public function claimAvailable(string $namespace, WorkerRegistration $worker, ?WorkerTaskDelivery $delivery = null): ?array
     {
-        return $this->claimAvailableResult($namespace, $worker, $this->workerFence($worker))['task'];
+        return $this->claimAvailableResult($namespace, $worker, $this->workerFence($worker), $delivery)['task'];
     }
 
     /**
      * @param  array<string, mixed>  $fence
      * @return array{task: array<string, mixed>|null, poll_status: string}
      */
-    private function claimAvailableResult(string $namespace, WorkerRegistration $worker, array $fence): array
+    private function claimAvailableResult(string $namespace, WorkerRegistration $worker, array $fence, ?WorkerTaskDelivery $delivery = null): array
     {
         app(StoragePressure::class)->requireNewWork();
 
@@ -189,10 +190,10 @@ private function claimAvailableResult(string $namespace, WorkerRegistration $wor
             return ['task' => null, 'poll_status' => 'empty'];
         }
 
-        $claim = $this->claim($namespace, $worker, $fence);
+        $claim = $this->claim($namespace, $worker, $fence, $delivery);
 
         return [
-            'task' => $claim['task'] instanceof WorkflowUpdateValidationTask ? $this->taskPayload($claim['task']) : null,
+            'task' => $claim['task'] instanceof WorkflowUpdateValidationTask ? $this->taskPayload($claim['task']) : $claim['task'],
             'poll_status' => $claim['poll_status'],
         ];
     }
@@ -386,50 +387,99 @@ private function findOrCreateTask(
      * @param  array<string, mixed>  $fence
      * @return array{task: WorkflowUpdateValidationTask|null, poll_status: string}
      */
-    private function claim(string $namespace, WorkerRegistration $worker, array $fence): array
+    private function claim(string $namespace, WorkerRegistration $worker, array $fence, ?WorkerTaskDelivery $delivery = null): array
     {
-        $claim = function () use ($namespace, $worker, $fence): array {
-            if (! $this->workerFenceIsCurrent($worker, $fence)) {
-                return ['task' => null, 'poll_status' => 'stale_worker_registration'];
-            }
+        if ($delivery?->accepting() && $delivery->offeredTaskId('update_validation') === null) {
+            return ['task' => null, 'poll_status' => 'empty'];
+        }
 
-            /** @var Collection<int, WorkflowUpdateValidationTask> $tasks */
-            $tasks = WorkflowUpdateValidationTask::query()
-                ->where('namespace', $namespace)
-                ->where('task_queue', $worker->task_queue)
-                ->where(static function ($query): void {
-                    $query->where('status', WorkflowUpdateValidationTask::STATUS_PENDING)
-                        ->orWhere(static function ($query): void {
-                            $query->where('status', WorkflowUpdateValidationTask::STATUS_LEASED)
-                                ->where('lease_expires_at', '<=', now());
-                        });
-                })
-                ->orderBy('created_at')
-                ->lockForUpdate()
-                ->limit(25)
-                ->get();
-
-            foreach ($tasks as $task) {
-                if (! $this->workerMatchesTask($worker, $task)) {
-                    continue;
-                }
+        return app(WorkerProtocolMutationRetrier::class)->run(
+            fn (): array => DB::transaction(fn (): array => $this->claimWithFence($namespace, $worker, $fence, $delivery)),
+        );
+    }
+
+    private function claimWithFence(string $namespace, WorkerRegistration $worker, array $fence, ?WorkerTaskDelivery $delivery): array
+    {
+        $resolved = $this->validationReplayOrRefusal($worker, $fence, $delivery);
+        if ($resolved !== null) {
+            return $resolved;
+        }
+        /** @var Collection<int, WorkflowUpdateValidationTask> $tasks */
+        $tasks = WorkflowUpdateValidationTask::query()
+            ->where('namespace', $namespace)
+            ->where('task_queue', $worker->task_queue)
+            ->when($delivery?->accepting(), fn ($query) => $query->whereKey($delivery->offeredTaskId('update_validation')))
+            ->where(static function ($query): void {
+                $query->where('status', WorkflowUpdateValidationTask::STATUS_PENDING)
+                    ->orWhere(static function ($query): void {
+                        $query->where('status', WorkflowUpdateValidationTask::STATUS_LEASED)
+                            ->where('lease_expires_at', '<=', now());
+                    });
+            })
+            ->orderBy('created_at')->lockForUpdate()->limit(25)->get();
+        foreach ($tasks as $task) {
+            if ($this->workerMatchesTask($worker, $task)) {
+                return $this->deliverValidationTask($task, $worker, $delivery);
+            }
+        }
 
-                $task->forceFill([
-                    'status' => WorkflowUpdateValidationTask::STATUS_LEASED,
-                    'attempt_count' => $task->attempt_count + 1,
-                    'lease_owner' => $worker->worker_id,
-                    'lease_expires_at' => now()->addSeconds($this->leaseTimeoutSeconds()),
-                ])->save();
+        return ['task' => null, 'poll_status' => 'empty'];
+    }
 
-                return ['task' => $task, 'poll_status' => 'leased'];
+    private function validationReplayOrRefusal(WorkerRegistration $worker, array $fence, ?WorkerTaskDelivery $delivery): ?array
+    {
+        if ($delivery === null && ! $this->workerFenceIsCurrent($worker, $fence)) {
+            return ['task' => null, 'poll_status' => 'stale_worker_registration'];
+        }
+        if ($delivery === null) {
+            return null;
+        }
+        $delivery->lockWorker();
+        $accepted = WorkflowUpdateValidationTask::query()->where('delivery_poll_key', $delivery->pollKey())->lockForUpdate()->first();
+        if ($accepted instanceof WorkflowUpdateValidationTask) {
+            if ($accepted->status !== WorkflowUpdateValidationTask::STATUS_LEASED
+                || $accepted->lease_owner !== $worker->worker_id || ! $accepted->lease_expires_at?->isFuture()) {
+                throw new TaskDeliveryOfferRejected('task_offer_lost_authority');
             }
+            $delivery->assertReceipt($accepted->delivery_receipt, $accepted->id, $accepted->attempt_count);
+        }
 
-            return ['task' => null, 'poll_status' => 'empty'];
-        };
+        return $accepted instanceof WorkflowUpdateValidationTask ? ['task' => $accepted, 'poll_status' => 'leased'] : null;
+    }
 
-        return app(WorkerProtocolMutationRetrier::class)->run(
-            fn (): array => DB::transaction($claim),
-        );
+    private function deliverValidationTask(WorkflowUpdateValidationTask $task, WorkerRegistration $worker, ?WorkerTaskDelivery $delivery): array
+    {
+        if ($delivery !== null && ! $delivery->accepting()) {
+            return ['task' => $delivery->issue($task->id, 'update_validation', $task->attempt_count, $this->deliveryFingerprint($task)), 'poll_status' => 'offered'];
+        }
+        if ($delivery !== null) {
+            $delivery->assertAcceptance($task->id, 'update_validation', $task->attempt_count, $this->deliveryFingerprint($task));
+            $delivery->assertSqlPollUnbound();
+        }
+        $task->forceFill([
+            'status' => WorkflowUpdateValidationTask::STATUS_LEASED,
+            'attempt_count' => $task->attempt_count + 1,
+            'lease_owner' => $worker->worker_id,
+            'lease_expires_at' => now()->addSeconds($this->leaseTimeoutSeconds()),
+        ])->save();
+        if ($delivery !== null) {
+            $task->forceFill(['delivery_poll_key' => $delivery->pollKey(),
+                'delivery_receipt' => $delivery->acceptedReceipt($task->attempt_count)])->save();
+            $delivery->acceptedReceipt($task->attempt_count);
+        }
+
+        return ['task' => $task, 'poll_status' => 'leased'];
+    }
+
+    private function deliveryFingerprint(WorkflowUpdateValidationTask $task): string
+    {
+        $attributes = $task->getRawOriginal();
+        foreach (['status', 'attempt_count', 'lease_owner', 'lease_expires_at', 'updated_at', 'delivery_poll_key', 'delivery_receipt'] as $key) {
+            unset($attributes[$key]);
+        }
+        ksort($attributes);
+
+        return hash('sha256', json_encode($attributes, JSON_THROW_ON_ERROR));
     }
 
     /** @param array<string, mixed> $fence */
diff --git a/bootstrap/app.php b/bootstrap/app.php
index 6d4d85540..4ba49fc60 100644
--- a/bootstrap/app.php
+++ b/bootstrap/app.php
@@ -13,6 +13,7 @@
 use App\Support\PollClientDisconnected;
 use App\Support\RuntimeExternalPayloadAudit;
 use App\Support\RuntimeExternalPayloadException;
+use App\Support\TaskDeliveryOfferRejected;
 use App\Support\WorkerProtocol;
 use Illuminate\Foundation\Application;
 use Illuminate\Foundation\Configuration\Exceptions;
@@ -43,7 +44,14 @@
         ]);
     })
     ->withExceptions(function (Exceptions $exceptions) {
-        $exceptions->dontReport([PollClientDisconnected::class]);
+        $exceptions->dontReport([PollClientDisconnected::class, TaskDeliveryOfferRejected::class]);
+
+        $exceptions->render(function (TaskDeliveryOfferRejected $exception) {
+            return WorkerProtocol::json([
+                'task' => null, 'poll_status' => $exception->reason,
+                'reason' => $exception->reason, 'retryable' => true,
+            ], 409);
+        });
 
         // Laravel selects renderers by the first parameter's reflected type.
         $exceptions->render(function (PollClientDisconnected $exception) {
diff --git a/config/server.php b/config/server.php
index 9e6c0d793..c779206ca 100644
--- a/config/server.php
+++ b/config/server.php
@@ -246,6 +246,8 @@
     */
 
     'polling' => [
+        // Offers start no lease. A worker must accept within this short budget.
+        'task_offer_accept_seconds' => 2,
         'timeout' => (int) EnvAuditor::env(
             'DW_WORKER_POLL_TIMEOUT',
             'WORKFLOW_SERVER_WORKER_POLL_TIMEOUT',
diff --git a/database/migrations/2026_10_25_000001_add_task_delivery_receipts.php b/database/migrations/2026_10_25_000001_add_task_delivery_receipts.php
new file mode 100644
index 000000000..fad223d88
--- /dev/null
+++ b/database/migrations/2026_10_25_000001_add_task_delivery_receipts.php
@@ -0,0 +1,35 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration
+{
+    public function up(): void
+    {
+        Schema::table('workflow_tasks', function (Blueprint $table): void {
+            $table->string('delivery_poll_key', 64)->nullable()->index();
+        });
+        Schema::table('workflow_update_validation_tasks', function (Blueprint $table): void {
+            $table->string('delivery_poll_key', 64)->nullable()->index();
+            $table->json('delivery_receipt')->nullable();
+        });
+    }
+
+    public function down(): void
+    {
+        if (Schema::hasTable('workflow_tasks')) {
+            Schema::table('workflow_tasks', function (Blueprint $table): void {
+                $table->dropIndex(['delivery_poll_key']);
+                $table->dropColumn('delivery_poll_key');
+            });
+        }
+        if (Schema::hasTable('workflow_update_validation_tasks')) {
+            Schema::table('workflow_update_validation_tasks', function (Blueprint $table): void {
+                $table->dropIndex(['delivery_poll_key']);
+                $table->dropColumn(['delivery_poll_key', 'delivery_receipt']);
+            });
+        }
+    }
+};
diff --git a/docs/worker-poll-progress.md b/docs/worker-poll-progress.md
index 54d6dee06..221c5eca9 100644
--- a/docs/worker-poll-progress.md
+++ b/docs/worker-poll-progress.md
@@ -1,7 +1,8 @@
 # Worker poll progress transport
 
-Development status: this transport is not in a published Server image or SDK.
-Task-delivery recovery and first-party SDK integration remain part
+Development status: this transport is not in a published Server image. PHP and
+Python SDK transport support is published; full task-delivery recovery and
+first-party SDK integration remain part
 of [Server #526](https://github.com/durable-workflow/server/issues/526).
 
 ## Negotiation
diff --git a/docs/worker-task-delivery.md b/docs/worker-task-delivery.md
new file mode 100644
index 000000000..8fde4599a
--- /dev/null
+++ b/docs/worker-task-delivery.md
@@ -0,0 +1,79 @@
+# Worker task delivery offers
+
+Development candidate for [Server #526](https://github.com/durable-workflow/server/issues/526).
+This contract is not available in a published Server image. SDK integration,
+concurrent SQL/shared-cache qualification, real proxy/restart tests and published
+acceptance remain required before delivery.
+
+Workers advertising `task_delivery_offers` with protocol 1.20 may use
+`POST /api/worker/{workflow,activity,query,update-validation}-tasks/offers`.
+Send `X-Durable-Workflow-Task-Delivery: offer-accept.v1`, the current
+`registration_token`, and a unique `poll_request_id`. Workflow requests retain
+their ordinary `task_kinds`, history and routing fields. Progress transport
+negotiation is independent and may also be used on these endpoints.
+
+An old Server has no offer endpoint, so it rejects the request before claiming
+work. A worker may choose ordinary polling before starting a logical poll on
+an older Server. Once acceptance has an uncertain outcome, it must keep retrying
+the same offer endpoint and identity. A response from a different, older node
+does not prove that the accepting node failed to commit.
+
+The first request returns an ordinary empty result or:
+
+```json
+{
+  "task": null,
+  "poll_status": "offered",
+  "task_offer": {
+    "schema": "durable-workflow.task-delivery.v1",
+    "task_id": "opaque-task-id",
+    "task_kind": "activity",
+    "token": "opaque-sealed-token",
+    "expires_at_ms": 1792908000000
+  }
+}
+```
+
+The sealed packet describes one observed task and its original attempt,
+registration incarnation, routing fence, namespace, queue, endpoint, logical
+poll and requested kinds. It starts no lease, activity attempt or history event.
+It reserves no task: another worker may claim it immediately. Losing the offer
+response requires no recovery job or expiry sweep.
+
+The worker selects one offer and repeats the original request with
+`task_offer_token`. Acceptance uses a zero-wait claim and revalidates the original
+task authority and normal routing/session admission. The acceptance budget is
+two seconds by default, controlled by `server.polling.task_offer_accept_seconds`
+and clamped to one through ten seconds. Original scheduling, query deadlines,
+lease lengths and business retry limits remain independent.
+
+SQL task acceptance records a receipt in the task's existing row in the same
+transaction as its normal claim. A late acceptance rolls back the claim,
+activity attempt and history together. An indexed poll key and the registration
+row lock prevent one multiplexed poll from claiming both task kinds. Query
+acceptance uses its existing queue lock and authoritative cache task record;
+its receipt stays in that record. Poll wait capacity and wake hints grant no
+task authority.
+
+Query acceptance prepares its discovery index before committing the authority
+record and confirms the stored receipt before returning a task. A failed or
+uncertain cache write returns a typed queue-unavailable response; retry the same
+packet. A committed receipt survives that failure, while an uncommitted task
+remains immediately available to peers. This mode creates no provisional lease
+hint, and an abandoned legacy hint cannot reserve an authoritative pending task.
+
+Only the accepted task response authorizes execution. An offer never does.
+If the response is lost, retry the exact selected token, registration, poll ID
+and endpoint. An already-accepted live claim can replay after the offer deadline
+without another attempt. A refused duplicate cannot revoke a peer's claim.
+After the original deadline, the Server serializes against in-flight acceptance
+under the registration lock and reads the actual task record. It either permits
+the original live receipt to replay or returns `task_offer_expired`; an empty
+acceptance response is not proof that no claim committed. This also resolves
+expired packets whose task was claimed by a peer or removed.
+Expired unaccepted packets, replaced registrations and changed task identities
+are refused. Query status remains a read and does not confirm delivery.
+
+Mixed-node routing, concurrent alternative offers, SQL/cache failure points,
+restart and worker replacement remain explicit qualification targets. The draft
+does not establish full #526 acceptance from source tests alone.
diff --git a/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml b/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml
index 022f00c17..8feb4a605 100644
--- a/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml
+++ b/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml
@@ -1,7 +1,7 @@
 openapi: 3.1.0
 info:
   title: durable-workflow.v2.worker-protocol-api
-  version: "56"
+  version: "57"
   summary: Durable Workflow worker-plane HTTP+JSON API
   description: >
     Normative OpenAPI specification for the worker-plane HTTP+JSON API:
@@ -481,6 +481,39 @@ paths:
         "200": { $ref: "#/components/responses/WorkerSessionOperation" }
         "404": { $ref: "#/components/responses/WorkerError" }
         "409": { $ref: "#/components/responses/WorkerError" }
+  /worker/workflow-tasks/offers:
+    post:
+      operationId: offerOrAcceptWorkflowTask
+      tags: [workflow-tasks]
+      description: >
+        Explicit protocol-1.20 offer/accept opt-in. An offer carries no lease and
+        reserves no task. Retry uncertain acceptance with the same endpoint,
+        poll_request_id, registration_token and task_offer_token.
+      parameters:
+        - $ref: "#/components/parameters/WorkerProtocolVersionHeader"
+        - $ref: "#/components/parameters/TaskDeliveryVersionHeader"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              allOf:
+                - $ref: "#/components/schemas/WorkflowTaskPollRequest"
+                - $ref: "#/components/schemas/TaskDeliveryRequestFields"
+      responses:
+        "200":
+          description: An unaccepted offer, the exact accepted task, or an ordinary empty result.
+          content:
+            application/json:
+              schema:
+                oneOf:
+                  - $ref: "#/components/schemas/TaskDeliveryOfferedResponse"
+                  - $ref: "#/components/schemas/WorkflowTaskPollResponse"
+        "400": { $ref: "#/components/responses/WorkerError" }
+        "409": { $ref: "#/components/responses/TaskDeliveryConflict" }
+        "422": { $ref: "#/components/responses/WorkerError" }
+        "429": { $ref: "#/components/responses/WorkerPollBackpressure" }
+        "503": { $ref: "#/components/responses/WorkerServiceUnavailable" }
   /worker/workflow-tasks/poll:
     post:
       operationId: pollWorkflowTask
@@ -963,6 +996,39 @@ paths:
       responses:
         "200": { $ref: "#/components/responses/WorkerEnvelope" }
         "409": { $ref: "#/components/responses/WorkerError" }
+  /worker/update-validation-tasks/offers:
+    post:
+      operationId: offerOrAcceptUpdateValidationTask
+      tags: [update-validation-tasks]
+      description: >
+        Explicit protocol-1.20 offer/accept opt-in. An offer carries no lease and
+        reserves no task. Retry uncertain acceptance with the same endpoint,
+        poll_request_id, registration_token and task_offer_token.
+      parameters:
+        - $ref: "#/components/parameters/WorkerProtocolVersionHeader"
+        - $ref: "#/components/parameters/TaskDeliveryVersionHeader"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              allOf:
+                - $ref: "#/components/schemas/UpdateValidationTaskPollRequest"
+                - $ref: "#/components/schemas/TaskDeliveryRequestFields"
+      responses:
+        "200":
+          description: An unaccepted offer, the exact accepted task, or an ordinary empty result.
+          content:
+            application/json:
+              schema:
+                oneOf:
+                  - $ref: "#/components/schemas/TaskDeliveryOfferedResponse"
+                  - $ref: "#/components/schemas/UpdateValidationTaskOfferAcceptedResponse"
+        "400": { $ref: "#/components/responses/WorkerError" }
+        "409": { $ref: "#/components/responses/TaskDeliveryConflict" }
+        "422": { $ref: "#/components/responses/WorkerError" }
+        "429": { $ref: "#/components/responses/WorkerPollBackpressure" }
+        "503": { $ref: "#/components/responses/WorkerServiceUnavailable" }
   /worker/update-validation-tasks/poll:
     post:
       operationId: pollUpdateValidationTask
@@ -1022,6 +1088,39 @@ paths:
         "404": { $ref: "#/components/responses/UpdateValidationTaskCompletionFailure" }
         "409": { $ref: "#/components/responses/UpdateValidationTaskCompletionFailure" }
         "422": { $ref: "#/components/responses/WorkerError" }
+  /worker/query-tasks/offers:
+    post:
+      operationId: offerOrAcceptQueryTask
+      tags: [query-tasks]
+      description: >
+        Explicit protocol-1.20 offer/accept opt-in. An offer carries no lease and
+        reserves no task. Retry uncertain acceptance with the same endpoint,
+        poll_request_id, registration_token and task_offer_token.
+      parameters:
+        - $ref: "#/components/parameters/WorkerProtocolVersionHeader"
+        - $ref: "#/components/parameters/TaskDeliveryVersionHeader"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              allOf:
+                - $ref: "#/components/schemas/PollRequest"
+                - $ref: "#/components/schemas/TaskDeliveryRequestFields"
+      responses:
+        "200":
+          description: An unaccepted offer, the exact accepted task, or an ordinary empty result.
+          content:
+            application/json:
+              schema:
+                oneOf:
+                  - $ref: "#/components/schemas/TaskDeliveryOfferedResponse"
+                  - $ref: "#/components/schemas/QueryTaskPollResult"
+        "400": { $ref: "#/components/responses/WorkerError" }
+        "409": { $ref: "#/components/responses/TaskDeliveryConflict" }
+        "422": { $ref: "#/components/responses/WorkerError" }
+        "429": { $ref: "#/components/responses/WorkerPollBackpressure" }
+        "503": { $ref: "#/components/responses/WorkerServiceUnavailable" }
   /worker/query-tasks/poll:
     post:
       operationId: pollQueryTask
@@ -1102,6 +1201,39 @@ paths:
         "200": { $ref: "#/components/responses/WorkerEnvelope" }
         "409": { $ref: "#/components/responses/WorkerError" }
         "503": { $ref: "#/components/responses/QueryTaskOutcomeUnavailable" }
+  /worker/activity-tasks/offers:
+    post:
+      operationId: offerOrAcceptActivityTask
+      tags: [activity-tasks]
+      description: >
+        Explicit protocol-1.20 offer/accept opt-in. An offer carries no lease and
+        reserves no task. Retry uncertain acceptance with the same endpoint,
+        poll_request_id, registration_token and task_offer_token.
+      parameters:
+        - $ref: "#/components/parameters/WorkerProtocolVersionHeader"
+        - $ref: "#/components/parameters/TaskDeliveryVersionHeader"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              allOf:
+                - $ref: "#/components/schemas/PollRequest"
+                - $ref: "#/components/schemas/TaskDeliveryRequestFields"
+      responses:
+        "200":
+          description: An unaccepted offer, the exact accepted task, or an ordinary empty result.
+          content:
+            application/json:
+              schema:
+                oneOf:
+                  - $ref: "#/components/schemas/TaskDeliveryOfferedResponse"
+                  - $ref: "#/components/schemas/ActivityTaskOfferAcceptedResponse"
+        "400": { $ref: "#/components/responses/WorkerError" }
+        "409": { $ref: "#/components/responses/TaskDeliveryConflict" }
+        "422": { $ref: "#/components/responses/WorkerError" }
+        "429": { $ref: "#/components/responses/WorkerPollBackpressure" }
+        "503": { $ref: "#/components/responses/WorkerServiceUnavailable" }
   /worker/activity-tasks/poll:
     post:
       operationId: pollActivityTask
@@ -1287,6 +1419,11 @@ components:
       in: header
       name: Authorization
   parameters:
+    TaskDeliveryVersionHeader:
+      name: X-Durable-Workflow-Task-Delivery
+      in: header
+      required: true
+      schema: { type: string, const: offer-accept.v1 }
     WorkerProtocolVersionHeader:
       name: X-Durable-Workflow-Protocol-Version
       in: header
@@ -1319,6 +1456,21 @@ components:
       description: Worker registration identity in the resolved namespace.
       schema: { type: string, minLength: 1, maxLength: 255 }
   responses:
+    TaskDeliveryConflict:
+      description: >
+        Offer negotiation, incarnation, identity, deadline or task authority
+        no longer permits this acceptance. This refusal never revokes a claim
+        already accepted through another response.
+      content:
+        application/json:
+          schema:
+            type: object
+            required: [task, poll_status, reason, retryable]
+            properties:
+              task: { type: "null" }
+              poll_status: { type: string }
+              reason: { type: string }
+              retryable: { type: boolean, const: true }
     WorkerEnvelope:
       description: Worker-protocol response envelope.
       headers:
@@ -1499,6 +1651,50 @@ components:
               - $ref: "#/components/schemas/WorkerEnvelope"
               - $ref: "#/components/schemas/WorkerDeregistrationResult"
   schemas:
+    TaskDeliveryRequestFields:
+      type: object
+      required: [poll_request_id, registration_token]
+      properties:
+        poll_request_id: { type: string, minLength: 1, maxLength: 255 }
+        registration_token: { type: string, pattern: "^[a-f0-9]{32}$" }
+        task_offer_token: { type: string, minLength: 1, maxLength: 8192 }
+    TaskDeliveryOfferedResponse:
+      type: object
+      required: [task, task_offer, poll_status]
+      properties:
+        task: { type: "null" }
+        poll_status: { type: string, const: offered }
+        task_offer:
+          type: object
+          required: [schema, task_id, task_kind, token, expires_at_ms]
+          properties:
+            schema: { type: string, const: durable-workflow.task-delivery.v1 }
+            task_id: { type: string, minLength: 1 }
+            task_kind: { type: string, enum: [workflow, activity, query, update_validation] }
+            token: { type: string, minLength: 1, maxLength: 8192 }
+            expires_at_ms: { type: integer }
+    ActivityTaskOfferAcceptedResponse:
+      oneOf:
+        - $ref: "#/components/schemas/ActivityTaskClaim"
+        - $ref: "#/components/schemas/EmptyPoll"
+        - $ref: "#/components/schemas/DrainingPoll"
+    UpdateValidationTaskOfferAcceptedResponse:
+      oneOf:
+        - $ref: "#/components/schemas/UpdateValidationTaskClaim"
+        - $ref: "#/components/schemas/EmptyPoll"
+    TaskDeliveryCapabilities:
+      type: object
+      required: [schema, supported, minimum_protocol_version, worker_capability, header, version, endpoint_suffix, acceptance_timeout_seconds, unaccepted_offer_reserves_task]
+      properties:
+        schema: { type: string, const: durable-workflow.task-delivery.v1 }
+        supported: { type: boolean }
+        minimum_protocol_version: { type: string, const: "1.20" }
+        worker_capability: { type: string, const: task_delivery_offers }
+        header: { type: string, const: X-Durable-Workflow-Task-Delivery }
+        version: { type: string, const: offer-accept.v1 }
+        endpoint_suffix: { type: string, const: /offers }
+        acceptance_timeout_seconds: { type: integer, minimum: 1, maximum: 10 }
+        unaccepted_offer_reserves_task: { type: boolean, const: false }
     AcceptedWorkerProtocolRequestVersion:
       type: string
       enum: ["1.0", "1.1", "1.2", "1.3", "1.4", "1.5", "1.6", "1.7", "1.8", "1.9", "1.10", "1.11", "1.12", "1.13", "1.14", "1.15", "1.16", "1.17", "1.18", "1.19", "1.20"]
@@ -1707,6 +1903,7 @@ components:
       additionalProperties: true
       required: [supported_workflow_task_commands, workflow_task_poll_request_idempotency, workflow_memo_updates, message_streams, typed_search_attributes, condition_wait_occurrence_identity, update_validation_tasks, synchronous_update_validation, local_activities, sticky_execution]
       properties:
+        task_delivery: { $ref: "#/components/schemas/TaskDeliveryCapabilities" }
         query_task_authority:
           type: boolean
           description: Original query timing and atomic outcome fences with non-renewing protocol-1.20 query status. A 1.19 status request is explicitly unavailable.
diff --git a/routes/api.php b/routes/api.php
index f9f79ccb5..a33bcde45 100644
--- a/routes/api.php
+++ b/routes/api.php
@@ -241,6 +241,7 @@
 
         // Workflow tasks (long-poll)
         Route::post('/workflow-tasks/poll', [WorkerController::class, 'pollWorkflowTasks']);
+        Route::post('/workflow-tasks/offers', [WorkerController::class, 'pollWorkflowTasks']);
         Route::post('/workflow-tasks/{taskId}/history', [WorkerController::class, 'workflowTaskHistory']);
         Route::post('/workflow-tasks/{taskId}/heartbeat', [WorkerController::class, 'heartbeatWorkflowTask']);
         Route::post('/workflow-tasks/{taskId}/deliver-cancellation', [CooperativeCancellationController::class, 'deliver']);
@@ -261,17 +262,20 @@
 
         // Query tasks (ephemeral worker-routed workflow queries)
         Route::post('/query-tasks/poll', [WorkerController::class, 'pollQueryTasks']);
+        Route::post('/query-tasks/offers', [WorkerController::class, 'pollQueryTasks']);
         Route::get('/query-tasks/{queryTaskId}/status', [WorkerController::class, 'queryTaskStatus']);
         Route::post('/query-tasks/{queryTaskId}/complete', [WorkerController::class, 'completeQueryTask']);
         Route::post('/query-tasks/{queryTaskId}/fail', [WorkerController::class, 'failQueryTask']);
 
         // Update validation tasks (durable, synchronous pre-acceptance checks)
         Route::post('/update-validation-tasks/poll', [WorkerController::class, 'pollUpdateValidationTasks']);
+        Route::post('/update-validation-tasks/offers', [WorkerController::class, 'pollUpdateValidationTasks']);
         Route::post('/update-validation-tasks/{taskId}/approve', [WorkerController::class, 'approveUpdateValidationTask']);
         Route::post('/update-validation-tasks/{taskId}/reject', [WorkerController::class, 'rejectUpdateValidationTask']);
 
         // Activity tasks (long-poll)
         Route::post('/activity-tasks/poll', [ActivityTaskController::class, 'poll']);
+        Route::post('/activity-tasks/offers', [ActivityTaskController::class, 'poll']);
         Route::post('/activity-tasks/{taskId}/status', [ActivityTaskController::class, 'status']);
         Route::post('/activity-tasks/{taskId}/acknowledge-cancellation', [ActivityTaskController::class, 'acknowledgeCancellation']);
         Route::post('/activity-tasks/{taskId}/complete', [ActivityTaskController::class, 'complete']);
diff --git a/tests/Feature/WorkerTaskDeliveryConcurrencyTest.php b/tests/Feature/WorkerTaskDeliveryConcurrencyTest.php
new file mode 100644
index 000000000..c81ab5d07
--- /dev/null
+++ b/tests/Feature/WorkerTaskDeliveryConcurrencyTest.php
@@ -0,0 +1,156 @@
+<?php
+
+namespace Tests\Feature;
+
+use App\Models\WorkerRegistration;
+use App\Support\WorkerTaskDelivery;
+use Illuminate\Foundation\Testing\DatabaseMigrations;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\File;
+use Illuminate\Support\Facades\Queue;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Symfony\Component\Process\Process;
+use Tests\Feature\Concerns\ServerTestHelpers;
+use Tests\Fixtures\ExternalGreetingWorkflow;
+use Tests\TestCase;
+use Workflow\V2\Models\ActivityAttempt;
+use Workflow\V2\Models\WorkflowTask;
+
+class WorkerTaskDeliveryConcurrencyTest extends TestCase
+{
+    use DatabaseMigrations {
+        runDatabaseMigrations as migrateDeliveryDatabase;
+    }
+    use ServerTestHelpers;
+
+    private ?string $sqliteDatabase = null;
+
+    public function runDatabaseMigrations(): void
+    {
+        if (config('database.default') === 'sqlite') {
+            $path = storage_path('framework/cache/delivery-concurrency-'.bin2hex(random_bytes(6)).'.sqlite');
+            File::ensureDirectoryExists(dirname($path));
+            touch($path);
+            $this->sqliteDatabase = $path;
+            config(['database.connections.sqlite.database' => $path,
+                'database.connections.sqlite.busy_timeout' => 1,
+                'database.connections.sqlite.journal_mode' => 'WAL',
+                'database.connections.sqlite.transaction_mode' => 'IMMEDIATE']);
+            DB::purge('sqlite');
+        }
+        $this->migrateDeliveryDatabase();
+    }
+
+    protected function tearDown(): void
+    {
+        try {
+            parent::tearDown();
+        } finally {
+            if ($this->sqliteDatabase !== null) {
+                foreach ([$this->sqliteDatabase, $this->sqliteDatabase.'-wal', $this->sqliteDatabase.'-shm'] as $path) {
+                    if (is_file($path)) {
+                        unlink($path);
+                    }
+                }
+            }
+        }
+    }
+
+    #[DataProvider('taskKinds')]
+    public function test_independent_processes_accepting_one_offer_start_one_attempt(string $kind): void
+    {
+        Queue::fake();
+        $this->createNamespace('default');
+        $this->configureWorkflowTypes(['tests.external-greeting-workflow' => ExternalGreetingWorkflow::class]);
+        $run = $this->postJson('/api/workflows', [
+            'workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'delivery', 'input' => ['Ada'],
+        ], $this->apiHeaders())->assertCreated()->json('run_id');
+        if ($kind === 'activity') {
+            $this->runReadyWorkflowTask($run);
+        }
+        $task = WorkflowTask::query()->where('workflow_run_id', $run)->where('task_type', $kind)->firstOrFail();
+        $this->registerWorker('delivery-worker', $task->queue);
+        $worker = WorkerRegistration::query()->where('worker_id', 'delivery-worker')->firstOrFail();
+        $worker->forceFill(['registration_token' => str_repeat('a', 32),
+            'capabilities' => [...$worker->capabilities, WorkerTaskDelivery::CAPABILITY]])->save();
+        $path = '/api/worker/'.$kind.'-tasks/offers';
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => $task->queue, 'poll_request_id' => 'concurrent',
+            'registration_token' => str_repeat('a', 32), 'timeout_seconds' => 0];
+        $barrier = storage_path('framework/cache/delivery-start-'.bin2hex(random_bytes(6)));
+        $bodyFile = $barrier.'.json';
+        $connection = config('database.default');
+        $db = config('database.connections.'.$connection);
+        $environment = [
+            'APP_ENV' => 'testing', 'APP_KEY' => config('app.key'), 'DW_AUTH_DRIVER' => 'none',
+            'DB_CONNECTION' => $connection, 'DB_HOST' => $db['host'] ?? '',
+            'DB_PORT' => (string) ($db['port'] ?? ''), 'DB_DATABASE' => $db['database'],
+            'DB_USERNAME' => $db['username'] ?? '', 'DB_PASSWORD' => $db['password'] ?? '',
+            'DB_SQLITE_BUSY_TIMEOUT' => (string) ($db['busy_timeout'] ?? 5000),
+            'DB_SQLITE_JOURNAL_MODE' => $db['journal_mode'] ?? 'WAL',
+            'DB_SQLITE_TRANSACTION_MODE' => $db['transaction_mode'] ?? 'IMMEDIATE',
+            'DW_POLLING_CACHE_PATH' => config('server.polling.cache_path'),
+        ];
+        $processes = [];
+        try {
+            for ($index = 0; $index < 2; $index++) {
+                $process = new Process([PHP_BINARY, base_path('tests/Support/TaskDeliveryAcceptProcess.php')], base_path(), $environment);
+                $process->setInput(json_encode(['path' => $path, 'barrier' => $barrier,
+                    'body_file' => $bodyFile, 'ready' => $barrier.'.ready'.$index], JSON_THROW_ON_ERROR));
+                $process->setTimeout(15);
+                $process->start();
+                $processes[] = $process;
+            }
+            $deadline = microtime(true) + 10;
+            do {
+                $ready = 0;
+                foreach ($processes as $index => $process) {
+                    if (! $process->isRunning()) {
+                        $this->fail($process->getErrorOutput());
+                    }
+                    clearstatcache(true, $barrier.'.ready'.$index);
+                    $ready += (int) is_file($barrier.'.ready'.$index);
+                }
+                if ($ready === 2) {
+                    break;
+                }
+                usleep(1000);
+            } while (microtime(true) < $deadline);
+            $this->assertSame(2, $ready, 'Both independent applications must be ready before issuing the offer.');
+            $offer = $this->postJson($path, $body, $headers)->assertOk()->assertJsonPath('poll_status', 'offered');
+            $body['task_offer_token'] = $offer->json('task_offer.token');
+            File::put($bodyFile, json_encode($body, JSON_THROW_ON_ERROR));
+            touch($barrier);
+            foreach ($processes as $process) {
+                $this->assertSame(0, $process->wait(), $process->getErrorOutput());
+                $lines = explode("\n", trim($process->getOutput()));
+                $result = json_decode(end($lines), true, flags: JSON_THROW_ON_ERROR);
+                $diagnostic = json_encode(['status' => $result['status'],
+                    'reason' => $result['body']['reason'] ?? null, 'task' => $result['body']['task']['task_id'] ?? null]);
+                $this->assertSame(200, $result['status'], $diagnostic);
+                $this->assertSame($task->id, $result['body']['task']['task_id'] ?? null, $diagnostic);
+                foreach ($result['attempts'] as $attempt) {
+                    if ($attempt['status'] === 503) {
+                        $this->assertSame('backend_lock_pressure', $attempt['reason']);
+                        $this->assertNull($attempt['task']);
+                    }
+                }
+            }
+            $this->assertSame(1, (int) $task->fresh()->attempt_count);
+            $this->assertSame($kind === 'activity' ? 1 : 0, ActivityAttempt::query()->count());
+        } finally {
+            foreach ($processes as $process) {
+                if ($process->isRunning()) {
+                    $process->stop(1);
+                }
+            }
+            File::delete([$barrier, $bodyFile, $barrier.'.ready0', $barrier.'.ready1']);
+            DB::purge();
+        }
+    }
+
+    public static function taskKinds(): array
+    {
+        return ['workflow' => ['workflow'], 'activity' => ['activity']];
+    }
+}
diff --git a/tests/Feature/WorkerTaskDeliveryTest.php b/tests/Feature/WorkerTaskDeliveryTest.php
new file mode 100644
index 000000000..a54be1d2c
--- /dev/null
+++ b/tests/Feature/WorkerTaskDeliveryTest.php
@@ -0,0 +1,348 @@
+<?php
+
+namespace Tests\Feature;
+
+use App\Models\WorkerRegistration;
+use App\Support\WorkerTaskDelivery;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Queue;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Tests\Feature\Concerns\ServerTestHelpers;
+use Tests\Fixtures\ExternalGreetingWorkflow;
+use Tests\Support\OpenApiSchema;
+use Tests\TestCase;
+use Workflow\V2\Enums\TaskStatus;
+use Workflow\V2\Models\ActivityAttempt;
+use Workflow\V2\Models\ActivityExecution;
+use Workflow\V2\Models\WorkflowHistoryEvent;
+use Workflow\V2\Models\WorkflowTask;
+
+class WorkerTaskDeliveryTest extends TestCase
+{
+    use RefreshDatabase;
+    use ServerTestHelpers;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+        Queue::fake();
+        $this->createNamespace('default');
+        $this->configureWorkflowTypes(['tests.external-greeting-workflow' => ExternalGreetingWorkflow::class]);
+        config(['cache.default' => 'array', 'server.polling.task_offer_accept_seconds' => 2]);
+    }
+
+    #[DataProvider('kinds')]
+    public function test_an_offer_never_claims_or_reserves_the_task(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $before = $task->getRawOriginal();
+        $counts = [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()];
+        $offer = $this->offer($kind, $task->queue);
+        $this->assertSame($task->id, $offer['task_id']);
+        $this->assertSame($kind, $offer['task_kind']);
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+        $this->assertSame($counts, [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()]);
+    }
+
+    #[DataProvider('kinds')]
+    public function test_a_lost_acceptance_reply_replays_exactly_one_live_claim_after_offer_expiry(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $body = $this->body($task->queue) + ['task_offer_token' => $offer['token']];
+        $first = $this->withHeaders($this->headers())->postJson($this->path($kind), $body)->assertOk();
+        $this->assertSame($task->id, $first->json('task.task_id'));
+        $this->assertSame(TaskStatus::Leased, $task->fresh()->status);
+        $this->assertSame(1, (int) $task->fresh()->attempt_count);
+        $counts = [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()];
+        $this->travel(3)->seconds();
+        WorkerRegistration::query()->where('worker_id', 'offer-worker')->firstOrFail()
+            ->forceFill(['last_heartbeat_at' => now()])->save();
+        $replay = $this->withHeaders($this->headers())->postJson($this->path($kind), $body)->assertOk();
+        foreach (['task_id', 'lease_owner', 'workflow_task_attempt', 'activity_attempt_id', 'attempt_number'] as $key) {
+            $this->assertSame($first->json('task.'.$key), $replay->json('task.'.$key));
+        }
+        $this->assertSame(1, (int) $task->fresh()->attempt_count);
+        $this->assertSame($counts, [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()]);
+    }
+
+    #[DataProvider('kinds')]
+    public function test_a_peer_can_claim_immediately_and_a_failed_acceptance_cannot_revoke_it(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $this->registerWorker('peer', $task->queue);
+        $this->flushHeaders();
+        $peer = $this->withHeaders($this->workerHeaders())->postJson('/api/worker/'.$kind.'-tasks/poll', [
+            'worker_id' => 'peer', 'task_queue' => $task->queue, 'poll_request_id' => 'peer-poll', 'timeout_seconds' => 0,
+        ])->assertOk();
+        $this->assertSame($task->id, $peer->json('task.task_id'));
+        $before = $task->fresh()->getRawOriginal();
+        $this->withHeaders($this->headers())->postJson($this->path($kind),
+            $this->body($task->queue) + ['task_offer_token' => $offer['token']])->assertConflict();
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+    }
+
+    #[DataProvider('kinds')]
+    public function test_an_unaccepted_expired_offer_starts_no_business_attempt(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $counts = [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()];
+        $this->travel(3)->seconds();
+        $this->withHeaders($this->headers())->postJson($this->path($kind),
+            $this->body($task->queue) + ['task_offer_token' => $offer['token']])
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertSame(TaskStatus::Ready, $task->fresh()->status);
+        $this->assertSame(0, (int) $task->fresh()->attempt_count);
+        $this->assertSame($counts, [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()]);
+    }
+
+    #[DataProvider('kinds')]
+    public function test_expiry_resolves_an_offer_even_after_a_peer_has_claimed_the_task(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $this->registerWorker('peer', $task->queue);
+        $this->flushHeaders();
+        $this->withHeaders($this->workerHeaders())->postJson('/api/worker/'.$kind.'-tasks/poll', [
+            'worker_id' => 'peer', 'task_queue' => $task->queue, 'poll_request_id' => 'peer-poll', 'timeout_seconds' => 0,
+        ])->assertOk()->assertJsonPath('task.task_id', $task->id);
+        $before = $task->fresh()->getRawOriginal();
+        $this->travel(3)->seconds();
+        $this->withHeaders($this->headers())->postJson($this->path($kind),
+            $this->body($task->queue) + ['task_offer_token' => $offer['token']])
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+    }
+
+    #[DataProvider('kinds')]
+    public function test_expiry_resolves_an_offer_for_a_deleted_task_without_claiming_another(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $task->delete();
+        $this->travel(3)->seconds();
+        $this->withHeaders($this->headers())->postJson($this->path($kind),
+            $this->body($task->queue) + ['task_offer_token' => $offer['token']])
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertSame(0, WorkflowTask::query()->where('status', TaskStatus::Leased)->count());
+    }
+
+    #[DataProvider('kinds')]
+    public function test_a_replacement_registration_cannot_accept_the_old_offer(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $worker = $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $worker->forceFill(['registration_token' => str_repeat('b', 32)])->save();
+        foreach ([str_repeat('a', 32), str_repeat('b', 32)] as $token) {
+            $body = $this->body($task->queue);
+            $body['registration_token'] = $token;
+            $this->withHeaders($this->headers())->postJson($this->path($kind), $body + ['task_offer_token' => $offer['token']])
+                ->assertConflict()->assertJsonPath('reason', 'worker_registration_lost_authority');
+        }
+        $this->assertSame(TaskStatus::Ready, $task->fresh()->status);
+        $this->assertSame(0, (int) $task->fresh()->attempt_count);
+    }
+
+    public function test_modified_offer_ciphertext_cannot_authorize_a_claim(): void
+    {
+        $task = $this->queuedTask('workflow');
+        $this->actor($task->queue);
+        $offer = $this->offer('workflow', $task->queue);
+        $token = $offer['token'];
+        $token[20] = $token[20] === 'A' ? 'B' : 'A';
+        $this->withHeaders($this->headers())->postJson($this->path('workflow'),
+            $this->body($task->queue) + ['task_offer_token' => $token])
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_invalid');
+        $this->assertSame(TaskStatus::Ready, $task->fresh()->status);
+    }
+
+    #[DataProvider('kinds')]
+    public function test_two_offer_packets_for_one_poll_replay_the_same_confirmed_claim(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $first = $this->offer($kind, $task->queue);
+        $second = $this->offer($kind, $task->queue);
+        $this->assertNotSame($first['token'], $second['token']);
+        $responses = [];
+        foreach ([$first, $second] as $offer) {
+            $responses[] = $this->withHeaders($this->headers())->postJson($this->path($kind),
+                $this->body($task->queue) + ['task_offer_token' => $offer['token']])->assertOk()->json('task');
+        }
+        $this->assertSame($responses[0]['task_id'], $responses[1]['task_id']);
+        $this->assertSame($responses[0]['lease_owner'], $responses[1]['lease_owner']);
+        $this->assertSame(1, (int) $task->fresh()->attempt_count);
+    }
+
+    #[DataProvider('kinds')]
+    public function test_a_claim_that_exceeds_the_acceptance_deadline_rolls_back_its_attempt_and_history(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $before = $task->getRawOriginal();
+        $counts = [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()];
+        $advanced = false;
+        DB::listen(function ($query) use (&$advanced): void {
+            if (! $advanced && str_starts_with(str_replace(['"', '`'], '', $query->sql), 'update workflow_tasks') && in_array('leased', $query->bindings, true)) {
+                $advanced = true;
+                $this->travel(3)->seconds();
+            }
+        });
+        $this->withHeaders($this->headers())->postJson($this->path($kind),
+            $this->body($task->queue) + ['task_offer_token' => $offer['token']])
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertTrue($advanced, 'The real bridge claim must run before the injected delay.');
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+        $this->assertSame($counts, [ActivityAttempt::query()->count(), WorkflowHistoryEvent::query()->count()]);
+        if ($kind === 'activity') {
+            $this->assertSame(0, (int) ActivityExecution::query()->firstOrFail()->attempt_count);
+            $this->assertNull(ActivityExecution::query()->firstOrFail()->current_attempt_id);
+        }
+    }
+
+    public static function kinds(): array
+    {
+        return ['workflow' => ['workflow'], 'activity' => ['activity']];
+    }
+
+    #[DataProvider('kinds')]
+    public function test_negotiation_cannot_silently_fall_back_to_an_immediate_claim(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $this->actor($task->queue);
+        $before = $task->getRawOriginal();
+        $this->withHeaders($this->headers())->postJson('/api/worker/'.$kind.'-tasks/poll', $this->body($task->queue))
+            ->assertConflict()->assertJsonPath('reason', 'task_delivery_offer_endpoint_required');
+        $this->flushHeaders();
+        $this->withHeaders($this->workerHeaders())->postJson($this->path($kind), $this->body($task->queue))
+            ->assertConflict()->assertJsonPath('reason', 'task_delivery_protocol_required');
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+    }
+
+    public function test_an_offer_is_bound_to_the_original_poll_and_requested_kinds(): void
+    {
+        $task = $this->queuedTask('workflow');
+        $worker = $this->actor($task->queue);
+        $worker->forceFill(['capabilities' => [...$worker->capabilities, 'update_validation_tasks']])->save();
+        $offer = $this->offer('workflow', $task->queue);
+        foreach ([['poll_request_id' => 'different-poll'], ['task_kinds' => ['workflow', 'update_validation']]] as $change) {
+            $this->withHeaders($this->headers())->postJson($this->path('workflow'),
+                array_replace($this->body($task->queue), $change, ['task_offer_token' => $offer['token']]))
+                ->assertConflict()->assertJsonPath('reason', 'task_offer_mismatch');
+        }
+        $this->assertSame(TaskStatus::Ready, $task->fresh()->status);
+        $this->assertSame(0, (int) $task->fresh()->attempt_count);
+    }
+
+    #[DataProvider('routingChanges')]
+    public function test_routing_changes_still_fence_an_offer_with_the_same_registration_token(string $kind, array $change): void
+    {
+        $task = $this->queuedTask($kind);
+        $worker = $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $worker->forceFill($change)->save();
+        $this->withHeaders($this->headers())->postJson($this->path($kind),
+            $this->body($task->queue) + ['task_offer_token' => $offer['token']])
+            ->assertConflict()->assertJsonPath('reason', 'worker_registration_lost_authority');
+        $this->assertSame(TaskStatus::Ready, $task->fresh()->status);
+        $this->assertSame(0, (int) $task->fresh()->attempt_count);
+    }
+
+    public static function routingChanges(): array
+    {
+        return [
+            'workflow-build' => ['workflow', ['build_id' => 'changed-build']],
+            'activity-build' => ['activity', ['build_id' => 'changed-build']],
+            'workflow-process' => ['workflow', ['process_metrics' => ['process_id' => 9876]]],
+            'activity-process' => ['activity', ['process_metrics' => ['process_id' => 9876]]],
+        ];
+    }
+
+    #[DataProvider('kinds')]
+    public function test_a_routine_heartbeat_during_acceptance_does_not_replace_the_incarnation(string $kind): void
+    {
+        $task = $this->queuedTask($kind);
+        $worker = $this->actor($task->queue);
+        $offer = $this->offer($kind, $task->queue);
+        $advanced = false;
+        DB::listen(function ($query) use (&$advanced, $worker): void {
+            if (! $advanced && str_contains($query->sql, 'select')
+                && str_contains(str_replace(['"', '`'], '', $query->sql), 'workflow_tasks')) {
+                $advanced = true;
+                $this->travel(1)->seconds();
+                $worker->forceFill(['last_heartbeat_at' => now()])->save();
+            }
+        });
+        $this->withHeaders($this->headers())->postJson($this->path($kind),
+            $this->body($task->queue) + ['task_offer_token' => $offer['token']])
+            ->assertOk()->assertJsonPath('task.task_id', $task->id);
+        $this->assertTrue($advanced);
+        $this->assertSame(1, (int) $task->fresh()->attempt_count);
+    }
+
+    private function queuedTask(string $kind): WorkflowTask
+    {
+        $result = $this->withHeaders($this->apiHeaders())->postJson('/api/workflows', [
+            'workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'delivery-queue', 'input' => ['Ada'],
+        ])->assertCreated();
+        $task = WorkflowTask::query()->where('workflow_run_id', $result->json('run_id'))->firstOrFail();
+        if ($kind === 'activity') {
+            $this->runReadyWorkflowTask($task->workflow_run_id);
+            $task = WorkflowTask::query()->where('workflow_run_id', $task->workflow_run_id)->where('task_type', 'activity')->firstOrFail();
+            $this->assertSame(0, (int) ActivityExecution::query()->firstOrFail()->attempt_count);
+        }
+
+        return $task;
+    }
+
+    private function actor(string $queue): WorkerRegistration
+    {
+        $this->registerWorker('offer-worker', $queue);
+        $worker = WorkerRegistration::query()->where('worker_id', 'offer-worker')->firstOrFail();
+        $worker->forceFill([
+            'registration_token' => str_repeat('a', 32),
+            'capabilities' => [...$worker->capabilities, WorkerTaskDelivery::CAPABILITY],
+        ])->save();
+
+        return $worker;
+    }
+
+    private function path(string $kind): string
+    {
+        return '/api/worker/'.$kind.'-tasks/offers';
+    }
+
+    private function headers(): array
+    {
+        return $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+    }
+
+    private function body(string $queue): array
+    {
+        return ['worker_id' => 'offer-worker', 'task_queue' => $queue, 'poll_request_id' => 'original-poll',
+            'registration_token' => str_repeat('a', 32), 'timeout_seconds' => 0];
+    }
+
+    private function offer(string $kind, string $queue): array
+    {
+        $response = $this->withHeaders($this->headers())->postJson($this->path($kind), $this->body($queue))
+            ->assertOk()->assertJsonPath('task', null)->assertJsonPath('poll_status', 'offered')
+            ->assertJsonPath('task_offer.schema', WorkerTaskDelivery::SCHEMA);
+        $this->assertIsArray($response->json('task_offer'));
+        OpenApiSchema::fromFile(base_path('resources/platform-protocol-specs/worker-protocol-api.openapi.yaml'))
+            ->assertReferenceMatches('#/components/schemas/TaskDeliveryOfferedResponse', json_decode($response->getContent()));
+
+        return $response->json('task_offer');
+    }
+}
diff --git a/tests/Feature/WorkflowQueryTaskBrokerTest.php b/tests/Feature/WorkflowQueryTaskBrokerTest.php
index f63945766..31c0e9a97 100644
--- a/tests/Feature/WorkflowQueryTaskBrokerTest.php
+++ b/tests/Feature/WorkflowQueryTaskBrokerTest.php
@@ -18,6 +18,7 @@
 use App\Support\RuntimeExternalPayloadRegistry;
 use App\Support\ServerPollingCache;
 use App\Support\WorkerProtocol;
+use App\Support\WorkerTaskDelivery;
 use App\Support\WorkflowQueryTaskBroker;
 use Illuminate\Cache\ArrayStore;
 use Illuminate\Cache\Repository as CacheRepository;
@@ -30,6 +31,7 @@
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\File;
 use Illuminate\Support\Facades\Queue;
+use PHPUnit\Framework\Attributes\DataProvider;
 use Symfony\Component\Process\Process;
 use Tests\Feature\Concerns\ServerTestHelpers;
 use Tests\Support\OpenApiSchema;
@@ -77,6 +79,204 @@ public function test_query_admission_snapshots_deadline_and_receipt_expiry(): vo
         $this->assertSame('2026-10-14T22:03:00.123456Z', $task['receipt_expires_at'] ?? null);
     }
 
+    public function test_query_offer_leaves_authority_pending_and_acceptance_replays_after_reply_loss(): void
+    {
+        Queue::fake();
+        $run = $this->startRemoteWorkflow('query-delivery');
+        $broker = app(WorkflowQueryTaskBroker::class);
+        $task = $broker->enqueue('default', $run, 'status', $this->queryArguments());
+        $this->registerPythonWorker('delivery-worker', 'python-queries', ['python.queryable'], ['query_tasks', WorkerTaskDelivery::CAPABILITY]);
+        WorkerRegistration::query()->where('worker_id', 'delivery-worker')->firstOrFail()
+            ->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => 'python-queries', 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $path = '/api/worker/query-tasks/offers';
+        $before = $broker->task($task['query_task_id']);
+        $offer = $this->postJson($path, $body, $headers)->assertOk()->assertJsonPath('poll_status', 'offered');
+        $this->assertSame($before, $broker->task($task['query_task_id']));
+        $acceptance = $body + ['task_offer_token' => $offer->json('task_offer.token')];
+        $first = $this->postJson($path, $acceptance, $headers)->assertOk()
+            ->assertJsonPath('task.query_task_id', $task['query_task_id'])->assertJsonPath('task.query_task_attempt', 1);
+        $this->travel(3)->seconds();
+        // Replay is sourced from the original domain record even if an index
+        // write or response was lost after the accepted record committed.
+        app(ServerPollingCache::class)->store()->forget('server:workflow-query-task:leased:'.sha1('default|python-queries|delivery-worker'));
+        WorkerRegistration::query()->where('worker_id', 'delivery-worker')->firstOrFail()
+            ->forceFill(['last_heartbeat_at' => now()])->save();
+        $this->postJson($path, $acceptance, $headers)->assertOk()
+            ->assertJsonPath('task.query_task_id', $first->json('task.query_task_id'))
+            ->assertJsonPath('task.query_task_attempt', 1);
+        $this->assertSame(1, $broker->task($task['query_task_id'])['attempt_count']);
+        $this->assertSame($task['query_deadline_at'], $broker->task($task['query_task_id'])['query_deadline_at']);
+    }
+
+    public function test_unaccepted_query_offer_does_not_block_a_peer(): void
+    {
+        Queue::fake();
+        $run = $this->startRemoteWorkflow('query-delivery-peer');
+        $broker = app(WorkflowQueryTaskBroker::class);
+        $task = $broker->enqueue('default', $run, 'status', $this->queryArguments());
+        $this->registerPythonWorker('delivery-worker', 'python-queries', ['python.queryable'], ['query_tasks', WorkerTaskDelivery::CAPABILITY]);
+        WorkerRegistration::query()->where('worker_id', 'delivery-worker')->firstOrFail()
+            ->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => 'python-queries', 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $offer = $this->postJson('/api/worker/query-tasks/offers', $body, $headers)->assertOk();
+        $this->registerPythonWorker('peer', 'python-queries', ['python.queryable']);
+        $this->postJson('/api/worker/query-tasks/poll', ['worker_id' => 'peer', 'task_queue' => 'python-queries', 'timeout_seconds' => 0], $this->workerHeaders())
+            ->assertOk()->assertJsonPath('task.query_task_id', $task['query_task_id']);
+        $before = $broker->task($task['query_task_id']);
+        $this->postJson('/api/worker/query-tasks/offers', $body + ['task_offer_token' => $offer->json('task_offer.token')], $headers)
+            ->assertConflict();
+        $this->assertSame($before, $broker->task($task['query_task_id']));
+    }
+
+    public function test_expiry_resolves_a_query_offer_after_a_peer_claim_without_mutating_its_authority(): void
+    {
+        Queue::fake();
+        $run = $this->startRemoteWorkflow('query-expired-offer-peer');
+        $broker = app(WorkflowQueryTaskBroker::class);
+        $task = $broker->enqueue('default', $run, 'status', $this->queryArguments());
+        $this->registerPythonWorker('delivery-worker', 'python-queries', ['python.queryable'], ['query_tasks', WorkerTaskDelivery::CAPABILITY]);
+        WorkerRegistration::query()->where('worker_id', 'delivery-worker')->firstOrFail()
+            ->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => 'python-queries', 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $path = '/api/worker/query-tasks/offers';
+        $offer = $this->postJson($path, $body, $headers)->assertOk();
+        $this->registerPythonWorker('peer', 'python-queries', ['python.queryable']);
+        $this->postJson('/api/worker/query-tasks/poll', ['worker_id' => 'peer', 'task_queue' => 'python-queries', 'timeout_seconds' => 0], $this->workerHeaders())
+            ->assertOk()->assertJsonPath('task.query_task_id', $task['query_task_id']);
+        $before = $broker->task($task['query_task_id']);
+        $this->travel(3)->seconds();
+        $this->postJson($path, $body + ['task_offer_token' => $offer->json('task_offer.token')], $headers)
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertSame($before, $broker->task($task['query_task_id']));
+    }
+
+    #[DataProvider('deliveryCacheFailures')]
+    public function test_query_acceptance_cache_failures_preserve_only_committed_authority(string $mode, bool $committed): void
+    {
+        Queue::fake();
+        $store = new class extends ArrayStore
+        {
+            public string $mode = 'none';
+
+            public function put($key, $value, $seconds)
+            {
+                if ($this->mode === 'index' && str_starts_with($key, 'server:workflow-query-task:leased:')) {
+                    throw new \RuntimeException('Fixture lease-index failure.');
+                }
+                if ($this->mode === 'index_false' && str_starts_with($key, 'server:workflow-query-task:leased:')) {
+                    return false;
+                }
+                if ($this->mode === 'pending_index' && str_starts_with($key, 'server:workflow-query-task:queue:')) {
+                    throw new \RuntimeException('Fixture pending-index failure.');
+                }
+                if ($this->mode !== 'none' && is_array($value) && ($value['status'] ?? null) === 'leased'
+                    && str_starts_with($key, 'server:workflow-query-task:task:')) {
+                    if ($this->mode === 'before') {
+                        throw new \RuntimeException('Fixture task write failure before commit.');
+                    }
+                    if ($this->mode === 'false') {
+                        return false;
+                    }
+                    if ($this->mode === 'after') {
+                        parent::put($key, $value, $seconds);
+                        throw new \RuntimeException('Fixture task write failure after commit.');
+                    }
+                }
+
+                return parent::put($key, $value, $seconds);
+            }
+        };
+        $this->bindPollingCacheStore($store);
+        $run = $this->startRemoteWorkflow('query-delivery-cache-'.$mode);
+        $broker = app(WorkflowQueryTaskBroker::class);
+        $task = $broker->enqueue('default', $run, 'status', $this->queryArguments());
+        $this->registerPythonWorker('delivery-worker', 'python-queries', ['python.queryable'], ['query_tasks', WorkerTaskDelivery::CAPABILITY]);
+        WorkerRegistration::query()->where('worker_id', 'delivery-worker')->firstOrFail()
+            ->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => 'python-queries', 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $path = '/api/worker/query-tasks/offers';
+        $offer = $this->postJson($path, $body, $headers)->assertOk()->assertJsonPath('poll_status', 'offered');
+        $acceptance = $body + ['task_offer_token' => $offer->json('task_offer.token')];
+        $store->mode = $mode;
+        $response = $this->postJson($path, $acceptance, $headers);
+        $stored = $broker->task($task['query_task_id']);
+        $this->assertSame($committed ? 'leased' : 'pending', $stored['status']);
+        $this->assertSame($committed ? 1 : 0, $stored['attempt_count']);
+        $response->assertStatus(503)->assertJsonPath('task', null)->assertJsonPath('reason', 'query_task_queue_unavailable');
+        $store->mode = 'none';
+        if ($committed) {
+            $this->registerPythonWorker('peer', 'python-queries', ['python.queryable']);
+            $this->postJson('/api/worker/query-tasks/poll', ['worker_id' => 'peer', 'task_queue' => 'python-queries', 'timeout_seconds' => 0], $this->workerHeaders())
+                ->assertOk()->assertJsonPath('task', null);
+            $this->postJson($path, $acceptance, $headers)->assertOk()
+                ->assertJsonPath('task.query_task_id', $task['query_task_id'])->assertJsonPath('task.query_task_attempt', 1);
+        } else {
+            $this->registerPythonWorker('peer', 'python-queries', ['python.queryable']);
+            $this->postJson('/api/worker/query-tasks/poll', ['worker_id' => 'peer', 'task_queue' => 'python-queries', 'timeout_seconds' => 0], $this->workerHeaders())
+                ->assertOk()->assertJsonPath('task.query_task_id', $task['query_task_id']);
+        }
+    }
+
+    public static function deliveryCacheFailures(): array
+    {
+        return ['before commit' => ['before', false], 'false write' => ['false', false],
+            'after commit' => ['after', true], 'lease index' => ['index', false],
+            'false lease index' => ['index_false', false],
+            'pending index' => ['pending_index', true]];
+    }
+
+    public function test_a_failed_legacy_query_write_cannot_make_its_hint_authoritative_for_offers(): void
+    {
+        Queue::fake();
+        $store = new class extends ArrayStore
+        {
+            public bool $failClaims = false;
+
+            public function put($key, $value, $seconds)
+            {
+                if ($this->failClaims && is_array($value) && ($value['status'] ?? null) === 'leased'
+                    && str_starts_with($key, 'server:workflow-query-task:task:')) {
+                    throw new \RuntimeException('Fixture legacy claim write failure.');
+                }
+
+                return parent::put($key, $value, $seconds);
+            }
+        };
+        $this->bindPollingCacheStore($store);
+        $run = $this->startRemoteWorkflow('query-legacy-hint');
+        $broker = app(WorkflowQueryTaskBroker::class);
+        $task = $broker->enqueue('default', $run, 'status', $this->queryArguments());
+        $this->registerPythonWorker('legacy', 'python-queries', ['python.queryable']);
+        $store->failClaims = true;
+        $this->postJson('/api/worker/query-tasks/poll', ['worker_id' => 'legacy', 'task_queue' => 'python-queries', 'timeout_seconds' => 0], $this->workerHeaders())
+            ->assertStatus(500);
+        $this->assertSame('pending', $broker->task($task['query_task_id'])['status']);
+        $hint = 'server:workflow-query-task:lease:'.$task['query_task_id'];
+        $this->assertSame('legacy', $store->get($hint));
+        $store->failClaims = false;
+        $this->registerPythonWorker('delivery-worker', 'python-queries', ['python.queryable'], ['query_tasks', WorkerTaskDelivery::CAPABILITY]);
+        WorkerRegistration::query()->where('worker_id', 'delivery-worker')->firstOrFail()
+            ->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => 'python-queries', 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $path = '/api/worker/query-tasks/offers';
+        $offer = $this->postJson($path, $body, $headers)->assertOk()->assertJsonPath('poll_status', 'offered');
+        $this->postJson($path, $body + ['task_offer_token' => $offer->json('task_offer.token')], $headers)
+            ->assertOk()->assertJsonPath('task.query_task_id', $task['query_task_id'])->assertJsonPath('task.query_task_attempt', 1);
+        $this->assertSame('legacy', $store->get($hint), 'The new claimant must not delete another owner\'s hint.');
+        $this->assertSame('delivery-worker', $broker->task($task['query_task_id'])['lease_owner']);
+    }
+
     public function test_late_query_claim_keeps_original_deadline_despite_configuration_change(): void
     {
         Queue::fake();
diff --git a/tests/Feature/WorkflowUpdateValidationTaskTest.php b/tests/Feature/WorkflowUpdateValidationTaskTest.php
index d7fe76699..949be6549 100644
--- a/tests/Feature/WorkflowUpdateValidationTaskTest.php
+++ b/tests/Feature/WorkflowUpdateValidationTaskTest.php
@@ -8,6 +8,7 @@
 use App\Support\LongPollSignalStore;
 use App\Support\LongPollWaitSlotStore;
 use App\Support\ServerWorkflowControlPlane;
+use App\Support\WorkerTaskDelivery;
 use App\Support\WorkflowQueryTaskBroker;
 use App\Support\WorkflowUpdateValidationTaskBroker;
 use Illuminate\Foundation\Testing\RefreshDatabase;
@@ -33,6 +34,131 @@ class WorkflowUpdateValidationTaskTest extends TestCase
 
     private const TASK_QUEUE = 'validated-updates';
 
+    public function test_validation_offers_do_not_claim_and_lost_acceptance_replays_one_attempt(): void
+    {
+        foreach (['update-validation', 'workflow'] as $route) {
+            $workerId = 'delivery-'.$route;
+            $this->registerValidatorWorker($workerId, ['workflow_tasks', 'update_validation_tasks', WorkerTaskDelivery::CAPABILITY]);
+            $this->worker($workerId)->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+            $run = $this->startRemoteWorkflow('delivery-'.$route);
+            $this->createPendingValidationTask($run, 'delivery-request-'.$route);
+            $task = WorkflowUpdateValidationTask::query()->where('workflow_run_id', $run->id)->firstOrFail();
+            $before = $task->getRawOriginal();
+            $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+            $body = ['worker_id' => $workerId, 'task_queue' => self::TASK_QUEUE, 'timeout_seconds' => 0,
+                'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32),
+                'task_kinds' => ['workflow', 'update_validation']];
+            $path = '/api/worker/'.$route.'-tasks/offers';
+            $offer = $this->postJson($path, $body, $headers)->assertOk()->assertJsonPath('poll_status', 'offered');
+            $this->assertSame($before, $task->fresh()->getRawOriginal());
+            $acceptance = $body + ['task_offer_token' => $offer->json('task_offer.token')];
+            $first = $this->postJson($path, $acceptance, $headers)->assertOk()
+                ->assertJsonPath('task.update_validation_task_id', $task->id)->assertJsonPath('task.update_validation_attempt', 1);
+            $this->travel(3)->seconds();
+            $this->worker($workerId)->forceFill(['last_heartbeat_at' => now()])->save();
+            $this->postJson($path, $acceptance, $headers)->assertOk()
+                ->assertJsonPath('task.update_validation_task_id', $first->json('task.update_validation_task_id'))
+                ->assertJsonPath('task.update_validation_attempt', 1);
+            $this->assertSame(1, $task->fresh()->attempt_count);
+            $task->forceFill(['status' => WorkflowUpdateValidationTask::STATUS_APPROVED])->save();
+            $this->travelBack();
+        }
+    }
+
+    public function test_expired_validation_acceptance_starts_no_attempt(): void
+    {
+        $this->registerValidatorWorker('delivery-worker', ['workflow_tasks', 'update_validation_tasks', WorkerTaskDelivery::CAPABILITY]);
+        $this->worker('delivery-worker')->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $run = $this->startRemoteWorkflow('delivery-expired');
+        $this->createPendingValidationTask($run, 'delivery-expired');
+        $task = WorkflowUpdateValidationTask::query()->where('workflow_run_id', $run->id)->firstOrFail();
+        $before = $task->getRawOriginal();
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => self::TASK_QUEUE, 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $path = '/api/worker/update-validation-tasks/offers';
+        $offer = $this->postJson($path, $body, $headers)->assertOk();
+        $this->travel(3)->seconds();
+        $this->postJson($path, $body + ['task_offer_token' => $offer->json('task_offer.token')], $headers)
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+    }
+
+    public function test_one_multiplexed_poll_cannot_accept_both_a_validation_and_a_workflow_offer(): void
+    {
+        $this->registerValidatorWorker('delivery-worker', ['workflow_tasks', 'update_validation_tasks', WorkerTaskDelivery::CAPABILITY]);
+        $this->worker('delivery-worker')->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $run = $this->startRemoteWorkflow('delivery-competing-kinds');
+        $this->createPendingValidationTask($run, 'delivery-competing-kinds');
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => self::TASK_QUEUE, 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32),
+            'task_kinds' => ['workflow', 'update_validation']];
+        $path = '/api/worker/workflow-tasks/offers';
+        $validation = $this->postJson($path, $body, $headers)->assertOk()->assertJsonPath('task_offer.task_kind', 'update_validation');
+        $this->assertSame('update_validation', DB::table('workflow_task_poll_cursors')->value('next_task_kind'));
+        DB::table('workflow_task_poll_cursors')->update(['next_task_kind' => 'workflow']);
+        $workflow = $this->postJson($path, $body, $headers)->assertOk()->assertJsonPath('task_offer.task_kind', 'workflow');
+        $this->assertSame('workflow', DB::table('workflow_task_poll_cursors')->value('next_task_kind'));
+        $this->postJson($path, $body + ['task_offer_token' => $validation->json('task_offer.token')], $headers)
+            ->assertOk()->assertJsonPath('task.task_kind', 'update_validation');
+        $before = WorkflowTask::query()->where('workflow_run_id', $run->id)->firstOrFail()->getRawOriginal();
+        $this->postJson($path, $body + ['task_offer_token' => $workflow->json('task_offer.token')], $headers)
+            ->assertConflict()->assertJsonPath('reason', 'task_poll_already_accepted');
+        $this->assertSame($before, WorkflowTask::query()->where('workflow_run_id', $run->id)->firstOrFail()->getRawOriginal());
+        $this->assertSame(1, WorkflowUpdateValidationTask::query()->where('status', 'leased')->count());
+    }
+
+    public function test_expiry_resolves_a_validation_offer_after_a_peer_claim(): void
+    {
+        $this->registerValidatorWorker('delivery-worker', ['workflow_tasks', 'update_validation_tasks', WorkerTaskDelivery::CAPABILITY]);
+        $this->worker('delivery-worker')->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $run = $this->startRemoteWorkflow('delivery-expired-peer');
+        $this->createPendingValidationTask($run, 'delivery-expired-peer');
+        $task = WorkflowUpdateValidationTask::query()->where('workflow_run_id', $run->id)->firstOrFail();
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => self::TASK_QUEUE, 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $path = '/api/worker/update-validation-tasks/offers';
+        $offer = $this->postJson($path, $body, $headers)->assertOk();
+        $this->registerValidatorWorker('peer', ['workflow_tasks', 'update_validation_tasks']);
+        $this->postJson('/api/worker/update-validation-tasks/poll', [
+            'worker_id' => 'peer', 'task_queue' => self::TASK_QUEUE, 'timeout_seconds' => 0,
+        ], $this->workerHeaders())->assertOk()->assertJsonPath('task.update_validation_task_id', $task->id);
+        $before = $task->fresh()->getRawOriginal();
+        $this->travel(3)->seconds();
+        $this->postJson($path, $body + ['task_offer_token' => $offer->json('task_offer.token')], $headers)
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+    }
+
+    public function test_validation_claim_that_finishes_after_the_offer_budget_rolls_back(): void
+    {
+        $this->registerValidatorWorker('delivery-worker', ['workflow_tasks', 'update_validation_tasks', WorkerTaskDelivery::CAPABILITY]);
+        $this->worker('delivery-worker')->forceFill(['registration_token' => str_repeat('a', 32)])->save();
+        $run = $this->startRemoteWorkflow('delivery-late-validation');
+        $this->createPendingValidationTask($run, 'delivery-late-validation');
+        $task = WorkflowUpdateValidationTask::query()->where('workflow_run_id', $run->id)->firstOrFail();
+        $before = $task->getRawOriginal();
+        $headers = $this->workerHeaders() + [WorkerTaskDelivery::HEADER => WorkerTaskDelivery::VERSION];
+        $body = ['worker_id' => 'delivery-worker', 'task_queue' => self::TASK_QUEUE, 'timeout_seconds' => 0,
+            'poll_request_id' => 'delivery-poll', 'registration_token' => str_repeat('a', 32)];
+        $path = '/api/worker/update-validation-tasks/offers';
+        $offer = $this->postJson($path, $body, $headers)->assertOk();
+        $advanced = false;
+        DB::listen(function ($query) use (&$advanced): void {
+            if (! $advanced && str_starts_with(str_replace(['"', '`'], '', $query->sql), 'update workflow_update_validation_tasks')
+                && in_array('leased', $query->bindings, true)) {
+                $advanced = true;
+                $this->travel(3)->seconds();
+            }
+        });
+        $this->postJson($path, $body + ['task_offer_token' => $offer->json('task_offer.token')], $headers)
+            ->assertConflict()->assertJsonPath('reason', 'task_offer_expired');
+        $this->assertTrue($advanced);
+        $this->assertSame($before, $task->fresh()->getRawOriginal());
+    }
+
     protected function setUp(): void
     {
         parent::setUp();
diff --git a/tests/Support/TaskDeliveryAcceptProcess.php b/tests/Support/TaskDeliveryAcceptProcess.php
new file mode 100644
index 000000000..21bb91248
--- /dev/null
+++ b/tests/Support/TaskDeliveryAcceptProcess.php
@@ -0,0 +1,46 @@
+<?php
+
+use Illuminate\Contracts\Http\Kernel;
+use Illuminate\Http\Request;
+use Tests\Fixtures\ExternalGreetingWorkflow;
+
+require dirname(__DIR__, 2).'/vendor/autoload.php';
+$app = require dirname(__DIR__, 2).'/bootstrap/app.php';
+$kernel = $app->make(Kernel::class);
+$input = json_decode(stream_get_contents(STDIN), true, flags: JSON_THROW_ON_ERROR);
+$kernel->bootstrap();
+config(['workflows.v2.types.workflows' => ['tests.external-greeting-workflow' => ExternalGreetingWorkflow::class]]);
+fwrite(STDOUT, "READY\n");
+touch($input['ready']);
+$deadline = microtime(true) + 10;
+while (! is_file($input['barrier'])) {
+    clearstatcache(true, $input['barrier']);
+    if (microtime(true) >= $deadline) {
+        throw new RuntimeException('Concurrent acceptance barrier timed out.');
+    }
+    usleep(1000);
+}
+$body = file_get_contents($input['body_file']);
+$attempts = [];
+for ($attempt = 0; $attempt < 3; $attempt++) {
+    $request = Request::create($input['path'], 'POST', [], [], [], [
+        'CONTENT_TYPE' => 'application/json', 'HTTP_ACCEPT' => 'application/json',
+        'HTTP_X_NAMESPACE' => 'default',
+        'HTTP_X_DURABLE_WORKFLOW_PROTOCOL_VERSION' => '1.20',
+        'HTTP_X_DURABLE_WORKFLOW_TASK_DELIVERY' => 'offer-accept.v1',
+    ], $body);
+    $response = $kernel->handle($request);
+    $result = ['status' => $response->getStatusCode(),
+        'body' => json_decode($response->getContent(), true, flags: JSON_THROW_ON_ERROR)];
+    $kernel->terminate($request, $response);
+    $attempts[] = ['status' => $result['status'], 'reason' => $result['body']['reason'] ?? null,
+        'task' => $result['body']['task']['task_id'] ?? null];
+    if ($result['status'] !== 503 || ($result['body']['reason'] ?? null) !== 'backend_lock_pressure' || $attempt === 2) {
+        break;
+    }
+    // This is the normal typed retry contract, with identical request bytes.
+    // The offer's original two-second deadline is never extended.
+    sleep(max(1, (int) ($result['body']['retry_after_seconds'] ?? 1)));
+}
+$result['attempts'] = $attempts;
+fwrite(STDOUT, json_encode($result, JSON_THROW_ON_ERROR)."\n");