Qualify published SDK external completions during draining

tests/Integration/StorageAdmission/README.md

@@ -89,8 +89,10 @@ new capacity measurement or a production reserve qualification.
 Each activity computes a value larger than the 64-KiB external-payload threshold
 and records its execution before waiting at a test barrier. The test fences
 Server, releases all three handlers and verifies that no acknowledgement changes
-the original attempts. During draining, PHP/Rust inline completions may finish;
-Python's separate payload upload remains paused because uploads are producers.
+the original attempts. During draining, every SDK must externalize its result
+under the current activity lease and complete the original attempt. Three
+independent completion budgets prove the external path was exercised. New
+workflow starts and unbound uploads remain refused.
 After normal admission resumes, every workflow must consume its original result,
 verify the bytes and complete without re-executing its activity. A fresh-process
 check after Server/MySQL restart verifies the same results and identities.

tests/Integration/StorageAdmission/sdk/experiment.php

@@ -1,6 +1,7 @@
 <?php
 
 declare(strict_types=1);
+use Workflow\Serializers\Serializer;
 
 const STORAGE_ADMISSION_HELPERS_ONLY = true;
 require '/experiment.php';
@@ -87,18 +88,39 @@ function assertResults(array $runs): void
         check(attempts($db) === $before, 'Fenced completion changed an activity attempt.');
         echo "Fenced attempts unchanged; permitting bounded acknowledgements to drain.\n";
 
-        // Inline acknowledgements may drain; new standalone uploads must wait.
+        // Bound completion uploads may drain; unbound producers must still wait.
         observe('draining');
+        call('POST', '/workflows', startBody('draining-start'), false, 503);
+        $payload = Serializer::serializeWithCodec('avro', 'unbound');
+        $upload = curl_init('http://server:8080/api/external-payloads/v1');
+        curl_setopt_array($upload, [CURLOPT_POST => true, CURLOPT_RETURNTRANSFER => true,
+            CURLOPT_CONNECTTIMEOUT => 5, CURLOPT_TIMEOUT => 15, CURLOPT_POSTFIELDS => $payload,
+            CURLOPT_HTTPHEADER => ['Content-Type: application/octet-stream', 'Authorization: Bearer storage-fixture',
+                'X-Namespace: default', 'X-Durable-Workflow-Payload-Codec: avro',
+                'X-Durable-Workflow-Payload-Size: '.strlen($payload),
+                'X-Durable-Workflow-Payload-SHA256: '.hash('sha256', $payload)]]);
+        $response = curl_exec($upload);
+        check(curl_errno($upload) === 0, 'Unbound upload transport failed.');
+        check(curl_getinfo($upload, CURLINFO_RESPONSE_CODE) === 503, 'Unbound upload was not refused while draining.');
+        curl_close($upload);
+        check((json_decode($response, true, flags: JSON_THROW_ON_ERROR)['request_admitted'] ?? null) === false,
+            'Unbound upload was admitted while draining.');
         waitFor(static function () use ($db, $runs): bool {
             $statuses = array_column(attempts($db), 'status', 'workflow_run_id');
+            foreach ($runs as $run) {
+                if (($statuses[$run] ?? null) !== 'completed') {
+                    return false;
+                }
+            }
 
-            return ($statuses[$runs['php']] ?? null) === 'completed'
-                && ($statuses[$runs['rust']] ?? null) === 'completed';
-        }, 'Inline SDK completions did not drain.', 30, 'draining');
+            return true;
+        }, 'Lease-bound SDK completions did not drain.', 30, 'draining');
         hold('draining', 10);
         $draining = attempts($db);
-        $statuses = array_column($draining, 'status', 'workflow_run_id');
-        check(($statuses[$runs['python']] ?? null) === 'running', 'Python upload should still be paused while draining.');
+        $budgets = $db->query('SELECT context FROM runtime_payload_completion_budgets')->fetchAll(PDO::FETCH_COLUMN);
+        $contexts = array_map(static fn (string $value): array => json_decode($value, true, flags: JSON_THROW_ON_ERROR), $budgets);
+        $activityBudgets = array_filter($contexts, static fn (array $context): bool => $context['kind'] === 'activity');
+        check(count($activityBudgets) === 3, 'Every SDK must use its own bounded external-completion upload.');
         foreach (LANGUAGES as $language) {
             check(file_get_contents('/observation/'.$language.'.effects') === "executed\n", 'Handler was repeated during pressure.');
         }
@@ -125,7 +147,7 @@ function assertResults(array $runs): void
             'activity_tasks' => $db->query("SELECT id, workflow_run_id FROM workflow_tasks WHERE task_type = 'activity'")->fetchAll(PDO::FETCH_ASSOC)];
         file_put_contents('/observation/sdk-evidence.json', json_encode($evidence, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT));
         echo json_encode(['sdk_recovery' => true, 'handlers_executed' => 3, 'payload_bytes_each' => 262144,
-            'separate_python_upload_paused_until_normal' => true], JSON_THROW_ON_ERROR).PHP_EOL;
+            'lease_bound_external_completions_drained' => 3], JSON_THROW_ON_ERROR).PHP_EOL;
     } elseif ($phase === 'recover') {
         $before = json_decode(file_get_contents('/observation/sdk-evidence.json'), true, 512, JSON_THROW_ON_ERROR);
         observe('normal');

tests/Integration/StorageAdmission/sdk/versions.env

@@ -1,3 +1,3 @@
-PHP_SDK_VERSION=2.0.7
-PYTHON_SDK_VERSION=2.0.3
-RUST_SDK_VERSION=2.0.3
+PHP_SDK_VERSION=2.0.9
+PYTHON_SDK_VERSION=2.0.4
+RUST_SDK_VERSION=2.0.5