Streamed response support

app/controllers/general.php

@@ -560,6 +560,29 @@ function router(Http $utopia, Database $dbForPlatform, callable $getProjectDB, S
                 'site' => '',
             };
 
+            $streamingDetected = false;
+            $streamCallback = function (?string $streamData, ?array $streamHeaders) use ($response, &$streamingDetected, $execution, $deployment): void {
+                if ($streamHeaders !== null) {
+                    // Headers signal — fired once before body, only when SSE detected
+                    $streamingDetected = true;
+                    $statusCode = \intval($streamHeaders['x-open-runtimes-status-code'] ?? 200);
+                    $response->setStatusCode($statusCode);
+                    foreach ($streamHeaders as $key => $value) {
+                        if (\in_array(\strtolower($key), FUNCTION_ALLOWLIST_HEADERS_RESPONSE)) {
+                            $response->addHeader($key, \is_array($value) ? \implode(', ', $value) : $value);
+                        }
+                    }
+                    if ($deployment->getAttribute('resourceType') === 'functions') {
+                        $response->addHeader('x-appwrite-execution-id', $execution->getId());
+                    } elseif ($deployment->getAttribute('resourceType') === 'sites') {
+                        $response->addHeader('x-appwrite-log-id', $execution->getId());
+                    }
+                }
+                if ($streamData !== null) {
+                    $response->chunk($streamData);
+                }
+            };
+
             $executionResponse = $executor->createExecution(
                 projectId: $project->getId(),
                 deploymentId: $deployment->getId(),
@@ -578,9 +601,26 @@ function router(Http $utopia, Database $dbForPlatform, callable $getProjectDB, S
                 memory: $spec['memory'] ?? APP_COMPUTE_MEMORY_DEFAULT,
                 logging: $resource->getAttribute('logging', true),
                 requestTimeout: 30,
-                responseFormat: Executor::RESPONSE_FORMAT_ARRAY_HEADERS
+                responseFormat: Executor::RESPONSE_FORMAT_ARRAY_HEADERS,
+                streamCallback: $streamCallback,
             );
 
+            // If SSE streaming was detected, body was already forwarded chunk by chunk.
+            // Close the response and populate execution metadata from what we have.
+            // Note: logs/errors are not available for streaming responses (the executor
+            // streams the body and does not send metadata back over the same channel).
+            if ($streamingDetected) {
+                $response->chunk('', true);
+
+                $execution->setAttribute('status', 'completed');
+                $execution->setAttribute('logs', '');
+                $execution->setAttribute('errors', '');
+                $execution->setAttribute('responseStatusCode', 200);
+                $execution->setAttribute('responseHeaders', []);
+                $execution->setAttribute('duration', \microtime(true) - $durationStart);
+                return true;
+            }
+
             $headerOverrides = [];
 
             // Branded 404 override

src/Executor/Executor.php

@@ -129,7 +129,15 @@ class Executor
             'timeout' => $timeout
         ];
 
-        $this->call($this->endpoint, self::METHOD_GET, $route, [ 'x-opr-runtime-id' => $runtimeId ], $params, true, $timeout, $callback);
+        // Wrap callback to match two-arg signature (?string $data, ?array $headers)
+        // getLogs only cares about data chunks, not headers signal
+        $wrappedCallback = function (?string $data, ?array $headers) use ($callback): void {
+            if ($data !== null) {
+                $callback($data);
+            }
+        };
+
+        $this->call($this->endpoint, self::METHOD_GET, $route, [ 'x-opr-runtime-id' => $runtimeId ], $params, true, $timeout, $wrappedCallback);
     }
 
     /**
@@ -204,7 +212,8 @@ class Executor
         bool $logging,
         string $runtimeEntrypoint = '',
         ?int $requestTimeout = null,
-        string $responseFormat = self::RESPONSE_FORMAT_OBJECT_HEADERS
+        string $responseFormat = self::RESPONSE_FORMAT_OBJECT_HEADERS,
+        ?callable $streamCallback = null,
     ) {
         $runtimeId = "$projectId-$deploymentId";
         $route = '/runtimes/' . $runtimeId . '/executions';
@@ -242,6 +251,13 @@ class Executor
             $requestTimeout = $timeout + 15;
         }
 
+        // Streaming path: tell executor to stream and forward chunks via callback
+        if ($streamCallback !== null) {
+            $params['stream'] = 'true';
+            $this->call($this->endpoint, self::METHOD_POST, $route, [ 'x-opr-runtime-id' => $runtimeId, 'content-type' => 'multipart/form-data', 'x-executor-response-format' => $responseFormat ], $params, false, $requestTimeout, $streamCallback);
+            return [];
+        }
+
         $response = $this->call($this->endpoint, self::METHOD_POST, $route, [ 'x-opr-runtime-id' => $runtimeId, 'content-type' => 'multipart/form-data', 'accept' => 'multipart/form-data', 'x-executor-response-format' => $responseFormat ], $params, true, $requestTimeout);
 
         $status = $response['headers']['status-code'];
@@ -337,8 +353,14 @@ class Executor
         if (isset($callback)) {
             $headers[] = 'accept: text/event-stream';
 
-            $handleEvent = function ($ch, $data) use ($callback) {
-                $callback($data);
+            $callbackHeadersFired = false;
+            $handleEvent = function ($ch, $data) use ($callback, &$callbackHeadersFired, &$responseHeaders) {
+                if (!$callbackHeadersFired) {
+                    // Fire headers signal once before the first data chunk
+                    $callback(null, $responseHeaders);
+                    $callbackHeadersFired = true;
+                }
+                $callback($data, null);
                 return \strlen($data);
             };