diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml
index 30606976..1c244f01 100644
--- a/.github/workflows/phpunit-feature.yml
+++ b/.github/workflows/phpunit-feature.yml
@@ -215,7 +215,9 @@ jobs:
                 -e DW_TEST_S3_ACCESS_KEY_ID=dw-minio-access \
                 -e DW_TEST_S3_SECRET_ACCESS_KEY=dw-minio-secret-key \
                 -e DW_TEST_BACKUP_MYSQL_HOST="dw-server-backup-mysql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+                -e DW_TEST_COMPLETION_MYSQL_HOST="dw-server-backup-mysql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
                 -e DW_TEST_BACKUP_PGSQL_HOST="dw-server-backup-pgsql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+                -e DW_TEST_COMPLETION_PGSQL_HOST="dw-server-backup-pgsql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
                 -w /app \
                 --entrypoint sh \
                 durable-workflow-server-corpus-validator \
diff --git a/app/Http/Controllers/Api/RuntimeExternalPayloadController.php b/app/Http/Controllers/Api/RuntimeExternalPayloadController.php
index 5916af24..3f234f2f 100644
--- a/app/Http/Controllers/Api/RuntimeExternalPayloadController.php
+++ b/app/Http/Controllers/Api/RuntimeExternalPayloadController.php
@@ -7,6 +7,10 @@
 use App\Support\RuntimeExternalPayloadReference;
 use App\Support\RuntimeExternalPayloadRegistry;
 use App\Support\RuntimeExternalPayloadUploadBody;
+use App\Support\RuntimePayloadCompletionContext;
+use App\Support\RuntimePayloadCompletionUploads;
+use App\Support\StorageAdmissionPaused;
+use App\Support\StoragePressure;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
 use Symfony\Component\HttpFoundation\Response;
@@ -62,12 +66,24 @@ public function store(Request $request): JsonResponse
         }
 
         $namespace = (string) $request->attributes->get('namespace', config('server.default_namespace'));
-        $reference = $this->registry->upload(
-            $namespace,
-            $data,
-            (string) $request->header('X-Durable-Workflow-Payload-Codec'),
-            strtolower((string) $request->header('X-Durable-Workflow-Payload-SHA256')),
-        );
+        $context = $request->attributes->get(RuntimePayloadCompletionContext::class);
+        $codec = (string) $request->header('X-Durable-Workflow-Payload-Codec');
+        $sha256 = strtolower((string) $request->header('X-Durable-Workflow-Payload-SHA256'));
+        try {
+            if ($context instanceof RuntimePayloadCompletionContext) {
+                $reference = app(RuntimePayloadCompletionUploads::class)->upload($namespace, $context, $data, $codec, $sha256, $this->registry);
+            } else {
+                // Pressure may change while an admitted HTTP upload is being read.
+                $snapshot = app(StoragePressure::class)->snapshot();
+                if ($snapshot['state'] !== 'normal') {
+                    throw new StorageAdmissionPaused($snapshot);
+                }
+                $reference = $this->registry->upload($namespace, $data, $codec, $sha256,
+                    fn () => app(StoragePressure::class)->requireNewWork());
+            }
+        } finally {
+            $data->close();
+        }
 
         $this->audit->record($request, 'external_payload.uploaded', [
             'reference_identity_sha256' => hash('sha256', $reference['reference_id']),
diff --git a/app/Http/Middleware/EnforceStorageAdmission.php b/app/Http/Middleware/EnforceStorageAdmission.php
index 96088125..4e9baf79 100644
--- a/app/Http/Middleware/EnforceStorageAdmission.php
+++ b/app/Http/Middleware/EnforceStorageAdmission.php
@@ -3,6 +3,8 @@
 namespace App\Http\Middleware;
 
 use App\Support\ControlPlaneProtocol;
+use App\Support\RuntimePayloadCompletionContext;
+use App\Support\RuntimePayloadCompletionUploads;
 use App\Support\StoragePressure;
 use App\Support\WorkerProtocol;
 use Closure;
@@ -39,6 +41,21 @@ public function handle(Request $request, Closure $next): Response
     {
         $snapshot = $this->pressure->snapshot();
         $action = class_basename((string) $request->route()?->getActionName());
+        if ($action === 'RuntimeExternalPayloadController@store'
+            && $request->headers->has(RuntimePayloadCompletionContext::HEADER)
+            && in_array($snapshot['state'], ['normal', 'draining'], true)) {
+            return app(RequireRole::class)->handle($request, function (Request $request) use ($next): Response {
+                $context = RuntimePayloadCompletionContext::parse((string) $request->header(RuntimePayloadCompletionContext::HEADER));
+                app(RuntimePayloadCompletionUploads::class)->authorize(
+                    (string) $request->attributes->get('namespace'), $context,
+                    strtolower((string) $request->header('X-Durable-Workflow-Payload-SHA256')),
+                    (int) $request->header('X-Durable-Workflow-Payload-Size'),
+                );
+                $request->attributes->set(RuntimePayloadCompletionContext::class, $context);
+
+                return $next($request);
+            }, 'worker');
+        }
         if ($snapshot['state'] === 'normal' || $request->isMethodSafe()
             || $action === 'WorkerController@workflowTaskHistory'
             || ($snapshot['state'] === 'draining' && in_array($action, self::DRAIN_ACTIONS, true))) {
diff --git a/app/Models/RuntimePayloadCompletionBudget.php b/app/Models/RuntimePayloadCompletionBudget.php
new file mode 100644
index 00000000..0eff3d8a
--- /dev/null
+++ b/app/Models/RuntimePayloadCompletionBudget.php
@@ -0,0 +1,18 @@
+<?php
+
+namespace App\Models;
+
+use Illuminate\Database\Eloquent\Model;
+
+class RuntimePayloadCompletionBudget extends Model
+{
+    public $incrementing = false;
+
+    protected $keyType = 'string';
+
+    protected $guarded = [];
+
+    protected $casts = [
+        'context' => 'array', 'slots' => 'array', 'objects' => 'array', 'expires_at' => 'immutable_datetime',
+    ];
+}
diff --git a/app/Support/RuntimeExternalPayloadCleanup.php b/app/Support/RuntimeExternalPayloadCleanup.php
index d8c731c0..700c8d2f 100644
--- a/app/Support/RuntimeExternalPayloadCleanup.php
+++ b/app/Support/RuntimeExternalPayloadCleanup.php
@@ -44,6 +44,8 @@ public function runPass(
         $limit = min(self::MAX_BATCH_SIZE, max(1, $limit));
         $cutoff ??= now();
 
+        app(RuntimePayloadCompletionUploads::class)->cleanup($namespace, $limit, $cutoff);
+
         $candidates = $this->expiredQuery($namespace, $cutoff)
             ->orderBy('expires_at')
             ->orderBy('id')
diff --git a/app/Support/RuntimeExternalPayloadObjectLock.php b/app/Support/RuntimeExternalPayloadObjectLock.php
index 66752921..a9ad8701 100644
--- a/app/Support/RuntimeExternalPayloadObjectLock.php
+++ b/app/Support/RuntimeExternalPayloadObjectLock.php
@@ -22,11 +22,14 @@ final class RuntimeExternalPayloadObjectLock
      * @param  Closure(): TReturn  $callback
      * @return TReturn
      */
-    public function transaction(string $uri, Closure $callback): mixed
+    public function transaction(string $uri, Closure $callback, ?Closure $beforeLock = null): mixed
     {
         $bucket = hexdec(substr(hash('sha256', $uri), 0, 4)) % self::BUCKETS;
 
-        return DB::transaction(function () use ($bucket, $callback): mixed {
+        return DB::transaction(function () use ($bucket, $callback, $beforeLock): mixed {
+            // Completion ownership locks must precede payload locks, as they do
+            // when the engine commits an activity result.
+            $beforeLock?->__invoke();
             $lock = DB::table('runtime_external_payload_object_locks')
                 ->where('bucket', $bucket)
                 ->lockForUpdate()
diff --git a/app/Support/RuntimeExternalPayloadReference.php b/app/Support/RuntimeExternalPayloadReference.php
index def5af90..91931776 100644
--- a/app/Support/RuntimeExternalPayloadReference.php
+++ b/app/Support/RuntimeExternalPayloadReference.php
@@ -72,6 +72,14 @@ public static function transportManifest(): array
                     'X-Durable-Workflow-Payload-SHA256',
                 ],
                 'idempotency' => 'content_addressed_per_namespace',
+                'completion_context' => [
+                    'schema' => RuntimePayloadCompletionContext::SCHEMA,
+                    'header' => RuntimePayloadCompletionContext::HEADER,
+                    'use' => 'retry_draining_refusal_for_current_worker_completion_only',
+                    'max_bytes_per_lease' => RuntimePayloadCompletionUploads::maxBytes(),
+                    'max_slots_per_lease' => RuntimePayloadCompletionUploads::MAX_SLOTS,
+                    'fenced_or_stale_admission' => 'refuse',
+                ],
             ],
             'fetch' => [
                 'method' => 'GET',
@@ -121,6 +129,10 @@ public static function transportManifest(): array
                 'external_payload_namespace_bytes_exhausted' => ['status' => 429, 'retryable' => true],
                 'external_payload_namespace_objects_exhausted' => ['status' => 429, 'retryable' => true],
                 'external_payload_namespace_quota_unavailable' => ['status' => 503, 'retryable' => true],
+                'external_payload_completion_invalid' => ['status' => 422, 'retryable' => false],
+                'external_payload_completion_lease_rejected' => ['status' => 409, 'retryable' => false],
+                'external_payload_completion_conflict' => ['status' => 409, 'retryable' => false],
+                'external_payload_completion_budget_exhausted' => ['status' => 429, 'retryable' => true],
             ],
             'audit_events' => [
                 'external_payload.uploaded',
diff --git a/app/Support/RuntimeExternalPayloadRegistry.php b/app/Support/RuntimeExternalPayloadRegistry.php
index 39759a09..8dd2e253 100644
--- a/app/Support/RuntimeExternalPayloadRegistry.php
+++ b/app/Support/RuntimeExternalPayloadRegistry.php
@@ -19,7 +19,7 @@ public function __construct(
     /**
      * @return array{schema: string, reference_id: string, codec: string, size_bytes: int, sha256: string}
      */
-    public function upload(string $namespace, string|ExternalPayloadStream $data, string $codec, string $sha256): array
+    public function upload(string $namespace, string|ExternalPayloadStream $data, string $codec, string $sha256, ?\Closure $beforeWrite = null): array
     {
         $namespace = $this->namespace($namespace);
         try {
@@ -56,6 +56,7 @@ public function upload(string $namespace, string|ExternalPayloadStream $data, st
             sizeBytes: $sizeBytes,
             retained: false,
             expiresAt: $expiresAt,
+            beforeWrite: $beforeWrite,
         ));
     }
 
@@ -465,6 +466,7 @@ private function store(
         int $sizeBytes,
         bool $retained,
         mixed $expiresAt,
+        ?\Closure $beforeWrite = null,
     ): RuntimeExternalPayload {
         $observedSize = $data instanceof ExternalPayloadStream ? $data->sizeBytes : strlen($data);
         $observedHash = $data instanceof ExternalPayloadStream ? $data->sha256 : hash('sha256', $data);
@@ -479,17 +481,12 @@ private function store(
         }
 
         try {
-            $this->objectLock->transaction($uri, fn (): RuntimeExternalPayload => $this->track(
-                $namespace,
-                $uri,
-                $codec,
-                $sha256,
-                $sizeBytes,
-                false,
-                $expiresAt,
-                RuntimeExternalPayload::UPLOAD_WRITING,
-            ));
-        } catch (RuntimeExternalPayloadException $exception) {
+            $this->objectLock->transaction($uri, function () use ($namespace, $uri, $codec, $sha256, $sizeBytes, $expiresAt, $beforeWrite): RuntimeExternalPayload {
+                $beforeWrite?->__invoke();
+
+                return $this->track($namespace, $uri, $codec, $sha256, $sizeBytes, false, $expiresAt, RuntimeExternalPayload::UPLOAD_WRITING);
+            }, $beforeWrite);
+        } catch (RuntimeExternalPayloadException|StorageAdmissionPaused $exception) {
             throw $exception;
         } catch (Throwable $exception) {
             throw $this->unavailable('External payload reference registration failed before storage commit.', $exception);
@@ -506,7 +503,9 @@ private function store(
                 $sizeBytes,
                 $retained,
                 $expiresAt,
+                $beforeWrite,
             ): RuntimeExternalPayload {
+                $beforeWrite?->__invoke();
                 $row = RuntimeExternalPayload::query()
                     ->where('namespace', $namespace)
                     ->where('storage_uri_sha256', hash('sha256', $uri))
@@ -546,8 +545,8 @@ private function store(
                     $expiresAt,
                     RuntimeExternalPayload::UPLOAD_READY,
                 );
-            });
-        } catch (RuntimeExternalPayloadException $exception) {
+            }, $beforeWrite);
+        } catch (RuntimeExternalPayloadException|StorageAdmissionPaused $exception) {
             throw $exception;
         } catch (Throwable $exception) {
             // The writing row was committed before the backing write. If this
diff --git a/app/Support/RuntimePayloadCompletionContext.php b/app/Support/RuntimePayloadCompletionContext.php
new file mode 100644
index 00000000..a38a0de8
--- /dev/null
+++ b/app/Support/RuntimePayloadCompletionContext.php
@@ -0,0 +1,109 @@
+<?php
+
+namespace App\Support;
+
+use JsonException;
+
+final readonly class RuntimePayloadCompletionContext
+{
+    public const HEADER = 'X-Durable-Workflow-Payload-Completion';
+
+    public const SCHEMA = 'durable-workflow.v2.payload-completion-context.v1';
+
+    /** @param list<int|string> $slot */
+    private function __construct(
+        public string $kind,
+        public string $taskId,
+        public int|string $attempt,
+        public string $leaseOwner,
+        public string $operation,
+        public array $slot,
+    ) {}
+
+    public static function parse(string $header): self
+    {
+        if (strlen($header) > 4096) {
+            throw self::invalid();
+        }
+        try {
+            $value = json_decode($header, true, 8, JSON_THROW_ON_ERROR);
+        } catch (JsonException) {
+            throw self::invalid();
+        }
+        $keys = ['schema', 'kind', 'task_id', 'attempt', 'lease_owner', 'operation', 'slot'];
+        if (! is_array($value) || count($value) !== count($keys)
+            || array_diff($keys, array_keys($value)) !== []
+            || ($value['schema'] ?? null) !== self::SCHEMA
+            || ! in_array($value['kind'] ?? null, ['activity', 'workflow', 'query'], true)
+            || ! is_string($value['operation'])
+            || ! self::identifier($value['task_id']) || ! self::identifier($value['lease_owner'])
+            || ! is_array($value['slot']) || ! array_is_list($value['slot'])) {
+            throw self::invalid();
+        }
+        if ($value['kind'] === 'activity'
+            ? ! self::identifier($value['attempt'])
+            : (! is_int($value['attempt']) || $value['attempt'] < 1)) {
+            throw self::invalid();
+        }
+        $validSlot = match ($value['kind'].'.'.($value['operation'] ?? '')) {
+            'activity.complete' => $value['slot'] === ['result'],
+            'activity.fail' => $value['slot'] === ['failure', 'details'],
+            'query.complete' => $value['slot'] === ['result_envelope'],
+            'workflow.complete' => self::workflowSlot($value['slot']),
+            default => false,
+        };
+        if (! $validSlot) {
+            throw self::invalid();
+        }
+
+        return new self($value['kind'], $value['task_id'], $value['attempt'],
+            $value['lease_owner'], $value['operation'], $value['slot']);
+    }
+
+    public function scope(string $namespace): string
+    {
+        // Complete/fail and every payload slot share one allowance for this lease.
+        return hash('sha256', json_encode([$namespace, $this->kind, $this->taskId,
+            $this->attempt, $this->leaseOwner], JSON_THROW_ON_ERROR));
+    }
+
+    public function slotIdentity(): string
+    {
+        return hash('sha256', json_encode([$this->operation, $this->slot], JSON_THROW_ON_ERROR));
+    }
+
+    public function toArray(): array
+    {
+        return ['schema' => self::SCHEMA, 'kind' => $this->kind, 'task_id' => $this->taskId,
+            'attempt' => $this->attempt, 'lease_owner' => $this->leaseOwner,
+            'operation' => $this->operation, 'slot' => $this->slot];
+    }
+
+    private static function identifier(mixed $value): bool
+    {
+        return is_string($value) && $value !== '' && strlen($value) <= 255
+            && preg_match('/[\x00-\x1f\x7f]/', $value) === 0;
+    }
+
+    private static function workflowSlot(array $slot): bool
+    {
+        if (($slot[0] ?? null) !== 'commands' || ! is_int($slot[1] ?? null) || $slot[1] < 0) {
+            return false;
+        }
+        if (count($slot) === 3) {
+            return in_array($slot[2], ['arguments', 'entries', 'request_payload', 'result'], true);
+        }
+        if (count($slot) === 4) {
+            return array_slice($slot, 2) === ['exception', 'details'];
+        }
+
+        return count($slot) === 6 && $slot[2] === 'workflow_stream' && $slot[3] === 'items'
+            && is_int($slot[4]) && $slot[4] >= 0 && $slot[5] === 'payload';
+    }
+
+    private static function invalid(): RuntimeExternalPayloadException
+    {
+        return new RuntimeExternalPayloadException('external_payload_completion_invalid', 422, false,
+            'A canonical worker completion context is required.');
+    }
+}
diff --git a/app/Support/RuntimePayloadCompletionLease.php b/app/Support/RuntimePayloadCompletionLease.php
new file mode 100644
index 00000000..2d20e4ed
--- /dev/null
+++ b/app/Support/RuntimePayloadCompletionLease.php
@@ -0,0 +1,119 @@
+<?php
+
+namespace App\Support;
+
+use App\Models\WorkerRegistration;
+use Carbon\CarbonImmutable;
+use Workflow\V2\Contracts\ActivityTaskBridge;
+use Workflow\V2\Enums\TaskType;
+use Workflow\V2\Support\WorkflowTaskOwnership;
+
+final class RuntimePayloadCompletionLease
+{
+    public function __construct(
+        private readonly ActivityTaskBridge $activities,
+        private readonly WorkflowTaskOwnership $workflows,
+        private readonly WorkflowQueryTaskBroker $queries,
+    ) {}
+
+    public function expiresAt(string $namespace, RuntimePayloadCompletionContext $context): CarbonImmutable
+    {
+        $worker = WorkerRegistration::query()->where('namespace', $namespace)
+            ->where('worker_id', $context->leaseOwner)->first();
+        if (! $worker instanceof WorkerRegistration || ! WorkerPollFence::isFresh($worker)) {
+            throw self::rejected();
+        }
+
+        $expires = match ($context->kind) {
+            'activity' => $this->activity($namespace, $context),
+            'workflow' => $this->workflow($namespace, $context),
+            'query' => $this->query($namespace, $context),
+        };
+        if (! is_string($expires) || $expires === '') {
+            throw self::rejected();
+        }
+        try {
+            $expiresAt = CarbonImmutable::parse($expires);
+        } catch (\Exception) {
+            throw self::rejected();
+        }
+        if ($expiresAt->lte(now())) {
+            throw self::rejected();
+        }
+
+        return $expiresAt;
+    }
+
+    public function mayResume(string $namespace, RuntimePayloadCompletionContext $context): bool
+    {
+        // An expired activity can still be renewed before lease recovery. Keep its
+        // budget until its identity is replaced or closed, not merely until TTL.
+        if ($context->kind === 'query') {
+            $task = $this->queries->task($context->taskId);
+
+            return ($task['namespace'] ?? null) === $namespace && ($task['status'] ?? null) === 'leased'
+                && ($task['lease_owner'] ?? null) === $context->leaseOwner
+                && ($task['attempt_count'] ?? null) === $context->attempt;
+        }
+        $task = NamespaceWorkflowScope::task($namespace, $context->taskId);
+        if ($task === null || $task->status->value !== 'leased' || $task->lease_owner !== $context->leaseOwner) {
+            return false;
+        }
+        if ($context->kind === 'workflow') {
+            return $task->attempt_count === $context->attempt;
+        }
+        $status = $this->activities->status($context->attempt);
+
+        return ($status['can_continue'] ?? false) === true
+            && ($status['workflow_task_id'] ?? null) === $context->taskId
+            && ($status['lease_owner'] ?? null) === $context->leaseOwner;
+    }
+
+    private function activity(string $namespace, RuntimePayloadCompletionContext $context): ?string
+    {
+        $task = NamespaceWorkflowScope::task($namespace, $context->taskId);
+        if ($task === null || $task->task_type !== TaskType::Activity || $task->lease_owner !== $context->leaseOwner
+            || $task->lease_expires_at === null || $task->lease_expires_at->lte(now())) {
+            throw self::rejected();
+        }
+        // The bridge checks the current attempt, task and run without renewing the lease.
+        $status = $this->activities->status($context->attempt);
+        if (($status['can_continue'] ?? false) !== true
+            || ($status['workflow_task_id'] ?? null) !== $context->taskId
+            || ($status['lease_owner'] ?? null) !== $context->leaseOwner) {
+            throw self::rejected();
+        }
+
+        return $status['lease_expires_at'] ?? null;
+    }
+
+    private function workflow(string $namespace, RuntimePayloadCompletionContext $context): ?string
+    {
+        $guard = $this->workflows->guard(
+            fn (string $ns, string $id) => NamespaceWorkflowScope::task($ns, $id),
+            $namespace, $context->taskId, $context->attempt, $context->leaseOwner,
+        );
+        if (! $guard['valid'] || $guard['task']?->task_type !== TaskType::Workflow || in_array($guard['status']['run_status'] ?? null,
+            ['completed', 'failed', 'cancelled', 'terminated'], true)) {
+            throw self::rejected();
+        }
+
+        return $guard['status']['lease_expires_at'] ?? null;
+    }
+
+    private function query(string $namespace, RuntimePayloadCompletionContext $context): ?string
+    {
+        if ($this->queries->guardCompletion($namespace, $context->taskId,
+            $context->leaseOwner, $context->attempt) !== null) {
+            throw self::rejected();
+        }
+
+        return $this->queries->task($context->taskId)['lease_expires_at'] ?? null;
+    }
+
+    public static function rejected(): RuntimeExternalPayloadException
+    {
+        return new RuntimeExternalPayloadException('external_payload_completion_lease_rejected', 409, false,
+            'The payload upload does not belong to a current worker completion lease.');
+    }
+}
diff --git a/app/Support/RuntimePayloadCompletionUploads.php b/app/Support/RuntimePayloadCompletionUploads.php
new file mode 100644
index 00000000..c7deea05
--- /dev/null
+++ b/app/Support/RuntimePayloadCompletionUploads.php
@@ -0,0 +1,157 @@
+<?php
+
+namespace App\Support;
+
+use App\Models\RuntimeExternalPayload;
+use App\Models\RuntimePayloadCompletionBudget;
+use Carbon\CarbonInterface;
+
+final class RuntimePayloadCompletionUploads
+{
+    public const MAX_SLOTS = 128;
+
+    public function __construct(
+        private readonly RuntimePayloadCompletionLease $leases,
+        private readonly RuntimeExternalPayloadObjectLock $locks,
+        private readonly StoragePressure $pressure,
+    ) {}
+
+    public function authorize(string $namespace, RuntimePayloadCompletionContext $context, string $sha256, int $size): void
+    {
+        $this->requireWritable();
+        if ($this->reference($namespace, $context, $sha256, $size) === null) {
+            $this->leases->expiresAt($namespace, $context);
+        }
+    }
+
+    public function upload(
+        string $namespace, RuntimePayloadCompletionContext $context, ExternalPayloadStream $data,
+        string $codec, string $sha256, RuntimeExternalPayloadRegistry $registry,
+    ): array {
+        $this->requireWritable();
+        try {
+            PayloadCodecContract::canonicalize($codec);
+        } catch (\InvalidArgumentException $exception) {
+            throw new RuntimeExternalPayloadException('external_payload_unsupported', 422, false, $exception->getMessage());
+        }
+        if (! hash_equals($data->sha256, $sha256)) {
+            throw new RuntimeExternalPayloadException('external_payload_integrity_mismatch', 422, false,
+                'Declared external payload integrity metadata does not match the uploaded bytes.');
+        }
+        if (($reference = $this->reference($namespace, $context, $sha256, $data->sizeBytes)) !== null) {
+            // Lost upload/completion responses may be reconciled after the lease closes.
+            // This path never creates, rewrites, retains or extends an object.
+            return $reference;
+        }
+
+        $expiresAt = $this->leases->expiresAt($namespace, $context);
+        $scope = $context->scope($namespace);
+        $this->locks->transaction($scope, function () use ($namespace, $context, $sha256, $data, $expiresAt, $scope): void {
+            $this->requireWritable();
+            $budget = RuntimePayloadCompletionBudget::query()->lockForUpdate()->find($scope)
+                ?? new RuntimePayloadCompletionBudget(['id' => $scope, 'namespace' => $namespace,
+                    'context' => $context->toArray(), 'slots' => [], 'objects' => []]);
+            $slots = $budget->slots;
+            $objects = $budget->objects;
+            $slot = $context->slotIdentity();
+            $identity = ['sha256' => $sha256, 'size_bytes' => $data->sizeBytes];
+            if (isset($slots[$slot]) && array_intersect_key($slots[$slot], $identity) !== $identity) {
+                throw new RuntimeExternalPayloadException('external_payload_completion_conflict', 409, false,
+                    'A completion payload slot is already bound to different bytes.');
+            }
+            if (! isset($slots[$slot]) && count($slots) >= self::MAX_SLOTS
+                || ! isset($objects[$sha256]) && $data->sizeBytes > self::maxBytes() - array_sum($objects)) {
+                $snapshot = $this->pressure->snapshot();
+                if ($snapshot['state'] === 'draining') {
+                    // No slot or object mutation was admitted. Preserve the worker's
+                    // existing pause/retry contract, not an activity failure.
+                    throw new StorageAdmissionPaused($snapshot, unadmitted: true);
+                }
+                throw new RuntimeExternalPayloadException('external_payload_completion_budget_exhausted', 429, true,
+                    'The bounded payload allowance for this completion lease is exhausted.', retryAfterSeconds: 5);
+            }
+            $slots[$slot] ??= $identity;
+            $objects[$sha256] = $data->sizeBytes;
+            $budget->forceFill(['slots' => $slots, 'objects' => $objects,
+                'expires_at' => $expiresAt->addSeconds($this->retryRetention())])->save();
+        });
+
+        $reference = $registry->upload($namespace, $data, $codec, $sha256, function () use ($namespace, $context): void {
+            $this->requireWritable();
+            $this->leases->expiresAt($namespace, $context);
+        });
+        $this->locks->transaction($scope, function () use ($scope, $context, $reference): void {
+            $this->requireWritable();
+            $budget = RuntimePayloadCompletionBudget::query()->lockForUpdate()->findOrFail($scope);
+            $slots = $budget->slots;
+            $slots[$context->slotIdentity()]['reference'] = $reference;
+            $budget->forceFill(['slots' => $slots])->save();
+        });
+
+        return $reference;
+    }
+
+    public static function maxBytes(): int
+    {
+        return max(1, (int) (config('server.external_payload_transport.completion_max_bytes')
+            ?? config('server.external_payload_transport.max_payload_bytes')));
+    }
+
+    public function cleanup(?string $namespace, int $limit, CarbonInterface $cutoff): void
+    {
+        $candidates = RuntimePayloadCompletionBudget::query()->where('expires_at', '<=', $cutoff)
+            ->when($namespace !== null, fn ($query) => $query->where('namespace', $namespace))
+            ->orderBy('expires_at')->limit($limit)->get();
+        foreach ($candidates as $candidate) {
+            $context = RuntimePayloadCompletionContext::parse(json_encode($candidate->context, JSON_THROW_ON_ERROR));
+            $mayResume = true;
+            $this->locks->transaction($candidate->id, function () use ($candidate, $cutoff, &$mayResume): void {
+                $this->pressure->requireNewWork();
+                $row = RuntimePayloadCompletionBudget::query()->lockForUpdate()->find($candidate->id);
+                if ($row === null || $row->expires_at->gt($cutoff)) {
+                    return;
+                }
+                // A heartbeat may keep an old lease alive: never reset its allowance.
+                if ($mayResume) {
+                    $row->forceFill(['expires_at' => now()->addSeconds($this->retryRetention())])->save();
+                } else {
+                    $row->delete();
+                }
+            }, function () use ($candidate, $context, &$mayResume): void {
+                $this->pressure->requireNewWork();
+                $mayResume = $this->leases->mayResume($candidate->namespace, $context);
+            });
+        }
+    }
+
+    private function reference(string $namespace, RuntimePayloadCompletionContext $context, string $sha256, int $size): ?array
+    {
+        $budget = RuntimePayloadCompletionBudget::query()->find($context->scope($namespace));
+        $slot = $budget?->slots[$context->slotIdentity()] ?? null;
+        $reference = $slot['reference'] ?? null;
+        if (! is_array($reference) || ($slot['sha256'] ?? null) !== $sha256 || ($slot['size_bytes'] ?? null) !== $size) {
+            return null;
+        }
+        $row = RuntimeExternalPayload::query()->whereKey($reference['reference_id'])->where('namespace', $namespace)
+            ->where('sha256', $sha256)->where('size_bytes', $size)->where('codec', 'avro')
+            ->where('upload_status', RuntimeExternalPayload::UPLOAD_READY)->first();
+        if ($row === null || ($row->retained_at === null && ($row->expires_at === null || $row->expires_at->lte(now())))) {
+            return null;
+        }
+
+        return $reference;
+    }
+
+    private function requireWritable(): void
+    {
+        $snapshot = $this->pressure->snapshot();
+        if (! in_array($snapshot['state'], ['normal', 'draining'], true)) {
+            throw new StorageAdmissionPaused($snapshot);
+        }
+    }
+
+    private function retryRetention(): int
+    {
+        return max(1, (int) config('server.external_payload_transport.abandoned_upload_expiry_seconds'));
+    }
+}
diff --git a/app/Support/StorageAdmissionPaused.php b/app/Support/StorageAdmissionPaused.php
index 96800bd0..2d09f4af 100644
--- a/app/Support/StorageAdmissionPaused.php
+++ b/app/Support/StorageAdmissionPaused.php
@@ -9,13 +9,13 @@
 
 final class StorageAdmissionPaused extends RuntimeException
 {
-    public function __construct(public readonly array $snapshot)
+    public function __construct(public readonly array $snapshot, private readonly bool $unadmitted = false)
     {
         parent::__construct('Storage admission paused before a new claim.');
     }
 
     public function render(Request $request): Response
     {
-        return app(EnforceStorageAdmission::class)->reject($request, $this->snapshot, false);
+        return app(EnforceStorageAdmission::class)->reject($request, $this->snapshot, $this->unadmitted);
     }
 }
diff --git a/composer.json b/composer.json
index 333e7122..5340112d 100644
--- a/composer.json
+++ b/composer.json
@@ -48,7 +48,7 @@
     },
     "extra": {
         "durable-workflow": {
-            "product-train": "2.3.4"
+            "product-train": "2.3.5"
         },
         "laravel": {
             "dont-discover": []
diff --git a/composer.lock b/composer.lock
index 908c17a6..7310934a 100644
--- a/composer.lock
+++ b/composer.lock
@@ -4,7 +4,7 @@
         "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
         "This file is @generated automatically"
     ],
-    "content-hash": "3e5f896b0401b41e5a397549f446227c",
+    "content-hash": "113d9348b3c41acbd32018eb77476363",
     "packages": [
         {
             "name": "apache/avro",
diff --git a/config/server.php b/config/server.php
index f0119516..8b6be75e 100644
--- a/config/server.php
+++ b/config/server.php
@@ -487,6 +487,11 @@
 
     'external_payload_transport' => [
         's3_disk' => 'external-payload-s3',
+        'completion_max_bytes' => EnvAuditor::env(
+            'DW_EXTERNAL_PAYLOAD_COMPLETION_MAX_BYTES',
+            'WORKFLOW_SERVER_EXTERNAL_PAYLOAD_COMPLETION_MAX_BYTES',
+            null,
+        ),
         'max_payload_bytes' => (int) EnvAuditor::env(
             'DW_EXTERNAL_PAYLOAD_MAX_BYTES',
             'WORKFLOW_SERVER_EXTERNAL_PAYLOAD_MAX_BYTES',
diff --git a/database/migrations/2026_09_08_000000_create_runtime_payload_completion_budgets.php b/database/migrations/2026_09_08_000000_create_runtime_payload_completion_budgets.php
new file mode 100644
index 00000000..08e3fe97
--- /dev/null
+++ b/database/migrations/2026_09_08_000000_create_runtime_payload_completion_budgets.php
@@ -0,0 +1,26 @@
+<?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::create('runtime_payload_completion_budgets', function (Blueprint $table): void {
+            $table->char('id', 64)->primary();
+            $table->string('namespace', 128)->index();
+            $table->json('context');
+            $table->json('slots');
+            $table->json('objects');
+            $table->timestamp('expires_at')->index();
+            $table->timestamps();
+        });
+    }
+
+    public function down(): void
+    {
+        Schema::dropIfExists('runtime_payload_completion_budgets');
+    }
+};
diff --git a/docker-compose.dedicated-matching.yml b/docker-compose.dedicated-matching.yml
index 7b252a5b..9d34c18c 100644
--- a/docker-compose.dedicated-matching.yml
+++ b/docker-compose.dedicated-matching.yml
@@ -32,13 +32,13 @@ name: durable-workflow-server
 # daemon reports `shape: dedicated`.
 
 # Generated by scripts/ci/sync-source-release.mjs. Do not edit the fallback.
-x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.4}}
+x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.5}}
 
 x-server-environment: &server-environment
   APP_NAME: "Durable Workflow Server"
   APP_ENV: ${APP_ENV:-local}
   DW_SERVER_KEY: ${DW_SERVER_KEY:-}
-  APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.4}}
+  APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.5}}
   APP_DEBUG: ${APP_DEBUG:-false}
   DB_CONNECTION: mysql
   DB_HOST: mysql
diff --git a/docker-compose.memo-rolling.yml b/docker-compose.memo-rolling.yml
index c171366f..158058a0 100644
--- a/docker-compose.memo-rolling.yml
+++ b/docker-compose.memo-rolling.yml
@@ -49,14 +49,14 @@ services:
     command: ["server-bootstrap"]
     environment:
       <<: *runtime-environment
-      APP_VERSION: ${APP_VERSION:-2.3.4}
+      APP_VERSION: ${APP_VERSION:-2.3.5}
 
   successor:
     image: ${DW_MEMO_SUCCESSOR_IMAGE:-durable-workflow/server-memo-rolling:local}
     ports: !override []
     environment:
       <<: *runtime-environment
-      APP_VERSION: ${APP_VERSION:-2.3.4}
+      APP_VERSION: ${APP_VERSION:-2.3.5}
       DW_SERVER_ID: memo-successor
       DW_SERVER_TOPOLOGY_SHAPE: standalone_server
       DW_SERVER_PROCESS_CLASS: server_http_node
diff --git a/docker-compose.published.yml b/docker-compose.published.yml
index 90e7184a..2203209e 100644
--- a/docker-compose.published.yml
+++ b/docker-compose.published.yml
@@ -1,13 +1,13 @@
 name: durable-workflow-server
 
 # Generated by scripts/ci/sync-source-release.mjs. Do not edit the fallback.
-x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.4}}
+x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.5}}
 
 x-server-environment: &server-environment
   APP_NAME: "Durable Workflow Server"
   APP_ENV: ${APP_ENV:-local}
   DW_SERVER_KEY: ${DW_SERVER_KEY:-}
-  APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.4}}
+  APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.5}}
   APP_DEBUG: ${APP_DEBUG:-false}
   LOG_CHANNEL: ${LOG_CHANNEL:-stderr}
   LOG_LEVEL: ${LOG_LEVEL:-info}
diff --git a/docker-compose.small-cluster.yml b/docker-compose.small-cluster.yml
index 476410c2..27fb87e9 100644
--- a/docker-compose.small-cluster.yml
+++ b/docker-compose.small-cluster.yml
@@ -12,7 +12,7 @@ x-server-build: &server-build
 x-server-environment: &server-environment
   APP_NAME: "Durable Workflow Server"
   APP_ENV: testing
-  APP_VERSION: ${APP_VERSION:-2.3.4}
+  APP_VERSION: ${APP_VERSION:-2.3.5}
   APP_DEBUG: "false"
   DW_SERVER_KEY: ${DW_SERVER_KEY:-base64:5Zt4nUhlCm3DD0nLXZJQdHiwPfb56yGo9gNV/g3jYbY=}
   DB_CONNECTION: ${DW_SMALL_CLUSTER_DB:-mysql}
diff --git a/docker-compose.yml b/docker-compose.yml
index feb61343..d22bea7b 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -15,7 +15,7 @@ services:
       DW_SERVER_KEY: "${DW_SERVER_KEY:-}"
       DW_SERVER_TOPOLOGY_SHAPE: standalone_server
       DW_SERVER_PROCESS_CLASS: server_http_node
-      APP_VERSION: "${APP_VERSION:-2.3.4}"
+      APP_VERSION: "${APP_VERSION:-2.3.5}"
       APP_DEBUG: "false"
       DB_CONNECTION: mysql
       DB_HOST: mysql
@@ -62,7 +62,7 @@ services:
       APP_NAME: "Durable Workflow Server"
       APP_ENV: local
       DW_SERVER_KEY: "${DW_SERVER_KEY:-}"
-      APP_VERSION: "${APP_VERSION:-2.3.4}"
+      APP_VERSION: "${APP_VERSION:-2.3.5}"
       APP_DEBUG: "false"
       DB_CONNECTION: mysql
       DB_HOST: mysql
@@ -123,7 +123,7 @@ services:
       DW_SERVER_KEY: "${DW_SERVER_KEY:-}"
       DW_SERVER_TOPOLOGY_SHAPE: standalone_server
       DW_SERVER_PROCESS_CLASS: worker_node
-      APP_VERSION: "${APP_VERSION:-2.3.4}"
+      APP_VERSION: "${APP_VERSION:-2.3.5}"
       DB_CONNECTION: mysql
       DB_HOST: mysql
       DB_PORT: 3306
@@ -176,7 +176,7 @@ services:
       DW_SERVER_KEY: "${DW_SERVER_KEY:-}"
       DW_SERVER_TOPOLOGY_SHAPE: standalone_server
       DW_SERVER_PROCESS_CLASS: scheduler_node
-      APP_VERSION: "${APP_VERSION:-2.3.4}"
+      APP_VERSION: "${APP_VERSION:-2.3.5}"
       DB_CONNECTION: mysql
       DB_HOST: mysql
       DB_PORT: 3306
diff --git a/docs/storage-admission.md b/docs/storage-admission.md
index d193ab9d..10d4a6d4 100644
--- a/docs/storage-admission.md
+++ b/docs/storage-admission.md
@@ -54,6 +54,43 @@ history pruning, and payload cleanup do not start a new pass under pressure.
 
 ## Deployment Requirements
 
+### External Completion Payloads
+
+When discovery advertises `upload.completion_context`, a worker may retry a
+draining upload refusal with `X-Durable-Workflow-Payload-Completion`. The header
+is a JSON object, at most 4096 bytes, containing exactly:
+
+```json
+{"schema":"durable-workflow.v2.payload-completion-context.v1","kind":"activity","task_id":"task-id","attempt":"activity-attempt-id","lease_owner":"worker-id","operation":"complete","slot":["result"]}
+```
+
+`activity` uses its attempt ID; `workflow` and `query` use their positive integer
+attempt counter. The context must describe the current namespace-scoped lease.
+An activity allows `complete`/`["result"]` or `fail`/`["failure","details"]`;
+a query allows `complete`/`["result_envelope"]`. Workflow completion slots are
+`["commands", index, field]` for `arguments`, `entries`, `request_payload`, or
+`result`, plus `exception/details` and `workflow_stream/items/index/payload`.
+Indices are non-negative integers, not numeric strings.
+
+This is not a general upload permission. The worker role and current lease are
+checked before reading and before committing new bytes. Every slot is immutable
+within a lease. All its slots and complete/fail outcomes share at most 128 slots
+and `DW_EXTERNAL_PAYLOAD_COMPLETION_MAX_BYTES` distinct bytes, defaulting to the
+ordinary maximum external payload size. Matching retries do not consume another
+allowance. A recorded ready reference can be returned without a new write after
+the lease closes. Expired references cannot be recreated without a current lease.
+Existing namespace quotas still apply, and fenced/stale admission refuses uploads.
+An exhausted drain allowance returns the ordinary retryable storage-pressure
+refusal, so workers hold their result instead of reporting an activity failure.
+After capacity recovers, retry the ordinary upload without the completion header.
+
+Normal uploads without this header keep their existing behavior. SDKs must not
+attach it to workflow starts, signals, updates, or other new input. The allowance
+does not reserve physical disk space or promise that arbitrarily many leased
+results fit; operators still need the headroom qualification below.
+
+### Physical Reserve
+
 This is cooperative admission, **not a database write fence or an exact byte
 reservation system**. A request, queue job, or maintenance pass already in
 progress can still write. A storage transition can race a successful check.
diff --git a/k8s/README.md b/k8s/README.md
index 4baf0110..3bf96497 100644
--- a/k8s/README.md
+++ b/k8s/README.md
@@ -13,7 +13,7 @@ The checked-in manifests are synchronized with the repository's stable source
 release and pin its Docker Hub tag:
 
 ```text
-durableworkflow/server:2.3.4
+durableworkflow/server:2.3.5
 ```
 
 Before production use, patch every workload image to the exact published tag or
@@ -21,15 +21,15 @@ digest you intend to run:
 
 ```bash
 kubectl set image -n durable-workflow deploy/durable-workflow-server \
-  server=durableworkflow/server:2.3.4
+  server=durableworkflow/server:2.3.5
 kubectl set image -n durable-workflow deploy/durable-workflow-worker \
-  worker=durableworkflow/server:2.3.4
+  worker=durableworkflow/server:2.3.5
 kubectl set image -n durable-workflow cronjob/durable-workflow-scheduler \
-  scheduler=durableworkflow/server:2.3.4
+  scheduler=durableworkflow/server:2.3.5
 ```
 
 GitHub Container Registry publishes the same release line at
-`ghcr.io/durable-workflow/server:2.3.4`. Digest pinning is preferred for strict
+`ghcr.io/durable-workflow/server:2.3.5`. Digest pinning is preferred for strict
 change control.
 
 The manifests expect you to provide:
diff --git a/k8s/helm/durable-workflow/Chart.yaml b/k8s/helm/durable-workflow/Chart.yaml
index d4312328..4fdf850a 100644
--- a/k8s/helm/durable-workflow/Chart.yaml
+++ b/k8s/helm/durable-workflow/Chart.yaml
@@ -5,11 +5,11 @@ type: application
 # The chart's own semver version. Bumped on every chart release; treated as
 # independent of the server image version (appVersion). Breaking-change rules
 # for this version live in docs/helm-upgrading.md alongside the chart.
-version: 0.1.83
+version: 0.1.84
 # The immutable Durable Workflow Server identity this chart release packages.
 # The onboarding default in values.yaml and appVersion are generated from the
 # checked-in source release record.
-appVersion: "2.3.4"
+appVersion: "2.3.5"
 kubeVersion: ">=1.27.0-0"
 home: https://durable-workflow.github.io/docs/2.0/deployment
 sources:
@@ -30,7 +30,7 @@ annotations:
   # exact commit that most recently changed the packaged chart.
   org.opencontainers.image.source: https://github.com/durable-workflow/server
   dev.durable-workflow.source-revision: "unreleased"
-  dev.durable-workflow.image-reference: "docker.io/durableworkflow/server:2.3.4"
+  dev.durable-workflow.image-reference: "docker.io/durableworkflow/server:2.3.5"
   artifacthub.io/license: MIT
   artifacthub.io/category: integration-delivery
   # Free-form changelog for the current chart release shown by Artifact Hub.
diff --git a/k8s/helm/durable-workflow/README.md b/k8s/helm/durable-workflow/README.md
index d189b6a3..e39ca26c 100644
--- a/k8s/helm/durable-workflow/README.md
+++ b/k8s/helm/durable-workflow/README.md
@@ -63,7 +63,7 @@ helm install durable-workflow ./k8s/helm/durable-workflow \
 
 ```yaml
 image:
-  tag: "2.3.4"
+  tag: "2.3.5"
   # Pin a digest in production:
   # digest: "sha256:abc123..."
   # memoPayloadStorage: "raw-json-v1" # Required for a digest or custom image.
diff --git a/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml b/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml
index df6f3395..3559eb19 100644
--- a/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml
+++ b/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml
@@ -1,7 +1,7 @@
 # CI fixture: GitOps / externally-managed-secret path. The chart consumes
 # existing Secrets and renders no Secret resources of its own.
 image:
-  tag: "2.3.4"
+  tag: "2.3.5"
 
 externalDatabase:
   connection: pgsql
diff --git a/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml b/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml
index 18ccae0a..199d9de3 100644
--- a/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml
+++ b/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml
@@ -1,6 +1,6 @@
 # CI fixture: ingress + autoscaling enabled. Exercises optional templates.
 image:
-  tag: "2.3.4"
+  tag: "2.3.5"
 
 externalDatabase:
   connection: mysql
diff --git a/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml b/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml
index 0d4c6c74..57fad99a 100644
--- a/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml
+++ b/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml
@@ -2,7 +2,7 @@
 # chart's render path is exercised end-to-end. Real deployments should use
 # existingSecret instead.
 image:
-  tag: "2.3.4"
+  tag: "2.3.5"
 
 externalDatabase:
   connection: mysql
diff --git a/k8s/helm/durable-workflow/templates/_helpers.tpl b/k8s/helm/durable-workflow/templates/_helpers.tpl
index 5d6c25aa..f75df08b 100644
--- a/k8s/helm/durable-workflow/templates/_helpers.tpl
+++ b/k8s/helm/durable-workflow/templates/_helpers.tpl
@@ -88,7 +88,7 @@ resolved by an explicit capability declaration or an existing workload marker.
 {{- define "durable-workflow.memoPayloadStorageForImage" -}}
 {{- $image := toString . -}}
 {{- $normalized := regexReplaceAll "^index\\.docker\\.io/" $image "docker.io/" -}}
-{{- if eq $normalized "docker.io/durableworkflow/server:2.3.4" -}}
+{{- if eq $normalized "docker.io/durableworkflow/server:2.3.5" -}}
 dual-v1
 {{- else if regexMatch "^docker\\.io/durableworkflow/server:2\\.0\\.0-rc\\.[0-9]+$" $normalized -}}
 {{- $releaseCandidate := atoi (regexFind "[0-9]+$" $normalized) -}}
diff --git a/k8s/helm/durable-workflow/values.yaml b/k8s/helm/durable-workflow/values.yaml
index 2f9a707a..acc8d6e9 100644
--- a/k8s/helm/durable-workflow/values.yaml
+++ b/k8s/helm/durable-workflow/values.yaml
@@ -21,7 +21,7 @@ image:
   registry: docker.io
   repository: durableworkflow/server
   # Generated by scripts/ci/sync-source-release.mjs. Do not edit this default.
-  tag: "2.3.4"
+  tag: "2.3.5"
   # Optional digest pin. When set, takes precedence over tag for change control.
   # Example: "sha256:abc123..."
   digest: ""
diff --git a/k8s/helm/examples/values-dev.yaml b/k8s/helm/examples/values-dev.yaml
index 5ee36577..f0fcc481 100644
--- a/k8s/helm/examples/values-dev.yaml
+++ b/k8s/helm/examples/values-dev.yaml
@@ -3,7 +3,7 @@
 # shape in production.
 
 image:
-  tag: "2.3.4"
+  tag: "2.3.5"
 
 externalDatabase:
   connection: mysql
diff --git a/k8s/helm/examples/values-external-secrets-operator.yaml b/k8s/helm/examples/values-external-secrets-operator.yaml
index f2c3b3e9..f0558429 100644
--- a/k8s/helm/examples/values-external-secrets-operator.yaml
+++ b/k8s/helm/examples/values-external-secrets-operator.yaml
@@ -5,7 +5,7 @@
 # concern.
 
 image:
-  tag: "2.3.4"
+  tag: "2.3.5"
 
 externalDatabase:
   connection: pgsql
diff --git a/k8s/helm/examples/values-production-existing-secrets.yaml b/k8s/helm/examples/values-production-existing-secrets.yaml
index 1ebd2e05..731483cf 100644
--- a/k8s/helm/examples/values-production-existing-secrets.yaml
+++ b/k8s/helm/examples/values-production-existing-secrets.yaml
@@ -10,7 +10,7 @@ image:
   repository: durable-workflow/server
   # Pin a digest in production for change-control auditability.
   digest: ""    # e.g. "sha256:abc123..."
-  tag: "2.3.4"
+  tag: "2.3.5"
 
 externalDatabase:
   connection: pgsql
diff --git a/k8s/migration-job.yaml b/k8s/migration-job.yaml
index 51443aa4..82ba17c1 100644
--- a/k8s/migration-job.yaml
+++ b/k8s/migration-job.yaml
@@ -13,7 +13,7 @@ spec:
       restartPolicy: OnFailure
       containers:
         - name: migrate
-          image: durableworkflow/server:2.3.4
+          image: durableworkflow/server:2.3.5
           command: ["server-entrypoint"]
           args: ["server-bootstrap"]
           envFrom:
diff --git a/k8s/scheduler-cronjob.yaml b/k8s/scheduler-cronjob.yaml
index 5197075e..2cf08f79 100644
--- a/k8s/scheduler-cronjob.yaml
+++ b/k8s/scheduler-cronjob.yaml
@@ -24,7 +24,7 @@ spec:
           restartPolicy: Never
           containers:
             - name: scheduler
-              image: durableworkflow/server:2.3.4
+              image: durableworkflow/server:2.3.5
               command: ["server-entrypoint"]
               args: ["sh", "-c", "php artisan schedule:evaluate --limit=100 --json; php artisan activity:timeout-enforce --limit=100; if php artisan list --raw | grep -q '^external-payloads:cleanup '; then php artisan external-payloads:cleanup --limit=100 --json; fi; php artisan history:prune --limit=100"]
               envFrom:
diff --git a/k8s/secret.yaml b/k8s/secret.yaml
index 542baa30..43f6631b 100644
--- a/k8s/secret.yaml
+++ b/k8s/secret.yaml
@@ -12,7 +12,7 @@ metadata:
     app.kubernetes.io/name: durable-workflow
 data:
   APP_NAME: "Durable Workflow Server"
-  APP_VERSION: "2.3.4"
+  APP_VERSION: "2.3.5"
   APP_ENV: production
   APP_DEBUG: "false"
   DB_CONNECTION: mysql
diff --git a/k8s/server-deployment.yaml b/k8s/server-deployment.yaml
index 3cadef37..d665b18b 100644
--- a/k8s/server-deployment.yaml
+++ b/k8s/server-deployment.yaml
@@ -23,7 +23,7 @@ spec:
     spec:
       containers:
         - name: server
-          image: durableworkflow/server:2.3.4
+          image: durableworkflow/server:2.3.5
           ports:
             - containerPort: 8080
               name: http
diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml
index 2a5d6303..e518f591 100644
--- a/k8s/worker-deployment.yaml
+++ b/k8s/worker-deployment.yaml
@@ -19,7 +19,7 @@ spec:
     spec:
       containers:
         - name: worker
-          image: durableworkflow/server:2.3.4
+          image: durableworkflow/server:2.3.5
           command: ["server-entrypoint"]
           args: ["php", "artisan", "queue:work", "--sleep=1", "--tries=3", "--max-time=3600"]
           envFrom:
diff --git a/resources/platform-protocol-specs/external-payload-transport.openapi.yaml b/resources/platform-protocol-specs/external-payload-transport.openapi.yaml
index 4060c391..4f874b14 100644
--- a/resources/platform-protocol-specs/external-payload-transport.openapi.yaml
+++ b/resources/platform-protocol-specs/external-payload-transport.openapi.yaml
@@ -32,6 +32,7 @@ paths:
         - $ref: "#/components/parameters/PayloadCodecHeader"
         - $ref: "#/components/parameters/PayloadSizeHeader"
         - $ref: "#/components/parameters/PayloadSha256Header"
+        - $ref: "#/components/parameters/PayloadCompletionHeader"
       requestBody:
         required: true
         content:
@@ -51,6 +52,7 @@ paths:
         "401": { $ref: "#/components/responses/Unauthorized" }
         "403": { $ref: "#/components/responses/Unauthorized" }
         "404": { $ref: "#/components/responses/NotFound" }
+        "409": { $ref: "#/components/responses/InvalidReference" }
         "413": { $ref: "#/components/responses/Oversized" }
         "415": { $ref: "#/components/responses/Unsupported" }
         "422": { $ref: "#/components/responses/InvalidReference" }
@@ -122,6 +124,19 @@ components:
       in: header
       required: true
       schema: { type: string, pattern: '^[a-f0-9]{64}$' }
+    PayloadCompletionHeader:
+      name: X-Durable-Workflow-Payload-Completion
+      in: header
+      required: false
+      description: >
+        Optional JSON worker completion context, at most 4096 UTF-8 bytes.
+        Requires the worker role and a current namespace-scoped task lease for
+        new writes. Use only after a draining refusal when discovery advertises
+        upload.completion_context. Contains exactly schema, kind, task_id,
+        attempt, lease_owner, operation and slot; see docs/storage-admission.md.
+        Fenced or stale admission remains closed. Recorded matching references
+        can be reconciled read-only after the lease closes.
+      schema: { type: string, maxLength: 4096, contentMediaType: application/json }
   headers:
     PayloadCodec:
       schema: { type: string, const: avro }
@@ -182,6 +197,10 @@ components:
             - external_payload_namespace_bytes_exhausted
             - external_payload_namespace_objects_exhausted
             - external_payload_namespace_quota_unavailable
+            - external_payload_completion_invalid
+            - external_payload_completion_lease_rejected
+            - external_payload_completion_conflict
+            - external_payload_completion_budget_exhausted
         message: { type: string, minLength: 1 }
         retryable: { type: boolean }
         status: { type: integer }
diff --git a/resources/release/source-release.json b/resources/release/source-release.json
index 5779b8db..f4765b34 100644
--- a/resources/release/source-release.json
+++ b/resources/release/source-release.json
@@ -1,9 +1,9 @@
 {
   "schema": "durable-workflow.server.source-release/v1",
   "server": {
-    "version": "2.3.4"
+    "version": "2.3.5"
   },
   "helm_chart": {
-    "version": "0.1.83"
+    "version": "0.1.84"
   }
 }
diff --git a/scripts/k8s-kind-smoke.sh b/scripts/k8s-kind-smoke.sh
index 825921f6..ffb6698d 100755
--- a/scripts/k8s-kind-smoke.sh
+++ b/scripts/k8s-kind-smoke.sh
@@ -7,7 +7,7 @@ cluster="${K8S_SMOKE_CLUSTER:-durable-workflow-server-smoke}"
 image="${K8S_SMOKE_IMAGE:-durableworkflow/server:k8s-smoke}"
 # Generated by scripts/ci/sync-source-release.mjs so the smoke replaces the
 # same default shipped by the public manifests.
-manifest_image="durableworkflow/server:2.3.4"
+manifest_image="durableworkflow/server:2.3.5"
 kind_node_image="${K8S_SMOKE_KIND_NODE_IMAGE:-kindest/node:v1.29.4}"
 artifact_dir="${K8S_SMOKE_ARTIFACT_DIR:-/tmp/durable-workflow-k8s-kind-smoke-artifacts}"
 rendered_dir="${artifact_dir}/rendered-manifests"
diff --git a/tests/Feature/RuntimePayloadCompletionProcessTest.php b/tests/Feature/RuntimePayloadCompletionProcessTest.php
new file mode 100644
index 00000000..e5f0257a
--- /dev/null
+++ b/tests/Feature/RuntimePayloadCompletionProcessTest.php
@@ -0,0 +1,152 @@
+<?php
+
+namespace Tests\Feature;
+
+use Illuminate\Filesystem\Filesystem;
+use PDO;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Process\Process;
+
+class RuntimePayloadCompletionProcessTest extends TestCase
+{
+    private array $processes = [];
+
+    private array $environment;
+
+    private ?PDO $databaseAdmin = null;
+
+    private string $database;
+
+    private string $directory;
+
+    public static function cases(): array
+    {
+        return [['sqlite', 'activity'], ['sqlite', 'workflow'], ['mysql', 'activity'], ['mysql', 'workflow'],
+            ['pgsql', 'activity'], ['pgsql', 'workflow']];
+    }
+
+    #[DataProvider('cases')]
+    public function test_concurrent_uploads_and_cold_retries_preserve_one_bounded_lease(string $driver, string $kind): void
+    {
+        $this->initialize($driver);
+        $this->runProbe('init', $kind);
+        $first = $this->probe('upload', $kind, 'alpha', '0', 'first');
+        $second = $this->probe('upload', $kind, 'bravo', '0', 'second');
+        $first->start();
+        $second->start();
+        foreach (['first' => $first, 'second' => $second] as $name => $process) {
+            $deadline = microtime(true) + 10;
+            while (! is_file($this->directory.'/'.$name.'.ready') && $process->isRunning() && microtime(true) < $deadline) {
+                usleep(10000);
+            }
+            self::assertFileExists($this->directory.'/'.$name.'.ready', $process->getErrorOutput());
+        }
+        touch($this->directory.'/go');
+        $a = $this->completedResult($first);
+        $b = $this->completedResult($second);
+        foreach ([$a, $b] as $response) {
+            if ($response['status'] === 503) {
+                self::assertSame('sqlite', $driver);
+                self::assertSame('backend_lock_pressure', $response['body']['reason']);
+                self::assertTrue($response['body']['retryable']);
+            }
+        }
+        $statuses = [$a['status'], $b['status']];
+        sort($statuses);
+        self::assertContains($statuses, [[201, 409], [201, 503]], json_encode([$a, $b]));
+        $winner = $a['status'] === 201 ? 'alpha' : 'bravo';
+        $loser = $winner === 'alpha' ? 'bravo' : 'alpha';
+        $accepted = $a['status'] === 201 ? $a : $b;
+        self::assertSame($accepted, $this->runProbe('upload', $kind, $winner));
+        self::assertSame(409, $this->runProbe('upload', $kind, $loser)['status']);
+        $exhausted = $this->runProbe('upload', $kind, $loser, '1');
+        self::assertSame(503, $exhausted['status']);
+        self::assertSame('storage_pressure', $exhausted['body']['reason']);
+        self::assertFalse($exhausted['body']['request_admitted']);
+        $status = $this->runProbe('status', $kind);
+        self::assertSame(['budgets' => 1, 'slots' => 1, 'objects' => 1,
+            'bytes' => $accepted['body']['reference']['size_bytes'], 'rows' => 1], $status);
+        self::assertSame(200, $this->runProbe('complete', $kind)['status']);
+        // Every operation above and below boots a new application/database connection.
+        self::assertSame($accepted, $this->runProbe('upload', $kind, $winner));
+        self::assertSame($status, $this->runProbe('status', $kind));
+        self::assertSame(409, $this->runProbe('upload', $kind, $loser, '1')['status']);
+    }
+
+    protected function tearDown(): void
+    {
+        foreach ($this->processes as $process) {
+            if ($process->isRunning()) {
+                $process->stop(1);
+            }
+        }
+        if ($this->databaseAdmin !== null) {
+            $this->databaseAdmin->exec('DROP DATABASE '.$this->database);
+        }
+        if (isset($this->directory)) {
+            (new Filesystem)->deleteDirectory($this->directory);
+        }
+        parent::tearDown();
+    }
+
+    private function initialize(string $driver): void
+    {
+        $prefix = $driver === 'pgsql' ? 'DW_TEST_COMPLETION_PGSQL_' : 'DW_TEST_COMPLETION_MYSQL_';
+        $host = getenv($prefix.'HOST');
+        if ($driver !== 'sqlite' && ! $host) {
+            $this->markTestSkipped('Set '.$prefix.'HOST to a disposable database test server.');
+        }
+        $this->directory = sys_get_temp_dir().'/dw-completion-process-'.bin2hex(random_bytes(6));
+        mkdir($this->directory, 0700);
+        $this->environment = ['APP_ENV' => 'testing', 'APP_CONFIG_CACHE' => $this->directory.'/config.php',
+            'CACHE_STORE' => 'array', 'QUEUE_CONNECTION' => 'database', 'DW_AUTH_DRIVER' => 'none',
+            'DW_WORKER_POLL_TIMEOUT' => '0', 'DB_URL' => false, 'DB_CONNECTION' => $driver,
+            'DB_DATABASE' => $this->directory.'/database.sqlite'];
+        if ($driver === 'sqlite') {
+            touch($this->environment['DB_DATABASE']);
+        } else {
+            $this->database = 'dw_completion_'.bin2hex(random_bytes(6));
+            $user = getenv($prefix.'USER') ?: ($driver === 'pgsql' ? 'postgres' : 'root');
+            $password = getenv($prefix.'PASSWORD') ?: '';
+            $port = $driver === 'pgsql' ? '5432' : '3306';
+            $this->databaseAdmin = new PDO($driver.':host='.$host.';port='.$port.($driver === 'pgsql' ? ';dbname=postgres' : ''), $user, $password,
+                [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
+            $this->databaseAdmin->exec('CREATE DATABASE '.$this->database);
+            $this->environment = array_replace($this->environment, ['DB_HOST' => $host, 'DB_PORT' => $port,
+                'DB_DATABASE' => $this->database, 'DB_USERNAME' => $user, 'DB_PASSWORD' => $password]);
+        }
+    }
+
+    private function probe(string $action, string $kind, string $variant = 'alpha', string $slot = '0', string $barrier = ''): Process
+    {
+        $process = new Process([PHP_BINARY, 'tests/Support/RuntimePayloadCompletionProcess.php', $action,
+            $this->directory, $kind, $variant, $slot, $barrier], dirname(__DIR__, 2), $this->environment,
+            timeout: $action === 'init' ? 120 : 30);
+        $this->processes[] = $process;
+
+        return $process;
+    }
+
+    private function runProbe(string $action, string $kind, string $variant = 'alpha', string $slot = '0'): array
+    {
+        $process = $this->probe($action, $kind, $variant, $slot);
+        $process->start();
+
+        return $this->completedResult($process);
+    }
+
+    private function completedResult(Process $process): array
+    {
+        $process->wait();
+        self::assertSame(0, $process->getExitCode(), $process->getErrorOutput());
+
+        $result = json_decode($process->getOutput(), true, flags: JSON_THROW_ON_ERROR);
+        if (isset($result['body']['reference'])) {
+            // MySQL JSON may reorder object keys; preserve strict scalar identity.
+            ksort($result['body']['reference']);
+        }
+
+        return $result;
+    }
+}
diff --git a/tests/Feature/RuntimePayloadCompletionUploadsTest.php b/tests/Feature/RuntimePayloadCompletionUploadsTest.php
new file mode 100644
index 00000000..ba876afd
--- /dev/null
+++ b/tests/Feature/RuntimePayloadCompletionUploadsTest.php
@@ -0,0 +1,345 @@
+<?php
+
+namespace Tests\Feature;
+
+use App\Models\RuntimeExternalPayload;
+use App\Models\RuntimePayloadCompletionBudget;
+use App\Models\WorkerRegistration;
+use App\Models\WorkflowNamespace;
+use App\Support\RuntimeExternalPayloadCleanup;
+use App\Support\RuntimeExternalPayloadRegistry;
+use App\Support\RuntimePayloadCompletionContext;
+use App\Support\WorkflowQueryTaskBroker;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\File;
+use Illuminate\Support\Facades\Queue;
+use Illuminate\Testing\TestResponse;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Tests\Feature\Concerns\ServerTestHelpers;
+use Tests\Feature\Concerns\StoragePressureFixture;
+use Tests\TestCase;
+use Workflow\Serializers\Serializer;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Models\WorkflowTask;
+
+class RuntimePayloadCompletionUploadsTest extends TestCase
+{
+    use RefreshDatabase;
+    use ServerTestHelpers;
+    use StoragePressureFixture;
+
+    private string $directory;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+        Queue::fake();
+        $this->directory = storage_path('framework/testing/completion-uploads');
+        File::deleteDirectory($this->directory);
+        foreach (['default', 'other'] as $namespace) {
+            $this->createNamespace($namespace);
+            WorkflowNamespace::query()->where('name', $namespace)->update(['external_payload_storage' => [
+                'driver' => 'local', 'enabled' => true, 'threshold_bytes' => 32,
+                'config' => ['uri' => 'file://'.$this->directory.'/'.$namespace],
+            ]]);
+        }
+        $this->registerWorker('worker', 'queue');
+        config(['server.external_payload_transport.max_payload_bytes' => 4096,
+            'server.external_payload_transport.completion_max_bytes' => null]);
+        $this->configureStoragePressure();
+    }
+
+    protected function tearDown(): void
+    {
+        RuntimePayloadCompletionBudget::flushEventListeners();
+        $this->removeStoragePressure();
+        File::deleteDirectory($this->directory);
+        parent::tearDown();
+    }
+
+    public function test_owned_activity_result_can_upload_complete_and_reconcile_after_lease_closes(): void
+    {
+        $context = $this->activity();
+        $payload = Serializer::serializeWithCodec('avro', str_repeat('result', 200));
+        $this->observeStoragePressure('draining');
+        $reference = $this->upload($payload, $context)->assertCreated()->json('reference');
+        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+        $this->postJson('/api/worker/activity-tasks/'.$context['task_id'].'/complete', [
+            'activity_attempt_id' => $context['attempt'], 'lease_owner' => 'worker',
+            'result' => ['codec' => 'avro', 'external_payload' => $reference],
+        ], $this->workerHeaders())->assertOk();
+        $before = RuntimeExternalPayload::query()->sole()->getRawOriginal();
+        $budgetBefore = RuntimePayloadCompletionBudget::query()->sole()->getRawOriginal();
+        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+        $this->assertSame($before, RuntimeExternalPayload::query()->sole()->getRawOriginal());
+        $this->assertSame($budgetBefore, RuntimePayloadCompletionBudget::query()->sole()->getRawOriginal());
+        $this->getJson('/api/activities/activity', $this->apiHeaders())->assertOk()
+            ->assertJsonPath('activity_status', 'completed')
+            ->assertJsonPath('result.external_payload.reference_id', $reference['reference_id']);
+        $this->call('GET', '/api/external-payloads/v1/'.$reference['reference_id'], [], [], [], [
+            'HTTP_X_NAMESPACE' => 'default', 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_CODEC' => 'avro',
+            'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SIZE' => (string) strlen($payload),
+            'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SHA256' => hash('sha256', $payload),
+        ])->assertOk()->assertStreamedContent($payload);
+    }
+
+    public function test_workflow_completion_upload_can_be_committed_during_draining(): void
+    {
+        $context = $this->workflow();
+        $payload = Serializer::serializeWithCodec('avro', ['result' => str_repeat('x', 100)]);
+        $this->observeStoragePressure('draining');
+        $reference = $this->upload($payload, $context)->assertCreated()->json('reference');
+        $this->postJson('/api/worker/workflow-tasks/'.$context['task_id'].'/complete', [
+            'lease_owner' => 'worker', 'workflow_task_attempt' => $context['attempt'],
+            'commands' => [['type' => 'complete_workflow', 'sequence' => 1,
+                'result' => ['codec' => 'avro', 'external_payload' => $reference]]],
+        ], $this->workerHeaders())->assertOk();
+        $this->assertNotNull(RuntimeExternalPayload::query()->sole()->retained_at);
+        $this->assertSame('completed', WorkflowRun::query()->sole()->status->value);
+        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+    }
+
+    public function test_query_completion_upload_uses_its_own_current_lease(): void
+    {
+        $this->postJson('/api/workflows', ['workflow_id' => 'query-workflow',
+            'workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'queue', 'input' => []],
+            $this->apiHeaders())->assertCreated();
+        WorkerRegistration::query()->where('worker_id', 'worker')->update(['capabilities' => ['query_tasks']]);
+        $broker = app(WorkflowQueryTaskBroker::class);
+        $broker->enqueue('default', WorkflowRun::query()->sole(), 'status', [
+            'codec' => 'avro', 'blob' => Serializer::serializeWithCodec('avro', []),
+        ]);
+        $task = $this->postJson('/api/worker/query-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue'],
+            $this->workerHeaders())->assertOk()->json('task');
+        $this->assertIsArray($task);
+        $context = ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'query',
+            'task_id' => $task['query_task_id'], 'attempt' => $task['query_task_attempt'],
+            'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['result_envelope']];
+        $this->observeStoragePressure('draining');
+        $wrong = $context;
+        $wrong['attempt']++;
+        $this->upload('bytes', $wrong)->assertStatus(409);
+        $payload = Serializer::serializeWithCodec('avro', ['status' => str_repeat('ready', 50)]);
+        $reference = $this->upload($payload, $context)->assertCreated()->json('reference');
+        $this->postJson('/api/worker/query-tasks/'.$context['task_id'].'/complete', [
+            'lease_owner' => 'worker', 'query_task_attempt' => $context['attempt'],
+            'result_envelope' => ['codec' => 'avro', 'external_payload' => $reference],
+        ], $this->workerHeaders())->assertOk()->assertJsonPath('outcome', 'completed');
+        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+    }
+
+    public function test_activity_cannot_claim_another_budget_as_a_workflow_task(): void
+    {
+        $context = $this->activity();
+        $context['kind'] = 'workflow';
+        $context['attempt'] = 1;
+        $context['slot'] = ['commands', 0, 'result'];
+        $this->observeStoragePressure('draining');
+        $this->upload('bytes', $context)->assertStatus(409);
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 0);
+    }
+
+    public function test_workflow_slots_are_bounded_even_when_the_bytes_are_deduplicated(): void
+    {
+        $context = $this->workflow();
+        $this->observeStoragePressure('draining');
+        for ($index = 0; $index < 128; $index++) {
+            $context['slot'] = ['commands', $index, 'arguments'];
+            $this->upload('same', $context)->assertCreated();
+        }
+        $context['slot'] = ['commands', 128, 'arguments'];
+        $this->upload('same', $context)->assertStatus(503)
+            ->assertJsonPath('reason', 'storage_pressure')->assertJsonPath('request_admitted', false);
+        $this->assertDatabaseCount('runtime_external_payloads', 1);
+        $this->assertCount(128, RuntimePayloadCompletionBudget::query()->sole()->slots);
+    }
+
+    public function test_renewed_worker_registration_does_not_authorize_a_wrong_workflow_attempt(): void
+    {
+        $context = $this->workflow();
+        $context['attempt']++;
+        $this->observeStoragePressure('draining');
+        $this->upload('bytes', $context)->assertStatus(409);
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 0);
+    }
+
+    #[DataProvider('wrongContexts')]
+    public function test_foreign_or_unrelated_context_rejects_without_reserving_storage(array $changes, string $namespace): void
+    {
+        $context = array_replace($this->activity(), $changes);
+        $this->observeStoragePressure('draining');
+        $this->upload('bytes', $context, $namespace)->assertStatus(409)
+            ->assertJsonPath('reason', 'external_payload_completion_lease_rejected');
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 0);
+        $this->assertDatabaseCount('runtime_external_payloads', 0);
+    }
+
+    public static function wrongContexts(): array
+    {
+        return [[['task_id' => 'missing'], 'default'], [['attempt' => 'missing'], 'default'],
+            [['lease_owner' => 'another-worker'], 'default'], [[], 'other']];
+    }
+
+    public function test_operator_credentials_cannot_claim_the_worker_completion_exception(): void
+    {
+        config(['server.auth.driver' => 'token', 'server.auth.token' => null,
+            'server.auth.role_tokens' => ['worker' => 'worker-fixture', 'operator' => 'operator-fixture']]);
+        $context = ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'activity',
+            'task_id' => 'missing', 'attempt' => 'missing', 'lease_owner' => 'worker',
+            'operation' => 'complete', 'slot' => ['result']];
+        $this->observeStoragePressure('draining');
+        $this->upload('bytes', $context, token: 'operator-fixture')->assertForbidden()
+            ->assertJsonPath('reason', 'external_payload_unauthorized');
+        $this->upload('bytes', $context, token: 'worker-fixture')->assertStatus(409)
+            ->assertJsonPath('reason', 'external_payload_completion_lease_rejected');
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 0);
+    }
+
+    public function test_registry_ownership_guards_precede_object_locks_in_each_write_transaction(): void
+    {
+        $order = [];
+        DB::listen(function ($query) use (&$order): void {
+            if (str_starts_with(strtolower($query->sql), 'select')
+                && str_contains($query->sql, 'runtime_external_payload_object_locks')) {
+                $order[] = 'object-lock';
+            }
+        });
+        app(RuntimeExternalPayloadRegistry::class)->upload('default', 'bytes', 'avro', hash('sha256', 'bytes'),
+            function () use (&$order): void {
+                $this->assertGreaterThan(0, DB::transactionLevel());
+                $order[] = 'ownership';
+            });
+        $this->assertSame(['ownership', 'object-lock', 'ownership',
+            'ownership', 'object-lock', 'ownership'], $order);
+    }
+
+    public function test_expired_lease_cannot_create_an_upload(): void
+    {
+        $context = $this->activity();
+        WorkflowTask::query()->whereKey($context['task_id'])->update(['lease_expires_at' => now()->subSecond()]);
+        $this->observeStoragePressure('draining');
+        $this->upload('bytes', $context)->assertStatus(409);
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 0);
+        $this->assertDatabaseCount('runtime_external_payloads', 0);
+    }
+
+    public function test_one_slot_cannot_rotate_content_and_complete_fail_share_the_budget(): void
+    {
+        $context = $this->activity();
+        config(['server.external_payload_transport.completion_max_bytes' => 6]);
+        $this->observeStoragePressure('draining');
+        $this->upload('first', $context)->assertCreated();
+        $this->upload('other', $context)->assertStatus(409)->assertJsonPath('reason', 'external_payload_completion_conflict');
+        $context['operation'] = 'fail';
+        $context['slot'] = ['failure', 'details'];
+        $this->upload('other', $context)->assertStatus(503)->assertJsonPath('reason', 'storage_pressure')
+            ->assertJsonPath('request_admitted', false)->assertJsonPath('retryable', true);
+        $this->upload('first', $context)->assertCreated();
+        $this->assertDatabaseCount('runtime_external_payloads', 1);
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 1);
+        $this->observeStoragePressure('normal');
+        $this->upload('other', null)->assertCreated();
+    }
+
+    #[DataProvider('closedAdmission')]
+    public function test_fenced_or_stale_state_does_not_admit_even_an_owned_upload(string $state): void
+    {
+        $context = $this->activity();
+        $this->observeStoragePressure($state === 'stale' ? 'normal' : $state);
+        if ($state === 'stale') {
+            $this->travel(120)->seconds();
+        }
+        $this->upload('bytes', $context)->assertStatus(503);
+        $this->assertDatabaseCount('runtime_external_payloads', 0);
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 0);
+    }
+
+    public static function closedAdmission(): array
+    {
+        return [['fenced'], ['stale']];
+    }
+
+    public function test_lease_is_rechecked_after_reservation_and_before_object_write(): void
+    {
+        $context = $this->activity();
+        $this->observeStoragePressure('draining');
+        RuntimePayloadCompletionBudget::created(function () use ($context): void {
+            WorkflowTask::query()->whereKey($context['task_id'])->update(['lease_expires_at' => now()->subSecond()]);
+        });
+        $this->upload('bytes', $context)->assertStatus(409);
+        $this->assertDatabaseCount('runtime_external_payloads', 0);
+    }
+
+    public function test_hard_fence_is_rechecked_before_object_write(): void
+    {
+        $context = $this->activity();
+        $this->observeStoragePressure('draining');
+        RuntimePayloadCompletionBudget::created(fn () => $this->observeStoragePressure('fenced'));
+        $this->upload('bytes', $context)->assertStatus(503);
+        $this->assertDatabaseCount('runtime_external_payloads', 0);
+    }
+
+    public function test_namespace_quota_still_applies_to_owned_uploads(): void
+    {
+        $context = $this->activity();
+        config(['server.external_payload_transport.hard_max_bytes_per_namespace' => 2]);
+        $this->observeStoragePressure('draining');
+        $this->upload('bytes', $context)->assertStatus(429)->assertJsonPath('reason', 'external_payload_namespace_bytes_exhausted');
+        $this->assertDatabaseCount('runtime_external_payloads', 0);
+    }
+
+    public function test_cleanup_does_not_reset_an_expired_but_renewable_activity_budget(): void
+    {
+        $context = $this->activity();
+        $this->observeStoragePressure('draining');
+        $this->upload('bytes', $context)->assertCreated();
+        $this->observeStoragePressure('normal');
+        RuntimePayloadCompletionBudget::query()->update(['expires_at' => now()->subHour()]);
+        WorkflowTask::query()->whereKey($context['task_id'])->update(['lease_expires_at' => now()->subMinute()]);
+        app(RuntimeExternalPayloadCleanup::class)->runPass();
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 1);
+        WorkflowTask::query()->whereKey($context['task_id'])->update(['status' => 'completed']);
+        RuntimePayloadCompletionBudget::query()->update(['expires_at' => now()->subHour()]);
+        app(RuntimeExternalPayloadCleanup::class)->runPass();
+        $this->assertDatabaseCount('runtime_payload_completion_budgets', 0);
+    }
+
+    private function activity(): array
+    {
+        $this->postJson('/api/activities', ['activity_id' => 'activity',
+            'activity_type' => 'tests.external-greeting-activity', 'task_queue' => 'queue', 'input' => []], $this->apiHeaders())->assertCreated();
+        $task = $this->postJson('/api/worker/activity-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue'],
+            $this->workerHeaders())->assertOk()->json('task');
+        $this->assertIsArray($task);
+
+        return ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'activity',
+            'task_id' => $task['task_id'], 'attempt' => $task['activity_attempt_id'],
+            'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['result']];
+    }
+
+    private function workflow(): array
+    {
+        $this->postJson('/api/workflows', ['workflow_id' => 'workflow',
+            'workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'queue', 'input' => []],
+            $this->apiHeaders())->assertCreated();
+        $task = $this->postJson('/api/worker/workflow-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue'],
+            $this->workerHeaders())->assertOk()->json('task');
+        $this->assertIsArray($task);
+
+        return ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'workflow',
+            'task_id' => $task['task_id'], 'attempt' => $task['workflow_task_attempt'],
+            'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['commands', 0, 'result']];
+    }
+
+    private function upload(string $bytes, ?array $context, string $namespace = 'default', ?string $token = null): TestResponse
+    {
+        return $this->call('POST', '/api/external-payloads/v1', [], [], [], [
+            'CONTENT_TYPE' => 'application/octet-stream', 'HTTP_X_NAMESPACE' => $namespace,
+            'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_CODEC' => 'avro', 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SIZE' => (string) strlen($bytes),
+            'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SHA256' => hash('sha256', $bytes),
+            ...($context === null ? [] : ['HTTP_X_DURABLE_WORKFLOW_PAYLOAD_COMPLETION' => json_encode($context, JSON_THROW_ON_ERROR)]),
+            ...($token === null ? [] : ['HTTP_AUTHORIZATION' => 'Bearer '.$token]),
+        ], $bytes);
+    }
+}
diff --git a/tests/Support/RuntimePayloadCompletionProcess.php b/tests/Support/RuntimePayloadCompletionProcess.php
new file mode 100644
index 00000000..393e07b9
--- /dev/null
+++ b/tests/Support/RuntimePayloadCompletionProcess.php
@@ -0,0 +1,122 @@
+<?php
+
+use App\Models\RuntimeExternalPayload;
+use App\Models\RuntimePayloadCompletionBudget;
+use App\Models\WorkerRegistration;
+use App\Models\WorkflowNamespace;
+use App\Support\RuntimeExternalPayloadRegistry;
+use App\Support\RuntimePayloadCompletionContext;
+use App\Support\StoragePressure;
+use App\Support\WorkerProtocol;
+use Illuminate\Contracts\Console\Kernel;
+use Illuminate\Http\Request;
+use Illuminate\Support\Facades\Artisan;
+use Illuminate\Support\Facades\Queue;
+use Workflow\Serializers\Serializer;
+
+require dirname(__DIR__, 2).'/vendor/autoload.php';
+$app = require dirname(__DIR__, 2).'/bootstrap/app.php';
+$app->make(Kernel::class)->bootstrap();
+Queue::fake();
+
+try {
+    [$script, $action, $directory, $kind, $variant, $slot, $barrier] = $argv;
+    config(['server.storage_admission.file' => $directory.'/pressure.json',
+        'server.storage_admission.source' => 'completion-process-test',
+        'server.storage_admission.max_age_seconds' => 300,
+        'server.external_payload_transport.max_payload_bytes' => 4096,
+        'server.external_payload_transport.completion_max_bytes' => strlen(Serializer::serializeWithCodec('avro', str_repeat('alpha', 100)))]);
+    $request = static function (string $path, array|string $body, array $headers = []): array {
+        $request = Request::create($path, 'POST', [], [], [], array_replace([
+            'CONTENT_TYPE' => is_array($body) ? 'application/json' : 'application/octet-stream',
+            'HTTP_ACCEPT' => 'application/json', 'HTTP_X_NAMESPACE' => 'default',
+            'HTTP_X_DURABLE_WORKFLOW_CONTROL_PLANE_VERSION' => '2',
+            'HTTP_'.str_replace('-', '_', strtoupper(WorkerProtocol::HEADER)) => WorkerProtocol::VERSION,
+        ], $headers), is_array($body) ? json_encode($body, JSON_THROW_ON_ERROR) : $body);
+        $kernel = app(Illuminate\Contracts\Http\Kernel::class);
+        $response = $kernel->handle($request);
+        $kernel->terminate($request, $response);
+
+        return ['status' => $response->getStatusCode(), 'body' => json_decode($response->getContent(), true, flags: JSON_THROW_ON_ERROR)];
+    };
+    if ($action === 'init') {
+        file_put_contents($directory.'/pressure.json', json_encode(['schema' => StoragePressure::SCHEMA,
+            'source' => 'completion-process-test', 'observed_at' => time(), 'state' => 'normal'], JSON_THROW_ON_ERROR));
+        Artisan::call('migrate:fresh', ['--force' => true]);
+        WorkflowNamespace::query()->updateOrCreate(['name' => 'default'], ['status' => 'active', 'retention_days' => 30,
+            'external_payload_storage' => ['driver' => 'local', 'enabled' => true, 'threshold_bytes' => 32,
+                'config' => ['uri' => 'file://'.$directory.'/objects']]]);
+        WorkerRegistration::query()->create(['worker_id' => 'worker', 'namespace' => 'default', 'task_queue' => 'queue',
+            'runtime' => 'php', 'supported_workflow_types' => ['test.workflow'], 'supported_activity_types' => ['test.activity'],
+            'last_heartbeat_at' => now(), 'status' => 'active']);
+        $plural = $kind === 'activity' ? 'activities' : 'workflows';
+        $started = $request('/api/'.$plural, [$kind.'_id' => 'test', $kind.'_type' => 'test.'.$kind,
+            'task_queue' => 'queue', 'input' => []]);
+        if ($started['status'] !== 201) {
+            throw new RuntimeException(json_encode($started, JSON_THROW_ON_ERROR));
+        }
+        $polled = $request('/api/worker/'.$kind.'-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue']);
+        $task = $polled['body']['task'] ?? null;
+        if (! is_array($task)) {
+            throw new RuntimeException(json_encode($polled, JSON_THROW_ON_ERROR));
+        }
+        $context = ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => $kind,
+            'task_id' => $task['task_id'], 'attempt' => $task[$kind === 'activity' ? 'activity_attempt_id' : 'workflow_task_attempt'],
+            'lease_owner' => 'worker', 'operation' => 'complete',
+            'slot' => $kind === 'activity' ? ['result'] : ['commands', 0, 'result']];
+        file_put_contents($directory.'/context.json', json_encode($context, JSON_THROW_ON_ERROR));
+        file_put_contents($directory.'/pressure.json', json_encode(['schema' => StoragePressure::SCHEMA,
+            'source' => 'completion-process-test', 'observed_at' => time(), 'state' => 'draining'], JSON_THROW_ON_ERROR));
+        echo '{}';
+        exit(0);
+    }
+    $context = json_decode(file_get_contents($directory.'/context.json'), true, flags: JSON_THROW_ON_ERROR);
+    if ($action === 'upload') {
+        if ($slot !== '0') {
+            if ($kind === 'activity') {
+                $context['operation'] = 'fail';
+                $context['slot'] = ['failure', 'details'];
+            } else {
+                $context['slot'] = ['commands', (int) $slot, 'result'];
+            }
+        }
+        if ($barrier !== '') {
+            touch($directory.'/'.$barrier.'.ready');
+            $deadline = microtime(true) + 10;
+            while (! is_file($directory.'/go')) {
+                if (microtime(true) > $deadline) {
+                    throw new RuntimeException('Upload barrier timed out.');
+                }
+                usleep(10000);
+            }
+        }
+        $bytes = Serializer::serializeWithCodec('avro', str_repeat($variant, 100));
+        echo json_encode($request('/api/external-payloads/v1', $bytes, [
+            'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_COMPLETION' => json_encode($context, JSON_THROW_ON_ERROR),
+            'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_CODEC' => 'avro', 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SIZE' => (string) strlen($bytes),
+            'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SHA256' => hash('sha256', $bytes),
+        ]), JSON_THROW_ON_ERROR);
+    } elseif ($action === 'complete') {
+        $row = RuntimeExternalPayload::query()->sole();
+        $reference = app(RuntimeExternalPayloadRegistry::class)->referenceForUri('default', $row->storage_uri);
+        $envelope = ['codec' => 'avro', 'external_payload' => $reference];
+        $body = ['lease_owner' => 'worker'];
+        if ($kind === 'activity') {
+            $body += ['activity_attempt_id' => $context['attempt'], 'result' => $envelope];
+        } else {
+            $body += ['workflow_task_attempt' => $context['attempt'],
+                'commands' => [['type' => 'complete_workflow', 'sequence' => 1, 'result' => $envelope]]];
+        }
+        echo json_encode($request('/api/worker/'.$kind.'-tasks/'.$context['task_id'].'/complete', $body), JSON_THROW_ON_ERROR);
+    } elseif ($action === 'status') {
+        $budget = RuntimePayloadCompletionBudget::query()->sole();
+        echo json_encode(['budgets' => RuntimePayloadCompletionBudget::query()->count(),
+            'slots' => count($budget->slots), 'objects' => count($budget->objects),
+            'bytes' => array_sum($budget->objects), 'rows' => RuntimeExternalPayload::query()->count()], JSON_THROW_ON_ERROR);
+    } else {
+        throw new RuntimeException('Unknown completion probe action.');
+    }
+} catch (Throwable $exception) {
+    fwrite(STDERR, $exception::class.': '.$exception->getMessage().PHP_EOL);
+    exit(1);
+}
diff --git a/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php b/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php
index 130e28b4..d5e8b1c7 100644
--- a/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php
+++ b/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php
@@ -49,6 +49,10 @@ public function test_runtime_external_payload_openapi_is_parseable_and_complete(
             'external_payload_namespace_bytes_exhausted',
             'external_payload_namespace_objects_exhausted',
             'external_payload_namespace_quota_unavailable',
+            'external_payload_completion_invalid',
+            'external_payload_completion_lease_rejected',
+            'external_payload_completion_conflict',
+            'external_payload_completion_budget_exhausted',
         ], $reasons);
 
         $this->assertSame(
diff --git a/tests/Unit/RuntimePayloadCompletionContextTest.php b/tests/Unit/RuntimePayloadCompletionContextTest.php
new file mode 100644
index 00000000..868eec12
--- /dev/null
+++ b/tests/Unit/RuntimePayloadCompletionContextTest.php
@@ -0,0 +1,88 @@
+<?php
+
+namespace Tests\Unit;
+
+use App\Support\RuntimeExternalPayloadException;
+use App\Support\RuntimePayloadCompletionContext;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+
+class RuntimePayloadCompletionContextTest extends TestCase
+{
+    #[DataProvider('validSlots')]
+    public function test_only_existing_worker_payload_purposes_have_a_context(string $kind, string $operation, array $slot): void
+    {
+        $value = $this->value($kind, $operation, $slot);
+        $context = RuntimePayloadCompletionContext::parse(json_encode($value, JSON_THROW_ON_ERROR));
+        $this->assertSame($kind, $context->kind);
+        $this->assertSame($operation, $context->operation);
+        $this->assertSame($slot, $context->slot);
+        $this->assertSame($context->scope('default'),
+            RuntimePayloadCompletionContext::parse(json_encode(array_reverse($value, true), JSON_THROW_ON_ERROR))->scope('default'));
+        $this->assertNotSame($context->scope('default'), $context->scope('other'));
+    }
+
+    public static function validSlots(): array
+    {
+        return [
+            ['activity', 'complete', ['result']],
+            ['activity', 'fail', ['failure', 'details']],
+            ['query', 'complete', ['result_envelope']],
+            ['workflow', 'complete', ['commands', 0, 'arguments']],
+            ['workflow', 'complete', ['commands', 1, 'entries']],
+            ['workflow', 'complete', ['commands', 2, 'request_payload']],
+            ['workflow', 'complete', ['commands', 3, 'result']],
+            ['workflow', 'complete', ['commands', 4, 'exception', 'details']],
+            ['workflow', 'complete', ['commands', 5, 'workflow_stream', 'items', 0, 'payload']],
+        ];
+    }
+
+    public function test_all_slots_and_outcomes_for_the_same_lease_share_one_budget(): void
+    {
+        $complete = RuntimePayloadCompletionContext::parse(json_encode($this->value(), JSON_THROW_ON_ERROR));
+        $fail = RuntimePayloadCompletionContext::parse(json_encode($this->value('activity', 'fail', ['failure', 'details']), JSON_THROW_ON_ERROR));
+        $this->assertSame($complete->scope('default'), $fail->scope('default'));
+        $this->assertNotSame($complete->slotIdentity(), $fail->slotIdentity());
+        foreach (['task_id', 'attempt', 'lease_owner'] as $key) {
+            $other = $this->value();
+            $other[$key] .= '-other';
+            $this->assertNotSame($complete->scope('default'),
+                RuntimePayloadCompletionContext::parse(json_encode($other, JSON_THROW_ON_ERROR))->scope('default'));
+        }
+    }
+
+    #[DataProvider('invalidContexts')]
+    public function test_malformed_or_unrelated_context_cannot_authorize_writes(array $changes): void
+    {
+        $this->expectException(RuntimeExternalPayloadException::class);
+        RuntimePayloadCompletionContext::parse(json_encode(array_replace($this->value(), $changes), JSON_THROW_ON_ERROR));
+    }
+
+    public static function invalidContexts(): array
+    {
+        return array_map(static fn (array $changes): array => [$changes], [
+            ['schema' => 'unknown'], ['extra' => true], ['kind' => 'client'], ['operation' => 'start'],
+            ['operation' => []], ['operation' => null], ['slot' => null], ['task_id' => []],
+            ['task_id' => ''], ['task_id' => str_repeat('a', 256)], ['lease_owner' => "worker\n"],
+            ['attempt' => 1], ['slot' => ['input']], ['slot' => ['failure', 'details']],
+            ['kind' => 'workflow', 'attempt' => '1', 'slot' => ['commands', 0, 'result']],
+            ['kind' => 'workflow', 'attempt' => 0, 'slot' => ['commands', 0, 'result']],
+            ['kind' => 'workflow', 'attempt' => 1, 'slot' => ['commands', '0', 'result']],
+            ['kind' => 'workflow', 'attempt' => 1, 'slot' => ['commands', -1, 'result']],
+            ['kind' => 'query', 'attempt' => 1, 'slot' => ['result']],
+        ]);
+    }
+
+    public function test_oversized_header_is_rejected_before_json_decode(): void
+    {
+        $this->expectException(RuntimeExternalPayloadException::class);
+        RuntimePayloadCompletionContext::parse(str_repeat(' ', 4097));
+    }
+
+    private function value(string $kind = 'activity', string $operation = 'complete', array $slot = ['result']): array
+    {
+        return ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => $kind,
+            'task_id' => 'task', 'attempt' => $kind === 'activity' ? 'attempt' : 1,
+            'lease_owner' => 'worker', 'operation' => $operation, 'slot' => $slot];
+    }
+}