Plan activity-continuation payloads within existing slot authority

CHANGELOG.md

@@ -8,6 +8,10 @@ project follows [Semantic Versioning](https://semver.org/).
 
 ### Fixed
 
+- Activity continuations honor the completion aggregate and command per-value
+  limits. Commands stay inline; only the activity result may be offloaded.
+  An aggregate that cannot fit through an authorized rewrite reaches the
+  Server unchanged once, without partial uploads.
 - Workflow completion planning honors the Server's advertised
   `limits.max_workflow_task_completion_bytes` without changing ordinary request
   or per-value limits (.github#228, server#1136). When external storage is

src/Transport/RuntimePayloadUploads.php

@@ -57,35 +57,92 @@ public function request(array $body, string $method, string $path, bool $worker,
             $body = $this->failureMetadata($body, $payloads, $path, $failureReports);
             $payloads = $this->payloads($body, $slots);
         }
-        if ($payloads === []) {
+        $commandCompletion = $worker && preg_match('~\A/worker/(?:workflow-tasks/[^/]+/complete|activity-tasks/[^/]+/continuation/complete)\z~', $path) === 1;
+        if ($payloads === [] && !$commandCompletion) {
             return $body;
         }
 
         $policy = $this->requestPolicy($body, $path, $worker, $headers, $budget);
-        if ($worker && preg_match('~\A/worker/workflow-tasks/[^/]+/complete\z~', $path) === 1
-            && array_key_exists('completion_request_bytes', $policy)) {
-            $limit = $policy['completion_request_bytes'];
-            if (!is_int($limit) || $limit < 1) {
-                throw new ExternalPayloadException('Invalid workflow-task completion limit in runtime discovery.', 422, 'external_payload_unsupported');
-            }
-            $inlineLimit = $policy['request_bytes'];
-            $policy['request_bytes'] = $limit;
-            if ($requestBytesLimit === null && in_array($policy['status'], ['unconfigured', 'disabled', 'unsupported'], true)
-                && array_filter($payloads, static fn (array $payload): bool => strlen($payload['blob']) > $inlineLimit) === []) {
-                // Without configured storage, individually valid values can still fit the
-                // completion's larger aggregate limit. Let the Server return its typed
-                // structural refusal if the actual serialized batch exceeds that limit.
-                return $body;
-            }
+        $inlineLimit = $policy['request_bytes'];
+        $policy = $this->completionPolicy($policy, $commandCompletion);
+        if ($this->maySubmitInlineCompletion($policy, $payloads, $inlineLimit, $requestBytesLimit, $commandCompletion)) {
+            return $body;
         }
         if ($requestBytesLimit !== null) {
             $policy['request_bytes'] = min($policy['request_bytes'], $requestBytesLimit);
         }
+        if ($commandCompletion && str_starts_with($path, '/worker/activity-tasks/')) {
+            $payloads = $this->continuationUploads($body, $payloads, $policy, $inlineLimit, $requestBytesLimit);
+            if ($payloads === null) {
+                return $body;
+            }
+        }
         $selected = $this->plan($body, $payloads, $policy);
         return $this->uploadSelected($body, $payloads, $selected, $policy,
             ['headers' => $headers, 'path' => $path, 'worker' => $worker, 'budget' => $budget]);
     }
 
+    /** @param UploadPolicy $policy
+     * @return UploadPolicy
+     */
+    private function completionPolicy(array $policy, bool $completion): array
+    {
+        if (!$completion || !array_key_exists('completion_request_bytes', $policy)) {
+            return $policy;
+        }
+        $limit = $policy['completion_request_bytes'];
+        if (!is_int($limit) || $limit < 1) {
+            throw new ExternalPayloadException('Invalid workflow-task completion limit in runtime discovery.', 422, 'external_payload_unsupported');
+        }
+        $policy['request_bytes'] = $limit;
+
+        return $policy;
+    }
+
+    /** @param UploadPolicy $policy
+     * @param list<PayloadSlot> $payloads
+     */
+    private function maySubmitInlineCompletion(array $policy, array $payloads, int $inlineLimit, ?int $reportLimit, bool $completion): bool
+    {
+        // Individually valid values need no unavailable external storage. The
+        // Server settles an oversized aggregate through its typed refusal.
+        return $completion && $reportLimit === null && array_key_exists('completion_request_bytes', $policy)
+            && in_array($policy['status'], ['unconfigured', 'disabled', 'unsupported'], true)
+            && array_filter($payloads, static fn (array $payload): bool => strlen($payload['blob']) > $inlineLimit) === [];
+    }
+
+    /** @param array<string, mixed> $body
+     * @param list<PayloadSlot> $payloads
+     * @param UploadPolicy $policy
+     * @return list<PayloadSlot>|null Null means no authorized rewrite fits; submit unchanged once.
+     */
+    private function continuationUploads(array $body, array $payloads, array $policy, int $inlineLimit, ?int $reportLimit): ?array
+    {
+        $results = [];
+        $smallest = $body;
+        foreach ($payloads as $payload) {
+            // Commands stay inline even when their bytes equal the result.
+            if ($payload['path'] !== ['result']) {
+                if (strlen($payload['blob']) > $inlineLimit) {
+                    throw new ExternalPayloadException('Continuation command payload exceeds the inline value limit.', 413, 'external_payload_oversized');
+                }
+                continue;
+            }
+            if (strlen($payload['blob']) > $policy['max_bytes']) {
+                throw new ExternalPayloadException('Encoded payload exceeds the runtime upload limit.', 413, 'external_payload_oversized');
+            }
+            $results[] = $payload;
+            $this->placeholder($smallest, $payload);
+        }
+        if ($reportLimit === null && array_key_exists('completion_request_bytes', $policy)
+            && $this->jsonSize($smallest) > $policy['request_bytes']) {
+            // Avoid a futile partial result upload before structural settlement.
+            return null;
+        }
+
+        return $results;
+    }
+
     /** @param array<string, mixed> $body
      * @param array<string, string> $headers
      * @return UploadPolicy
@@ -371,19 +428,25 @@ private function paths(array $body, string $path, bool $worker): array
     {
         $paths = [];
         if ($worker) {
-            if (preg_match('~\A/worker/workflow-tasks/[^/]+/(?:complete|local-activities/checkpoint(?:-group)?|cancellation-scopes/checkpoint)\z~', $path)) {
-                foreach ($body['commands'] ?? [] as $index => $command) {
+            $continuation = preg_match('~\A/worker/activity-tasks/[^/]+/continuation/complete\z~', $path) === 1;
+            if ($continuation || preg_match('~\A/worker/workflow-tasks/[^/]+/(?:complete|local-activities/checkpoint(?:-group)?|cancellation-scopes/checkpoint)\z~', $path)) {
+                $prefix = $continuation ? ['continuation', 'commands'] : ['commands'];
+                $commands = $continuation ? ($body['continuation']['commands'] ?? []) : ($body['commands'] ?? []);
+                if ($continuation) {
+                    $paths[] = ['result'];
+                }
+                foreach ($commands as $index => $command) {
                     if (!is_array($command)) {
                         continue;
                     }
                     foreach (self::COMMAND_FIELDS[$command['type'] ?? ''] ?? [] as $field) {
-                        $paths[] = ['commands', $index, $field];
+                        $paths[] = [...$prefix, $index, $field];
                     }
                     if (($command['type'] ?? null) === 'fail_workflow') {
-                        $paths[] = ['commands', $index, 'exception', 'details'];
+                        $paths[] = [...$prefix, $index, 'exception', 'details'];
                     }
                     foreach ($command['workflow_stream']['items'] ?? [] as $item => $_) {
-                        $paths[] = ['commands', $index, 'workflow_stream', 'items', $item, 'payload'];
+                        $paths[] = [...$prefix, $index, 'workflow_stream', 'items', $item, 'payload'];
                     }
                 }
             } elseif (preg_match('~\A/worker/workflow-tasks/[^/]+/local-activities/(?:prepare|recover)\z~', $path)) {
@@ -459,7 +522,15 @@ private function completionContext(array $body, string $path, array $slot, bool
 
             return strlen($context) <= 4096 ? $context : null;
         }
-        if (!preg_match('~\A/worker/(activity|workflow|query)-tasks/([^/]+)/(complete|fail)\z~', explode('?', $path)[0], $match)) {
+        if (preg_match('~\A/worker/activity-tasks/([^/]+)/continuation/complete\z~', $path, $match)) {
+            // V1 continuation uploads inherit only the activity result's lease.
+            // Proposed workflow commands have no completion upload authority.
+            if ($slot !== ['result']) {
+                return null;
+            }
+            $path = '/worker/activity-tasks/'.$match[1].'/complete';
+        }
+        if (!preg_match('~\A/worker/(activity|workflow|query)-tasks/([^/]+)/(complete|fail)\z~', $path, $match)) {
             return null;
         }
         $kind = $match[1];