fix: stabilize history and open sessions from the latest turn

.github/workflows/ci.yml

@@ -122,17 +122,23 @@ jobs:
     needs: build-impact
     if: ${{ !cancelled() && needs.build-impact.outputs.rust_required != 'false' }}
     runs-on: ${{ matrix.os }}
-    timeout-minutes: 30
+    timeout-minutes: ${{ matrix.timeout_minutes }}
     strategy:
       fail-fast: false
       matrix:
         include:
           - os: ubuntu-latest
             cache_key: ubuntu
-          - os: macos-15
-            cache_key: macos
+            timeout_minutes: 30
+          # Pin the Intel pool for macOS OS-contract validation; release jobs
+          # select their Apple Silicon / Intel targets independently.
+          - os: macos-15-intel
+            cache_key: macos-intel
+            # A cold Intel build can still be compiling the CLI at 30 minutes.
+            timeout_minutes: 75
           - os: windows-latest
             cache_key: windows
+            timeout_minutes: 30
     steps:
       - uses: actions/checkout@v5
 
@@ -211,7 +217,7 @@ jobs:
       matrix:
         os:
           - ubuntu-latest
-          - macos-15
+          - macos-15-intel
           - windows-latest
     steps:
       - uses: actions/checkout@v5

docs/architecture/peer-device-mode.md

@@ -125,6 +125,30 @@ records and responses from a device that is no longer rendered. Desktop
 `RelaySessionHistory` owns the subscription across initial loading, realtime
 delivery and older-page prefetch.
 
+Desktop/Web FlowChat history publishes complete Turns. The Rust subscriber offers
+`turn_history: {version: 1}` on `read_stream`; a response acknowledgement enables
+single-Turn snapshots. Hosts also advertise `host_stream_turn_history_v1`. The
+first read selects only the last Turn, independently of the ordering of live
+events. Large Turns travel in byte/event-bounded packets of one immutable snapshot;
+packet offsets and the older-Turn boundary are separate. The live cursor captured
+before source IO fences snapshot revisions, and normal live publication resumes
+after it. Snapshots are ephemeral, controller scoped, epoch fenced and byte bounded
+(one oversized Turn may exceed the budget); they add no persisted format or relay
+storage. Identity-only tombstone tracking preserves deletions without suppressing
+live publication to another controller.
+
+Hosts that omit the acknowledgement keep the record-page path: optional root
+`turnRecordIds` membership proves completeness across pages, including unchanged
+items older than a republished root. Older hosts without membership remain
+readable by finishing their available history before publication, so their first
+open can be slower. Missing records, non-advancing packets and epoch changes fail
+without exposing fragments. A terminal header alone never proves completeness.
+`RelaySessionHistory` stages controller-annotated `historyReplay` records outside
+the replica until ready. Live records remain incremental even while an older Turn
+is loading. After the last Turn is visibly presented, a visible view may prefetch
+one adjacent Turn; further reads follow reader demand. Switching back to a partial
+session does not automatically drain its history.
+
 Native mobile history keeps these record-page boundaries; a page is not a
 complete conversation turn. The initial replay and each older-history request
 reduce all received records before publishing one transcript projection. A turn
@@ -251,6 +275,16 @@ the same before, during and after a switch. Seeds carry no history or state
 machine; they are promoted through the normal metadata path before any open,
 and a seed that has been opened is never replaced by a later seed.
 
+The displayed row count belongs to the surface/workspace/source view and is
+published when the user expands or collapses it. Live and background lists
+read the same snapshot during replacement; an unmount-time handoff is too late
+for React's incoming render. Collapsing a workspace resets its source views;
+Collapse All resets every view. Counts are window-local presentation state.
+Disconnecting a control attachment disposes transcripts but leaves the
+independent catalog read lease intact. Disposal invalidates the projection's
+array deduplication and immediately reseeds metadata from that catalog, so
+the still-visible sidebar does not empty while waiting for another read.
+
 A cross-device open publishes a `DeviceOpenIntent` for its whole duration. While
 it is set, every device keeps its background rows (the rendered device hands
 over to its live list once, after the open, so the live list's reads never

docs/architecture/rust-build-dependency-boundaries.md

@@ -223,6 +223,7 @@ CI 负责 workspace 级检查、真实产品 feature 组合、跨平台、完整
 
 ### 7.1 Hosted CI 关键路径与缓存
 
+- 日常 Rust/CLI 验证保留 Linux、Windows 和 macOS 三个平台，macOS 固定使用标准 Intel macOS 15 runner（`macos-15-intel`），避免默认标签的架构迁移和 ARM64 托管资源池分配失败阻塞 OS 契约验证。CLI 与 Rust 使用相同平台矩阵，缓存键区分 Intel。这提供 macOS x64 原生测试证据，不代表 Apple Silicon 运行验证；ARM64/x64 发布目标仍由发布 workflow 的矩阵拥有。
 - 验证 job 只依赖自身的编译期前置条件。Tauri `check`/`test` 只要求配置中的前端和资源目录存在时，Rust job 自行创建空目录，不等待或传递可发布前端产物；真实静态资源仍由前端构建和产品打包 owner 负责。
 - Rust/CLI 影响分类必须保守且 fail-closed：只有活动路径全部位于 `src/web-ui/**` 时才允许跳过 Rust 与 CLI matrix；仓库根目录/`docs/**` 的 Markdown 和 `png/**` 可作为已知中性伴随路径。其他嵌套 Markdown 可能是 `include_str!`/`include_bytes!` 输入，workflow 不得宽泛忽略全部 Markdown。空变更、非法或不可用 diff range、Rust/build 输入、CI 脚本、其他产品 surface 和所有未识别路径都必须运行完整验证，不能依赖易漏维护的“已知 Rust 目录”名单。PR 使用 base/head 的 merge-base range，只分类 PR 自身改动；`main` push 使用事件 before/head 的 direct range，保留 force-push 或回退提交的实际影响。
 - “纯 Web”前提必须是可执行边界：影响分类 job 在计算 diff 前，从 Git 跟踪文件列表扫描全仓 `.rs`（包括 workspace 外 Installer 和根 build script）。除逐行精确登记的既有测试 fixture 外，非注释 Rust 源码不得出现 `web-ui` 路径 token；因此 `include_dir!`、编译嵌入、分段 `PathBuf::join` 和间接路径变量都会在分类前失败。Frontend job 同时显式运行边界 contract tests 和真实 repository boundary check；跨 surface 的命令注册和调用分别由 owner 测试看护。新增类似源码读取时应修复所有权，而不是给影响分类器追加隐式例外。

scripts/check-github-config.test.mjs

@@ -429,7 +429,7 @@ test('keeps Rust CI independent, restore-only on PRs, and target-focused', () =>
   );
   assert.deepEqual(
     rustJob.strategy.matrix.os,
-    ['ubuntu-latest', 'macos-15', 'windows-latest'],
+    ['ubuntu-latest', 'macos-15-intel', 'windows-latest'],
     'Rust validation must retain the reviewed Linux, macOS, and Windows matrix',
   );
   assert.equal(
@@ -472,7 +472,21 @@ test('keeps Rust CI independent, restore-only on PRs, and target-focused', () =>
   }
 
   const cliJob = workflow.jobs['cli-test'];
-  assert.equal(cliJob['timeout-minutes'], 30);
+  assert.deepEqual(
+    cliJob.strategy.matrix.include.map((entry) => entry.os),
+    rustJob.strategy.matrix.os,
+    'CLI and Rust validation must exercise the same Linux, Intel macOS, and Windows hosts',
+  );
+  assert.equal(
+    cliJob.strategy.matrix.include.find((entry) => entry.os === 'macos-15-intel').cache_key,
+    'macos-intel',
+  );
+  assert.equal(cliJob['timeout-minutes'], '${{ matrix.timeout_minutes }}');
+  assert.deepEqual(
+    cliJob.strategy.matrix.include.map((entry) => entry.timeout_minutes),
+    [30, 75, 30],
+    'Only the cold Intel macOS CLI build needs a longer job budget',
+  );
   assert.ok(
     cliJob.strategy.matrix.include.some((entry) => entry.os === 'windows-latest'),
     'Windows ConPTY contracts must run before Nightly',

src/crates/services/services-integrations/AGENTS.md

@@ -35,16 +35,32 @@ slices that are outside pure product logic but still platform-neutral.
   reader. Neither the relay nor any client persists stream content; do not add
   relay-stored history, `get_session_key`, or a durable stream cache here.
 - Session history reads use `HostStreamHub::read_history` with a Core-owned
-  source loader. Backfill allocates decreasing JS-safe sequences below the
+  source loader. The optional `turn_history` v1 offer selects exactly one Turn;
+  its response acknowledgement enables the mode, while older hosts retain the
+  existing record-page reader. Hosts advertise `host_stream_turn_history_v1`.
+  The immutable snapshot spans bounded packets, carries an older-Turn boundary
+  separately from its packet offset, and uses the live cursor captured before
+  source IO as its revision fence. It never allocates live event sequences.
+  Snapshot IDs are controller/stream scoped and epoch fenced; retained bodies
+  are byte bounded and expire with leases/grace (one oversized Turn may exceed
+  the budget). Identity-only tombstone tracking must not suppress subsequent
+  live publication to other controllers. The subscriber publishes only a
+  validated complete snapshot and catches up live events during older reads.
+  Legacy record-page backfill allocates decreasing JS-safe sequences below the
   live sequence range; it never advances the forward cursor. One source gate
   orders backfill, live publication, and history invalidation. Undo/import and
   cache misses across evicted pages require an epoch fence, never silent skips.
   Only the requested page's bodies enter the bounded log; the pending source
-  batch is at most one persisted turn, which can itself be large. No wire or
-  persisted-format migration is required; all controllers retain `read_stream`.
-  `limit` is a budget for history reads: a page grows to the first record of its
+  batch is at most one persisted turn, which can itself be large. No persisted
+  migration is required; the optional wire fields keep `read_stream` compatible.
+  In legacy record paging, `limit` is a budget: a page grows to the first record of its
   oldest turn so no turn splits across pages, within the hard event and byte
   bounds. Controllers must not assume a page holds at most `limit` events.
+  Root records carry optional `turnRecordIds` membership. The Rust subscriber
+  joins bounded history pages until those members are present, even if a live
+  root moved ahead of its unchanged children; legacy hosts without membership
+  finish their available history read before presentation. Never treat a page
+  limit, a terminal header or a timeout as proof of a complete Turn.
 - The `remote-persistence` feature is the lightweight persisted-shape owner shared
   by Remote Connect, remote SSH, and offline migration. Keep it free of network,
   SSH transport, and runtime orchestration dependencies so owner readers and
@@ -144,7 +160,7 @@ cargo test -p openbitfun-services-integrations --no-default-features --features
 cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-ssh-concrete --lib remote_ssh::relay_deploy::tests::
 cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::account::tests
 cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::relay_client::tests::
-cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::host_stream::tests::
+cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::host_stream::
 cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::host_stream_subscriber::tests::
 cargo test -p openbitfun-services-integrations --no-default-features --features file-watch --test file_watch_contracts
 cargo test --locked -p openbitfun-services-integrations --no-default-features --features workspace-search --test workspace_search_contracts

src/crates/services/services-integrations/src/remote_connect.rs

@@ -572,7 +572,7 @@ pub const REMOTE_CAPABILITY_PLAN_BUILD_V1: &str = "plan_build_v1";
 /// must check this string before offering the write; a host without it can
 /// still answer `get_model_catalog` read-only.
 pub const REMOTE_CAPABILITY_DEVICE_MODEL_DEFAULTS_V1: &str = "device_model_defaults_v1";
-pub use host_stream::REMOTE_CAPABILITY_HOST_STREAM_V1;
+pub use host_stream::{REMOTE_CAPABILITY_HOST_STREAM_V1, REMOTE_CAPABILITY_TURN_HISTORY_V1};
 
 fn remote_host_capabilities() -> Vec<String> {
     vec![
@@ -586,6 +586,7 @@ fn remote_host_capabilities() -> Vec<String> {
         REMOTE_CAPABILITY_USER_QUESTION_INTERACTION_V1.to_string(),
         REMOTE_CAPABILITY_BACKGROUND_RESULT_DELIVERY_V1.to_string(),
         REMOTE_CAPABILITY_HOST_STREAM_V1.to_string(),
+        REMOTE_CAPABILITY_TURN_HISTORY_V1.to_string(),
         REMOTE_CAPABILITY_DEVICE_MODEL_DEFAULTS_V1.to_string(),
     ]
 }

src/crates/services/services-integrations/src/remote_connect/host_stream.rs

@@ -14,13 +14,18 @@ use std::{
     time::{Duration, Instant},
 };
 
+mod turn_history;
+use turn_history::TurnSnapshot;
+pub use turn_history::{TurnHistoryPage, TurnHistoryRequest};
+
 /// Contains a separator rejected by runtime Session IDs, so control metadata
 /// cannot alias a user's session.
 pub const HOST_CATALOG_ID: &str = "@host/catalog";
 /// `DeviceEvent` name carrying a stream change hint to a subscribed controller.
 pub const HOST_STREAM_CHANGED_EVENT: &str = "host-stream-changed";
 /// Advertised in `get_workspace_info.capabilities` by hosts serving `read_stream`.
 pub const REMOTE_CAPABILITY_HOST_STREAM_V1: &str = "host_stream_v1";
+pub const REMOTE_CAPABILITY_TURN_HISTORY_V1: &str = "host_stream_turn_history_v1";
 /// Returned to controllers that still ask for a relay-stored session key.
 pub const RELAY_SESSION_HISTORY_RETIRED_MESSAGE: &str = "Relay-stored session history has been retired; session content is now read directly from the online host. Update the controlling app to continue.";
 
@@ -59,6 +64,9 @@ pub struct StreamPage {
     /// Older history was evicted from host memory; `has_more=false` then means
     /// "unavailable", not "complete".
     pub truncated: bool,
+    /// Present only when the host accepted the optional single-Turn read offer.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub turn_history: Option<TurnHistoryPage>,
 }
 
 /// A stream read request as carried by `RemoteCommand::ReadStream`.
@@ -79,6 +87,9 @@ pub struct StreamReadRequest {
     /// Keep (or start) receiving change hints for this stream.
     #[serde(default)]
     pub subscribe: bool,
+    /// Optional read negotiation. Older hosts ignore it and return ordinary pages.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub turn_history: Option<TurnHistoryRequest>,
 }
 
 /// Delivers an encrypted `DeviceEvent` hint to one subscribed device. Hosts own
@@ -143,6 +154,10 @@ struct StreamLog {
     truncated: bool,
     subscribers: HashMap<String, Instant>,
     last_activity: Instant,
+    turn_snapshots: HashMap<String, TurnSnapshot>,
+    /// Snapshot identities participate in tombstones, but must not suppress a
+    /// later live publication that other subscribers have not received yet.
+    snapshot_record_turns: HashMap<String, String>,
 }
 
 impl StreamLog {
@@ -158,6 +173,8 @@ impl StreamLog {
             truncated: false,
             subscribers: HashMap::new(),
             last_activity: Instant::now(),
+            turn_snapshots: HashMap::new(),
+            snapshot_record_turns: HashMap::new(),
         }
     }
 
@@ -358,6 +375,7 @@ impl StreamLog {
             cursor: self.latest(),
             oldest_seq: self.oldest(),
             truncated: self.truncated,
+            turn_history: None,
         })
     }
 }
@@ -471,6 +489,10 @@ impl HostStreamHub {
                 let now = Instant::now();
                 state.streams.retain(|_, log| {
                     log.subscribers.retain(|_, expires| *expires > now);
+                    log.turn_snapshots.retain(|_, snapshot| {
+                        now.duration_since(snapshot.last_activity) < STREAM_GRACE
+                            && log.subscribers.contains_key(&snapshot.device)
+                    });
                     !log.subscribers.is_empty()
                         || now.duration_since(log.last_activity) < STREAM_GRACE
                 });
@@ -553,7 +575,10 @@ impl HostStreamHub {
         let Some(log) = state.streams.get_mut(session_id) else {
             return false;
         };
-        if log.history.is_none() {
+        if log.history.is_none()
+            && log.turn_snapshots.is_empty()
+            && log.snapshot_record_turns.is_empty()
+        {
             return false;
         }
         let subscribers = std::mem::take(&mut log.subscribers);
@@ -595,6 +620,9 @@ impl HostStreamHub {
         F: FnMut(Option<usize>) -> Fut,
         Fut: std::future::Future<Output = Result<HistoryBatch>>,
     {
+        if let Some(history) = &request.turn_history {
+            return self.read_turn_history(device, request, history, load).await;
+        }
         if request.after.is_some() && request.before.is_some() {
             bail!("read_stream accepts either after or before, not both");
         }
@@ -758,6 +786,8 @@ impl HostStreamHub {
         let mut state = self.lock();
         if let Some(log) = state.streams.get_mut(stream_id) {
             log.subscribers.remove(source_device_id);
+            log.turn_snapshots
+                .retain(|_, snapshot| snapshot.device != source_device_id);
             log.last_activity = Instant::now();
         }
     }
@@ -769,6 +799,8 @@ impl HostStreamHub {
         for log in state.streams.values_mut() {
             log.subscribers
                 .retain(|device, _| online_device_ids.iter().any(|id| id == device));
+            log.turn_snapshots
+                .retain(|_, snapshot| online_device_ids.contains(&snapshot.device));
         }
     }
 
@@ -930,11 +962,16 @@ fn diff_records(
     let removed: Vec<String> = log
         .record_index
         .iter()
-        .filter(|(id, entry)| !present.contains(*id) && (full || scope.contains(&entry.turn)))
+        .map(|(id, entry)| (id, &entry.turn))
+        .chain(log.snapshot_record_turns.iter())
+        .filter(|(id, turn)| !present.contains(*id) && (full || scope.contains(*turn)))
         .map(|(id, _)| id.clone())
+        .collect::<HashSet<_>>()
+        .into_iter()
         .collect();
     for id in removed {
         log.record_index.remove(&id);
+        log.snapshot_record_turns.remove(&id);
         changed.push(serde_json::json!({"sessionId":session_id,"id":id,"deleted":true}));
     }
     if changed.is_empty() {