diff --git a/README.md b/README.md
index c56d74462b..bd97ec97bf 100644
--- a/README.md
+++ b/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
diff --git a/apps/platform-cloudflare/__tests__/r2-file-store_test.ts b/apps/platform-cloudflare/__tests__/r2-file-store_test.ts
index 9b5e800536..99ce9d388c 100644
--- a/apps/platform-cloudflare/__tests__/r2-file-store_test.ts
+++ b/apps/platform-cloudflare/__tests__/r2-file-store_test.ts
@@ -1,11 +1,29 @@
-import { test } from 'vitest';
+import { expect, test } from 'vitest';
 
-import { R2FileStore, type R2BucketLike } from '../src/r2-file-store.ts';
+import { R2FileStore, type R2BucketLike, type R2MultipartUploadLike, type R2UploadedPartLike } from '../src/r2-file-store.ts';
 import { assertEquals } from '@floway-dev/test-utils';
 
 class FakeR2Bucket implements R2BucketLike {
   store = new Map<string, Uint8Array>();
   deleteCalls: string[][] = [];
+  parts: Uint8Array[] = [];
+  aborted = false;
+
+  async createMultipartUpload(key: string): Promise<R2MultipartUploadLike> {
+    return {
+      uploadPart: async (partNumber, value) => {
+        this.parts.push(value.slice());
+        return { partNumber, etag: String(partNumber) };
+      },
+      complete: async (_parts: R2UploadedPartLike[]) => {
+        const bytes = new Uint8Array(this.parts.reduce((size, part) => size + part.byteLength, 0));
+        let offset = 0;
+        for (const part of this.parts) { bytes.set(part, offset); offset += part.byteLength; }
+        this.store.set(key, bytes);
+      },
+      abort: async () => { this.aborted = true; },
+    };
+  }
 
   async put(key: string, value: ReadableStream | ArrayBuffer | ArrayBufferView | string | null): Promise<unknown> {
     if (!(value instanceof Uint8Array)) throw new Error('FakeR2Bucket only supports Uint8Array');
@@ -37,3 +55,53 @@ test('R2FileStore deletes exact keys in one R2 batch', async () => {
   assertEquals([...bucket.store.keys()], ['drop/ab']);
   assertEquals(bucket.deleteCalls, [['drop/a', 'missing']]);
 });
+
+test('unknown-length streaming writes keep parts bounded and publish exact bytes at EOF', async () => {
+  const bucket = new FakeR2Bucket();
+  const store = new R2FileStore(bucket);
+  const size = 5 * 1024 * 1024;
+  let finish!: () => void;
+  const gate = new Promise<void>(resolve => { finish = resolve; });
+  let chunk = 0;
+  const source = new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      if (chunk++ === 0) controller.enqueue(new Uint8Array(size + 3).fill(7));
+      else { await gate; controller.close(); }
+    },
+  }, { highWaterMark: 0 });
+  const writing = store.put('stream', source);
+  await expect.poll(() => bucket.parts.length).toBe(1);
+  expect(bucket.parts[0]?.byteLength).toBe(size);
+  expect(await store.get('stream')).toBeNull();
+  finish();
+  await writing;
+  expect(bucket.parts.map(part => part.byteLength)).toEqual([size, 3]);
+  const stored = (await store.get('stream'))!;
+  expect(stored.byteLength).toBe(size + 3);
+  expect(stored.every(byte => byte === 7)).toBe(true);
+  expect(bucket.aborted).toBe(false);
+});
+
+test.each([0, 3])('a short stream of %i bytes uses a complete single put', async size => {
+  const bucket = new FakeR2Bucket();
+  const store = new R2FileStore(bucket);
+  await store.put('short', new ReadableStream({ start(controller) { controller.enqueue(new Uint8Array(size)); controller.close(); } }));
+  expect(await store.get('short')).toEqual(new Uint8Array(size));
+  expect(bucket.parts).toEqual([]);
+});
+
+test('a failed streamed upload leaves the previous object and preserves the original error', async () => {
+  const bucket = new FakeR2Bucket();
+  const store = new R2FileStore(bucket);
+  await store.put('failed', new Uint8Array([9]));
+  const error = new Error('source interrupted');
+  let chunk = 0;
+  await expect(store.put('failed', new ReadableStream({
+    pull(controller) {
+      if (chunk++ === 0) controller.enqueue(new Uint8Array(5 * 1024 * 1024));
+      else controller.error(error);
+    },
+  }, { highWaterMark: 0 }))).rejects.toBe(error);
+  expect(bucket.aborted).toBe(true);
+  expect(await store.get('failed')).toEqual(new Uint8Array([9]));
+});
diff --git a/apps/platform-cloudflare/src/r2-file-store.ts b/apps/platform-cloudflare/src/r2-file-store.ts
index ab1c018e16..721fa22123 100644
--- a/apps/platform-cloudflare/src/r2-file-store.ts
+++ b/apps/platform-cloudflare/src/r2-file-store.ts
@@ -1,9 +1,17 @@
-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.
@@ -11,11 +19,54 @@ export interface R2BucketLike {
 // 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> {
diff --git a/apps/platform-node/__tests__/fixtures/concurrent-file-writers.ts b/apps/platform-node/__tests__/fixtures/concurrent-file-writers.ts
new file mode 100644
index 0000000000..ad7541ceb6
--- /dev/null
+++ b/apps/platform-node/__tests__/fixtures/concurrent-file-writers.ts
@@ -0,0 +1,26 @@
+import fsPromises from 'node:fs/promises';
+import { syncBuiltinESMExports } from 'node:module';
+
+const originalOpen = fsPromises.open;
+let pause = true;
+let staged!: () => void;
+const started = new Promise<void>(resolve => { staged = resolve; });
+let resume!: () => void;
+const gate = new Promise<void>(resolve => { resume = resolve; });
+fsPromises.open = async (...args: Parameters<typeof fsPromises.open>) => {
+  if (pause) { pause = false; staged(); await gate; }
+  return await originalOpen(...args);
+};
+syncBuiltinESMExports();
+const { FsFileStore } = await import('../../src/fs-file-store.ts');
+const store = new FsFileStore(process.argv[2]!);
+try {
+  const first = store.put('file', new Uint8Array([1]));
+  await started;
+  await store.put('file', new Uint8Array([2]));
+  resume();
+  await first;
+} finally {
+  fsPromises.open = originalOpen;
+  syncBuiltinESMExports();
+}
diff --git a/apps/platform-node/__tests__/fixtures/killed-file-writer.ts b/apps/platform-node/__tests__/fixtures/killed-file-writer.ts
new file mode 100644
index 0000000000..88b584ca84
--- /dev/null
+++ b/apps/platform-node/__tests__/fixtures/killed-file-writer.ts
@@ -0,0 +1,13 @@
+import { FsFileStore } from '../../src/fs-file-store.ts';
+
+const store = new FsFileStore(process.argv[2]!);
+let chunk = 0;
+await store.put('file', new ReadableStream({
+  async pull(controller) {
+    if (chunk++ === 0) controller.enqueue(new Uint8Array([1, 2]));
+    else {
+      process.send!('staged');
+      await new Promise<void>(() => {});
+    }
+  },
+}, { highWaterMark: 0 }));
diff --git a/apps/platform-node/__tests__/fs-file-store_test.ts b/apps/platform-node/__tests__/fs-file-store_test.ts
index 34d1b2e114..060390c4f7 100644
--- a/apps/platform-node/__tests__/fs-file-store_test.ts
+++ b/apps/platform-node/__tests__/fs-file-store_test.ts
@@ -1,8 +1,11 @@
-import { mkdtemp, rm } from 'node:fs/promises';
+import { spawn } from 'node:child_process';
+import { once } from 'node:events';
+import { mkdtemp, readdir, rm, stat } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
+import { fileURLToPath } from 'node:url';
 
-import { test } from 'vitest';
+import { expect, test } from 'vitest';
 
 import { FsFileStore } from '../src/fs-file-store.ts';
 import { assertEquals } from '@floway-dev/test-utils';
@@ -47,3 +50,83 @@ test('put creates intermediate directories', () => withTempRoot(async root => {
   const read = await store.get('deeply/nested/path/file.bin');
   assertEquals(read, new Uint8Array([42]));
 }));
+
+test('a streamed replacement stays private until EOF and publishes complete binary bytes', () => withTempRoot(async root => {
+  const store = new FsFileStore(root);
+  await store.put('file', new Uint8Array([9]));
+  let finish!: () => void;
+  const gate = new Promise<void>(resolve => { finish = resolve; });
+  let chunk = 0;
+  const writing = store.put('file', new ReadableStream({
+    async pull(controller) {
+      if (chunk++ === 0) controller.enqueue(new Uint8Array([1, 2]));
+      else { await gate; controller.enqueue(new Uint8Array([3, 4])); controller.close(); }
+    },
+  }, { highWaterMark: 0 }));
+  await expect.poll(() => chunk).toBe(2);
+  expect(await store.get('file')).toEqual(new Uint8Array([9]));
+  finish();
+  await writing;
+  expect(await store.get('file')).toEqual(new Uint8Array([1, 2, 3, 4]));
+  expect(await readdir(root)).toEqual(['.floway-staging', 'file']);
+  const directories = await readdir(join(root, '.floway-staging'));
+  expect(directories).toHaveLength(1);
+  expect(await readdir(join(root, '.floway-staging', directories[0]!))).toEqual([]);
+}));
+
+test('source failure removes staging data without replacing the prior file', () => withTempRoot(async root => {
+  const store = new FsFileStore(root);
+  await store.put('file', new Uint8Array([9]));
+  const error = new Error('source interrupted');
+  let chunk = 0;
+  await expect(store.put('file', new ReadableStream({
+    pull(controller) {
+      if (chunk++ === 0) controller.enqueue(new Uint8Array([1, 2]));
+      else controller.error(error);
+    },
+  }, { highWaterMark: 0 }))).rejects.toBe(error);
+  expect(await store.get('file')).toEqual(new Uint8Array([9]));
+  expect(await readdir(root)).toEqual(['.floway-staging', 'file']);
+  const directories = await readdir(join(root, '.floway-staging'));
+  expect(directories).toHaveLength(1);
+  expect(await readdir(join(root, '.floway-staging', directories[0]!))).toEqual([]);
+}));
+
+test('deleteKeys collects an upload whose writer was killed without matching another key by name', () => withTempRoot(async root => {
+  const store = new FsFileStore(root);
+  const namedLikeStaging = 'file.00000000-0000-4000-8000-000000000000.tmp';
+  await store.put(namedLikeStaging, new Uint8Array([9]));
+  const child = spawn(process.execPath, ['--import', import.meta.resolve('tsx'), fileURLToPath(new URL('./fixtures/killed-file-writer.ts', import.meta.url)), root], { stdio: ['ignore', 'pipe', 'pipe', 'ipc'] });
+  const exited = once(child, 'exit');
+  try {
+    await Promise.race([once(child, 'message'), exited.then(([code]) => { throw new Error(`Writer exited before staging: ${String(code)}`); })]);
+    expect((await stat(join(root, '.floway-staging'))).dev).toBe((await stat(root)).dev);
+    child.kill('SIGKILL');
+    await exited;
+    await store.deleteKeys(['file']);
+    expect(await store.get('file')).toBeNull();
+    expect(await store.get(namedLikeStaging)).toEqual(new Uint8Array([9]));
+    expect(await readdir(root)).toEqual(['.floway-staging', namedLikeStaging]);
+    expect(await readdir(join(root, '.floway-staging'))).toHaveLength(1);
+  } finally {
+    if (child.exitCode === null && child.signalCode === null) { child.kill('SIGKILL'); await exited; }
+  }
+}));
+
+test.each(['.floway-staging', '.FLOWAY-STAGING', '.Floway-Staging'])('Node reserves private segment %s while retaining published key paths', segment => withTempRoot(async root => {
+  const store = new FsFileStore(root);
+  await expect(store.put(`nested/${segment}/key`, new Uint8Array([1]))).rejects.toThrow('reserved staging segment');
+  await expect(store.get(`${segment}/key`)).rejects.toThrow('reserved staging segment');
+  await expect(store.deleteKeys([`${segment}/key`])).rejects.toThrow('reserved staging segment');
+  await store.put('nested/key', new Uint8Array([7]));
+  expect(await readdir(join(root, 'nested'))).toEqual(['.floway-staging', 'key']);
+}));
+
+test('a finishing writer leaves the shared staging directory available to a writer opening its file', () => withTempRoot(async root => {
+  const child = spawn(process.execPath, ['--import', import.meta.resolve('tsx'), fileURLToPath(new URL('./fixtures/concurrent-file-writers.ts', import.meta.url)), root], { stdio: ['ignore', 'pipe', 'pipe'] });
+  let stderr = '';
+  child.stderr.on('data', chunk => { stderr += String(chunk); });
+  const [code] = await once(child, 'exit');
+  expect({ code, stderr }).toEqual({ code: 0, stderr: '' });
+  expect(await new FsFileStore(root).get('file')).toEqual(new Uint8Array([1]));
+}));
diff --git a/apps/platform-node/src/fs-file-store.ts b/apps/platform-node/src/fs-file-store.ts
index 7560d16202..42ec8f3fa6 100644
--- a/apps/platform-node/src/fs-file-store.ts
+++ b/apps/platform-node/src/fs-file-store.ts
@@ -1,38 +1,58 @@
+import { createHash, randomUUID } from 'node:crypto';
 import { mkdirSync } from 'node:fs';
-import { mkdir, readFile, rm, writeFile } from 'node:fs/promises';
-import { dirname, isAbsolute, resolve, sep } from 'node:path';
-
-import type { FileStore } from '@floway-dev/platform';
-
-// Filesystem-backed FileStore. Every key resolves to a path under `root`.
-// Keys use forward-slash POSIX separators (matching R2's surface) and are
-// translated to native path segments on the way in/out so the same key reads
-// identically on Windows and POSIX hosts.
-//
-// Threat model: `root` (`FLOWAY_FILES_DIR`) is gateway-trusted. Everything
-// dumped here is data the gateway already holds in its database (API keys,
-// upstream credentials, request payloads); fs-level access to this directory
-// is already equivalent to gateway compromise. We deliberately do not mode
-// 0o600 / 0o700 the writes — bodies are stored verbatim and the OS-level
-// confidentiality boundary belongs to the operator (umask, mount perms,
-// dedicated user). The dashboard redacts sensitive headers at render time
-// for human display, but the on-disk record stays untouched so an operator
-// can replay or diff against upstream byte-for-byte.
+import { mkdir, open, readFile, rename, rm } from 'node:fs/promises';
+import { basename, dirname, isAbsolute, join, relative, resolve, sep } from 'node:path';
+
+import type { FileBody, FileStore } from '@floway-dev/platform';
+
+const STAGING_SEGMENT = '.floway-staging';
+const isStagingSegment = (segment: string): boolean => segment.toLowerCase() === STAGING_SEGMENT;
+
+// Keys use POSIX separators so filesystem and object-storage deployments agree.
+// The operator's umask, mount permissions and service account own confidentiality;
+// a file contains the writer's already-encoded bytes, including pipeline redaction.
+// Access to this directory exposes recorded request contents and is equivalent to
+// access to the gateway's stored configuration. We do not add a second permission policy.
 export class FsFileStore implements FileStore {
   private readonly root: string;
 
   constructor(root: string) {
-    // Resolve once so `pathFor` can verify resolved paths still live under it.
     this.root = resolve(root);
-    // Ensure the root exists so the first put() doesn't race against a missing
-    // directory and so tests / fresh deploys see a consistent structure.
     mkdirSync(this.root, { recursive: true });
   }
 
-  async put(key: string, body: Uint8Array): Promise<void> {
-    const path = this.pathFor(key);
-    await mkdir(dirname(path), { recursive: true });
-    await writeFile(path, body);
+  async put(key: string, body: FileBody): Promise<void> {
+    const reader = body instanceof Uint8Array ? null : body.getReader();
+    let staging: string | undefined;
+    try {
+      const path = this.pathFor(key);
+      await mkdir(dirname(path), { recursive: true });
+      const stagingDirectory = this.stagingFor(path);
+      await mkdir(stagingDirectory, { recursive: true });
+      staging = join(stagingDirectory, randomUUID());
+      {
+        await using file = await open(staging, 'wx');
+        if (body instanceof Uint8Array) await file.writeFile(body);
+        else {
+          for (;;) {
+            const { done, value } = await reader!.read();
+            if (done) break;
+            await file.writeFile(value);
+          }
+        }
+      }
+      await rename(staging, path);
+    } catch (error) {
+      const errors: unknown[] = [error];
+      if (reader !== null) {
+        try { await reader.cancel(error); } catch (cleanupError) { if (cleanupError !== error) errors.push(cleanupError); }
+      }
+      if (staging !== undefined) {
+        try { await rm(staging, { force: true }); } catch (cleanupError) { errors.push(cleanupError); }
+      }
+      if (errors.length > 1) throw new AggregateError(errors, 'File write failed and cleanup also failed', { cause: error });
+      throw error;
+    } finally { reader?.releaseLock(); }
   }
 
   async get(key: string): Promise<Uint8Array | null> {
@@ -45,7 +65,18 @@ export class FsFileStore implements FileStore {
   }
 
   async deleteKeys(keys: readonly string[]): Promise<void> {
-    await Promise.all(keys.map(async key => await rm(this.pathFor(key), { force: true })));
+    await Promise.all(keys.map(async key => {
+      const path = this.pathFor(key);
+      const staging = this.stagingFor(path);
+      await Promise.all([rm(path, { force: true }), rm(staging, { recursive: true, force: true })]);
+    }));
+  }
+
+  // Staging lives on the target's filesystem, including separately mounted key
+  // prefixes. Its reserved namespace lets key-based orphan cleanup collect a
+  // killed writer without interpreting another public key as a temporary file.
+  private stagingFor(path: string): string {
+    return join(dirname(path), STAGING_SEGMENT, createHash('sha256').update(basename(path)).digest('hex'));
   }
 
   // Resolve a key against `root` and reject paths that escape it. Even though
@@ -53,11 +84,13 @@ export class FsFileStore implements FileStore {
   // to scrub user-controlled segments and a `..`-laden key would otherwise
   // walk to arbitrary host paths under R2 it would simply be a strange key.
   private pathFor(key: string): string {
+    if (key.split('/').some(isStagingSegment)) throw new Error(`FsFileStore: reserved staging segment (${key})`);
     if (isAbsolute(key)) throw new Error(`FsFileStore: absolute keys are not supported (${key})`);
     const path = resolve(this.root, ...key.split('/'));
     if (path !== this.root && !path.startsWith(this.root + sep)) {
       throw new Error(`FsFileStore: key escapes root (${key})`);
     }
+    if (relative(this.root, path).split(sep).some(isStagingSegment)) throw new Error(`FsFileStore: reserved staging segment (${key})`);
     return path;
   }
 }
diff --git a/packages/dump/README.md b/packages/dump/README.md
index c5c82c7e54..4cf0521102 100644
--- a/packages/dump/README.md
+++ b/packages/dump/README.md
@@ -1,33 +1,74 @@
 # Run recording
 
-`@floway-dev/dump` stores pipeline runs as encoded NDJSON and publishes completed
-metadata. It owns recording contracts, stream recording, broker framing and the
-portable recorder. Its dependencies are supplied by the application.
+`@floway-dev/dump` owns the portable run writer, protocol-stream references,
+metadata contracts, broker framing and wire conversion. Pipeline owns stage
+execution and object-space encoding/reading. Platform owns FileStore, LogStream
+and their transport bindings. Gateway supplies fixed ports and owns request
+admission, identity generation, account/retention policy, model attribution,
+owned-work drain, HTTP measurement, SQL adapters and authenticated routes.
+
+## Recorder lifetime
+
+There is one streaming recorder. The application supplies the run ID and start
+time; storage receives an encoded byte stream and a promise for final metadata.
+The writer serializes each event's complete object batch, so concurrent deferred
+outcomes cannot overtake the objects they reference. Batches are emitted in at
+most 64 KiB byte chunks. Durable consumption applies backpressure to execution.
 
 ```ts
 const recorder = createRunRecorder({
-  write: record => store.put(keyId, record),
+  id,
+  startedAt,
+  write: run => store.putRun(keyId, run),
   publish: meta => broker.publish(keyId, meta),
+  openLive: () => liveStore.open(runStreamId(keyId, id)),
+  background,
 });
 const result = await run(chain, facts, { dump: recorder.sink });
-await result.drain();
-await recorder.finish(metadata);
+await recorder.finish(async () => {
+  await finishOwnedWork(result.drain);
+  await recorder.flush();
+  return buildMetadata();
+});
 ```
 
-The application owns drain/error handling and final metadata assembly, including
-failed requests. Metadata is published after storage succeeds. An encoder failure
-rejects completion with the original error so a partial object space is never
-published as a completed record.
+The application handles drain failures and assembles its business metadata in
+the finish callback. `flush` awaits the current encoded event tail, allowing the
+completion timestamp and attribution to follow all owned work and writes. The
+callback runs inside the recorder's abort boundary. Encoding or metadata failures
+abort the native durable reader and retain their original error. Cleanup failures
+retain the original error in an AggregateError chain.
+
+The artifact is published before broker notification; notification must succeed
+before clean live EOF. A notification failure retains the stored artifact and
+rejects completion with its original error. Background storage failures are
+observed even before finish begins. Failed recordings stop their heartbeat,
+including a live carrier that opens after the failure.
+
+## Transient bytes
+
+Durable and live readers receive identical NDJSON bytes. A live append retries
+three times at its original position, so a lost acknowledgement cannot duplicate
+content. Persistent live failure is reported while durable recording continues.
+Empty heartbeats use the acknowledged position independently of the data tail;
+a blocked durable reader cannot expire an otherwise active writer. Completion
+awaits any in-flight heartbeat before ending the carrier.
+
+`runStreamId(scopeId, runId)` shares the writer/reader namespace. The application
+owns scope authorization. FileStore stages complete atomic objects; the SQL
+adapter renews unfinished file leases and publishes completed rows. Storage
+failures leave unfinished artifacts eligible for cleanup.
 
-`recordStream` records frames before forwarding them. Only natural exhaustion
-records the stream end marker; returning early or throwing leaves it incomplete.
-The symbol-backed stream reference survives transport wrappers through
-`streamReferenceOf`.
+## Protocol references and readers
 
-Pipeline events, object-space encoding/decoding, secrets and deferred values remain
-in `@floway-dev/pipeline`. FileStore and LogStream contracts remain in
-`@floway-dev/platform`. Gateway owns request admission, account/retention policy,
-model attribution, HTTP measurement, SQL adapters and authenticated routes.
+`recordStream` records frames before forwarding the same source values. Only
+natural exhaustion records the stream end marker; returning early or throwing
+leaves it incomplete. Its nullable projector excludes transport-only values
+without removing those values from the source. `streamReferenceOf` carries the
+same Symbol-backed reference across transport wrappers.
 
-Use `/types` for browser metadata/record contracts and `/codec` for platform broker
-composition. Run records contain facts and protocol frames.
+Use `/types` for browser metadata and record contracts and `/codec` for platform
+broker composition. `StoredDumpRecord` contains completed bytes; `DumpRunWrite`
+contains the writer-owned stream and deferred metadata. Run records contain
+facts and protocol frames in the pipeline object space. Live discovery,
+reconnection and list interaction are separate application work.
diff --git a/packages/dump/__tests__/fixtures.ts b/packages/dump/__tests__/fixtures.ts
index 11e6ebb579..bafe0d6f5e 100644
--- a/packages/dump/__tests__/fixtures.ts
+++ b/packages/dump/__tests__/fixtures.ts
@@ -1,4 +1,5 @@
-import type { DumpMetadata, StoredDumpRecord } from '../src/index.ts';
+import { testLogStreamStore } from './log-stream.ts';
+import { createRunRecorder, type RecordingPorts, type DumpMetadata, type StoredDumpRecord } from '../src/index.ts';
 import type { DumpEvent } from '@floway-dev/pipeline';
 
 export const metadata = (overrides: Partial<DumpMetadata> = {}): DumpMetadata => ({
@@ -23,3 +24,22 @@ export const metadata = (overrides: Partial<DumpMetadata> = {}): DumpMetadata =>
 
 export const eventsOf = (record: StoredDumpRecord): DumpEvent[] =>
   new TextDecoder('utf-8', { fatal: true }).decode(record.events).trimEnd().split('\n').map(line => JSON.parse(line) as DumpEvent);
+
+export const recordingFixture = (overrides: Partial<RecordingPorts> = {}) => {
+  const records: StoredDumpRecord[] = [];
+  const published: DumpMetadata[] = [];
+  const work: Promise<unknown>[] = [];
+  const live = testLogStreamStore();
+  const recorder = createRunRecorder({
+    id: 'recorded-run', startedAt: 100,
+    write: async run => {
+      const events = new Uint8Array(await new Response(run.events).arrayBuffer());
+      records.push({ meta: await run.metadata, events });
+    },
+    publish: async meta => { published.push(meta); },
+    openLive: () => live.open('recorded-run'),
+    background: task => { work.push(task); void task.catch(() => {}); },
+    ...overrides,
+  });
+  return { recorder, records, published, work, live };
+};
diff --git a/packages/dump/__tests__/log-stream.ts b/packages/dump/__tests__/log-stream.ts
new file mode 100644
index 0000000000..f48629313e
--- /dev/null
+++ b/packages/dump/__tests__/log-stream.ts
@@ -0,0 +1,43 @@
+import type { LogStream, LogStreamStore } from '@floway-dev/platform';
+
+export const testLogStreamStore = (): LogStreamStore => {
+  const streams = new Map<string, LogStream>();
+  return {
+    async open(id) {
+      const chunks: { offset: number; bytes: Uint8Array }[] = [];
+      let length = 0;
+      let ended = false;
+      const listeners = new Set<() => void>();
+      const wake = () => { for (const listener of listeners) listener(); };
+      const stream: LogStream = {
+        async append(offset, bytes) {
+          if (ended) throw new Error('Test LogStream is ended');
+          if (offset > length) throw new Error('Test LogStream append would leave a hole');
+          const tail = bytes.slice(length - offset);
+          if (tail.byteLength > 0) { chunks.push({ offset: length, bytes: tail }); length += tail.byteLength; }
+          wake();
+        },
+        async end() { ended = true; wake(); },
+        async *read(offset, signal) {
+          while (!signal.aborted) {
+            const chunk = chunks.find(chunk => chunk.offset + chunk.bytes.byteLength > offset);
+            if (chunk) {
+              const bytes = chunk.bytes.slice(Math.max(0, offset - chunk.offset));
+              offset += bytes.byteLength;
+              yield bytes;
+            } else if (ended) return;
+            else await new Promise<void>(resolve => {
+              const finish = () => { listeners.delete(finish); signal.removeEventListener('abort', finish); resolve(); };
+              listeners.add(finish);
+              signal.addEventListener('abort', finish, { once: true });
+            });
+          }
+          signal.throwIfAborted();
+        },
+      };
+      streams.set(id, stream);
+      return stream;
+    },
+    async get(id) { return streams.get(id) ?? null; },
+  };
+};
diff --git a/packages/dump/__tests__/recorder_test.ts b/packages/dump/__tests__/recorder_test.ts
index a0fd3c681a..80e1fa6213 100644
--- a/packages/dump/__tests__/recorder_test.ts
+++ b/packages/dump/__tests__/recorder_test.ts
@@ -1,12 +1,11 @@
 import { expect, test } from 'vitest';
 
-import { eventsOf, metadata } from './fixtures.ts';
-import { createRunRecorder, type StoredDumpRecord } from '../src/index.ts';
+import { eventsOf, metadata, recordingFixture } from './fixtures.ts';
+import type { StoredDumpRecord } from '../src/index.ts';
 import { compose, createRunReader, defer, defineStage, move, run } from '@floway-dev/pipeline';
 
 test('a native pipeline round-trips facts, shared frame values and deferred outcomes through one object space', async () => {
-  const written: StoredDumpRecord[] = [];
-  const recorder = createRunRecorder({ write: async record => { written.push(record); }, publish: async () => {} });
+  const { recorder, records: written } = recordingFixture();
   const shared = move({ text: 'Recorded 中', nested: { count: 7 } });
   const pending = Promise.withResolvers<typeof shared>();
   const deferred = defer(pending.promise);
@@ -31,7 +30,7 @@ test('a native pipeline round-trips facts, shared frame values and deferred outc
   await stream.end();
   pending.resolve(shared);
   await result.drain();
-  await recorder.finish(metadata());
+  await recorder.finish(async () => metadata());
 
   expect(written).toHaveLength(1);
   const events = eventsOf(written[0]!);
@@ -57,12 +56,18 @@ test('publication waits for an observed storage write to complete', async () =>
   const stored = Promise.withResolvers<void>();
   const order: string[] = [];
   const meta = metadata();
-  const recorder = createRunRecorder({
-    write: async record => { order.push('writing'); entered.resolve(record); await stored.promise; order.push('stored'); },
+  const { recorder } = recordingFixture({
+    write: async run => {
+      order.push('writing');
+      const events = new Uint8Array(await new Response(run.events).arrayBuffer());
+      entered.resolve({ meta: await run.metadata, events });
+      await stored.promise;
+      order.push('stored');
+    },
     publish: async value => { expect(value).toBe(meta); order.push('published'); },
   });
-  recorder.sink({ type: 'stage.entered', stageId: 1, name: 'answer', parentStageId: null, facts: move({ input: '中' }) });
-  const finishing = recorder.finish(meta);
+  await recorder.sink({ type: 'stage.entered', stageId: 1, name: 'answer', parentStageId: null, facts: move({ input: '中' }) });
+  const finishing = recorder.finish(async () => meta);
   const record = await entered.promise;
   expect(record.meta).toBe(meta);
   expect(eventsOf(record).some(event => event.type === 'stage.entered')).toBe(true);
@@ -73,54 +78,121 @@ test('publication waits for an observed storage write to complete', async () =>
 });
 
 test('an encoder failure retains its original exception and never stores or publishes a partial record', async () => {
-  let writes = 0;
-  let publications = 0;
-  const recorder = createRunRecorder({ write: async () => { writes++; }, publish: async () => { publications++; } });
-  let original: unknown;
-  try {
-    recorder.sink({ type: 'stage.entered', stageId: 1, name: 'invalid', parentStageId: null, facts: move({ upload: new Blob(['source bytes']) }) });
-  } catch (error) { original = error; }
+  const { recorder, records, published, work } = recordingFixture();
+  const original = await recorder.sink({ type: 'stage.entered', stageId: 1, name: 'invalid', parentStageId: null, facts: move({ upload: new Blob(['source bytes']) }) }).catch(error => error as Error);
   expect(original).toBeInstanceOf(TypeError);
-  expect((original as Error).message).toContain('Blob');
-  let subsequent: unknown;
-  try {
-    recorder.sink({ type: 'stage.failed', stageId: 1, error: original });
-  } catch (error) { subsequent = error; }
-  expect(subsequent).toBe(original);
-  await expect(recorder.finish(metadata({ status: 500 }))).rejects.toBe(original);
-  expect(writes).toBe(0);
-  expect(publications).toBe(0);
+  if (!(original instanceof Error)) throw new Error('Expected the encoder failure');
+  expect(original.message).toContain('Blob');
+  await expect(recorder.sink({ type: 'stage.failed', stageId: 1, error: original })).rejects.toBe(original);
+  await expect(recorder.finish(async () => metadata({ status: 500 }))).rejects.toBe(original);
+  const outcomes = await Promise.allSettled(work);
+  expect(outcomes).toEqual([{ status: 'rejected', reason: original }]);
+  expect(records).toHaveLength(0);
+  expect(published).toHaveLength(0);
 });
 
 test('storage rejection propagates the original error and prevents publication', async () => {
   const original = new Error('durable storage failed');
   let publications = 0;
-  const recorder = createRunRecorder({ write: async () => { throw original; }, publish: async () => { publications++; } });
-  await recorder.frame({ type: 'event', event: { text: 'recorded' } });
-  await expect(recorder.finish(metadata())).rejects.toBe(original);
+  const { recorder } = recordingFixture({ write: async () => { throw original; }, publish: async () => { publications++; } });
+  await expect(recorder.frame({ type: 'event', event: { text: 'recorded' } })).rejects.toBe(original);
+  await expect(recorder.finish(async () => metadata())).rejects.toBe(original);
   expect(publications).toBe(0);
 });
 
-test('publication rejection preserves its original error after the complete record exists', async () => {
+test('publication rejection preserves its original error after the complete record exists and omits live EOF', async () => {
   const original = new Error('broker failed');
-  const written: StoredDumpRecord[] = [];
-  const recorder = createRunRecorder({ write: async record => { written.push(record); }, publish: async () => { throw original; } });
+  const entered = Promise.withResolvers<void>();
+  const publication = Promise.withResolvers<void>();
+  const { recorder, records: written, live } = recordingFixture({ publish: async () => { entered.resolve(); await publication.promise; } });
   await recorder.frame({ type: 'event', event: { text: 'recorded' } });
-  await expect(recorder.finish(metadata())).rejects.toBe(original);
+  const controller = new AbortController();
+  const stream = (await live.get('recorded-run'))!;
+  const reader = stream.read(0, controller.signal)[Symbol.asyncIterator]();
+  expect((await reader.next()).done).toBe(false);
+  const finishing = recorder.finish(async () => metadata());
+  await entered.promise;
   expect(written).toHaveLength(1);
+  publication.reject(original);
+  await expect(finishing).rejects.toBe(original);
+  let ended = false;
+  const completion = reader.next().then(value => { ended = true; return value; });
+  await new Promise<void>(resolve => setImmediate(resolve));
+  expect(ended).toBe(false);
+  const cancelled = new Error('reader cancelled after interrupted recording');
+  controller.abort(cancelled);
+  await expect(completion).rejects.toBe(cancelled);
   expect(eventsOf(written[0]!).filter(event => event.type === 'stream.frame')).toHaveLength(1);
 });
 
 test('direct frames share a stream and separately opened streams retain distinct references', async () => {
-  const written: StoredDumpRecord[] = [];
-  const recorder = createRunRecorder({ write: async record => { written.push(record); }, publish: async () => {} });
+  const { recorder, records: written } = recordingFixture();
   await recorder.frame({ type: 'event', event: 'first' });
   await recorder.frame({ type: 'event', event: 'second' });
   const independent = recorder.openStream();
   await independent.frame({ type: 'event', event: 'other' });
   await independent.end();
-  await recorder.finish(metadata());
+  await recorder.finish(async () => metadata());
   const events = eventsOf(written[0]!);
   expect(events.filter(event => event.type === 'stream.frame').map(event => event.streamId)).toEqual([1, 1, 2]);
   expect(events.filter(event => event.type === 'stream.end').map(event => event.streamId)).toEqual([2]);
 });
+
+test('clean live EOF waits for an entered metadata publication gate', async () => {
+  const publishing = Promise.withResolvers<void>();
+  const release = Promise.withResolvers<void>();
+  const { recorder, records, live } = recordingFixture({ publish: async () => { publishing.resolve(); await release.promise; } });
+  await recorder.frame({ type: 'event', event: { text: 'published last' } });
+  const stream = (await live.get('recorded-run'))!;
+  const iterator = stream.read(0, new AbortController().signal)[Symbol.asyncIterator]();
+  expect((await iterator.next()).done).toBe(false);
+  const finishing = recorder.finish(async () => metadata());
+  await publishing.promise;
+  expect(records).toHaveLength(1);
+  let ended = false;
+  const completion = iterator.next().then(value => { ended = true; return value; });
+  await new Promise<void>(resolve => setImmediate(resolve));
+  expect(ended).toBe(false);
+  release.resolve();
+  await finishing;
+  expect((await completion).done).toBe(true);
+});
+
+test('a metadata factory failure aborts the actual durable reader with its original error', async () => {
+  const original = new Error('metadata lookup failed', { cause: new Error('database unavailable') });
+  const bodyFailure = Promise.withResolvers<unknown>();
+  const { recorder, records, published, work } = recordingFixture({
+    write: async run => {
+      try { await new Response(run.events).arrayBuffer(); } catch (error) { bodyFailure.resolve(error); throw error; }
+      await run.metadata;
+    },
+  });
+  await recorder.frame({ type: 'event', event: { text: 'before metadata' } });
+  await expect(recorder.finish(async () => { throw original; })).rejects.toBe(original);
+  expect(await bodyFailure.promise).toBe(original);
+  expect(await Promise.allSettled(work)).toEqual([{ status: 'rejected', reason: original }]);
+  expect(records).toHaveLength(0);
+  expect(published).toHaveLength(0);
+});
+
+test('large concurrent event batches retain shared identity and exact multibyte live bytes', async () => {
+  const { recorder, records, live } = recordingFixture();
+  const shared = move({ text: '中β'.repeat(40_000) });
+  await Promise.all([
+    recorder.sink({ type: 'stage.entered', stageId: 1, name: 'first', parentStageId: null, facts: move({ payload: shared }) }),
+    recorder.sink({ type: 'stage.entered', stageId: 2, name: 'second', parentStageId: 1, facts: move({ payload: shared }) }),
+  ]);
+  await recorder.finish(async () => metadata());
+  const chunks: Uint8Array[] = [];
+  const stream = (await live.get('recorded-run'))!;
+  for await (const bytes of stream.read(0, new AbortController().signal)) chunks.push(bytes);
+  expect(chunks.length).toBeGreaterThan(2);
+  expect(chunks.every(bytes => bytes.byteLength <= 64 * 1024)).toBe(true);
+  const bytes = new Uint8Array(chunks.reduce((sum, value) => sum + value.byteLength, 0));
+  let offset = 0;
+  for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.byteLength; }
+  expect(bytes).toEqual(records[0]!.events);
+  const events = eventsOf(records[0]!);
+  expect(events.filter(event => event.type === 'stage.entered').map(event => event.name)).toEqual(['first', 'second']);
+  expect(events.filter(event => event.type === 'object')).toHaveLength(1);
+});
diff --git a/packages/dump/__tests__/stream_test.ts b/packages/dump/__tests__/stream_test.ts
index e9a6f77a1a..353a93a7ff 100644
--- a/packages/dump/__tests__/stream_test.ts
+++ b/packages/dump/__tests__/stream_test.ts
@@ -1,7 +1,7 @@
 import { expect, test } from 'vitest';
 
-import { eventsOf, metadata } from './fixtures.ts';
-import { createRunRecorder, recordStream, streamReferenceOf, type StoredDumpRecord, type StreamRecorder } from '../src/index.ts';
+import { eventsOf, metadata, recordingFixture } from './fixtures.ts';
+import { recordStream, streamReferenceOf, type StreamRecorder } from '../src/index.ts';
 import { isStreamFact, streamFact } from '@floway-dev/pipeline';
 import type { ProtocolFrame } from '@floway-dev/protocols/common';
 
@@ -11,8 +11,7 @@ const frames = [
 ] as const;
 
 const fixture = () => {
-  const records: StoredDumpRecord[] = [];
-  const recorder = createRunRecorder({ write: async record => { records.push(record); }, publish: async () => {} });
+  const { recorder, records } = recordingFixture();
   return { recorder, records };
 };
 
@@ -26,7 +25,7 @@ test('natural source EOF records the same values and its completion marker', asy
   expect(values[1]).toBe(frames[1]);
   expect(isStreamFact(recorded)).toBe(true);
   expect(isStreamFact(streamReferenceOf(recorded))).toBe(true);
-  await recorder.finish(metadata());
+  await recorder.finish(async () => metadata());
   const events = eventsOf(records[0]!);
   expect(events.filter(event => event.type === 'stream.frame')).toHaveLength(2);
   expect(events.filter(event => event.type === 'stream.end').map(event => event.streamId)).toEqual([1]);
@@ -40,7 +39,7 @@ test('an early return closes the actual source and leaves its recording incomple
   expect((await iterator.next()).value).toBe(frames[0]);
   await iterator.return?.();
   expect(closed).toBe(true);
-  await recorder.finish(metadata());
+  await recorder.finish(async () => metadata());
   const events = eventsOf(records[0]!);
   expect(events.filter(event => event.type === 'stream.frame')).toHaveLength(1);
   expect(events.some(event => event.type === 'stream.end')).toBe(false);
@@ -57,7 +56,7 @@ test('a source failure preserves its original error and omits clean completion',
   expect((await iterator.next()).value).toBe(frames[0]);
   await expect(iterator.next()).rejects.toBe(original);
   expect(closed).toBe(true);
-  await recorder.finish(metadata());
+  await recorder.finish(async () => metadata());
   const events = eventsOf(records[0]!);
   expect(events.filter(event => event.type === 'stream.frame')).toHaveLength(1);
   expect(events.some(event => event.type === 'stream.end')).toBe(false);
@@ -73,7 +72,7 @@ test('projection can skip recording a value while forwarding every source value
   for await (const value of recordStream(source, recorder, value => value.kind === 'heartbeat' ? null : projected)) values.push(value);
   expect(values[0]).toBe(skipped);
   expect(values[1]).toBe(content);
-  await recorder.finish(metadata());
+  await recorder.finish(async () => metadata());
   const events = eventsOf(records[0]!);
   expect(events.filter(event => event.type === 'stream.frame')).toHaveLength(1);
   expect(events.filter(event => event.type === 'stream.end')).toHaveLength(1);
diff --git a/packages/dump/src/recorder.ts b/packages/dump/src/recorder.ts
index 3f77546e71..915fd9d6a9 100644
--- a/packages/dump/src/recorder.ts
+++ b/packages/dump/src/recorder.ts
@@ -1,5 +1,6 @@
-import type { DumpMetadata, StoredDumpRecord } from './types.ts';
-import { createRunEncoder, streamFact, toNdjson, type DumpEvent, type Event, type StreamFact } from '@floway-dev/pipeline';
+import type { DumpMetadata, DumpRunWrite } from './types.ts';
+import { createRunEncoder, streamFact, toNdjson, type Event, type StreamFact } from '@floway-dev/pipeline';
+import { LOG_STREAM_IDLE_MS, type BackgroundScheduler, type LogStream } from '@floway-dev/platform';
 
 export interface StreamRecording {
   readonly fact: StreamFact;
@@ -12,43 +13,141 @@ export interface StreamRecorder {
 }
 
 export interface RecordingPorts {
-  write(record: StoredDumpRecord): Promise<void>;
+  readonly id: string;
+  readonly startedAt: number;
+  write(record: DumpRunWrite): Promise<void>;
   publish(meta: DumpMetadata): Promise<void>;
+  openLive(): Promise<LogStream>;
+  background: BackgroundScheduler;
 }
 
-export const createRunRecorder = ({ write, publish }: RecordingPorts) => {
+const LIVE_APPEND_ATTEMPTS = 3;
+const CHUNK_BYTES = 64 * 1024;
+
+export const runStreamId = (scopeId: string, runId: string): string => `${scopeId}/${runId}`;
+
+export const createRunRecorder = (ports: RecordingPorts) => {
   const encode = createRunEncoder();
-  const events: DumpEvent[] = [];
-  let failure: { readonly error: unknown } | null = null;
+  const metadata = Promise.withResolvers<DumpMetadata>();
+  const bytes = new TransformStream<Uint8Array, Uint8Array>();
+  const writer = bytes.writable.getWriter();
+  let live: LogStream | null = null;
+  let offset = 0;
+  let heartbeat: ReturnType<typeof setTimeout> | undefined;
+  let heartbeatWrite: Promise<void> = Promise.resolve();
+  let closing = false;
+  let tail: Promise<void> = Promise.resolve();
   let streams = 0;
   let answerStream: StreamRecording | undefined;
-  const sink = (event: Event): void => {
-    if (failure !== null) throw failure.error;
-    try {
-      for (const encoded of encode(event)) events.push(encoded);
-    } catch (error) {
-      failure = { error };
-      throw error;
+
+  // One event owns its entire object batch; concurrent deferred outcomes cannot overtake it.
+  const sink = (event: Event): Promise<void> => {
+    const write = tail.then(async () => {
+      const bytes = new TextEncoder().encode(toNdjson(encode(event)));
+      await liveReady;
+      for (let offset = 0; offset < bytes.byteLength; offset += CHUNK_BYTES) {
+        const chunk = bytes.subarray(offset, offset + CHUNK_BYTES);
+        await writer.write(chunk);
+        await appendLive(chunk);
+      }
+    });
+    tail = write;
+    return write;
+  };
+
+  const flush = (): Promise<void> => tail;
+
+  const appendLive = async (bytes: Uint8Array): Promise<void> => {
+    const stream = live;
+    if (stream === null) return;
+    const atOffset = offset;
+    for (let attempt = 0; attempt < LIVE_APPEND_ATTEMPTS; attempt += 1) {
+      try {
+        await stream.append(atOffset, bytes);
+        offset += bytes.byteLength;
+        return;
+      } catch (error) {
+        if (attempt + 1 === LIVE_APPEND_ATTEMPTS) {
+          live = null;
+          console.error('[dump] live stream unavailable; durable recording continues', error);
+        }
+      }
     }
   };
+
+  const scheduleHeartbeat = (): void => {
+    if (closing || live === null) return;
+    heartbeat = setTimeout(() => {
+      // The acknowledged offset is safe during an in-flight data append: an empty write
+      // adds no bytes, and durable backpressure must not block the temporary stream lease.
+      const heartbeat = appendLive(new Uint8Array());
+      heartbeatWrite = heartbeat.then(() => { scheduleHeartbeat(); });
+    }, LOG_STREAM_IDLE_MS / 2);
+  };
+
+  const frame = async (value: unknown): Promise<void> => {
+    answerStream ??= openStream();
+    await answerStream.frame(value);
+  };
+
   const openStream = (): StreamRecording => {
     const streamId = ++streams;
     return {
       fact: streamFact(streamId),
-      frame: frame => { sink({ type: 'stream.frame', streamId, frames: [frame] }); },
-      end: () => { sink({ type: 'stream.end', streamId }); },
+      frame: frame => sink({ type: 'stream.frame', streamId, frames: [frame] }),
+      end: () => sink({ type: 'stream.end', streamId }),
     };
   };
-  const frame = (value: unknown): void | Promise<void> => {
-    answerStream ??= openStream();
-    return answerStream.frame(value);
+
+  const abort = async (error: unknown): Promise<never> => {
+    closing = true;
+    clearTimeout(heartbeat);
+    metadata.reject(error);
+    try { await writer.abort(error); } catch (cleanupError) {
+      if (cleanupError !== error) throw new AggregateError([error, cleanupError], 'Run recording failed and writer abort also failed', { cause: error });
+    }
+    throw error;
   };
-  const finish = async (meta: DumpMetadata): Promise<void> => {
-    if (failure !== null) throw failure.error;
-    await write({ meta, events: new TextEncoder().encode(toNdjson(events)) });
-    await publish(meta);
+
+  const finish = async (metadataFactory: () => Promise<DumpMetadata>): Promise<void> => {
+    try {
+      const meta = await metadataFactory();
+      await flush();
+      metadata.resolve(meta);
+      await writer.close();
+      await stored;
+      await ports.publish(meta);
+      await liveReady;
+      closing = true;
+      clearTimeout(heartbeat);
+      await heartbeatWrite;
+      if (live !== null) {
+        try { await live.end(); } catch (error) { console.error('[dump] live stream end failed', error); }
+      }
+    } catch (error) {
+      await abort(error);
+    } finally {
+      closing = true;
+      clearTimeout(heartbeat);
+    }
   };
-  return { sink, frame, openStream, finish };
+
+  // The metadata wait can reject before the storage reader reaches it; storage owns reporting.
+  void metadata.promise.catch(() => {});
+  const stored = ports.write({
+    id: ports.id, startedAt: ports.startedAt, events: bytes.readable, metadata: metadata.promise,
+  }).catch(error => abort(error));
+  ports.background(stored);
+  const liveReady = ports.openLive().then(stream => {
+    live = stream;
+    scheduleHeartbeat();
+    return stream;
+  }).catch(error => {
+    console.error('[dump] live stream open failed', error);
+    return null;
+  });
+
+  return { sink, flush, frame, openStream, finish };
 };
 
 export type RunRecorder = ReturnType<typeof createRunRecorder>;
diff --git a/packages/dump/src/types.ts b/packages/dump/src/types.ts
index 5ff2339514..6db982b4c4 100644
--- a/packages/dump/src/types.ts
+++ b/packages/dump/src/types.ts
@@ -26,8 +26,15 @@ export interface DumpListOptions {
   limit: number;
 }
 
+export interface DumpRunWrite {
+  readonly id: string;
+  readonly startedAt: number;
+  readonly events: ReadableStream<Uint8Array>;
+  readonly metadata: Promise<DumpMetadata>;
+}
+
 export interface DumpStore {
-  put(keyId: string, record: StoredDumpRecord): Promise<void>;
+  putRun(keyId: string, record: DumpRunWrite): Promise<void>;
   list(keyId: string, options: DumpListOptions): Promise<DumpMetadata[]>;
   get(keyId: string, recordId: DumpRecordId): Promise<StoredDumpRecord | null>;
   deleteExpiredBatch(keyId: string, now: number, limit: number): Promise<number>;
diff --git a/packages/gateway/__tests__/control-plane/dump_test.ts b/packages/gateway/__tests__/control-plane/dump_test.ts
index aa763d1f78..20a3323ce7 100644
--- a/packages/gateway/__tests__/control-plane/dump_test.ts
+++ b/packages/gateway/__tests__/control-plane/dump_test.ts
@@ -1,9 +1,12 @@
-import { test } from 'vitest';
+import { expect, test, vi } from 'vitest';
 
 import { initDumpBroker, initDumpStore } from '../../src/dump/registry.ts';
 import { fakeMeta as baseFakeMeta, fakeRunRecord, installDumpStubs } from '../dump/test-fixtures.ts';
 import { requestApp, setupAppTest } from '../test-utils/app.ts';
+import { testLogStreamStore } from '../test-utils/log-stream.ts';
+import { runStreamId } from '@floway-dev/dump';
 import type { DumpStore, DumpMetadata, DumpRecord, StoredDumpRecord } from '@floway-dev/dump/types';
+import { initLogStreamStore, readLogStream } from '@floway-dev/platform';
 import { assertEquals, assertExists } from '@floway-dev/test-utils';
 
 const fakeMeta = (id: string, completedAt: number): DumpMetadata =>
@@ -254,3 +257,44 @@ test('GET /api/dump/keys/:keyId/stream delivers a frame appended between snapsho
   // it.
   assertEquals((parseSseFrame(appended).data as { id: string }).id, '01HZZ0000000000000000000A2');
 });
+
+test('live record reads are framed, seekable, key-scoped and never create an unknown stream', async () => {
+  const { repo, apiKey } = await setupAppTest();
+  await repo.apiKeys.save({ ...apiKey, dumpRetentionSeconds: 3600 });
+  const store = testLogStreamStore();
+  initLogStreamStore(store);
+  const first = new TextEncoder().encode('{"before":"中"}\n');
+  const second = new TextEncoder().encode('{"after":"β"}\n');
+  const stream = await store.open(runStreamId(apiKey.id, 'active'));
+  await stream.append(0, first); await stream.append(first.byteLength, second); await stream.end();
+  const open = vi.spyOn(store, 'open');
+  const response = await requestApp(`/api/dump/keys/${apiKey.id}/records/active/live?offset=${first.byteLength}`, { headers: { 'x-api-key': apiKey.key } });
+  expect(response.status).toBe(200);
+  const chunks: Uint8Array[] = [];
+  for await (const bytes of readLogStream(response)) chunks.push(bytes);
+  expect(chunks.map(bytes => new TextDecoder().decode(bytes)).join('')).toBe(new TextDecoder().decode(second));
+  const missing = await requestApp(`/api/dump/keys/${apiKey.id}/records/absent/live`, { headers: { 'x-api-key': apiKey.key } });
+  expect(missing.status).toBe(404);
+  expect(open).not.toHaveBeenCalled();
+  const invalid = await requestApp(`/api/dump/keys/${apiKey.id}/records/active/live?offset=-1`, { headers: { 'x-api-key': apiKey.key } });
+  expect(invalid.status).toBe(400);
+  await store.open(runStreamId('foreign-key', 'foreign'));
+  const foreign = await requestApp(`/api/dump/keys/${apiKey.id}/records/foreign/live`, { headers: { 'x-api-key': apiKey.key } });
+  expect(foreign.status).toBe(404);
+});
+
+test('live readers require authentication, key ownership and enabled retention before attaching', async () => {
+  const { repo, apiKey, adminSession } = await setupAppTest();
+  await repo.apiKeys.save({ ...apiKey, dumpRetentionSeconds: 3600 });
+  const store = testLogStreamStore();
+  const stream = await store.open(runStreamId(apiKey.id, 'private'));
+  await stream.end();
+  initLogStreamStore(store);
+  const get = vi.spyOn(store, 'get');
+  const path = `/api/dump/keys/${apiKey.id}/records/private/live`;
+  expect((await requestApp(path, {})).status).toBe(401);
+  expect((await requestApp(path, { headers: { 'x-floway-session': adminSession } })).status).toBe(404);
+  await repo.apiKeys.save({ ...apiKey, dumpRetentionSeconds: null });
+  expect((await requestApp(path, { headers: { 'x-api-key': apiKey.key } })).status).toBe(404);
+  expect(get).not.toHaveBeenCalled();
+});
diff --git a/packages/gateway/__tests__/dump/run-sink_test.ts b/packages/gateway/__tests__/dump/run-sink_test.ts
index 9358b37fc8..dd46028453 100644
--- a/packages/gateway/__tests__/dump/run-sink_test.ts
+++ b/packages/gateway/__tests__/dump/run-sink_test.ts
@@ -1,12 +1,15 @@
-import { expect, test } from 'vitest';
+import { expect, test, vi } from 'vitest';
 
 import { installDumpStubs } from './test-fixtures.ts';
 import { initDumpBroker, initDumpStore } from '../../src/dump/registry.ts';
 import { openRunDump } from '../../src/dump/run-sink.ts';
 import type { ApiKey } from '../../src/repo/types.ts';
 import { flushBackground, trackBackground } from '../test-utils/background-tracker.ts';
+import { testLogStreamStore } from '../test-utils/log-stream.ts';
+import { runStreamId } from '@floway-dev/dump';
 import type { StoredDumpRecord } from '@floway-dev/dump/types';
 import { compose, defineStage, defer, move, run, type DumpEvent, type Event } from '@floway-dev/pipeline';
+import { getLogStreamStore, initLogStreamStore, LOG_STREAM_IDLE_MS } from '@floway-dev/platform';
 import { assertEquals } from '@floway-dev/test-utils';
 
 const apiKey = (dumpRetentionSeconds: number | null): ApiKey => ({
@@ -184,7 +187,120 @@ test('a run reads the stable timing state at completion alongside its stage hist
   assertEquals(lines(record).some(event => event.type === 'stage.entered'), true);
 });
 
+test('live bytes are visible before completion and equal the durable NDJSON artifact', async () => {
+  initLogStreamStore(testLogStreamStore());
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+  await dump.sink({ type: 'stage.entered', stageId: 1, name: 'open', parentStageId: null, facts: move({ input: '中' }) });
+  const live = await getLogStreamStore().get(runStreamId(apiKey(3600).id, dump.id));
+  const reader = live!.read(0, new AbortController().signal)[Symbol.asyncIterator]();
+  const first = await reader.next();
+  expect(new TextDecoder().decode(first.value)).toContain('stage.entered');
+  expect(stubs.stored).toHaveLength(0);
+  dump.finalize(200, 0);
+  await flushBackground();
+  const rest: Uint8Array[] = [first.value!];
+  for (;;) { const item = await reader.next(); if (item.done) break; rest.push(item.value); }
+  expect(rest.map(bytes => new TextDecoder().decode(bytes)).join('')).toBe(ndjson(runRecordOf(stubs.stored[0])));
+  expect(stubs.stored[0]!.record.meta.id).toBe(dump.id);
+});
+
+test('retries a lost live acknowledgement at the same offset without duplicating bytes', async () => {
+  const underlying = testLogStreamStore();
+  const positions: number[] = [];
+  initLogStreamStore({
+    get: id => underlying.get(id),
+    open: async id => {
+      const stream = await underlying.open(id);
+      return {
+        ...stream, append: async (offset, bytes) => {
+          positions.push(offset);
+          await stream.append(offset, bytes);
+          if (positions.length === 1) throw new Error('lost acknowledgement');
+        },
+      };
+    },
+  });
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+  await dump.sink({ type: 'stage.entered', stageId: 1, name: 'retry', parentStageId: null, facts: move({ input: 'x' }) });
+  dump.finalize(200, 0);
+  await flushBackground();
+  expect(positions).toEqual([0, 0]);
+  const stream = await underlying.get(runStreamId(apiKey(3600).id, dump.id));
+  const text: string[] = [];
+  for await (const bytes of stream!.read(0, new AbortController().signal)) text.push(new TextDecoder().decode(bytes));
+  expect(text.join('')).toBe(ndjson(runRecordOf(stubs.stored[0])));
+});
+
+test('a repeatedly failed live stream still stores the complete durable artifact', async () => {
+  const original = new Error('live transport unavailable');
+  const append = vi.fn(async () => { throw original; });
+  initLogStreamStore({
+    open: async () => ({ append, end: async () => {}, read: () => (async function* () {})() }),
+    get: async () => null,
+  });
+  const reported = vi.spyOn(console, 'error').mockImplementation(() => {});
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  try {
+    const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+    const { drain } = await run(pipeline, move({ 'in.text': 'hello' }), { dump: dump.sink });
+    dump.afterRun(drain);
+    dump.finalize(200, 0);
+    await flushBackground();
+    expect(append).toHaveBeenCalledTimes(3);
+    expect(lines(runRecordOf(stubs.stored[0])).filter(event => event.type === 'stage.entered')).toHaveLength(2);
+    expect(reported.mock.calls.some(call => call[1] === original)).toBe(true);
+  } finally { reported.mockRestore(); initLogStreamStore(testLogStreamStore()); }
+});
+
+test('durable backpressure blocks a producer behind an acquired native reader', async () => {
+  const underlying = testLogStreamStore();
+  const append = vi.fn();
+  initLogStreamStore({
+    get: id => underlying.get(id), open: async id => {
+      const stream = await underlying.open(id);
+      return { ...stream, append: async (offset, bytes) => { append(); await stream.append(offset, bytes); } };
+    },
+  });
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  const gate = Promise.withResolvers<void>();
+  const acquired = Promise.withResolvers<void>();
+  initDumpStore({
+    ...stubs.store, putRun: async (keyId, writing) => {
+      const reader = writing.events.getReader();
+      acquired.resolve();
+      await gate.promise;
+      try {
+        await stubs.store.putRun(keyId, {
+          ...writing, events: new ReadableStream({
+            async pull(controller) {
+              const { value, done } = await reader.read();
+              if (done) controller.close(); else controller.enqueue(value);
+            },
+          }),
+        });
+      } finally { reader.releaseLock(); }
+    },
+  });
+  const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+  await acquired.promise;
+  let wrote = false;
+  const write = dump.sink({ type: 'stage.entered', stageId: 1, name: 'blocked', parentStageId: null, facts: move({}) }).then(() => { wrote = true; });
+  await new Promise<void>(resolve => setImmediate(resolve));
+  expect(wrote).toBe(false);
+  expect(append).not.toHaveBeenCalled();
+  gate.resolve();
+  await write;
+  expect(wrote).toBe(true);
+  expect(append).toHaveBeenCalledOnce();
+  dump.finalize(200, 0);
+  await flushBackground();
+  expect(stubs.stored).toHaveLength(1);
+});
+
 test('closing waits for deferred outcomes and persists their settlement event', async () => {
+  initLogStreamStore(testLogStreamStore());
   const stubs = installDumpStubs(initDumpStore, initDumpBroker);
   const result = Promise.withResolvers<string>();
   const deferred = defer(result.promise);
@@ -202,3 +318,195 @@ test('closing waits for deferred outcomes and persists their settlement event',
   expect(lines(runRecordOf(stubs.stored[0])).some(event => event.type === 'deferred.settled')).toBe(true);
   expect(ndjson(runRecordOf(stubs.stored[0]))).toContain('late value');
 });
+
+test('heartbeats renew the acknowledged offset while an event append awaits acknowledgement', async () => {
+  vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] });
+  const underlying = testLogStreamStore();
+  const entered = Promise.withResolvers<void>();
+  const release = Promise.withResolvers<void>();
+  const positions: { offset: number; length: number }[] = [];
+  initLogStreamStore({
+    get: id => underlying.get(id),
+    open: async id => {
+      const stream = await underlying.open(id);
+      return {
+        ...stream, append: async (offset, bytes) => {
+          positions.push({ offset, length: bytes.byteLength });
+          if (positions.length === 1) { entered.resolve(); await release.promise; }
+          await stream.append(offset, bytes);
+        },
+      };
+    },
+  });
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  try {
+    const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+    const event = dump.sink({ type: 'stage.entered', stageId: 1, name: 'slow append', parentStageId: null, facts: move({}) });
+    await entered.promise;
+    await vi.advanceTimersByTimeAsync(LOG_STREAM_IDLE_MS / 2);
+    expect(positions).toEqual([{ offset: 0, length: positions[0]!.length }, { offset: 0, length: 0 }]);
+    release.resolve();
+    await event;
+    await vi.advanceTimersByTimeAsync(LOG_STREAM_IDLE_MS / 2);
+    dump.finalize(200, 0);
+    await flushBackground();
+    expect(positions).toEqual([
+      { offset: 0, length: positions[0]!.length }, { offset: 0, length: 0 },
+      { offset: positions[0]!.length, length: 0 },
+    ]);
+    expect(stubs.stored).toHaveLength(1);
+  } finally { release.resolve(); vi.useRealTimers(); initLogStreamStore(testLogStreamStore()); }
+});
+
+test('an encoding failure aborts the durable reader instead of leaving it waiting for EOF', async () => {
+  initLogStreamStore(testLogStreamStore());
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  const rejected = Promise.withResolvers<unknown>();
+  initDumpStore({
+    ...stubs.store, putRun: async (_keyId, writing) => {
+      try { await new Response(writing.events).arrayBuffer(); } catch (error) { rejected.resolve(error); throw error; }
+      await writing.metadata;
+    },
+  });
+  const work: Promise<unknown>[] = [];
+  const dump = openRunDump(apiKey(3600), turn, pending => { work.push(pending); }, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+  const error = await dump.sink({ type: 'stage.entered', stageId: 1, name: 'invalid', parentStageId: null, facts: move({ upload: new Blob(['data']) }) }).catch(error => error as Error);
+  expect(error).toBeInstanceOf(TypeError);
+  expect((error as Error).message).toContain('Blob');
+  dump.finalize(500, 0);
+  expect(await rejected.promise).toBe(error);
+  const outcomes = await Promise.allSettled(work);
+  expect(outcomes).toHaveLength(2);
+  expect(outcomes.every(outcome => outcome.status === 'rejected' && outcome.reason === error)).toBe(true);
+  expect(stubs.stored).toEqual([]);
+});
+
+test('an early durable failure stops heartbeats even when the live stream opens later', async () => {
+  vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] });
+  const original = new Error('durable store unavailable');
+  const opening = Promise.withResolvers<void>();
+  const underlying = testLogStreamStore();
+  initLogStreamStore({ get: id => underlying.get(id), open: async id => { await opening.promise; return await underlying.open(id); } });
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  initDumpStore({ ...stubs.store, putRun: async () => { throw original; } });
+  const work: Promise<unknown>[] = [];
+  try {
+    const dump = openRunDump(apiKey(3600), turn, pending => { work.push(pending); }, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+    expect(await Promise.allSettled(work)).toEqual([{ status: 'rejected', reason: original }]);
+    opening.resolve();
+    await getLogStreamStore().open('test-open-barrier');
+    expect(vi.getTimerCount()).toBe(0);
+    await expect(dump.sink({ type: 'stage.entered', stageId: 1, name: 'failed storage', parentStageId: null, facts: move({}) })).rejects.toBe(original);
+  } finally { opening.resolve(); vi.clearAllTimers(); vi.useRealTimers(); initLogStreamStore(testLogStreamStore()); }
+});
+
+test('live EOF waits until the durable artifact has been published', async () => {
+  const underlying = testLogStreamStore();
+  const end = vi.fn();
+  initLogStreamStore({
+    get: id => underlying.get(id), open: async id => {
+      const stream = await underlying.open(id);
+      return { ...stream, end: async () => { end(); await stream.end(); } };
+    },
+  });
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  const consumed = Promise.withResolvers<void>();
+  const publish = Promise.withResolvers<void>();
+  initDumpStore({
+    ...stubs.store, putRun: async (keyId, writing) => {
+      const bytes = new Uint8Array(await new Response(writing.events).arrayBuffer());
+      await writing.metadata;
+      consumed.resolve();
+      await publish.promise;
+      await stubs.store.putRun(keyId, {
+        ...writing, events: new ReadableStream({ start(controller) { controller.enqueue(bytes); controller.close(); } }),
+      });
+    },
+  });
+  const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+  await dump.sink({ type: 'stage.entered', stageId: 1, name: 'publication', parentStageId: null, facts: move({}) });
+  dump.finalize(200, 0);
+  await consumed.promise;
+  expect(stubs.stored).toHaveLength(0);
+  expect(end).not.toHaveBeenCalled();
+  publish.resolve();
+  await flushBackground();
+  expect(stubs.stored).toHaveLength(1);
+  expect(end).toHaveBeenCalledOnce();
+});
+
+test('a durable publication failure never ends the live stream and preserves its original error', async () => {
+  const underlying = testLogStreamStore();
+  const end = vi.fn();
+  initLogStreamStore({
+    get: id => underlying.get(id), open: async id => {
+      const stream = await underlying.open(id);
+      return { ...stream, end: async () => { end(); await stream.end(); } };
+    },
+  });
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  const consumed = Promise.withResolvers<void>();
+  const publish = Promise.withResolvers<void>();
+  const original = new Error('durable publication failed');
+  initDumpStore({
+    ...stubs.store, putRun: async (_keyId, writing) => {
+      await new Response(writing.events).arrayBuffer();
+      consumed.resolve();
+      await publish.promise;
+      await writing.metadata;
+    },
+  });
+  const work: Promise<unknown>[] = [];
+  const dump = openRunDump(apiKey(3600), turn, pending => { work.push(pending); }, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+  await dump.sink({ type: 'stage.entered', stageId: 1, name: 'publication failure', parentStageId: null, facts: move({}) });
+  dump.finalize(200, 0);
+  await consumed.promise;
+  publish.reject(original);
+  const outcomes = await Promise.allSettled(work);
+  expect(outcomes).toHaveLength(2);
+  expect(outcomes.every(outcome => outcome.status === 'rejected' && outcome.reason === original)).toBe(true);
+  expect(stubs.stored).toHaveLength(0);
+  expect(stubs.published).toHaveLength(0);
+  expect(end).not.toHaveBeenCalled();
+});
+
+test('heartbeats remain active through a long durable reader backpressure gate', async () => {
+  vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] });
+  const underlying = testLogStreamStore();
+  const positions: { offset: number; length: number }[] = [];
+  initLogStreamStore({
+    get: id => underlying.get(id), open: async id => {
+      const stream = await underlying.open(id);
+      return { ...stream, append: async (offset, bytes) => { positions.push({ offset, length: bytes.byteLength }); await stream.append(offset, bytes); } };
+    },
+  });
+  const stubs = installDumpStubs(initDumpStore, initDumpBroker);
+  const acquired = Promise.withResolvers<void>();
+  const release = Promise.withResolvers<void>();
+  initDumpStore({
+    ...stubs.store, putRun: async (_keyId, writing) => {
+      const reader = writing.events.getReader();
+      acquired.resolve();
+      await release.promise;
+      try { while (!(await reader.read()).done) {} } finally { reader.releaseLock(); }
+      await writing.metadata;
+    },
+  });
+  try {
+    const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!;
+    await acquired.promise;
+    let wrote = false;
+    const writing = dump.sink({ type: 'stage.entered', stageId: 1, name: 'long backpressure', parentStageId: null, facts: move({}) }).then(() => { wrote = true; });
+    await new Promise<void>(resolve => setImmediate(resolve));
+    expect(wrote).toBe(false);
+    await vi.advanceTimersByTimeAsync(LOG_STREAM_IDLE_MS * 2);
+    expect(wrote).toBe(false);
+    expect(positions).toEqual(Array.from({ length: 4 }, () => ({ offset: 0, length: 0 })));
+    release.resolve();
+    await writing;
+    expect(positions[4]!.length).toBeGreaterThan(0);
+    dump.finalize(200, 0);
+    await flushBackground();
+    expect(await underlying.get(runStreamId(apiKey(3600).id, dump.id))).not.toBeNull();
+  } finally { release.resolve(); vi.useRealTimers(); initLogStreamStore(testLogStreamStore()); }
+});
diff --git a/packages/gateway/__tests__/dump/test-fixtures.ts b/packages/gateway/__tests__/dump/test-fixtures.ts
index d365b4e1c0..7699f4e3e2 100644
--- a/packages/gateway/__tests__/dump/test-fixtures.ts
+++ b/packages/gateway/__tests__/dump/test-fixtures.ts
@@ -38,7 +38,7 @@ export const eventsOf = (record: StoredDumpRecord): readonly Record<string, unkn
     .map(line => JSON.parse(line) as Record<string, unknown>);
 
 type DumpStubFailMethod =
-  | 'put'
+  | 'putRun'
   | 'list'
   | 'get'
   | 'publish'
@@ -66,9 +66,11 @@ export const installDumpStubs = (
   const throws: Partial<Record<DumpStubFailMethod, Error>> = {};
 
   const store: DumpStore = {
-    async put(keyId, record) {
-      if (throws.put) throw throws.put;
-      const storedRecord: StoredDumpRecord = record;
+    async putRun(keyId, writing) {
+      const events = new Uint8Array(await new Response(writing.events).arrayBuffer());
+      const meta = await writing.metadata;
+      if (throws.putRun) throw throws.putRun;
+      const storedRecord: StoredDumpRecord = { meta, events };
       stored.push({ keyId, record: storedRecord });
       const list = records.get(keyId) ?? [];
       list.unshift(storedRecord);
diff --git a/packages/gateway/__tests__/dump/write-run.ts b/packages/gateway/__tests__/dump/write-run.ts
new file mode 100644
index 0000000000..b5f30ec43d
--- /dev/null
+++ b/packages/gateway/__tests__/dump/write-run.ts
@@ -0,0 +1,8 @@
+import type { DumpStore, StoredDumpRecord } from '@floway-dev/dump/types';
+
+export const writeRun = (store: DumpStore, keyId: string, record: StoredDumpRecord): Promise<void> => store.putRun(keyId, {
+  id: record.meta.id,
+  startedAt: record.meta.startedAt,
+  events: new ReadableStream<Uint8Array>({ start(controller) { controller.enqueue(record.events); controller.close(); } }),
+  metadata: Promise.resolve(record.meta),
+});
diff --git a/packages/gateway/__tests__/repo/dump-store_test.ts b/packages/gateway/__tests__/repo/dump-store_test.ts
index bcf3a00317..f62c13104b 100644
--- a/packages/gateway/__tests__/repo/dump-store_test.ts
+++ b/packages/gateway/__tests__/repo/dump-store_test.ts
@@ -2,14 +2,16 @@ import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { dirname, join, resolve } from 'node:path';
 
-import { expect, test } from 'vitest';
+import { expect, test, vi } from 'vitest';
 
 import { createSqliteTestDb, mapRunChangeCount } from './test-sqlite.ts';
 import { decodeDumpBodyDescriptor } from '../../src/dump/storage-codec.ts';
 import { FileDumpStore } from '../../src/repo/dump-store.ts';
 import { initRepo } from '../../src/repo/index.ts';
+import { SPILLED_FILE_STAGE_GRACE_MS } from '../../src/repo/spilled-files-policy.ts';
 import { SqlRepo } from '../../src/repo/sql.ts';
 import { collectSpilledFiles } from '../../src/scheduled/spilled-files.ts';
+import { writeRun } from '../dump/write-run.ts';
 import type { StoredDumpRecord } from '@floway-dev/dump/types';
 import { encodeRun, toNdjson, type Facts } from '@floway-dev/pipeline';
 import { initFileStore, MemoryFileStore } from '@floway-dev/platform';
@@ -59,7 +61,7 @@ test('FileDumpStore round-trips a run record as its NDJSON event stream', async
   const store = new FileDumpStore(db, files);
   const record = runRecord('01HZZ00000000000000000RUN1', Date.UTC(2026, 5, 1, 12, 0, 0));
 
-  await store.put('key_x', record);
+  await writeRun(store, 'key_x', record);
   const fetched = await store.get('key_x', record.meta.id);
   assertExists(fetched);
   assertEquals(new TextDecoder().decode(fetched.events), new TextDecoder().decode(record.events));
@@ -80,8 +82,8 @@ test('FileDumpStore lists what a row renders without opening the stream', async
   const db = await openDb();
   const store = new FileDumpStore(db, new MemoryFileStore());
   const base = Date.UTC(2026, 5, 1, 12, 0, 0);
-  await store.put('key_x', runRecord('01HZZ00000000000000000RUN1', base));
-  await store.put('key_x', runRecord('01HZZ00000000000000000RUN2', base + 1));
+  await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN1', base));
+  await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN2', base + 1));
 
   const listed = await store.list('key_x', { limit: 10 });
   assertEquals(listed.map(meta => meta.id), ['01HZZ00000000000000000RUN2', '01HZZ00000000000000000RUN1']);
@@ -98,8 +100,8 @@ test('FileDumpStore retires an expired run record and collects its stream file',
   initFileStore(files);
   const store = new FileDumpStore(db, files);
   const now = Date.UTC(2026, 5, 1, 12, 0, 0);
-  await store.put('key_x', runRecord('01HZZ00000000000000000RUN3', Date.UTC(2026, 5, 1, 9, 0, 0)));
-  await store.put('key_x', runRecord('01HZZ00000000000000000RUN4', now));
+  await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN3', Date.UTC(2026, 5, 1, 9, 0, 0)));
+  await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN4', now));
   await repo.apiKeys.update('key_x', { dumpRetentionSeconds: 2 * 3600 });
 
   const originalNow = Date.now;
@@ -124,7 +126,7 @@ test('FileDumpStore rejects malformed metadata with its row identity', async ()
   const db = await openDb();
   const store = new FileDumpStore(db, new MemoryFileStore());
   const record = runRecord('01HZZ000000000000000000BADM', Date.UTC(2026, 5, 1, 12));
-  await store.put('key_x', record);
+  await writeRun(store, 'key_x', record);
   await db.prepare('UPDATE dump_records SET meta_json = ? WHERE key_id = ? AND id = ?')
     .bind(JSON.stringify({ ...record.meta, upstream: undefined, status: '200' }), 'key_x', record.meta.id)
     .run();
@@ -137,7 +139,7 @@ test('FileDumpStore rejects a malformed body descriptor before file access', asy
   const db = await openDb();
   const store = new FileDumpStore(db, new MemoryFileStore());
   const record = runRecord('01HZZ000000000000000000BADD', Date.UTC(2026, 5, 1, 12));
-  await store.put('key_x', record);
+  await writeRun(store, 'key_x', record);
   await db.prepare('UPDATE dump_records SET response_body_descriptor = ? WHERE key_id = ? AND id = ?')
     .bind(JSON.stringify({ key: 'dumps/v1/key_x/run.gz', type: 'chunks' }), 'key_x', record.meta.id)
     .run();
@@ -153,7 +155,7 @@ test('FileDumpStore refuses a row that points at no run stream', async () => {
   const db = await openDb();
   const store = new FileDumpStore(db, new MemoryFileStore());
   const record = runRecord('01HZZ000000000000000000NORUN', Date.UTC(2026, 5, 1, 12));
-  await store.put('key_x', record);
+  await writeRun(store, 'key_x', record);
   await db.prepare('UPDATE dump_records SET response_body_descriptor = ? WHERE key_id = ? AND id = ?')
     .bind(JSON.stringify({ key: 'dumps/v1/key_x/run.gz', type: 'bytes' }), 'key_x', record.meta.id)
     .run();
@@ -168,7 +170,7 @@ test('FileDumpStore.list paginates newest-first with the (createdAt, id) cursor'
   const store = new FileDumpStore(db, files);
   const base = Date.UTC(2026, 5, 1, 12, 0, 0);
   for (let i = 0; i < 5; i++) {
-    await store.put('key_x', runRecord(`01HZZ000000000000000000A0${i}`, base + i));
+    await writeRun(store, 'key_x', runRecord(`01HZZ000000000000000000A0${i}`, base + i));
   }
   const first = await store.list('key_x', { limit: 2 });
   assertEquals(first.map(m => m.id), ['01HZZ000000000000000000A04', '01HZZ000000000000000000A03']);
@@ -185,8 +187,8 @@ test('FileDumpStore applies retention immediately and retires exact expired file
   const store = new FileDumpStore(db, files);
   const now = Date.UTC(2026, 5, 1, 12, 0, 0);
   // Old bucket 9:xx, current bucket 12:xx.
-  await store.put('key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0)));
-  await store.put('key_x', runRecord('01HZZ0000000000000000000A2', now));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0)));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A2', now));
   await repo.apiKeys.update('key_x', { dumpRetentionSeconds: 2 * 3600 });
   const originalNow = Date.now;
   Date.now = () => now + 1;
@@ -215,7 +217,7 @@ test('growing dump retention can reveal a row not yet physically deleted', async
   const files = new MemoryFileStore();
   const store = new FileDumpStore(db, files);
   const now = Date.UTC(2026, 5, 1, 12);
-  await store.put('key_x', runRecord('01HZZ0000000000000000000A4', now - 3 * 3600_000));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A4', now - 3 * 3600_000));
   const originalNow = Date.now;
   Date.now = () => now;
   try {
@@ -235,8 +237,8 @@ test('FileDumpStore retires every dump record when retention is disabled and col
   const files = new MemoryFileStore();
   initFileStore(files);
   const store = new FileDumpStore(db, files);
-  await store.put('key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0)));
-  await store.put('key_x', runRecord('01HZZ0000000000000000000A2', Date.UTC(2026, 5, 1, 12, 0, 0)));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0)));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A2', Date.UTC(2026, 5, 1, 12, 0, 0)));
   await repo.apiKeys.update('key_x', { dumpRetentionSeconds: null });
   assertEquals((await store.list('key_x', { limit: 10 })).length, 0);
   assertEquals(await store.deleteExpiredBatch('key_x', Date.now(), 100), 2);
@@ -250,8 +252,8 @@ test('FileDumpStore counts returned dump rows instead of trigger-amplified chang
   const repo = new SqlRepo(db);
   const files = new MemoryFileStore();
   const store = new FileDumpStore(db, files);
-  await store.put('key_x', runRecord('01HZZ0000000000000000000C1', Date.UTC(2026, 5, 1, 9)));
-  await store.put('key_x', runRecord('01HZZ0000000000000000000C2', Date.UTC(2026, 5, 1, 10)));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C1', Date.UTC(2026, 5, 1, 9)));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C2', Date.UTC(2026, 5, 1, 10)));
   await repo.apiKeys.update('key_x', { dumpRetentionSeconds: null });
 
   const d1LikeStore = new FileDumpStore(mapRunChangeCount(db, changes => changes * 3), files);
@@ -265,8 +267,8 @@ test('FileDumpStore counts returned active dump rows instead of trigger-amplifie
   const files = new MemoryFileStore();
   const store = new FileDumpStore(db, files);
   const now = Date.UTC(2026, 5, 1, 12);
-  await store.put('key_x', runRecord('01HZZ0000000000000000000C3', Date.UTC(2026, 5, 1, 9)));
-  await store.put('key_x', runRecord('01HZZ0000000000000000000C4', Date.UTC(2026, 5, 1, 10)));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C3', Date.UTC(2026, 5, 1, 9)));
+  await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C4', Date.UTC(2026, 5, 1, 10)));
   await repo.apiKeys.update('key_x', { dumpRetentionSeconds: 3600 });
 
   const d1LikeStore = new FileDumpStore(mapRunChangeCount(db, changes => changes * 3), files);
@@ -282,11 +284,10 @@ test('a record-ID race leaves only the losing write\'s uniquely keyed files coll
   initFileStore(files);
   const store = new FileDumpStore(db, files);
   const record = runRecord('01HZZ0000000000000000000A3', Date.now());
-  await store.put('key_x', record);
+  await writeRun(store, 'key_x', record);
 
-  // One file per record now — the run's own stream — so the winner owns one and the loser
-  // staged one under a key of its own.
-  await expect(store.put('key_x', record)).rejects.toThrow();
+  // The winner owns its artifact; the losing write leaves a distinct staged file.
+  await expect(writeRun(store, 'key_x', record)).rejects.toThrow();
   expect((await db.prepare("SELECT COUNT(*) AS count FROM spilled_files WHERE state = 'owned'").first<{ count: number }>())?.count).toBe(1);
   expect((await db.prepare("SELECT COUNT(*) AS count FROM spilled_files WHERE state = 'staged'").first<{ count: number }>())?.count).toBe(1);
 
@@ -334,14 +335,14 @@ class TmpDirFileStore implements FileStore {
   }
 }
 
-test('FileDumpStore: put + get round-trips through real-filesystem IO', async () => {
+test('FileDumpStore: streaming write + get round-trips through real-filesystem IO', async () => {
   const root = await mkdtemp(join(tmpdir(), 'dump-store-'));
   try {
     const db = await openDb();
     const store = new FileDumpStore(db, new TmpDirFileStore(join(root, 'files')));
     const record = runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 12, 0, 0));
 
-    await store.put('key_x', record);
+    await writeRun(store, 'key_x', record);
     const fetched = await store.get('key_x', '01HZZ0000000000000000000A1');
     assertExists(fetched);
     assertEquals(new TextDecoder().decode(fetched.events), new TextDecoder().decode(record.events));
@@ -358,10 +359,57 @@ test('FileDumpStore filters all retained history before applying the page limit'
     const record = runRecord(`filter-${i}`, now - i);
     record.meta.model = i === 4 ? 'needle-model' : 'other';
     record.meta.error = i === 4 ? { kind: 'failed', reason: 'socket reset' } : null;
-    await store.put('key_x', record);
+    await writeRun(store, 'key_x', record);
   }
   expect((await store.list('key_x', { q: 'needle', limit: 1 })).map(meta => meta.id)).toEqual(['filter-4']);
   expect((await store.list('key_x', { q: 'SOCKET', failures: true, limit: 1 })).map(meta => meta.id)).toEqual(['filter-4']);
   expect(await store.list('key_x', { q: "' OR 1=1 --", limit: 1 })).toEqual([]);
   expect(await store.list('other-key', { q: 'needle', limit: 1 })).toEqual([]);
 });
+
+test('FileDumpStore writes streaming run bytes before publishing its metadata row', async () => {
+  const db = await openDb();
+  const files = new MemoryFileStore();
+  const store = new FileDumpStore(db, files);
+  const metadata = Promise.withResolvers<StoredDumpRecord['meta']>();
+  const pipe = new TransformStream<Uint8Array>();
+  const writer = pipe.writable.getWriter();
+  const id = 'streamed-run';
+  const startedAt = Date.now();
+  const writing = store.putRun('key_x', { id, startedAt, events: pipe.readable, metadata: metadata.promise });
+  const bytes = new TextEncoder().encode('{"type":"stage.entered","facts":{"text":"中"}}\n');
+  await writer.write(bytes);
+  expect(await store.get('key_x', id)).toBeNull();
+  await writer.close();
+  metadata.resolve({ ...runRecord(id, Date.now()).meta, startedAt });
+  await writing;
+  const stored = await store.get('key_x', id);
+  expect(stored!.meta.id).toBe(id);
+  expect(stored!.events).toEqual(bytes);
+});
+
+test('a long active run keeps its staged file out of orphan collection', async () => {
+  const db = await openDb();
+  const files = new MemoryFileStore();
+  initRepo(new SqlRepo(db));
+  initFileStore(files);
+  const store = new FileDumpStore(db, files);
+  const metadata = Promise.withResolvers<StoredDumpRecord['meta']>();
+  const pipe = new TransformStream<Uint8Array>();
+  const writer = pipe.writable.getWriter();
+  const id = 'long-active-run';
+  const startedAt = Date.now();
+  vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] });
+  try {
+    const writing = store.putRun('key_x', { id, startedAt, events: pipe.readable, metadata: metadata.promise });
+    await writer.write(new TextEncoder().encode('{"type":"stage.entered"}\n'));
+    await vi.advanceTimersByTimeAsync(SPILLED_FILE_STAGE_GRACE_MS * 2 + 1);
+    await collectSpilledFiles(Date.now());
+    await writer.close();
+    metadata.resolve({ ...runRecord(id, Date.now()).meta, startedAt });
+    await writing;
+    const stored = await store.get('key_x', id);
+    expect(stored!.meta.id).toBe(id);
+    expect(new TextDecoder().decode(stored!.events)).toBe('{"type":"stage.entered"}\n');
+  } finally { vi.useRealTimers(); }
+});
diff --git a/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts b/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts
index 9a4c32f975..5a5393e294 100644
--- a/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts
+++ b/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts
@@ -7,6 +7,7 @@ import { quantizeOpenAIResponsesRefreshedAt, OPENAI_RESPONSES_REFRESH_GRANULARIT
 import { SqlRepo } from '../../src/repo/sql.ts';
 import type { ApiKey, StoredOpenAIResponsesItem } from '../../src/repo/types.ts';
 import { sweepExpirations } from '../../src/scheduled/expiration-sweeps.ts';
+import { writeRun } from '../dump/write-run.ts';
 import { InMemoryRepo } from '../repo/memory.ts';
 import { createSqliteTestDb, createSqlJsDatabase, migrationSqlByFilename, wrapSqlJsDatabase } from '../repo/test-sqlite.ts';
 import type { StoredDumpRecord } from '@floway-dev/dump/types';
@@ -80,9 +81,9 @@ test('one fair driver drains bounded OpenAI Responses and dump backlogs', async
   );
   await repo.openaiResponsesItems.insertMany([responseItem('msg-current', now)], 0);
   for (let index = 0; index < 150; index += 1) {
-    await dumps.put('key-a', dumpRecord(`01K00000000000000000${String(index).padStart(4, '0')}`, dumpExpiredAt));
+    await writeRun(dumps, 'key-a', dumpRecord(`01K00000000000000000${String(index).padStart(4, '0')}`, dumpExpiredAt));
   }
-  await dumps.put('key-a', dumpRecord('01K00000000000000000LIVE', now));
+  await writeRun(dumps, 'key-a', dumpRecord('01K00000000000000000LIVE', now));
   await repo.apiKeys.update('key-a', { dumpRetentionSeconds: 3600, openaiResponsesRetentionSeconds: OPENAI_RESPONSES_RETENTION_SECONDS });
 
   await sweepExpirations(now);
@@ -114,14 +115,14 @@ test('minute ticks catch up with a growing hot dump key without widening a batch
   await repo.apiKeys.save(key(start));
 
   for (let index = 0; index < 150; index += 1) {
-    await dumps.put('key-a', dumpRecord(`01K00000000000000001${String(index).padStart(4, '0')}`, start - 3600_001));
+    await writeRun(dumps, 'key-a', dumpRecord(`01K00000000000000001${String(index).padStart(4, '0')}`, start - 3600_001));
   }
 
   for (let tick = 0; tick < 4; tick += 1) {
     const now = start + tick * 60_000;
     vi.setSystemTime(now);
     for (let index = 0; index < 11; index += 1) {
-      await dumps.put('key-a', dumpRecord(`01K00000000000000002${tick}${String(index).padStart(3, '0')}`, now - 3600_001));
+      await writeRun(dumps, 'key-a', dumpRecord(`01K00000000000000002${tick}${String(index).padStart(3, '0')}`, now - 3600_001));
     }
     await sweepExpirations(now);
     const expected = Math.max(0, 150 + 11 * (tick + 1) - 50 * (tick + 1));
diff --git a/packages/gateway/__tests__/scheduled_test.ts b/packages/gateway/__tests__/scheduled_test.ts
index 5f206ae70b..f49366d4b4 100644
--- a/packages/gateway/__tests__/scheduled_test.ts
+++ b/packages/gateway/__tests__/scheduled_test.ts
@@ -1,5 +1,6 @@
 import { afterEach, expect, test, vi } from 'vitest';
 
+import { writeRun } from './dump/write-run.ts';
 import { createSqliteTestDb } from './repo/test-sqlite.ts';
 import { setupAppTest } from './test-utils/app.ts';
 import { initDumpStore } from '../src/dump/registry.ts';
@@ -127,7 +128,7 @@ test('one maintenance tick collects every file retired by its four dump units',
     const keyId = `key-${keyIndex}`;
     await repo.apiKeys.save(apiKey(keyId, now, keyIndex + 1));
     for (let rowIndex = 0; rowIndex < 50; rowIndex += 1) {
-      await dumps.put(keyId, fileBackedDumpRecord(`dump-${keyIndex}-${rowIndex}`, now - 3600_001));
+      await writeRun(dumps, keyId, fileBackedDumpRecord(`dump-${keyIndex}-${rowIndex}`, now - 3600_001));
     }
   }
   const { results: ownedFiles } = await db.prepare('SELECT file_key FROM spilled_files ORDER BY file_key')
diff --git a/packages/gateway/__tests__/test-utils/log-stream.ts b/packages/gateway/__tests__/test-utils/log-stream.ts
new file mode 100644
index 0000000000..f48629313e
--- /dev/null
+++ b/packages/gateway/__tests__/test-utils/log-stream.ts
@@ -0,0 +1,43 @@
+import type { LogStream, LogStreamStore } from '@floway-dev/platform';
+
+export const testLogStreamStore = (): LogStreamStore => {
+  const streams = new Map<string, LogStream>();
+  return {
+    async open(id) {
+      const chunks: { offset: number; bytes: Uint8Array }[] = [];
+      let length = 0;
+      let ended = false;
+      const listeners = new Set<() => void>();
+      const wake = () => { for (const listener of listeners) listener(); };
+      const stream: LogStream = {
+        async append(offset, bytes) {
+          if (ended) throw new Error('Test LogStream is ended');
+          if (offset > length) throw new Error('Test LogStream append would leave a hole');
+          const tail = bytes.slice(length - offset);
+          if (tail.byteLength > 0) { chunks.push({ offset: length, bytes: tail }); length += tail.byteLength; }
+          wake();
+        },
+        async end() { ended = true; wake(); },
+        async *read(offset, signal) {
+          while (!signal.aborted) {
+            const chunk = chunks.find(chunk => chunk.offset + chunk.bytes.byteLength > offset);
+            if (chunk) {
+              const bytes = chunk.bytes.slice(Math.max(0, offset - chunk.offset));
+              offset += bytes.byteLength;
+              yield bytes;
+            } else if (ended) return;
+            else await new Promise<void>(resolve => {
+              const finish = () => { listeners.delete(finish); signal.removeEventListener('abort', finish); resolve(); };
+              listeners.add(finish);
+              signal.addEventListener('abort', finish, { once: true });
+            });
+          }
+          signal.throwIfAborted();
+        },
+      };
+      streams.set(id, stream);
+      return stream;
+    },
+    async get(id) { return streams.get(id) ?? null; },
+  };
+};
diff --git a/packages/gateway/__tests__/vitest.setup.ts b/packages/gateway/__tests__/vitest.setup.ts
index 363aafbc2b..3d7bc1ee61 100644
--- a/packages/gateway/__tests__/vitest.setup.ts
+++ b/packages/gateway/__tests__/vitest.setup.ts
@@ -1,12 +1,12 @@
 import { trackBackground } from './test-utils/background-tracker.ts';
+import { testLogStreamStore } from './test-utils/log-stream.ts';
 import { initDumpBroker, initDumpStore } from '../src/dump/registry.ts';
 import { handleExecutionRequest } from '../src/execution/handler.ts';
 import { initBackgroundSchedulerResolver } from '../src/runtime/background.ts';
 import { initExecutionCellNamespace } from '../src/runtime/execution.ts';
-import type { DumpBroker } from '@floway-dev/dump';
-import type { StoredDumpRecord, DumpStore, DumpMetadata, DumpRecordId } from '@floway-dev/dump/types';
+import type { DumpMetadata, StoredDumpRecord, DumpRecordId, DumpStore, DumpBroker } from '@floway-dev/dump/types';
 import { isReplayableBody } from '@floway-dev/http';
-import { initEnv, initFetch, initRuntimeKind, initTimingSafeEqual, InProcessExecutionCellNamespace } from '@floway-dev/platform';
+import { initEnv, initFetch, initRuntimeKind, initTimingSafeEqual, InProcessExecutionCellNamespace, initLogStreamStore } from '@floway-dev/platform';
 
 // Production always initializes the environment getter at boot. Mirror that
 // here with a neutral default; tests needing real values (RUNTIME_LOCATION,
@@ -15,6 +15,7 @@ initEnv(() => '');
 // Tests run as 'node' by default. The few tests that exercise CF-specific
 // runtime behaviour re-init this with 'cloudflare'.
 initRuntimeKind('node');
+initLogStreamStore(testLogStreamStore());
 initTimingSafeEqual((a, b) => a.every((byte, index) => byte === b[index]));
 initFetch((url, init) => {
   const body = init.body;
@@ -32,7 +33,7 @@ initExecutionCellNamespace(new InProcessExecutionCellNamespace(handleExecutionRe
 // independent of that subsystem. Dump-specific tests install real or recording
 // implementations.
 const noopStore: DumpStore = {
-  async put(): Promise<void> {},
+  async putRun(_keyId, run) { await new Response(run.events).arrayBuffer(); await run.metadata; },
   async list(): Promise<DumpMetadata[]> { return []; },
   async get(_keyId: string, _id: DumpRecordId): Promise<StoredDumpRecord | null> { return null; },
   async deleteExpiredBatch(): Promise<number> { return 0; },
diff --git a/packages/gateway/src/control-plane/dump.ts b/packages/gateway/src/control-plane/dump.ts
index 7f7d84ebd9..1ef90fe1f3 100644
--- a/packages/gateway/src/control-plane/dump.ts
+++ b/packages/gateway/src/control-plane/dump.ts
@@ -6,7 +6,8 @@ import { z } from 'zod';
 import { ownedKeyForUser } from './shared/owned-key.ts';
 import { getDumpBroker, getDumpStore } from '../dump/registry.ts';
 import { zValidator } from '../middleware/zod-validator.ts';
-import { dumpRecordToWire } from '@floway-dev/dump';
+import { runStreamId, dumpRecordToWire } from '@floway-dev/dump';
+import { getLogStreamStore, serveLogStream } from '@floway-dev/platform';
 
 const LIST_LIMIT_DEFAULT = 100;
 const LIST_LIMIT_MAX = 200;
@@ -23,7 +24,7 @@ const ownedKey = async (c: Context): Promise<{ id: string; error: null } | { id:
   const owned = await ownedKeyForUser(c, keyId);
   if (!owned) return { id: null, error: 'Key not found' };
   if (owned.dumpRetentionSeconds === null) {
-    return { id: null, error: 'Dump capture is not enabled for this key.' };
+    return { id: null, error: 'Run recording is not enabled for this key.' };
   }
   return { id: owned.id, error: null };
 };
@@ -40,6 +41,15 @@ export const dumpRoutes = new Hono()
     });
     return c.json({ records });
   })
+  .get('/keys/:keyId/records/:recordId/live', zValidator('query', z.object({
+    offset: z.coerce.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER).optional(),
+  })), async c => {
+    const owned = await ownedKey(c);
+    if (owned.id === null) return c.json({ error: owned.error }, 404);
+    const stream = await getLogStreamStore().get(runStreamId(owned.id, c.req.param('recordId')!));
+    if (stream === null) return c.json({ error: 'Live recording is unavailable' }, 404);
+    return serveLogStream(stream, c.req.valid('query').offset ?? 0, c.req.raw.signal);
+  })
   .get('/keys/:keyId/records/:recordId', async c => {
     const owned = await ownedKey(c);
     if (owned.id === null) return c.json({ error: owned.error }, 404);
diff --git a/packages/gateway/src/dump/run-sink.ts b/packages/gateway/src/dump/run-sink.ts
index 67f46c6690..628250aca5 100644
--- a/packages/gateway/src/dump/run-sink.ts
+++ b/packages/gateway/src/dump/run-sink.ts
@@ -4,8 +4,8 @@ import { attemptTtftMs, type AttemptTiming } from '../data-plane/shared/attempt-
 import type { RequestBody } from '../data-plane/shared/request-body.ts';
 import type { ApiKey, TokenUsage } from '../repo/types.ts';
 import { ulid } from '../shared/ulid.ts';
-import { createRunRecorder, type DumpMetadata, type RunRecorder, type StreamRecording } from '@floway-dev/dump';
-import type { BackgroundScheduler } from '@floway-dev/platform';
+import { createRunRecorder, runStreamId, type RunRecorder, type StreamRecording } from '@floway-dev/dump';
+import { getLogStreamStore, type BackgroundScheduler } from '@floway-dev/platform';
 import type { ProtocolFrame } from '@floway-dev/protocols/common';
 import type { TelemetryModelIdentity } from '@floway-dev/provider';
 
@@ -18,6 +18,7 @@ interface RequestSnapshot {
 }
 
 export class RunDump {
+  readonly id: string;
   private readonly attribution = new DumpAttribution();
   private readonly recorder: RunRecorder;
   readonly sink: RunRecorder['sink'];
@@ -32,9 +33,13 @@ export class RunDump {
     private readonly wantsStream: boolean,
     private readonly timing: AttemptTiming,
   ) {
+    this.id = ulid(startedAt);
     this.recorder = createRunRecorder({
-      write: record => getDumpStore().put(apiKey.id, record),
+      id: this.id, startedAt,
+      write: record => getDumpStore().putRun(apiKey.id, record),
       publish: meta => getDumpBroker().publish(apiKey.id, meta),
+      openLive: () => getLogStreamStore().open(runStreamId(apiKey.id, this.id)),
+      background: backgroundScheduler,
     });
     this.sink = this.recorder.sink;
   }
@@ -120,25 +125,20 @@ export class RunDump {
   }
 
   private async write(status: number | null, responseBytes: number, responseStreamError: string | null): Promise<void> {
-    if (this.runDrain !== null) {
-      try { await this.runDrain(); } catch (error) { this.failed(error); }
-    }
-    const completedAt = Date.now();
-    const recordId = ulid(completedAt);
-    const meta: DumpMetadata = await this.attribution.metadata({
-      id: recordId,
-      startedAt: this.startedAt,
-      completedAt,
-      method: this.requestSnapshot.method,
-      path: this.requestSnapshot.path,
-      status,
-      requestBytes: this.requestSnapshot.bodyByteLength,
-      responseBytes,
-      ttftMs: this.wantsStream ? attemptTtftMs(this.timing) : null,
-      fallbackError: streamReadError(this.requestSnapshot.streamError, responseStreamError),
+    await this.recorder.finish(async () => {
+      if (this.runDrain !== null) {
+        try { await this.runDrain(); } catch (error) { this.failed(error); }
+      }
+      await this.recorder.flush();
+      const completedAt = Date.now();
+      return await this.attribution.metadata({
+        id: this.id, startedAt: this.startedAt, completedAt,
+        method: this.requestSnapshot.method, path: this.requestSnapshot.path, status,
+        requestBytes: this.requestSnapshot.bodyByteLength, responseBytes,
+        ttftMs: this.wantsStream ? attemptTtftMs(this.timing) : null,
+        fallbackError: streamReadError(this.requestSnapshot.streamError, responseStreamError),
+      });
     });
-
-    await this.recorder.finish(meta);
   }
 }
 
diff --git a/packages/gateway/src/repo/dump-store.ts b/packages/gateway/src/repo/dump-store.ts
index ab9d66f9ad..74c1e0d574 100644
--- a/packages/gateway/src/repo/dump-store.ts
+++ b/packages/gateway/src/repo/dump-store.ts
@@ -7,9 +7,9 @@ import {
   encodePersistedDumpMetadata,
 } from '../dump/storage-codec.ts';
 import type { DumpBodyDescriptor } from '../dump/storage-codec.ts';
-import { gunzipBytes, gzipBytes } from '../shared/gzip.ts';
+import { gunzipBytes, gzipStream } from '../shared/gzip.ts';
 import type {
-  DumpListOptions, DumpStore,
+  DumpListOptions, DumpStore, DumpRunWrite,
   DumpMetadata,
   DumpRecordId,
   DumpUpstreamRef,
@@ -75,23 +75,43 @@ const EMPTY_HEADERS_JSON = '[]';
 export class FileDumpStore implements DumpStore {
   constructor(private readonly db: SqlDatabase, private readonly files: FileStore) {}
 
-  async put(keyId: string, record: StoredDumpRecord): Promise<void> {
-    const { meta, events } = record;
-    const fileKey = bodyPath(keyId, hourBucket(meta.completedAt), meta.id);
+  async putRun(keyId: string, run: DumpRunWrite): Promise<void> {
+    const fileKey = bodyPath(keyId, hourBucket(run.startedAt), run.id);
     await this.db.prepare(
       `INSERT INTO spilled_files (file_key, owner_kind, owner_key, state, collect_after)
        VALUES (?, 'dump-response', ?, 'staged', ?)`,
-    ).bind(fileKey, JSON.stringify([keyId, meta.id]), Date.now() + SPILLED_FILE_STAGE_GRACE_MS).run();
-    await this.files.put(fileKey, await gzipBytes(events));
-    await this.db.prepare(
-      `INSERT INTO dump_records
-       (key_id, id, created_at, upstream_id, meta_json, request_headers_json, response_headers_json, request_body_descriptor, response_body_descriptor)
-       VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?)`,
-    ).bind(
-      keyId, meta.id, meta.completedAt, meta.upstream?.id ?? null,
-      encodePersistedDumpMetadata(meta, `dump record ${meta.id} metadata`), EMPTY_HEADERS_JSON,
-      encodeDumpBodyDescriptor({ key: fileKey, type: 'run' }, `dump record ${meta.id} run descriptor`),
-    ).run();
+    ).bind(fileKey, JSON.stringify([keyId, run.id]), Date.now() + SPILLED_FILE_STAGE_GRACE_MS).run();
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    let finished = false;
+    let renewal: Promise<void> = Promise.resolve();
+    const stopRenewing = () => { finished = true; clearTimeout(timer); };
+    const scheduleRenewal = () => {
+      timer = setTimeout(() => {
+        renewal = this.db.prepare(
+          "UPDATE spilled_files SET collect_after = ? WHERE file_key = ? AND state = 'staged'",
+        ).bind(Date.now() + SPILLED_FILE_STAGE_GRACE_MS, fileKey).run().then(() => {
+          if (!finished) scheduleRenewal();
+        });
+        // Awaited before publication even if the lease write failed while upstream was quiet.
+        void renewal.catch(() => {});
+      }, SPILLED_FILE_STAGE_GRACE_MS / 2);
+    };
+    scheduleRenewal();
+    try {
+      await this.files.put(fileKey, gzipStream(run.events));
+      const meta = await run.metadata;
+      stopRenewing();
+      await renewal;
+      await this.db.prepare(
+        `INSERT INTO dump_records
+         (key_id, id, created_at, upstream_id, meta_json, request_headers_json, response_headers_json, request_body_descriptor, response_body_descriptor, response_upstream_body_descriptor)
+         VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?, NULL)`,
+      ).bind(
+        keyId, meta.id, meta.completedAt, meta.upstream?.id ?? null,
+        encodePersistedDumpMetadata(meta, `dump record ${meta.id} metadata`), EMPTY_HEADERS_JSON,
+        encodeDumpBodyDescriptor({ key: fileKey, type: 'run' }, `dump record ${meta.id} run descriptor`),
+      ).run();
+    } finally { stopRenewing(); }
   }
 
   async list(keyId: string, opts: DumpListOptions): Promise<DumpMetadata[]> {
diff --git a/packages/gateway/src/shared/gzip.ts b/packages/gateway/src/shared/gzip.ts
index 74bda3d0e6..77a1f8241b 100644
--- a/packages/gateway/src/shared/gzip.ts
+++ b/packages/gateway/src/shared/gzip.ts
@@ -34,3 +34,6 @@ export const gzipBytes = async (bytes: Uint8Array): Promise<Uint8Array> =>
 
 export const gunzipBytes = async (bytes: Uint8Array): Promise<Uint8Array> =>
   await collect(bytesStream(bytes).pipeThrough(new DecompressionStream('gzip')));
+
+export const gzipStream = (stream: ReadableStream<Uint8Array>): ReadableStream<Uint8Array> =>
+  (stream as ReadableStream<BufferSource>).pipeThrough(new CompressionStream('gzip'));
diff --git a/packages/platform/README.md b/packages/platform/README.md
index 829472636e..509298c14a 100644
--- a/packages/platform/README.md
+++ b/packages/platform/README.md
@@ -4,12 +4,30 @@ This package defines portable runtime services. Implementations live in the
 platform applications; runtime composition installs them through the exported
 initializers.
 
-## Stored files
+## Durable file writes
 
-`FileStore.put(key, bytes)` stores a complete `Uint8Array`. `get` returns the
-stored bytes or null, and `deleteKeys` removes exact keys. Node uses the
-configured filesystem directory, Cloudflare uses R2, and the in-memory
-implementation stores byte copies.
+`FileStore.put(key, body)` accepts bytes or a `ReadableStream<Uint8Array>`.
+It consumes the stream with storage backpressure and publishes the complete
+object at EOF. Until then, readers see the previous value or a missing key.
+Failures preserve the prior object and expose the original error; cleanup
+failures retain it in the error chain. `get` still returns complete bytes.
+
+Node stages each write inside the target parent’s reserved `.floway-staging`
+namespace and atomically renames it after closing. This preserves the published
+key paths and the target filesystem when a key prefix is a separate mount.
+Node keys cannot contain that private segment in any ASCII case spelling,
+including components of the normalized filesystem path. `deleteKeys` removes the exact
+key’s unfinished writes as well as its published file, so retention cleanup
+also collects a writer killed before EOF. Public keys resembling UUID temporary
+filenames remain ordinary keys. Empty staging directories stay available to
+concurrent writers until key deletion; the common private directory stays in
+place like other parent directories.
+
+Cloudflare uses bounded 5 MiB multipart parts for unknown-length
+streams, since R2's single put requires a known length. Short and empty streams
+use a single byte put. A storage/source failure cancels the source and cleans
+up unfinished multipart state; live-view degradation does not affect this path.
+The in-memory implementation collects bytes as part of its storage semantics.
 
 ## Transient run streams
 
@@ -27,8 +45,8 @@ A `LogStream` contains bytes, independent of any event format:
   the stream ended; cancellation or expiry interrupts the read.
 
 `LOG_STREAM_IDLE_MS` is the shared idle lease. Append activity, including an
-empty heartbeat, and read progress renew it. To keep a quiet active stream open, its writer sends empty
-appends while work is pending; a completed or abandoned stream is
+empty heartbeat, and read progress renew it. A quiet active writer sends empty
+appends while upstream work is pending; a completed or abandoned stream is
 reclaimed after idle activity stops. Existing handles then raise
 `LogStreamExpiredError`. Expiration affects transient live viewing; the durable
 artifact is a separate writer-owned path.
diff --git a/packages/platform/__tests__/file-store_test.ts b/packages/platform/__tests__/file-store_test.ts
index f94fd61073..9d93c8011c 100644
--- a/packages/platform/__tests__/file-store_test.ts
+++ b/packages/platform/__tests__/file-store_test.ts
@@ -1,4 +1,4 @@
-import { test } from 'vitest';
+import { expect, test } from 'vitest';
 
 import { getFileStore, initFileStore, MemoryFileStore } from '../src/file-store.ts';
 import { assertEquals } from '@floway-dev/test-utils';
@@ -35,3 +35,16 @@ test('MemoryFileStore deletes exact keys without treating them as prefixes', asy
   assertEquals(await store.get('drop/a'), null);
   assertEquals(await store.get('drop/ab'), new Uint8Array([2]));
 });
+
+test('MemoryFileStore accepts streamed bytes and publishes only at completion', async () => {
+  const store = new MemoryFileStore();
+  const source = new TransformStream<Uint8Array, Uint8Array>();
+  const writing = store.put('stream', source.readable);
+  const writer = source.writable.getWriter();
+  await writer.write(new Uint8Array([1, 2]));
+  expect(await store.get('stream')).toBeNull();
+  await writer.write(new Uint8Array([3]));
+  await writer.close();
+  await writing;
+  expect(await store.get('stream')).toEqual(new Uint8Array([1, 2, 3]));
+});
diff --git a/packages/platform/src/file-store.ts b/packages/platform/src/file-store.ts
index 13bb33c2ce..ff4a45d995 100644
--- a/packages/platform/src/file-store.ts
+++ b/packages/platform/src/file-store.ts
@@ -1,5 +1,8 @@
+export type FileBody = Uint8Array | ReadableStream<Uint8Array>;
+
 export interface FileStore {
-  put(key: string, body: Uint8Array): Promise<void>;
+  /** Consumes the body and publishes the complete object atomically at EOF. */
+  put(key: string, body: FileBody): Promise<void>;
   get(key: string): Promise<Uint8Array | null>;
   deleteKeys(keys: readonly string[]): Promise<void>;
 }
@@ -18,8 +21,9 @@ export const getFileStore = (): FileStore => {
 export class MemoryFileStore implements FileStore {
   private readonly files = new Map<string, Uint8Array>();
 
-  async put(key: string, body: Uint8Array): Promise<void> {
-    this.files.set(key, body.slice());
+  async put(key: string, body: FileBody): Promise<void> {
+    const bytes = body instanceof Uint8Array ? body.slice() : new Uint8Array(await new Response(body).arrayBuffer());
+    this.files.set(key, bytes);
   }
 
   async get(key: string): Promise<Uint8Array | null> {