Skip to content

Commit 52780ea

Browse files
authored
Merge pull request #3203 from wgqqqqq/wgq/on-demand-session-history
perf(remote): load session history pages on demand
2 parents 3f8f5e8 + 925d539 commit 52780ea

9 files changed

Lines changed: 935 additions & 127 deletions

File tree

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

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -271,3 +271,34 @@ test('wire parsers reject foreign pages, invalid shapes, and non-hint device eve
271271
assert.throws(() => checkHostStreamPage('s', { resp: 'error', message: 'invalid RPC command' }), HostStreamUnsupportedError);
272272
assert.throws(() => checkHostStreamPage('s', { resp: 'ok' }), /Unexpected stream response/);
273273
});
274+
275+
276+
test('lazy disk history keeps JS-safe backward cursors separate from live updates', async () => {
277+
const ceiling = 2 ** 52;
278+
const c = callbacks(); const reads = []; let hint;
279+
const record = (seq, id, turn) => ({ seq, event: 'session-record', payload: { id, revision: seq, turn: { turnId: turn } } });
280+
const source = {
281+
target: 'desktop', unsubscribe: async () => {},
282+
onHint: listener => { hint = listener; return () => {}; }, onReconnect: () => () => {},
283+
read: async request => {
284+
reads.push({ ...request });
285+
const events = request.before !== undefined ? [record(ceiling - 3, 'old', 'old-turn')]
286+
: request.after !== undefined ? [record(ceiling, 'new', 'new-turn')]
287+
: [record(ceiling - 2, 'a', 'turn'), record(ceiling - 1, 'b', 'turn')];
288+
return { resp: 'stream_page', stream_id: 'session', epoch: 8, events,
289+
cursor: request.after !== undefined ? ceiling : ceiling - 1,
290+
oldest_seq: events[0].seq, has_more: request.before === undefined && request.after === undefined, truncated: false };
291+
}
292+
};
293+
const stream = new HostSessionStream('session', source, c.hooks);
294+
try {
295+
await settle(() => c.caught === 1);
296+
stream.loadOlder(); await settle(() => c.applied.length === 3);
297+
assert.deepEqual(reads[1], { before: ceiling - 2, epoch: 8 });
298+
hint({ sourceDeviceId: 'desktop', streamId: 'session', epoch: 8, cursor: ceiling });
299+
await settle(() => c.applied.length === 4);
300+
assert.deepEqual(reads[2], { after: ceiling - 1, epoch: 8 });
301+
assert.deepEqual(c.applied.map(e => e.payload.id), ['a', 'b', 'old', 'new']);
302+
assert.equal(c.errors.length, 0);
303+
} finally { stream.close(); }
304+
});

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,24 @@ class HostStreamTest {
7777
return received to job
7878
}
7979

80+
@Test fun jsSafeHistoryCursorsDoNotMoveTheForwardCursor() = runTest {
81+
val ceiling = 1L shl 52
82+
val host = FakeHost("s1", pageSize = 2)
83+
host.nextSeq = ceiling - 3
84+
repeat(3) { host.append("session-record", turnRecord("t$it", it)) }
85+
val hints = MutableSharedFlow<StreamHint>()
86+
val older = Channel<CompletableDeferred<Unit>>()
87+
val (received, _) = open(host, hints = hints, older = older)
88+
runCurrent()
89+
val page = CompletableDeferred<Unit>(); older.send(page); runCurrent()
90+
assertTrue(page.isCompleted)
91+
assertEquals(Triple<Long?,Long?,Long?>(null, ceiling - 2, 1L), host.reads.last())
92+
host.append("session-record", turnRecord("live", 3))
93+
hints.emit(StreamHint("desktop-1", "s1",host.epoch,host.cursor)); runCurrent()
94+
assertEquals(Triple<Long?,Long?,Long?>(ceiling - 1, null, 1L),host.reads.last())
95+
assertEquals(listOf("t1","t2","t0","live"), emittedTurnIds(received))
96+
}
97+
8098
@Test fun completionWaitsUntilTheCollectorConsumesTheWholePage() = runTest {
8199
val host = FakeHost("s1", pageSize = 2)
82100
repeat(4) { host.append("session-record", turnRecord("turn-$it", it)) }

‎src/crates/assembly/core/AGENTS.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,15 @@ or test-target layout. Workspace checks and product-wide tests are CI-backed and
212212
are not the default Core precheck. For documentation-only changes, run
213213
`git diff --check`.
214214

215+
For disk-backed history paging and legacy sessions without a catalog:
216+
`cargo test --locked -p openbitfun-core --no-default-features --features remote-connect,git --lib history_page_`.
217+
Also run the `staged_revert_catalog_projection` and `load_relay_session_turns_`
218+
filters for the same target when changing visibility. Paging must not parse
219+
unrelated turn bodies or rewrite history. To compare real-file first-page work
220+
against full materialization locally, use the same target with
221+
`history_page_benchmark -- --ignored --nocapture`; it checks content equivalence
222+
and reports timings without asserting a machine-dependent latency in CI.
223+
215224
For host-stream history reads and abandoned execution after a runtime restart:
216225
`cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git --lib load_relay_session_turns_`.
217226
The observer must preserve terminal history and another process's writer lease;

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

Lines changed: 56 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -8588,12 +8588,35 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
85888588
session_id: &str,
85898589
turn_id: Option<&str>,
85908590
) -> OpenBitFunResult<Vec<DialogTurnData>> {
8591+
self.load_relay_session_selection(storage, session_id, turn_id, None)
8592+
.await
8593+
.map(|page| page.0)
8594+
}
8595+
8596+
pub async fn load_relay_history_turn(
8597+
&self,
8598+
storage: &Path,
8599+
session_id: &str,
8600+
before: Option<usize>,
8601+
) -> OpenBitFunResult<(Vec<DialogTurnData>, Option<usize>)> {
8602+
self.load_relay_session_selection(storage, session_id, None, Some(before))
8603+
.await
8604+
}
8605+
8606+
async fn load_relay_session_selection(
8607+
&self,
8608+
storage: &Path,
8609+
session_id: &str,
8610+
turn_id: Option<&str>,
8611+
history: Option<Option<usize>>,
8612+
) -> OpenBitFunResult<(Vec<DialogTurnData>, Option<usize>)> {
85918613
let _mutation = self
85928614
.session_manager
85938615
.acquire_session_mutation(session_id)
85948616
.await?;
85958617
self.prepare_persisted_session_read_locked(storage, session_id)
85968618
.await?;
8619+
85978620
// Persisted InProgress is not proof of a live executor after restart.
85988621
// A loaded owner supplies runtime state; for an unloaded session an
85998622
// 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
86188641
SessionState::Idle | SessionState::Error { .. }
86198642
)
86208643
}) || observer_lease.is_some();
8621-
let (mut turns, read_mode) = if let Some(turn_id) = turn_id {
8644+
8645+
let (mut turns, next, read_mode) = if let Some(before) = history {
8646+
let (turns, next) = self
8647+
.session_manager
8648+
.persistence_manager()
8649+
.load_visible_history_turn(storage, session_id, before)
8650+
.await?;
8651+
(turns, next, "history")
8652+
} else if let Some(turn_id) = turn_id {
86228653
if let Some(turn) = self
86238654
.session_manager
86248655
.persistence_manager()
86258656
.load_visible_session_turn(storage, session_id, turn_id)
86268657
.await?
86278658
{
8628-
(vec![turn], "catalog")
8659+
(vec![turn], None, "catalog")
86298660
} else {
86308661
let mut turns = self
86318662
.session_manager
86328663
.persistence_manager()
86338664
.load_visible_session_turns(storage, session_id)
86348665
.await?;
86358666
turns.retain(|turn| turn.turn_id == turn_id);
8636-
(turns, "full-fallback")
8667+
(turns, None, "full-fallback")
86378668
}
86388669
} else {
86398670
(
86408671
self.session_manager
86418672
.persistence_manager()
86428673
.load_visible_session_turns(storage, session_id)
86438674
.await?,
8675+
None,
86448676
"full",
86458677
)
86468678
};
@@ -8700,10 +8732,16 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
87008732
}
87018733
}
87028734
}
8703-
let context = self
8704-
.session_manager
8705-
.get_context_messages(session_id)
8706-
.await?;
8735+
let context = if turns
8736+
.iter()
8737+
.any(|turn| turn.status == TurnStatus::InProgress)
8738+
{
8739+
self.session_manager
8740+
.get_context_messages(session_id)
8741+
.await?
8742+
} else {
8743+
Vec::new()
8744+
};
87078745
for turn in &mut turns {
87088746
if turn.status == TurnStatus::InProgress {
87098747
let messages: Vec<_> = context
@@ -8718,7 +8756,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
87188756
SessionManager::append_generation_rounds(turn, &id, &messages, timestamp);
87198757
}
87208758
}
8721-
Ok(turns)
8759+
Ok((turns, next))
87228760
}
87238761

87248762
/// Export a transcript while retaining the same Session history boundary
@@ -17206,84 +17244,6 @@ mod tests {
1720617244
.expect("clean up persisted test session");
1720717245
}
1720817246

17209-
#[tokio::test]
17210-
async fn load_relay_session_turns_marks_abandoned_execution_without_rewriting_history() {
17211-
let workspace = tempfile::tempdir().expect("workspace");
17212-
crate::service::workspace::legacy_compat::register_local_fixture_blocking(workspace.path());
17213-
let (coordinator, manager) = test_persistent_coordinator();
17214-
let session = manager
17215-
.create_session(
17216-
"Restart".into(),
17217-
"Standard".into(),
17218-
SessionConfig {
17219-
workspace_path: Some(workspace.path().to_string_lossy().into_owned()),
17220-
..Default::default()
17221-
},
17222-
)
17223-
.await
17224-
.unwrap();
17225-
let id = &session.session_id;
17226-
let storage = manager.effective_session_storage_path(id).await.unwrap();
17227-
let turn_id = manager
17228-
.start_dialog_turn(
17229-
id,
17230-
"Standard".into(),
17231-
"restart probe".into(),
17232-
Some("turn-restart-probe".into()),
17233-
None,
17234-
None,
17235-
)
17236-
.await
17237-
.unwrap();
17238-
let live = coordinator
17239-
.load_relay_session_turns(&storage, id, None)
17240-
.await
17241-
.unwrap();
17242-
assert_eq!(live[0].status, TurnStatus::InProgress);
17243-
// Simulate the executor disappearing while the durable turn still says
17244-
// InProgress, as happens across process shutdown/restore.
17245-
manager.reset_session_state_if_processing(id, &turn_id);
17246-
let idle = coordinator
17247-
.load_relay_session_turns(&storage, id, None)
17248-
.await
17249-
.unwrap();
17250-
assert_eq!(idle[0].status, TurnStatus::Cancelled);
17251-
manager.unload_session_from_memory(id).await.unwrap();
17252-
// Another writer, even in the same process, forbids inferring death
17253-
// from this coordinator's empty in-memory map.
17254-
let owner = manager
17255-
.persistence_manager()
17256-
.lock_session_writes(&storage, id)
17257-
.unwrap();
17258-
let observed = coordinator
17259-
.load_relay_session_turns(&storage, id, None)
17260-
.await
17261-
.unwrap();
17262-
assert_eq!(observed[0].status, TurnStatus::InProgress);
17263-
drop(owner);
17264-
let orphan = coordinator
17265-
.load_relay_session_turns(&storage, id, None)
17266-
.await
17267-
.unwrap();
17268-
assert_eq!(orphan[0].status, TurnStatus::Cancelled);
17269-
assert_eq!(orphan[0].finish_reason.as_deref(), Some("interrupted"));
17270-
let single = coordinator
17271-
.load_relay_session_turns(&storage, id, Some(&turn_id))
17272-
.await
17273-
.unwrap();
17274-
assert_eq!(single[0].status, TurnStatus::Cancelled);
17275-
let stored = manager
17276-
.persistence_manager()
17277-
.load_visible_session_turns(&storage, id)
17278-
.await
17279-
.unwrap();
17280-
assert_eq!(
17281-
stored[0].status,
17282-
TurnStatus::InProgress,
17283-
"observer must retain persisted evidence"
17284-
);
17285-
}
17286-
1728717247
#[tokio::test]
1728817248
async fn load_relay_session_turns_reads_history_after_the_session_is_unloaded() {
1728917249
let workspace = tempfile::tempdir().expect("workspace");
@@ -17364,6 +17324,16 @@ mod tests {
1736417324
assert_eq!(one.len(), 1);
1736517325
assert_eq!(one[0].turn_id, turn_id);
1736617326

17327+
let (page, next) = coordinator
17328+
.load_relay_history_turn(&storage, &session.session_id, None)
17329+
.await
17330+
.expect("paged history must not require an in-memory writer");
17331+
assert_eq!(
17332+
serde_json::to_value(&page).unwrap(),
17333+
serde_json::to_value(&one).unwrap()
17334+
);
17335+
assert_eq!(next, None);
17336+
1736717337
session_manager
1736817338
.restore_session(workspace.path(), &session.session_id)
1736917339
.await

0 commit comments

Comments
 (0)