feat(dump): stream active pipeline recordings

README.md

@@ -206,13 +206,26 @@ stage throws. Client JSON serialization runs inside settlement and publishes UTF
 while retaining the canonical content for inspection. Streaming quantities settle at the deferred exit. Explicit protocol and
 transport errors take precedence over the settlement's generic failure marker.
 
-`@floway-dev/dump` owns portable run recording, stream recording, metadata contracts,
-broker framing and wire conversion. Pipeline encodes stage facts, logs and protocol frames
-into one object space. Gateway owns attribution, admission, SQL storage and HTTP measurement.
-Closing waits for owned readers and deferred outcomes before storing the completed NDJSON
-record and publishing its metadata. The Collected view selects the recorded client stream and uses its
-`stream.end` event to establish recording completion. LogStream infrastructure is available
-on both platforms; live recording and its business readers remain outside these three PRs.
+Pipeline owns the stage-event format and object-space encoding. `@floway-dev/dump` owns
+run recording, stream references, durable NDJSON writing, transient byte streams and metadata
+publication. Gateway supplies storage/broker ports and applies request admission, model
+attribution, HTTP byte measurement and retention policy. Encoded batches flow to durable
+storage with backpressure and to a LogStream at the same byte positions. Three same-offset
+append attempts tolerate lost acknowledgements; persistent live failure leaves durable
+recording active. Empty heartbeats renew the acknowledged position independently of a
+blocked durable writer.
+
+Closing waits for owned readers, deferred outcomes and queued events before assembling
+metadata. The artifact and metadata notification are published before the live stream ends.
+Encoding, metadata or publication failures abort incomplete writes, stop leases and retain
+original errors. Run IDs are assigned at entry; listing remains ordered by completion time
+and ID. The Collected view uses the selected recorded client stream and its `stream.end`
+event to establish recording completion.
+
+The authenticated `GET /api/dump/keys/:keyId/records/:recordId/live?offset=0` endpoint attaches
+to an existing per-key stream through the platform framed byte protocol. Active-run
+discovery, reconnect/list semantics and live dashboard interactions remain separate design
+work; completed records retain the same run-only format.
 
 Pipeline endpoints are assembled from individual stage modules. Each stage owns
 its request and response contract; a large stage keeps local helpers in the same

apps/platform-cloudflare/src/r2-file-store.ts

@@ -1,21 +1,72 @@
-import type { FileStore } from '@floway-dev/platform';
+import type { FileBody, FileStore } from '@floway-dev/platform';
+
+export interface R2UploadedPartLike { partNumber: number; etag: string }
+export interface R2MultipartUploadLike {
+  uploadPart(partNumber: number, value: Uint8Array): Promise<R2UploadedPartLike>;
+  complete(parts: R2UploadedPartLike[]): Promise<unknown>;
+  abort(): Promise<void>;
+}
 
 export interface R2BucketLike {
   put(key: string, value: ReadableStream | ArrayBuffer | ArrayBufferView | string | null): Promise<unknown>;
   get(key: string): Promise<{ arrayBuffer(): Promise<ArrayBuffer> } | null>;
   delete(keys: string | string[]): Promise<void>;
+  createMultipartUpload(key: string): Promise<R2MultipartUploadLike>;
 }
 
 // R2 caps both `list` and `delete` at 1000 keys per call.
 // https://developers.cloudflare.com/r2/api/workers/workers-api-reference/#list
 // https://developers.cloudflare.com/r2/api/workers/workers-api-reference/#delete
 const R2_BATCH_LIMIT = 1000;
 
+// Generated streams have no known content length, so R2's single put cannot consume
+// them. Multipart parts must be at least 5 MiB except for the last part.
+// https://developers.cloudflare.com/r2/objects/upload-objects/#multipart-upload
+const R2_PART_BYTES = 5 * 1024 * 1024;
+
 export class R2FileStore implements FileStore {
   constructor(private readonly bucket: R2BucketLike) {}
 
-  async put(key: string, body: Uint8Array): Promise<void> {
-    await this.bucket.put(key, body);
+  async put(key: string, body: FileBody): Promise<void> {
+    if (body instanceof Uint8Array) { await this.bucket.put(key, body); return; }
+    const reader = body.getReader();
+    let upload: R2MultipartUploadLike | undefined;
+    const parts: R2UploadedPartLike[] = [];
+    let buffer = new Uint8Array(R2_PART_BYTES);
+    let length = 0;
+    const flush = async () => {
+      upload ??= await this.bucket.createMultipartUpload(key);
+      parts.push(await upload.uploadPart(parts.length + 1, length === buffer.byteLength ? buffer : buffer.subarray(0, length)));
+      buffer = new Uint8Array(R2_PART_BYTES);
+      length = 0;
+    };
+    try {
+      for (;;) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        let offset = 0;
+        while (offset < value.byteLength) {
+          const count = Math.min(buffer.byteLength - length, value.byteLength - offset);
+          buffer.set(value.subarray(offset, offset + count), length);
+          length += count;
+          offset += count;
+          if (length === buffer.byteLength) await flush();
+        }
+      }
+      if (upload === undefined) await this.bucket.put(key, buffer.subarray(0, length));
+      else {
+        if (length > 0) await flush();
+        await upload.complete(parts);
+      }
+    } catch (error) {
+      const errors: unknown[] = [error];
+      try { await reader.cancel(error); } catch (cleanupError) { if (cleanupError !== error) errors.push(cleanupError); }
+      if (upload !== undefined) {
+        try { await upload.abort(); } catch (cleanupError) { errors.push(cleanupError); }
+      }
+      if (errors.length > 1) throw new AggregateError(errors, 'R2 upload failed and cleanup also failed', { cause: error });
+      throw error;
+    } finally { reader.releaseLock(); }
   }
 
   async get(key: string): Promise<Uint8Array | null> {