feat: streaming Flight (NDJSON out-of-order payload)

.gitignore

@@ -16,3 +16,6 @@ docs/.docusaurus/
 
 # PHPUnit
 .phpunit.result.cache
+
+# Agent worktrees
+.claude/worktrees/

docs/docs/flight-navigation.md

@@ -73,7 +73,8 @@ async function navigate(url: string, push: boolean) {
 
 Every `data-flight-link` is a real `<a href>`. With JavaScript off, the links are ordinary navigation and each route renders as a full server page. With JavaScript on, the same clicks become instant Flight swaps. Same URLs, same server code — the enhancement is entirely additive.
 
+This page uses the non-streaming endpoint (`Flight::respond()`), which resolves the whole view before sending. To send it progressively — fallbacks first, boundaries filled in as they resolve — see [Streaming Flight](./streaming-flight).
+
 ## What's intentionally not here (yet)
 
-- **Streaming Flight.** The payload is resolved fully before sending. Streaming a Flight response (like the HTML [Suspense streaming](./suspense-streaming)) is future work.
 - **Root unmounting.** Navigating away replaces the DOM; React roots from the old view are dropped rather than formally unmounted. Fine for a demo, worth tightening for production.

docs/docs/streaming-flight.md

@@ -0,0 +1,60 @@
+---
+sidebar_position: 12
+---
+
+# Streaming Flight
+
+[Flight navigation](./flight-navigation) resolves the whole tree before sending it. **Streaming Flight** sends it progressively, the same way [Suspense streaming](./suspense-streaming) works for HTML — but over the data protocol, so client-driven navigation shows fallbacks immediately and fills them in as data resolves.
+
+> **React devs:** this is the RSC stream's out-of-order rows, over plain NDJSON instead of React's binary Flight wire.
+>
+> **PHP devs:** it's the same Fiber scheduler as HTML Suspense streaming, but each chunk is a piece of the tree as data, not rendered HTML.
+
+## The protocol: NDJSON
+
+`Flight::stream()` emits newline-delimited JSON:
+
+1. The **shell row** — the serialized tree, with each pending `Suspense` boundary replaced by a placeholder `["$","div",{"id":"F:<n>"},[<fallback>]]`.
+2. A **boundary row** for each boundary as it resolves, out of order: `{"b": <n>, "tree": <resolved tree>}`.
+
+```php
+use Attitude\PHPX\Server\Flight;
+
+if (Flight::wantsStream()) {           // ?__flight_stream=1 | X-Flight-Stream: 1 | Accept: application/x-ndjson
+    Flight::stream($viewContent, $components);
+}
+```
+
+A two-boundary page where the second declared boundary is faster streams like this:
+
+```
+["$","main",null,[["$","div",{"id":"F:0"},[["$","em",null,["load-a"]]]],["$","div",{"id":"F:1"},[["$","em",null,["load-b"]]]]]]
+{"b":1,"tree":[["$","p",{"id":"done-b"},["B"]]]}
+{"b":0,"tree":[["$","p",{"id":"done-a"},["A"]]]}
+```
+
+Boundary `1` arrives before `0` — out of order, as soon as its data is ready. Nested boundaries work too: resolving one can reveal more, which stream in turn.
+
+## The client reader
+
+`streamNavigate(url)` (in `examples/todo/src/client/flight.ts`) reads the stream row by row:
+
+```ts
+const reader = res.body.getReader()
+// first row -> build the view with fallbacks in place
+root.replaceChildren(toNode(shell)); mountIslands(root)
+// each later row -> patch its boundary as it arrives
+slot = document.getElementById('F:' + row.b)
+slot.replaceChildren(toNode(row.tree)); mountIslands(slot)
+```
+
+It reuses the same `toNode` DOM builder and `mountIslands` from Flight navigation, so islands inside a streamed boundary boot as soon as that boundary lands.
+
+## Relationship to the other streaming
+
+| | Transport | Chunks are |
+| --- | --- | --- |
+| [Suspense streaming](./suspense-streaming) | HTML + `<template>` swap | rendered HTML |
+| **Streaming Flight** | NDJSON rows | serialized tuple trees |
+
+Both are driven by the same Fiber scheduler; they differ only in what they put on the wire and who assembles the DOM (the browser vs. the client runtime).

examples/todo/src/client/flight.ts

@@ -92,6 +92,60 @@ async function navigate(url: string, push: boolean): Promise<void> {
   }
 }
 
+/**
+ * Streaming Flight navigation: read the NDJSON payload row by row. The first
+ * row is the shell (with fallback placeholders); each later row patches a
+ * `#F:<n>` boundary as it resolves — out of order, as they arrive.
+ */
+export async function streamNavigate(url: string, push = true): Promise<void> {
+  const root = document.getElementById('view-root')
+  if (!root) {
+    location.href = url
+    return
+  }
+
+  try {
+    const res = await fetch(url, { headers: { 'X-Flight-Stream': '1' } })
+    if (!res.body) throw new Error('no stream')
+
+    const reader = res.body.getReader()
+    const decoder = new TextDecoder()
+    let buffer = ''
+    let first = true
+
+    for (;;) {
+      const { done, value } = await reader.read()
+      if (done) break
+      buffer += decoder.decode(value, { stream: true })
+
+      let nl: number
+      while ((nl = buffer.indexOf('\n')) >= 0) {
+        const line = buffer.slice(0, nl).trim()
+        buffer = buffer.slice(nl + 1)
+        if (!line) continue
+
+        const msg = JSON.parse(line) as unknown
+        if (first) {
+          root.replaceChildren(toNode(msg))
+          mountIslands(root)
+          setActive(new URL(url).pathname)
+          if (push) history.pushState({ flight: true }, '', url)
+          first = false
+        } else {
+          const row = msg as { b: number; tree: unknown }
+          const slot = document.getElementById('F:' + row.b)
+          if (slot) {
+            slot.replaceChildren(toNode(row.tree))
+            mountIslands(slot)
+          }
+        }
+      }
+    }
+  } catch {
+    location.href = url
+  }
+}
+
 export function initFlight(): void {
   document.addEventListener('click', (event) => {
     if (event.defaultPrevented || event.button !== 0 || event.metaKey || event.ctrlKey || event.shiftKey || event.altKey) {

src/Flight.php

@@ -33,6 +33,14 @@ public static function wants(): bool
             || str_contains($_SERVER['HTTP_ACCEPT'] ?? '', 'application/x-component');
     }
 
+    /** Does the request want a *streaming* (NDJSON) Flight payload? */
+    public static function wantsStream(): bool
+    {
+        return isset($_GET['__flight_stream'])
+            || ($_SERVER['HTTP_X_FLIGHT_STREAM'] ?? '') === '1'
+            || str_contains($_SERVER['HTTP_ACCEPT'] ?? '', 'application/x-ndjson');
+    }
+
     /** Serialize a tree, emit it as JSON with the right header, and stop. */
     public static function respond(mixed $node, array $components = []): void
     {
@@ -41,6 +49,18 @@ public static function respond(mixed $node, array $components = []): void
         exit;
     }
 
+    /**
+     * Stream a tree as NDJSON Flight rows (shell first, then boundaries out of
+     * order as they resolve), then stop. See {@see FlightStream}.
+     */
+    public static function stream(mixed $node, array $components = []): void
+    {
+        header('Content-Type: application/x-ndjson; charset=utf-8');
+        header('X-Accel-Buffering: no');
+        (new FlightStream())->stream($node, $components);
+        exit;
+    }
+
     /** Resolve server components and return a JSON-serializable tuple tree. */
     public static function serialize(mixed $node, array $components = []): mixed
     {

src/FlightStream.php

@@ -0,0 +1,128 @@
+<?php declare(strict_types=1);
+
+namespace Attitude\PHPX\Server;
+
+use Fiber;
+
+/**
+ * Streaming Flight: the serialized-payload counterpart to {@see StreamingRenderer}.
+ *
+ * Where {@see Flight::serialize()} resolves the whole tree before returning JSON,
+ * this streams it as newline-delimited JSON (NDJSON):
+ *
+ *   1. the shell tree, with each pending Suspense boundary replaced by a
+ *      placeholder `['$','div',{'id':'F:<n>'},[<fallback>]]`;
+ *   2. then, out of order as each boundary resolves, a row
+ *      `{"b": <n>, "tree": <resolved tree>}`.
+ *
+ * The client reads the first line to build the view (fallbacks in place), then
+ * patches each `#F:<n>` as its row arrives — the Flight equivalent of React's
+ * out-of-order streaming, over a plain-text protocol. Suspend/resume is a Fiber,
+ * exactly as in {@see StreamingRenderer}; only the output differs (data, not HTML).
+ */
+final class FlightStream
+{
+    private int $seq = 0;
+
+    /** @var array<int, array{fiber: Fiber, delay: float, work: callable}> */
+    private array $pending = [];
+
+    /** @var array<string, callable> */
+    private array $components = [];
+
+    /** Stream $root as NDJSON Flight rows. */
+    public function stream(mixed $root, array $components = []): void
+    {
+        $this->components = $components;
+
+        while (ob_get_level() > 0) {
+            ob_end_flush();
+        }
+
+        // Shell row: server components resolved, pending boundaries as placeholders.
+        echo $this->line(Flight::serialize($root, $this->handlers()));
+        $this->flush();
+
+        // Boundary rows, soonest-ready first; nested boundaries appended here too.
+        while ($this->pending !== []) {
+            uasort($this->pending, static fn (array $a, array $b): int => $a['delay'] <=> $b['delay']);
+            $id = array_key_first($this->pending);
+            $job = $this->pending[$id];
+            unset($this->pending[$id]);
+
+            $this->sleep($job['delay']);
+            foreach ($this->pending as &$other) {
+                $other['delay'] = max(0.0, $other['delay'] - $job['delay']);
+            }
+            unset($other);
+
+            $tree = $this->drive($job['fiber'], ($job['work'])());
+            echo $this->line(['b' => $id, 'tree' => $tree]);
+            $this->flush();
+        }
+    }
+
+    /**
+     * Suspense handler used during serialization: run children in a fiber and
+     * serialize them to a tuple tree. Inline if synchronous; otherwise register
+     * the pending fiber and emit a placeholder keyed by boundary id.
+     *
+     * @internal
+     */
+    public function boundary(array $props): mixed
+    {
+        $id = $this->seq++;
+        $children = $props['children'] ?? [];
+
+        $fiber = new Fiber(fn (): mixed => Flight::serialize($children, $this->handlers()));
+        /** @var array{delay: float, work: callable} $signal */
+        $signal = $fiber->start();
+
+        if ($fiber->isTerminated()) {
+            return $fiber->getReturn();
+        }
+
+        $this->pending[$id] = ['fiber' => $fiber, 'delay' => $signal['delay'], 'work' => $signal['work']];
+
+        return ['$', 'div', ['id' => "F:{$id}"], [Flight::serialize($props['fallback'] ?? '')]];
+    }
+
+    /** @return array<string, callable> */
+    private function handlers(): array
+    {
+        return ['Suspense' => [$this, 'boundary']] + $this->components;
+    }
+
+    private function drive(Fiber $fiber, mixed $value): mixed
+    {
+        /** @var array{delay: float, work: callable} $signal */
+        $signal = $fiber->resume($value);
+        while (!$fiber->isTerminated()) {
+            $this->sleep($signal['delay']);
+            /** @var array{delay: float, work: callable} $signal */
+            $signal = $fiber->resume(($signal['work'])());
+        }
+
+        return $fiber->getReturn();
+    }
+
+    private function line(mixed $data): string
+    {
+        return json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE) . "\n";
+    }
+
+    private function sleep(float $seconds): void
+    {
+        if ($seconds > 0) {
+            usleep((int) round($seconds * 1_000_000));
+        }
+    }
+
+    private function flush(): void
+    {
+        if (ob_get_level() > 0) {
+            @ob_flush();
+        }
+        @flush();
+    }
+}

tests/FlightStreamTest.php

@@ -0,0 +1,58 @@
+<?php declare(strict_types=1);
+
+use PHPUnit\Framework\TestCase;
+
+require_once __DIR__ . '/Support/StreamHarness.php';
+
+final class FlightStreamTest extends TestCase
+{
+    /** Run a Flight::stream() page in a subprocess and capture the NDJSON. */
+    private function runStream(string $body): string
+    {
+        return StreamHarness::run("use Attitude\\PHPX\\Server\\Flight;\n" . $body);
+    }
+
+    public function testShellRowContainsPlaceholdersAndBoundariesResolveOutOfOrder(): void
+    {
+        $output = $this->runStream(<<<'PHP'
+        $slow = fn(string $label, float $d) => (function () use ($label, $d) {
+            $v = await($d, fn() => strtoupper($label));
+            return ['$', 'p', ['id' => "done-{$label}"], [$v]];
+        });
+        Flight::stream(['$', 'main', null, [
+            Suspense(['$', 'em', null, ['load-a']], ['$', $slow('a', 0.06)]),
+            Suspense(['$', 'em', null, ['load-b']], ['$', $slow('b', 0.02)]),
+        ]]);
+        PHP);
+
+        $lines = array_values(array_filter(explode("\n", trim($output))));
+        $this->assertCount(3, $lines, "expected shell + 2 boundary rows, got:\n{$output}");
+
+        // Shell row: placeholders with the fallbacks in place.
+        $shell = json_decode($lines[0], true);
+        $this->assertIsArray($shell);
+        $this->assertStringContainsString('F:0', $lines[0]);
+        $this->assertStringContainsString('F:1', $lines[0]);
+        $this->assertStringContainsString('load-a', $lines[0]);
+        $this->assertStringContainsString('load-b', $lines[0]);
+
+        // Boundary rows carry {b, tree}; the faster (b, 0.02) resolves first.
+        $first = json_decode($lines[1], true);
+        $second = json_decode($lines[2], true);
+        $this->assertSame(1, $first['b']);
+        $this->assertSame(0, $second['b']);
+        $this->assertStringContainsString('done-b', $lines[1]);
+        $this->assertStringContainsString('done-a', $lines[2]);
+    }
+
+    public function testSynchronousTreeStreamsAsASingleShellRow(): void
+    {
+        $output = $this->runStream(<<<'PHP'
+        Flight::stream(['$', 'div', null, [['$', 'span', null, ['hi']]]]);
+        PHP);
+
+        $lines = array_values(array_filter(explode("\n", trim($output))));
+        $this->assertCount(1, $lines);
+        $this->assertStringContainsString('hi', $lines[0]);
+    }
+}