diff --git a/docs/3.0/PROGRESS.md b/docs/3.0/PROGRESS.md
index 23bc3838..59dfe775 100644
--- a/docs/3.0/PROGRESS.md
+++ b/docs/3.0/PROGRESS.md
@@ -1,5 +1,10 @@
 # PROGRESS · Clawket 3.0 进度日志
 
+- 2026-10-16 Native Runtime 消费驱动事件缓冲（独立修复；RN 保持默认）。
+  - 保留公开 AsyncThrowingStream，内部单一64项/16MiB有界队列随真实next出队；仅同内部代际且UTF8精确路由/segment/lineage、无snapshot metadata的相邻纯delta合并，其他事件保持barrier。原始事件weight全部计费，容量溢出仍同步退休、不丢已准入正文、不提高限额、不自动重放；取消/finish单次通知并唤醒reader。
+  - 两个独立SPM文件覆盖慢consumer千片Unicode/空白、真实Hermes/Pi mapper→Coordinator→公开stream、tool/work/snapshot/final顺序、Stop不等UI消费、严格身份/代际隔离、65个非合并项、字节限/超限、取消/finish/owner释放与pending写退休无重放。首编strict Swift6/jobs2 39.21秒，新buffer9方法/10输入与真实runtime5方法首轮全绿；既有SessionLifecycle5方法/11输入、RuntimeOperations6方法/12输入逐文件全绿，16指令对/7docs与diff通过。strict Swift6/iOS16.4双arch实际App开发编译/codesign验证也已通过，私有dev副本交主线程，heavy已释放；不属发行包或设备验收。独立只读review与完整CI并行。
+  - 现场仅实采event_stream catch与随后activate断点；Runtime唯一throwing finish源码是capacity，但Swift表达式受stale PCM阻挡，未实采code或数量/字节细分，不将本切片冒称已确定现场全部根因。真实Hermes本轮最终final不等于Stop通过；跨工具aggregate final范围仍独立待处理。容量故障后的Host事件loop仅showFailure、已展示state可能仍ready，本轮不扩Host恢复或宣称全部capacity恢复完成。无RN、服务、设备审批、版本或发布变更。
+
 - 2026-10-16 Native 真实 Hermes Stop 定位与复验（独立探索；RN 保持默认）。
   - #274 exactca7f9393/base76bc45b9在current-main/CLEAN、全13 CI绿后合8afa5784并finish；App314/314零失败/跳过（57.765秒test/80.504秒wall）、package459、19语两表9082精确编译值。primary既有node_modules保留，自有开发副本/证据保存在SSD，不发行。工作入口/菜单实际验收边界见下条，不以CI替代真实后端。
   - 新任务从fresh main独立工作树开始，仅用既有owner QA route与baseline physical/首条自有输入比对。原生registry只在QA独立Keychain，未扫Vault或按剩余Hermes行猜observer身份；未启动原observer或覆写旧结果。独立loopback probe前后各定向history，先确认无活动，单次send，等真实非空delta，再核samephysical/inFlightRun精确归属后唯一次Stop。20秒绝对RPC预算、总145秒、事件只保留本QA的128KiB/256条紧凑类别/布尔，未知不重试，凭据只内存、不输出URL/正文。
diff --git a/docs/native-ios/RUNTIME.md b/docs/native-ios/RUNTIME.md
index 3b37dad7..8bde948c 100644
--- a/docs/native-ios/RUNTIME.md
+++ b/docs/native-ios/RUNTIME.md
@@ -28,6 +28,8 @@ History reads for one session serialize. OpenClaw's physical transcript offset c
 
 `RuntimeHistory.sessionID` is optional actual top-level `sessionId`, independent of the logical route and retained thinking level. OpenClaw's typed `transcriptPosition` preserves actual source/raw sequence; `transcriptBoundary` accepts only system rows with exact structured reset/compaction kind. They do not change wire order/clocks or infer that a reset occurred from prose. Participant attribution follows [the current participant contract](../3.0/25-message-participants.md): explicit structured authors first, then bounded native sender/transport metadata and recognized channel-session fallback. Web/client transport overrides that fallback. Channel, account and sender ID remain separate from backend/transport; names, owner claims and mentions never establish the viewer. Remote avatars accept bounded credential-free public HTTPS URLs, never relative Gateway paths, private/local addresses or ambiguous legacy numeric host forms. This parser performs no avatar fetch or directory lookup.
 
+The public `AsyncThrowingStream<RuntimeEvent, any Error>` pulls from one bounded queue: at most 64 entries and 16 MiB of original event weights. Consumption releases the exact retained weight. Adjacent pure text deltas may share one entry only with identical internal activation/socket/authentication revisions and byte-exact connection/session/run/segment/optional lineage; snapshot metadata must be absent. Every control, work, tool, snapshot, final and revision change is a boundary. Concatenation preserves whitespace and original UTF-8 without granting message identity. No timer, per-token task, UI wait or second prefetch queue is used. Nonmergeable overflow and byte overflow still fail closed and retire the connection; accepted events drain in order before the capacity error. Reader cancellation discards pending presentation work and retires once.
+
 Stream text declares delta versus snapshot and a stable assistant segment identifier. The Hermes Bridge forwards each `message.delta` fragment as `chat.state=delta`, with only that fragment in `message.content`; native maps it to `.delta`, preserving whitespace and original Unicode bytes. Those fragments bypass cumulative silent-prefix filtering: `NO` followed by `RMAL` must retain `NORMAL`, and a split sentinel-like word does not grant silent-reply semantics. OpenClaw's cumulative chat frames remain `.snapshot`. A full chat final remains a snapshot for either backend, including a final without preceding deltas. Fragments never carry `snapshotMetadata`, even if a peer supplies a message ID. Full-final, canonical SDK and subagent filtering is unchanged. Observed tool starts create a new segment; final-only or corrected final replies remain visible without replacing earlier segments. Stop acknowledgements do not invent completion. Canonical work also exposes commentary item snapshots, tool inputs/outputs and status, producer clocks, turn/input lineage, compaction, activity, usage and system metadata. Raw application data never becomes diagnostics.
 
 Complete `chat.message` snapshots also retain optional `snapshotMetadata`, using the same actual ID/role/lineage parser as finals and history. It is absent for missing IDs, true delta fragments, subagent text and canonical SDK aggregates, even when an envelope/item/tool ID exists. Conflicting explicit run/turn/input lineage remains marked. `RuntimeHistory.activeRunEvidence` separates actual producer evidence from the compatibility `hasActiveRun = false` default. An actual boolean is evidence unless contradictory or malformed fields are present; a bounded, correctly scoped actual `activeRun`/`inFlightRun.runId` also preserves the existing positive contract when the boolean is absent. Missing/malformed activity and legacy decoded caches remain unknown, never terminal proof. `activeRunID` exposes only that validated actual run ID. These fields add no backend capability or wire change.
diff --git a/packages/apple/AGENTS.md b/packages/apple/AGENTS.md
index e35e7a0e..e4dc8c94 100644
--- a/packages/apple/AGENTS.md
+++ b/packages/apple/AGENTS.md
@@ -32,7 +32,7 @@ Consent milliseconds may be fractional (Hermes forwards finite producer deadline
 
 Session management uses nonpersistent current-catalog references, consumed after local validation and revalidated before dispatch. Intersect the declared/negotiated capability with each actual row's permission; SDK native copy/continue does not grant reset/delete. Keep backend-specific source/active-run rules, original project inheritance, archive separate from deletion and unknown writes held without replay. An ACK retires catalog work started while its write was pending, so post-ACK recovery starts a fresh read. Old receipts and history must not mutate a replacement session. See `../../docs/native-ios/MANAGEMENT.md` for the source-backed action and remaining feature inventory.
 
-SDK catalogs are complete frozen snapshots: 64 KiB pages, 10,000 sessions/8 MiB total, at most 512 page reads/90 seconds and one expiry restart. Do not advertise delta/page-index support. Runtime events are a single-reader stream bounded to 64 events/16 MiB. Keep assistant segment identity across tool boundaries and distinguish snapshot text from deltas. Canonical work and its typed history projection preserve producer clocks/status, commentary item identity, turn/input lineage and raw review inputs/outputs. Emit the existing run-start event plus typed start lineage; normalize Gateway terminal tool phases at the adapter boundary. Optional malformed history items do not suppress a usable transcript; exact history-only identities never become live authority. Completion metadata stays a typed accessor on the existing final event; unknown status never proves execution. Long native model selection may use a 190-second RPC; the transport request ceiling is 300 seconds, separate from open/handshake limits.
+SDK catalogs are complete frozen snapshots: 64 KiB pages, 10,000 sessions/8 MiB total, at most 512 page reads/90 seconds and one expiry restart. Do not advertise delta/page-index support. Runtime events are a consumer-driven single-reader stream bounded to 64 queued entries/16 MiB of original event weights. Only adjacent pure deltas with no snapshot metadata and byte-exact route/segment/lineage plus identical internal connection revisions may coalesce; every other event is a barrier. Never wait for UI consumption inside the coordinator, add a second queue, or count merged fragments only once. Capacity failure retires synchronously; cancellation wakes the sole reader and retires once. Keep assistant segment identity across tool boundaries and distinguish snapshot text from deltas. Canonical work and its typed history projection preserve producer clocks/status, commentary item identity, turn/input lineage and raw review inputs/outputs. Emit the existing run-start event plus typed start lineage; normalize Gateway terminal tool phases at the adapter boundary. Optional malformed history items do not suppress a usable transcript; exact history-only identities never become live authority. Completion metadata stays a typed accessor on the existing final event; unknown status never proves execution. Long native model selection may use a 190-second RPC; the transport request ceiling is 300 seconds, separate from open/handshake limits.
 
 SDK history tool rows use the shared nested `tool` object (`name`, `callId`, status, input/output and clocks); Gateway top-level tool results remain compatible. `tests/native-contract/sdk-history.mjs --write|--check` generates synthetic vectors through the actual Bridge producers with Node 22.18 or later. Its source-drift and corrupted-input tests belong to the required gate; vectors are not device/inference evidence.
 The host may supply an accepted persistent display message ID to history work projection. Use it only for fallback display anchors/items; keep actual tool/run/turn/input lineage unchanged and preserve the existing API default for other consumers.
diff --git a/packages/apple/Sources/ClawketRuntime/RuntimeCoordinator.swift b/packages/apple/Sources/ClawketRuntime/RuntimeCoordinator.swift
index 44ba33db..c5275c9b 100644
--- a/packages/apple/Sources/ClawketRuntime/RuntimeCoordinator.swift
+++ b/packages/apple/Sources/ClawketRuntime/RuntimeCoordinator.swift
@@ -6,7 +6,7 @@ import ClawketTransport
 public actor RuntimeCoordinator {
     var historyExportPending = 0
     public nonisolated let events: AsyncThrowingStream<RuntimeEvent, any Error>
-    let eventContinuation: AsyncThrowingStream<RuntimeEvent, any Error>.Continuation
+    let eventBuffer: RuntimeEventBuffer
     let credentials: any DeviceCredentialProvider
     let socketFactory: any WebSocketFactory
     let clock: any TransportClock
@@ -59,7 +59,6 @@ public actor RuntimeCoordinator {
     var historyPending = 0
     var controlWaiter: ControlWaiter?
     var controlDispatches: [String: Task<Void, Never>] = [:]
-    var eventWeights: [Int] = []
     var streamAvailable = true
     public private(set) var state = RuntimeConnectionState()
 
@@ -85,9 +84,10 @@ public actor RuntimeCoordinator {
               ["iphone", "ipad"].contains(client.deviceFamily) else { throw RuntimeFailure(.invalidConnection) }
         self.credentials = credentials; self.socketFactory = socketFactory; self.clock = clock; self.client = client
         self.artifactHTTP = artifactHTTP
-        let stream = AsyncThrowingStream<RuntimeEvent, any Error>.makeStream(bufferingPolicy: .bufferingOldest(64))
-        events = stream.stream; eventContinuation = stream.continuation
-        eventContinuation.onTermination = { @Sendable [weak self] _ in Task { await self?.disconnect() } }
+        let buffer = RuntimeEventBuffer()
+        eventBuffer = buffer
+        events = AsyncThrowingStream(unfolding: { try await buffer.next() })
+        buffer.onTermination { @Sendable [weak self] in Task { await self?.eventStreamTerminated() } }
     }
 
     /// Waits for verified backend readiness. Repeated calls join the active connection.
@@ -430,20 +430,22 @@ public actor RuntimeCoordinator {
 
     func emit(_ value: RuntimeEvent, weight: Int = 128) {
         guard streamAvailable else { return }
-        eventWeights.append(weight)
-        let result = eventContinuation.yield(value)
-        if case .enqueued(let remaining) = result {
-            let buffered = max(0, 64 - remaining)
-            if eventWeights.count > buffered { eventWeights.removeFirst(eventWeights.count - buffered) }
-            if eventWeights.reduce(0, +) <= 16 * 1_024 * 1_024 { return }
-        }
-        streamAvailable = false; stopped = true
-        let old = retire()
-        eventContinuation.finish(throwing: RuntimeFailure(.capacity))
+        let context = RuntimeEventBuffer.Context(epoch: epoch, attempt: attempt, authentication: authentication, generation: generation)
+        let result = eventBuffer.offer(value, weight: weight, context: context)
+        if case .accepted = result { return }
+        eventStreamTerminated()
+    }
+
+    private func eventStreamTerminated() {
+        guard streamAvailable else { return }
+        streamAvailable = false
+        let old = retire(); stopped = true; connection = nil
+        publishState(.idle)
         Task { await old?.disconnect() }
     }
 
     deinit {
+        eventBuffer.finish()
         readerTask?.cancel(); openTask?.cancel(); handshakeTask?.cancel(); retryTask?.cancel()
         periodicProbeTask?.cancel(); probeTask?.cancel(); catalogTask?.cancel(); controlWaiter?.timeout.cancel()
         consentExpiryTask?.cancel()
@@ -451,6 +453,5 @@ public actor RuntimeCoordinator {
         controlWaiter?.continuation.resume(throwing: RuntimeFailure(.superseded))
         for waiter in waiters.values { waiter.resume(throwing: RuntimeFailure(.superseded)) }
         let transport = transport; Task { await transport?.disconnect() }
-        eventContinuation.finish()
     }
 }
diff --git a/packages/apple/Sources/ClawketRuntime/RuntimeEventBuffer.swift b/packages/apple/Sources/ClawketRuntime/RuntimeEventBuffer.swift
new file mode 100644
index 00000000..3647e051
--- /dev/null
+++ b/packages/apple/Sources/ClawketRuntime/RuntimeEventBuffer.swift
@@ -0,0 +1,179 @@
+import Foundation
+import ClawketTransport
+
+/// One queue, pulled by the public stream's single consumer. The lock protects
+/// short synchronous admission only; the coordinator never awaits that consumer.
+final class RuntimeEventBuffer: @unchecked Sendable {
+    static let maximumCount = 64
+    static let maximumBytes = 16 * 1_024 * 1_024
+
+    struct Context: Equatable, Sendable {
+        let epoch: UUID
+        let attempt: UUID
+        let authentication: UUID
+        let generation: ConnectionGeneration?
+    }
+    enum Admission: Equatable { case accepted, overflow, terminated }
+    private enum End { case finished, failure(RuntimeFailure), cancelled }
+    private struct Entry {
+        let event: RuntimeEvent
+        let context: Context
+        var weight: Int
+        var text: String?
+
+        init(_ event: RuntimeEvent, context: Context, weight: Int) {
+            self.context = context; self.weight = weight
+            if case let .update(connection, .text(update)) = event,
+               update.mode == .delta, update.snapshotMetadata == nil {
+                // Keep an empty template, so appending does not copy the entire
+                // accumulated String on every incoming fragment.
+                self.event = .update(connectionID: connection, .text(Self.replacingText(update, with: "")))
+                text = update.text
+            } else { self.event = event }
+        }
+
+        func accepts(_ event: RuntimeEvent, context: Context) -> Bool {
+            guard self.context == context, text != nil,
+                  case let .update(leftConnection, .text(left)) = self.event,
+                  case let .update(rightConnection, .text(right)) = event,
+                  right.mode == .delta, right.snapshotMetadata == nil else { return false }
+            return Self.same(leftConnection, rightConnection) && Self.same(left.sessionKey, right.sessionKey)
+                && Self.same(left.runID, right.runID) && Self.same(left.segmentID, right.segmentID)
+                && Self.same(left.turnID, right.turnID) && Self.same(left.inputMessageID, right.inputMessageID)
+                && Self.same(left.inputMessageKey, right.inputMessageKey)
+        }
+
+        func value() -> RuntimeEvent {
+            guard let text, case let .update(connection, .text(update)) = event else { return event }
+            return .update(connectionID: connection, .text(Self.replacingText(update, with: text)))
+        }
+
+        private static func replacingText(_ value: RuntimeTextUpdate, with text: String) -> RuntimeTextUpdate {
+            .init(sessionKey: value.sessionKey, runID: value.runID, segmentID: value.segmentID, text: text, mode: value.mode,
+                  turnID: value.turnID, inputMessageID: value.inputMessageID, inputMessageKey: value.inputMessageKey,
+                  snapshotMetadata: value.snapshotMetadata)
+        }
+        private static func same(_ left: String?, _ right: String?) -> Bool {
+            switch (left, right) {
+            case (nil, nil): true
+            case let (left?, right?): left.utf8.elementsEqual(right.utf8)
+            default: false
+            }
+        }
+    }
+
+    private let lock = NSLock()
+    private var entries: [Entry] = []
+    private var bytes = 0
+    private var waiter: CheckedContinuation<RuntimeEvent?, any Error>?
+    private var end: End?
+    private var terminationHandler: (@Sendable () -> Void)?
+    private var handlerInstalled = false
+    private var terminationDelivered = false
+
+    func onTermination(_ handler: @escaping @Sendable () -> Void) {
+        lock.lock()
+        precondition(!handlerInstalled)
+        handlerInstalled = true; terminationHandler = handler
+        let callback = takeTermination()
+        lock.unlock()
+        callback?()
+    }
+
+    func offer(_ event: RuntimeEvent, weight: Int, context: Context) -> Admission {
+        lock.lock()
+        guard end == nil else { lock.unlock(); return .terminated }
+        let merges = entries.last?.accepts(event, context: context) == true
+        guard weight >= 0, weight <= Self.maximumBytes - bytes,
+              merges || entries.count < Self.maximumCount else {
+            end = .failure(RuntimeFailure(.capacity))
+            let pending = waiter; waiter = nil
+            let callback = takeTermination()
+            lock.unlock()
+            pending?.resume(throwing: RuntimeFailure(.capacity)); callback?()
+            return .overflow
+        }
+        if let pending = waiter {
+            waiter = nil
+            lock.unlock()
+            pending.resume(returning: event)
+            return .accepted
+        }
+        if merges, case let .update(_, .text(update)) = event {
+            entries[entries.count - 1].text?.append(update.text)
+            entries[entries.count - 1].weight += weight
+        } else { entries.append(Entry(event, context: context, weight: weight)) }
+        bytes += weight
+        lock.unlock()
+        return .accepted
+    }
+
+    func next() async throws -> RuntimeEvent? {
+        try await withTaskCancellationHandler {
+            if Task.isCancelled { cancel(); throw CancellationError() }
+            return try await withCheckedThrowingContinuation { pull($0) }
+        } onCancel: { self.cancel() }
+    }
+
+    private func pull(_ continuation: CheckedContinuation<RuntimeEvent?, any Error>) {
+        lock.lock()
+        if !entries.isEmpty {
+            let entry = entries.removeFirst(); bytes -= entry.weight
+            lock.unlock()
+            continuation.resume(returning: entry.value())
+        } else if let end {
+            lock.unlock()
+            Self.resume(continuation, end: end)
+        } else if waiter == nil {
+            waiter = continuation; lock.unlock()
+        } else {
+            // The public contract has one reader. Fail closed rather than
+            // losing a second continuation or permitting concurrent reordering.
+            end = .failure(RuntimeFailure(.capacity))
+            let pending = waiter; waiter = nil
+            let callback = takeTermination()
+            lock.unlock()
+            pending?.resume(throwing: RuntimeFailure(.capacity))
+            continuation.resume(throwing: RuntimeFailure(.capacity)); callback?()
+        }
+    }
+
+    func finish(throwing failure: RuntimeFailure? = nil) {
+        lock.lock()
+        guard end == nil else { lock.unlock(); return }
+        let result: End = failure.map(End.failure) ?? .finished
+        end = result
+        let pending = waiter; waiter = nil
+        let callback = takeTermination()
+        lock.unlock()
+        // Accepted entries remain in order and drain before the terminal result.
+        if let pending { Self.resume(pending, end: result) }
+        callback?()
+    }
+
+    private func cancel() {
+        lock.lock()
+        end = .cancelled; entries.removeAll(); bytes = 0
+        let pending = waiter; waiter = nil
+        let callback = takeTermination()
+        lock.unlock()
+        pending?.resume(throwing: CancellationError()); callback?()
+    }
+
+    /// Called only with the lock held. Callbacks themselves always run unlocked.
+    private func takeTermination() -> (@Sendable () -> Void)? {
+        guard end != nil, !terminationDelivered, let handler = terminationHandler else { return nil }
+        terminationDelivered = true; terminationHandler = nil
+        return handler
+    }
+
+    private static func resume(_ continuation: CheckedContinuation<RuntimeEvent?, any Error>, end: End) {
+        switch end {
+        case .finished: continuation.resume(returning: nil)
+        case .failure(let failure): continuation.resume(throwing: failure)
+        case .cancelled: continuation.resume(throwing: CancellationError())
+        }
+    }
+
+    deinit { waiter?.resume(returning: nil) }
+}
diff --git a/packages/apple/Tests/ClawketRuntimeTests/RuntimeEventBufferTests.swift b/packages/apple/Tests/ClawketRuntimeTests/RuntimeEventBufferTests.swift
new file mode 100644
index 00000000..495524bf
--- /dev/null
+++ b/packages/apple/Tests/ClawketRuntimeTests/RuntimeEventBufferTests.swift
@@ -0,0 +1,168 @@
+import ClawketTransport
+import Foundation
+import Testing
+@testable import ClawketRuntime
+
+private final class Terminations: @unchecked Sendable {
+    private let lock = NSLock()
+    private var count = 0
+    func record() { lock.lock(); count += 1; lock.unlock() }
+    var value: Int { lock.lock(); defer { lock.unlock() }; return count }
+}
+
+@Suite(.serialized)
+struct RuntimeEventBufferTests {
+    private let context = RuntimeEventBuffer.Context(epoch: UUID(), attempt: UUID(), authentication: UUID(), generation: nil)
+    private func delta(_ text: String, connection: String = "connection", session: String = "session", run: String = "run",
+                       segment: String = "segment", turn: String? = "turn", inputID: String? = "input",
+                       inputKey: String? = "key", mode: RuntimeTextUpdate.Mode = .delta,
+                       metadata: RuntimeFinalMessageMetadata? = nil) -> RuntimeEvent {
+        .update(connectionID: connection, .text(.init(sessionKey: session, runID: run, segmentID: segment, text: text,
+            mode: mode, turnID: turn, inputMessageID: inputID, inputMessageKey: inputKey, snapshotMetadata: metadata)))
+    }
+    private func body(_ event: RuntimeEvent?) throws -> RuntimeTextUpdate {
+        guard case let .update(_, .text(update)) = try #require(event) else { throw RuntimeFailure(.invalidResponse) }
+        return update
+    }
+
+    @Test func pausedConsumerReceivesAllFragmentsWithoutChangingBytesOrIdentity() async throws {
+        let buffer = RuntimeEventBuffer()
+        let chunks = ["NO", "RMAL", " \r\n", "e", "\u{301}", "👩", "\u{200D}", "💻", "\t", "中"]
+        for _ in 0..<100 {
+            for chunk in chunks { #expect(buffer.offer(delta(chunk), weight: 128, context: context) == .accepted) }
+        }
+        buffer.finish()
+        let update = try body(await buffer.next())
+        #expect(Data(update.text.utf8) == Data(Array(repeating: chunks.joined(), count: 100).joined().utf8))
+        #expect(update.mode == .delta && update.snapshotMetadata == nil)
+        #expect(update.runID == "run" && update.segmentID == "segment" && update.inputMessageKey == "key")
+        #expect(try await buffer.next() == nil)
+    }
+
+    @Test func everyNonDeltaEventIsAnOrderedBarrier() async throws {
+        let buffer = RuntimeEventBuffer()
+        let barriers: [RuntimeEvent] = [
+            .state(.init(phase: .reconnecting)), .failure(connectionID: "connection", .init(.network)),
+            .update(connectionID: "connection", .runStarted(sessionKey: "session", runID: "run")),
+            .update(connectionID: "connection", .tool(sessionKey: "session", runID: "run", payload: .object([:]))),
+            .update(connectionID: "connection", .thought(sessionKey: "session", runID: "run", text: "thought")),
+            .update(connectionID: "connection", .other(.object([:]))),
+            delta("snapshot", mode: .snapshot),
+            .update(connectionID: "connection", .runFinished(sessionKey: "session", runID: "run", reason: "end_turn", segmentID: "segment", raw: .null))
+        ]
+        for (index, barrier) in barriers.enumerated() {
+            #expect(buffer.offer(delta("before-\(index)"), weight: 128, context: context) == .accepted)
+            #expect(buffer.offer(barrier, weight: 128, context: context) == .accepted)
+            #expect(buffer.offer(delta("after-\(index)", segment: "next-\(index)"), weight: 128, context: context) == .accepted)
+        }
+        buffer.finish()
+        for (index, barrier) in barriers.enumerated() {
+            #expect(try body(await buffer.next()).text == "before-\(index)")
+            #expect(try await buffer.next() == barrier)
+            #expect(try body(await buffer.next()).text == "after-\(index)")
+        }
+        #expect(try await buffer.next() == nil)
+    }
+
+    @Test func byteDistinctAndOptionalLineageNeverMerge() async throws {
+        func scoped(_ text: String, different field: Int? = nil, absent: Int? = nil,
+                    metadata: RuntimeFinalMessageMetadata? = nil) -> RuntimeEvent {
+            func value(_ index: Int) -> String { field == index ? "e\u{301}" : "é" }
+            return delta(text, connection: value(0), session: value(1), run: value(2), segment: value(3),
+                turn: absent == 4 ? nil : value(4), inputID: absent == 5 ? nil : value(5),
+                inputKey: absent == 6 ? nil : value(6), metadata: metadata)
+        }
+        let variants = (0..<7).map { scoped("second", different: $0) }
+            + (4..<7).map { scoped("second", absent: $0) }
+            + [scoped("second", metadata: .init(role: .assistant, providedID: "source"))]
+        for value in variants {
+            let buffer = RuntimeEventBuffer()
+            let first = scoped("first")
+            #expect(buffer.offer(first, weight: 128, context: context) == .accepted)
+            #expect(buffer.offer(value, weight: 128, context: context) == .accepted)
+            buffer.finish()
+            #expect(try await buffer.next() == first)
+            #expect(try await buffer.next() == value)
+            #expect(try await buffer.next() == nil)
+        }
+    }
+
+    @Test func anyInternalRevisionChangeSeparatesOtherwiseIdenticalFragments() async throws {
+        let revisions = [
+            RuntimeEventBuffer.Context(epoch: UUID(), attempt: context.attempt, authentication: context.authentication, generation: nil),
+            RuntimeEventBuffer.Context(epoch: context.epoch, attempt: UUID(), authentication: context.authentication, generation: nil),
+            RuntimeEventBuffer.Context(epoch: context.epoch, attempt: context.attempt, authentication: UUID(), generation: nil),
+            RuntimeEventBuffer.Context(epoch: context.epoch, attempt: context.attempt, authentication: context.authentication, generation: .init())
+        ]
+        for revision in revisions {
+            let buffer = RuntimeEventBuffer()
+            #expect(buffer.offer(delta("one"), weight: 128, context: context) == .accepted)
+            #expect(buffer.offer(delta("two"), weight: 128, context: revision) == .accepted)
+            buffer.finish()
+            #expect(try body(await buffer.next()).text == "one")
+            #expect(try body(await buffer.next()).text == "two")
+            #expect(try await buffer.next() == nil)
+        }
+    }
+
+    @Test func sixtyFifthNonmergeableEventFailsAfterDrainingTheAcceptedPrefix() async throws {
+        let buffer = RuntimeEventBuffer(), ended = Terminations()
+        buffer.onTermination { ended.record() }
+        for index in 0..<64 {
+            #expect(buffer.offer(delta("\(index)", segment: "segment-\(index)"), weight: 128, context: context) == .accepted)
+        }
+        #expect(buffer.offer(delta("not admitted", segment: "another"), weight: 128, context: context) == .overflow)
+        #expect(ended.value == 1)
+        buffer.finish()
+        #expect(buffer.offer(delta("late"), weight: 128, context: context) == .terminated)
+        for index in 0..<64 { #expect(try body(await buffer.next()).text == "\(index)") }
+        await #expect(throws: RuntimeFailure(.capacity)) { try await buffer.next() }
+        #expect(ended.value == 1)
+    }
+
+    @Test func mergedFragmentsKeepTheirOriginalWeightAndConsumptionReleasesIt() async throws {
+        let full = RuntimeEventBuffer.maximumBytes
+        let buffer = RuntimeEventBuffer()
+        #expect(buffer.offer(delta("one"), weight: full / 2, context: context) == .accepted)
+        #expect(buffer.offer(delta("two"), weight: full / 2, context: context) == .accepted)
+        #expect(try body(await buffer.next()).text == "onetwo")
+        #expect(buffer.offer(delta("new"), weight: full, context: context) == .accepted)
+        #expect(buffer.offer(delta("overflow"), weight: 1, context: context) == .overflow)
+        #expect(try body(await buffer.next()).text == "new")
+        await #expect(throws: RuntimeFailure(.capacity)) { try await buffer.next() }
+    }
+
+    @Test(arguments: [-1, RuntimeEventBuffer.maximumBytes + 1])
+    func invalidOrOversizedWeightFailsWithoutAdmittingTheBody(_ weight: Int) async throws {
+        let buffer = RuntimeEventBuffer(), ended = Terminations()
+        buffer.onTermination { ended.record() }
+        #expect(buffer.offer(delta("not admitted"), weight: weight, context: context) == .overflow)
+        await #expect(throws: RuntimeFailure(.capacity)) { try await buffer.next() }
+        buffer.finish(); #expect(ended.value == 1)
+    }
+
+    @Test func cancellationBeforeOrDuringPullTerminatesAndNotifiesOnlyOnce() async throws {
+        for iteration in 0..<100 {
+            let buffer = RuntimeEventBuffer(), ended = Terminations()
+            buffer.onTermination { ended.record() }
+            let task = Task { try await buffer.next() }
+            if iteration.isMultiple(of: 2) { await Task.yield() }
+            task.cancel()
+            do { _ = try await task.value; Issue.record("Cancelled consumer returned normally") }
+            catch is CancellationError { }
+            buffer.finish(); task.cancel()
+            #expect(ended.value == 1)
+            #expect(buffer.offer(delta("late"), weight: 128, context: context) == .terminated)
+        }
+    }
+
+    @Test func normalFinishWakesAPullAndRejectsLateAdmission() async throws {
+        let buffer = RuntimeEventBuffer(), ended = Terminations()
+        buffer.onTermination { ended.record() }
+        let task = Task { try await buffer.next() }
+        await Task.yield(); buffer.finish(); buffer.finish(throwing: RuntimeFailure(.capacity))
+        #expect(try await task.value == nil)
+        #expect(buffer.offer(delta("late"), weight: 128, context: context) == .terminated)
+        #expect(ended.value == 1)
+    }
+}
diff --git a/packages/apple/Tests/ClawketRuntimeTests/RuntimeLiveBufferTests.swift b/packages/apple/Tests/ClawketRuntimeTests/RuntimeLiveBufferTests.swift
new file mode 100644
index 00000000..565ac6d5
--- /dev/null
+++ b/packages/apple/Tests/ClawketRuntimeTests/RuntimeLiveBufferTests.swift
@@ -0,0 +1,163 @@
+import ClawketTransport
+import Foundation
+import Testing
+@testable import ClawketRuntime
+
+@Suite(.serialized)
+struct RuntimeLiveBufferTests {
+    private func ready(_ runtime: RuntimeCoordinator) async throws -> AsyncThrowingStream<RuntimeEvent, any Error>.Iterator {
+        var reader = runtime.events.makeAsyncIterator()
+        while let event = try await reader.next() {
+            if case let .state(state) = event, state.phase == .ready { return reader }
+        }
+        throw RuntimeFailure(.notConnected)
+    }
+
+    @Test func realHermesMapperDoesNotInsertPerTokenBarriersAndRetainsWholeSingleSegmentFinal() async throws {
+        let socket = ScriptSocket { request, socket in try await socket.reply(request, payload: health(.hermes)) }
+        let runtime = try RuntimeCoordinator(credentials: Credentials(), socketFactory: ScriptFactory(socket))
+        try await runtime.activate(connection(.hermes))
+        var reader = try await ready(runtime)
+        let scope = try await runtime.scope()
+        func chat(_ text: String, state: String = "delta") -> JSONValue {
+            event("chat", .object(["runId": .string("fixture-hermes"), "sessionKey": .string("main"),
+                "state": .string(state), "message": .object(["role": .string("assistant"), "content": .string(text)])]))
+        }
+        let pieces = ["NO", "RMAL", " \r\n", "e", "\u{301}", "👩", "\u{200D}", "💻"]
+        for index in 0..<1_024 {
+            await runtime.receive(.backendFrame(scope.generation, chat(pieces[index % pieces.count])), epoch: scope.epoch, attempt: scope.attempt)
+        }
+        #expect(await runtime.state.phase == .ready)
+        try await runtime.abort(sessionKey: "main", runID: "fixture-hermes")
+        #expect(await socket.requests("chat.abort").count == 1)
+        let full = (0..<1_024).map { pieces[$0 % pieces.count] }.joined() + " — corrected final"
+        await runtime.receive(.backendFrame(scope.generation, chat(full, state: "final")), epoch: scope.epoch, attempt: scope.attempt)
+        guard case let .update(_, .text(delta)) = try await reader.next(),
+              case let .update(_, .text(snapshot)) = try await reader.next(),
+              case .update(_, .runFinished) = try await reader.next() else {
+            Issue.record("Unexpected per-token event/barrier or lost terminal"); await runtime.disconnect(); return
+        }
+        #expect(delta.mode == .delta && snapshot.mode == .snapshot)
+        #expect(Data(delta.text.utf8) == Data((0..<1_024).map { pieces[$0 % pieces.count] }.joined().utf8))
+        #expect(Data(snapshot.text.utf8) == Data(full.utf8))
+        #expect(delta.segmentID == snapshot.segmentID && delta.snapshotMetadata == nil)
+        #expect(await socket.requests("chat.send").isEmpty)
+        await runtime.disconnect()
+    }
+
+    @Test func slowPublicConsumerKeepsMappedDeltasBarriersAndAllowsStopWithoutReplayingSend() async throws {
+        let socket = ScriptSocket { request, socket in
+            let payload: JSONValue = switch request["method"]?.stringValue {
+            case "connect": health(.pi)
+            case "chat.send": .object(["runId": .string("fixture-run")])
+            default: .object(["ok": .bool(true)])
+            }
+            try await socket.reply(request, payload: payload)
+        }
+        let runtime = try RuntimeCoordinator(credentials: Credentials(), socketFactory: ScriptFactory(socket))
+        try await runtime.activate(connection(.pi))
+        var reader = try await ready(runtime)
+        _ = try await runtime.send(sessionKey: "main", input: .init(text: "Synthetic", idempotencyKey: "one-send"))
+        let scope = try await runtime.scope()
+        func chunk(_ text: String, mode: String = "delta") -> JSONValue {
+            event("pi.update", .object(["type": .string("agent_message_chunk"), "runId": .string("fixture-run"),
+                "sessionKey": .string("main"), "textMode": .string(mode), "text": .string(text)]))
+        }
+        let fragments = ["前", " \r\n", "e", "\u{301}", "👩", "\u{200D}", "💻", "\t"]
+        // Pause the real public consumer while the actual mapper receives far
+        // more than 64 small fragments. No replacement accumulator is involved.
+        for index in 0..<1_024 {
+            await runtime.receive(.backendFrame(scope.generation, chunk(fragments[index % fragments.count])),
+                                  epoch: scope.epoch, attempt: scope.attempt)
+        }
+        #expect(await runtime.state.phase == .ready)
+        try await runtime.abort(sessionKey: "main", runID: "fixture-run")
+        #expect(await socket.requests("chat.abort").count == 1)
+        #expect(await socket.requests("chat.send").count == 1)
+        let tool = event("agent", .object(["runId": .string("fixture-run"), "sessionKey": .string("main"),
+            "stream": .string("tool"), "data": .object(["phase": .string("start"), "toolCallId": .string("tool-one"), "name": .string("synthetic")])]))
+        await runtime.receive(.backendFrame(scope.generation, tool), epoch: scope.epoch, attempt: scope.attempt)
+        await runtime.receive(.backendFrame(scope.generation, chunk("After")), epoch: scope.epoch, attempt: scope.attempt)
+        await runtime.receive(.backendFrame(scope.generation, chunk("Corrected", mode: "snapshot")), epoch: scope.epoch, attempt: scope.attempt)
+        let final = event("pi.update", .object(["type": .string("run_finished"), "runId": .string("fixture-run"),
+            "sessionKey": .string("main"), "stopReason": .string("end_turn")]))
+        await runtime.receive(.backendFrame(scope.generation, final), epoch: scope.epoch, attempt: scope.attempt)
+        var texts: [RuntimeTextUpdate] = [], toolSeen = false, ended = false
+        while let event = try await reader.next() {
+            guard case let .update(_, update) = event else { continue }
+            switch update {
+            case .text(let text):
+                if texts.isEmpty { #expect(!toolSeen) } else { #expect(toolSeen) }
+                texts.append(text)
+            case .tool: toolSeen = true
+            case .runFinished: ended = true
+            default: break
+            }
+            if ended { break }
+        }
+        #expect(texts.count == 3)
+        #expect(Data(try #require(texts.first).text.utf8) == Data((0..<1_024).map { fragments[$0 % fragments.count] }.joined().utf8))
+        if texts.count == 3 {
+            #expect(texts[0].mode == .delta && texts[1].mode == .delta && texts[2].mode == .snapshot)
+            #expect(texts[0].segmentID != texts[1].segmentID && texts[1].segmentID == texts[2].segmentID)
+            #expect(texts[1].text == "After" && texts[2].text == "Corrected")
+        }
+        #expect(toolSeen && ended)
+        #expect(await socket.requests("chat.send").count == 1)
+        await runtime.disconnect()
+    }
+
+    @Test func realOverflowRetiresPendingWriteAndNeverReplaysOrReactivatesTheClosedStream() async throws {
+        let socket = ScriptSocket { request, socket in
+            if request["method"]?.stringValue == "connect" { try await socket.reply(request, payload: health(.pi)) }
+        }
+        let runtime = try RuntimeCoordinator(credentials: Credentials(), socketFactory: ScriptFactory(socket))
+        try await runtime.activate(connection(.pi))
+        var reader = try await ready(runtime)
+        let send = Task { try await runtime.send(sessionKey: "main", input: .init(text: "Synthetic", idempotencyKey: "one-send")) }
+        try await eventually { await socket.requests("chat.send").count == 1 }
+        for index in 0..<65 {
+            await runtime.emit(.update(connectionID: "connection-a", .other(.number(Double(index)))))
+        }
+        do { _ = try await send.value; Issue.record("Pending write survived retirement") }
+        catch let failure as RuntimeFailure { #expect(failure.delivery == .uncertain && failure.code == .superseded) }
+        try await eventually { await socket.closed }
+        for index in 0..<64 {
+            #expect(try await reader.next() == .update(connectionID: "connection-a", .other(.number(Double(index)))))
+        }
+        await #expect(throws: RuntimeFailure(.capacity)) { try await reader.next() }
+        await #expect(throws: RuntimeFailure(.capacity)) { try await runtime.activate(connection(.pi)) }
+        #expect(await socket.requests("chat.send").count == 1)
+        #expect(await runtime.state.phase == .idle)
+    }
+
+    @Test func cancellingThePublicConsumerRetiresItsSocketAndLateEventsOnlyOnce() async throws {
+        let socket = ScriptSocket { request, socket in try await socket.reply(request, payload: health(.pi)) }
+        let runtime = try RuntimeCoordinator(credentials: Credentials(), socketFactory: ScriptFactory(socket))
+        try await runtime.activate(connection(.pi))
+        let reader = Task {
+            do { for try await _ in runtime.events { try Task.checkCancellation() } }
+            catch is CancellationError { }
+        }
+        await Task.yield(); reader.cancel(); try await reader.value
+        try await eventually { await socket.closed }
+        let epoch = await runtime.epoch
+        for _ in 0..<10 { await runtime.emit(.state(.init(phase: .ready))) }
+        #expect(await runtime.epoch == epoch)
+        #expect(await runtime.state.phase == .idle)
+        await #expect(throws: RuntimeFailure(.capacity)) { try await runtime.activate(connection(.pi)) }
+    }
+
+    @Test func retainedPublicStreamDoesNotRetainTheCoordinatorAndFinishesOnOwnerDeinit() async throws {
+        var runtime: RuntimeCoordinator? = try RuntimeCoordinator(credentials: Credentials())
+        weak var weakRuntime = runtime
+        let stream = try #require(runtime).events
+        let reader = Task {
+            var iterator = stream.makeAsyncIterator()
+            return try await iterator.next()
+        }
+        runtime = nil
+        #expect(weakRuntime == nil)
+        #expect(try await reader.value == nil)
+    }
+}