fix(ios): consume bounded live events without dropping delta fragments

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/正文。

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.

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.

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,27 +430,28 @@ 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()
         for task in controlDispatches.values { task.cancel() }
         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()
     }
 }