perf(remote): load session history pages on demand

src/apps/mobile/harmonyos/tools/tests/host-stream.test.cjs

@@ -271,3 +271,34 @@ test('wire parsers reject foreign pages, invalid shapes, and non-hint device eve
   assert.throws(() => checkHostStreamPage('s', { resp: 'error', message: 'invalid RPC command' }), HostStreamUnsupportedError);
   assert.throws(() => checkHostStreamPage('s', { resp: 'ok' }), /Unexpected stream response/);
 });
+
+
+test('lazy disk history keeps JS-safe backward cursors separate from live updates', async () => {
+  const ceiling = 2 ** 52;
+  const c = callbacks(); const reads = []; let hint;
+  const record = (seq, id, turn) => ({ seq, event: 'session-record', payload: { id, revision: seq, turn: { turnId: turn } } });
+  const source = {
+    target: 'desktop', unsubscribe: async () => {},
+    onHint: listener => { hint = listener; return () => {}; }, onReconnect: () => () => {},
+    read: async request => {
+      reads.push({ ...request });
+      const events = request.before !== undefined ? [record(ceiling - 3, 'old', 'old-turn')]
+        : request.after !== undefined ? [record(ceiling, 'new', 'new-turn')]
+        : [record(ceiling - 2, 'a', 'turn'), record(ceiling - 1, 'b', 'turn')];
+      return { resp: 'stream_page', stream_id: 'session', epoch: 8, events,
+        cursor: request.after !== undefined ? ceiling : ceiling - 1,
+        oldest_seq: events[0].seq, has_more: request.before === undefined && request.after === undefined, truncated: false };
+    }
+  };
+  const stream = new HostSessionStream('session', source, c.hooks);
+  try {
+    await settle(() => c.caught === 1);
+    stream.loadOlder(); await settle(() => c.applied.length === 3);
+    assert.deepEqual(reads[1], { before: ceiling - 2, epoch: 8 });
+    hint({ sourceDeviceId: 'desktop', streamId: 'session', epoch: 8, cursor: ceiling });
+    await settle(() => c.applied.length === 4);
+    assert.deepEqual(reads[2], { after: ceiling - 1, epoch: 8 });
+    assert.deepEqual(c.applied.map(e => e.payload.id), ['a', 'b', 'old', 'new']);
+    assert.equal(c.errors.length, 0);
+  } finally { stream.close(); }
+});

src/apps/mobile/shared/core-transport/src/commonTest/kotlin/com/openbitfun/mobile/core/transport/HostStreamTest.kt

@@ -77,6 +77,24 @@ class HostStreamTest {
         return received to job
     }
 
+    @Test fun jsSafeHistoryCursorsDoNotMoveTheForwardCursor() = runTest {
+        val ceiling = 1L shl 52
+        val host = FakeHost("s1", pageSize = 2)
+        host.nextSeq = ceiling - 3
+        repeat(3) { host.append("session-record", turnRecord("t$it", it)) }
+        val hints = MutableSharedFlow<StreamHint>()
+        val older = Channel<CompletableDeferred<Unit>>()
+        val (received, _) = open(host, hints = hints, older = older)
+        runCurrent()
+        val page = CompletableDeferred<Unit>(); older.send(page); runCurrent()
+        assertTrue(page.isCompleted)
+        assertEquals(Triple<Long?,Long?,Long?>(null, ceiling - 2, 1L), host.reads.last())
+        host.append("session-record", turnRecord("live", 3))
+        hints.emit(StreamHint("desktop-1", "s1",host.epoch,host.cursor)); runCurrent()
+        assertEquals(Triple<Long?,Long?,Long?>(ceiling - 1, null, 1L),host.reads.last())
+        assertEquals(listOf("t1","t2","t0","live"), emittedTurnIds(received))
+    }
+
     @Test fun completionWaitsUntilTheCollectorConsumesTheWholePage() = runTest {
         val host = FakeHost("s1", pageSize = 2)
         repeat(4) { host.append("session-record", turnRecord("turn-$it", it)) }

src/crates/assembly/core/AGENTS.md

@@ -212,6 +212,15 @@ or test-target layout. Workspace checks and product-wide tests are CI-backed and
 are not the default Core precheck. For documentation-only changes, run
 `git diff --check`.
 
+For disk-backed history paging and legacy sessions without a catalog:
+`cargo test --locked -p openbitfun-core --no-default-features --features remote-connect,git --lib history_page_`.
+Also run the `staged_revert_catalog_projection` and `load_relay_session_turns_`
+filters for the same target when changing visibility. Paging must not parse
+unrelated turn bodies or rewrite history. To compare real-file first-page work
+against full materialization locally, use the same target with
+`history_page_benchmark -- --ignored --nocapture`; it checks content equivalence
+and reports timings without asserting a machine-dependent latency in CI.
+
 For host-stream history reads and abandoned execution after a runtime restart:
 `cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git --lib load_relay_session_turns_`.
 The observer must preserve terminal history and another process's writer lease;

src/crates/assembly/core/src/agentic/coordination/coordinator.rs

@@ -8588,12 +8588,35 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
         session_id: &str,
         turn_id: Option<&str>,
     ) -> OpenBitFunResult<Vec<DialogTurnData>> {
+        self.load_relay_session_selection(storage, session_id, turn_id, None)
+            .await
+            .map(|page| page.0)
+    }
+
+    pub async fn load_relay_history_turn(
+        &self,
+        storage: &Path,
+        session_id: &str,
+        before: Option<usize>,
+    ) -> OpenBitFunResult<(Vec<DialogTurnData>, Option<usize>)> {
+        self.load_relay_session_selection(storage, session_id, None, Some(before))
+            .await
+    }
+
+    async fn load_relay_session_selection(
+        &self,
+        storage: &Path,
+        session_id: &str,
+        turn_id: Option<&str>,
+        history: Option<Option<usize>>,
+    ) -> OpenBitFunResult<(Vec<DialogTurnData>, Option<usize>)> {
         let _mutation = self
             .session_manager
             .acquire_session_mutation(session_id)
             .await?;
         self.prepare_persisted_session_read_locked(storage, session_id)
             .await?;
+
         // Persisted InProgress is not proof of a live executor after restart.
         // A loaded owner supplies runtime state; for an unloaded session an
         // exclusive writer lease proves that no other process is executing it.
@@ -8618,29 +8641,38 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
                 SessionState::Idle | SessionState::Error { .. }
             )
         }) || observer_lease.is_some();
-        let (mut turns, read_mode) = if let Some(turn_id) = turn_id {
+
+        let (mut turns, next, read_mode) = if let Some(before) = history {
+            let (turns, next) = self
+                .session_manager
+                .persistence_manager()
+                .load_visible_history_turn(storage, session_id, before)
+                .await?;
+            (turns, next, "history")
+        } else if let Some(turn_id) = turn_id {
             if let Some(turn) = self
                 .session_manager
                 .persistence_manager()
                 .load_visible_session_turn(storage, session_id, turn_id)
                 .await?
             {
-                (vec![turn], "catalog")
+                (vec![turn], None, "catalog")
             } else {
                 let mut turns = self
                     .session_manager
                     .persistence_manager()
                     .load_visible_session_turns(storage, session_id)
                     .await?;
                 turns.retain(|turn| turn.turn_id == turn_id);
-                (turns, "full-fallback")
+                (turns, None, "full-fallback")
             }
         } else {
             (
                 self.session_manager
                     .persistence_manager()
                     .load_visible_session_turns(storage, session_id)
                     .await?,
+                None,
                 "full",
             )
         };
@@ -8700,10 +8732,16 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
                 }
             }
         }
-        let context = self
-            .session_manager
-            .get_context_messages(session_id)
-            .await?;
+        let context = if turns
+            .iter()
+            .any(|turn| turn.status == TurnStatus::InProgress)
+        {
+            self.session_manager
+                .get_context_messages(session_id)
+                .await?
+        } else {
+            Vec::new()
+        };
         for turn in &mut turns {
             if turn.status == TurnStatus::InProgress {
                 let messages: Vec<_> = context
@@ -8718,7 +8756,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
                 SessionManager::append_generation_rounds(turn, &id, &messages, timestamp);
             }
         }
-        Ok(turns)
+        Ok((turns, next))
     }
 
     /// Export a transcript while retaining the same Session history boundary
@@ -17206,84 +17244,6 @@ mod tests {
             .expect("clean up persisted test session");
     }
 
-    #[tokio::test]
-    async fn load_relay_session_turns_marks_abandoned_execution_without_rewriting_history() {
-        let workspace = tempfile::tempdir().expect("workspace");
-        crate::service::workspace::legacy_compat::register_local_fixture_blocking(workspace.path());
-        let (coordinator, manager) = test_persistent_coordinator();
-        let session = manager
-            .create_session(
-                "Restart".into(),
-                "Standard".into(),
-                SessionConfig {
-                    workspace_path: Some(workspace.path().to_string_lossy().into_owned()),
-                    ..Default::default()
-                },
-            )
-            .await
-            .unwrap();
-        let id = &session.session_id;
-        let storage = manager.effective_session_storage_path(id).await.unwrap();
-        let turn_id = manager
-            .start_dialog_turn(
-                id,
-                "Standard".into(),
-                "restart probe".into(),
-                Some("turn-restart-probe".into()),
-                None,
-                None,
-            )
-            .await
-            .unwrap();
-        let live = coordinator
-            .load_relay_session_turns(&storage, id, None)
-            .await
-            .unwrap();
-        assert_eq!(live[0].status, TurnStatus::InProgress);
-        // Simulate the executor disappearing while the durable turn still says
-        // InProgress, as happens across process shutdown/restore.
-        manager.reset_session_state_if_processing(id, &turn_id);
-        let idle = coordinator
-            .load_relay_session_turns(&storage, id, None)
-            .await
-            .unwrap();
-        assert_eq!(idle[0].status, TurnStatus::Cancelled);
-        manager.unload_session_from_memory(id).await.unwrap();
-        // Another writer, even in the same process, forbids inferring death
-        // from this coordinator's empty in-memory map.
-        let owner = manager
-            .persistence_manager()
-            .lock_session_writes(&storage, id)
-            .unwrap();
-        let observed = coordinator
-            .load_relay_session_turns(&storage, id, None)
-            .await
-            .unwrap();
-        assert_eq!(observed[0].status, TurnStatus::InProgress);
-        drop(owner);
-        let orphan = coordinator
-            .load_relay_session_turns(&storage, id, None)
-            .await
-            .unwrap();
-        assert_eq!(orphan[0].status, TurnStatus::Cancelled);
-        assert_eq!(orphan[0].finish_reason.as_deref(), Some("interrupted"));
-        let single = coordinator
-            .load_relay_session_turns(&storage, id, Some(&turn_id))
-            .await
-            .unwrap();
-        assert_eq!(single[0].status, TurnStatus::Cancelled);
-        let stored = manager
-            .persistence_manager()
-            .load_visible_session_turns(&storage, id)
-            .await
-            .unwrap();
-        assert_eq!(
-            stored[0].status,
-            TurnStatus::InProgress,
-            "observer must retain persisted evidence"
-        );
-    }
-
     #[tokio::test]
     async fn load_relay_session_turns_reads_history_after_the_session_is_unloaded() {
         let workspace = tempfile::tempdir().expect("workspace");
@@ -17364,6 +17324,16 @@ mod tests {
         assert_eq!(one.len(), 1);
         assert_eq!(one[0].turn_id, turn_id);
 
+        let (page, next) = coordinator
+            .load_relay_history_turn(&storage, &session.session_id, None)
+            .await
+            .expect("paged history must not require an in-memory writer");
+        assert_eq!(
+            serde_json::to_value(&page).unwrap(),
+            serde_json::to_value(&one).unwrap()
+        );
+        assert_eq!(next, None);
+
         session_manager
             .restore_session(workspace.path(), &session.session_id)
             .await