authorgravatar for 62337003+Bil0000@users.noreply.github.comBilal Bakr <62337003+Bil0000@users.noreply.github.com> 2026-09-12 00:40:27+03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-11 14:40:27-07:00
loge145c5f22a10d7433420b447537ed3f29a66d7d9
treee06fe166764e09b20f99df67f7c12467602b48f5
parent18f7254e0adbd7d642f4d78f9b9c8e6cb569af56
signature Signed by PGP key B5690EEEBB952194

perf(client-runtime): speed up message sync on desktop and mobile (#11302)

3 files changed, 141 insertions(+), 3 deletions(-)

packages/client-runtime/src/state/threads-atoms.test.ts+62
......@@ -619,6 +619,68 @@ describe("createEnvironmentThreadStateAtoms", () => {
619619 }),
620620 );
621621
622 it.effect.each([1, 16, 500])("publishes each replay batch once (batch size: %i)", (batchSize) =>
623 Effect.gen(function* () {
624 const h = yield* makeHarness();
625 const unmount = h.registry.mount(h.stateAtom);
626 const first = yield* Queue.take(h.subscriptions);
627 let updates = 0;
628 const stop = h.registry.subscribe(h.details.messagesAtom(h.ref), () => updates++, {
629 immediate: true,
630 });
631 updates = 0;
632 const events: OrchestrationThreadStreamItem[] = Array.from({ length: 500 }, (_, index) => ({
633 kind: "event",
634 event: {
635 type: "thread.message-sent",
636 sequence: 8 + index,
637 eventId: EventId.make(`replay-${index}`),
638 aggregateKind: "thread",
639 aggregateId: THREAD_ID,
640 occurredAt: THREAD.createdAt,
641 commandId: null,
642 causationEventId: null,
643 correlationId: null,
644 metadata: {},
645 payload: {
646 threadId: THREAD_ID,
647 messageId: MessageId.make("replayed-message"),
648 role: "assistant",
649 text: `${index},`,
650 turnId: null,
651 streaming: true,
652 createdAt: THREAD.createdAt,
653 updatedAt: THREAD.createdAt,
654 },
655 },
656 }));
657 for (let offset = 0; offset < events.length; offset += batchSize) {
658 yield* Queue.offerAll(first.events, events.slice(offset, offset + batchSize));
659 const last = Math.min(offset + batchSize, events.length) - 1;
660 yield* observeState(
661 h.registry,
662 h.stateAtom,
663 (state) => Option.getOrNull(state.data)?.messages[0]?.text.endsWith(`${last},`) === true,
664 );
665 }
666 yield* Queue.offerAll(first.events, [events[499]!, events[0]!]);
667 yield* Queue.offer(first.events, { kind: "synchronized" });
668 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
669 expect(currentThread(h.registry, h.stateAtom).messages[0]?.text).toBe(
670 Array.from({ length: 500 }, (_, index) => `${index},`).join(""),
671 );
672 expect(updates).toBe(Math.ceil(500 / batchSize));
673 stop();
674 unmount();
675 yield* Deferred.await(first.closed);
676 const remount = h.registry.mount(h.stateAtom);
677 const next = yield* Queue.take(h.subscriptions);
678 expect(next.afterSequence).toBe(507);
679 remount();
680 yield* Deferred.await(next.closed);
681 }),
682 );
683
622684 it.effect("keeps warm data and resumes a completed cursor without loading another snapshot", () =>
623685 Effect.gen(function* () {
624686 const h = yield* makeHarness();
packages/client-runtime/src/state/threads-pagination.test.ts+31-2
......@@ -473,8 +473,37 @@ describe("thread pagination state", () => {
473473 );
474474
475475 // A live event at sequence 11 arrives; only then does the page merge.
476 yield* Queue.offer(harness.inputs, titleEvent("Advanced past watermark", 11));
477 const state = yield* harness.awaitState((value) => hasMessage(value, "message-old"));
476 yield* Queue.offerAll(harness.inputs, [
477 titleEvent("Advanced past watermark", 11),
478 {
479 kind: "event",
480 event: {
481 eventId: EventId.make("event-after-page"),
482 sequence: 12,
483 aggregateKind: "thread",
484 aggregateId: THREAD_ID,
485 occurredAt: BASE_THREAD.createdAt,
486 commandId: null,
487 causationEventId: null,
488 correlationId: null,
489 metadata: {},
490 type: "thread.message-sent",
491 payload: {
492 ...OLDER_MESSAGE,
493 threadId: THREAD_ID,
494 messageId: OLDER_MESSAGE.id,
495 text: " continued",
496 streaming: true,
497 },
498 },
499 },
500 ]);
501 const state = yield* harness.awaitState(
502 (value) => Option.getOrNull(value.data)?.messages[0]?.text.endsWith(" continued") === true,
503 );
504 expect(Option.getOrThrow(state.data).messages[0]?.text).toBe(
505 `${OLDER_MESSAGE.text} continued`,
506 );
478507 expect(hasMessage(state, "message-recent")).toBe(true);
479508 expect(Option.getOrThrow(state.page).loadingOlder).toBe(false);
480509 }),
packages/client-runtime/src/state/threads.ts+48-1
......@@ -513,6 +513,49 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
513513 yield* applyLock.withPermits(1)(applyItemLocked(item).pipe(Effect.andThen(remember)));
514514 });
515515
516 const applyItems = Effect.fn("EnvironmentThreadState.applyItems")(function* (
517 items: ReadonlyArray<OrchestrationThreadStreamItem>,
518 ) {
519 yield* applyLock.withPermits(1)(
520 Effect.gen(function* () {
521 const current = yield* SubscriptionRef.get(state);
522 if (
523 Option.isNone(current.data) ||
524 (yield* Ref.get(pendingOlderPage)) !== null ||
525 items.some(
526 (item) =>
527 item.kind === "snapshot" ||
528 (item.kind === "event" &&
529 (item.event.type === "thread.reverted" || item.event.type === "thread.deleted")),
530 )
531 ) {
532 for (const item of items) {
533 yield* applyItemLocked(item);
534 yield* remember;
535 }
536 return;
537 }
538
539 let thread = current.data.value;
540 let sequence = yield* SubscriptionRef.get(lastSequence);
541 let synchronized = false;
542 for (const item of items) {
543 if (item.kind === "synchronized") {
544 synchronized = true;
545 } else if (item.kind === "event" && item.event.sequence > sequence) {
546 sequence = item.event.sequence;
547 const result = applyThreadDetailEvent(thread, item.event);
548 if (result.kind === "updated") thread = result.thread;
549 }
550 }
551 yield* SubscriptionRef.set(lastSequence, sequence);
552 if (thread !== current.data.value) yield* setThread(thread, "keep");
553 if (synchronized) yield* applyItemLocked({ kind: "synchronized" });
554 yield* remember;
555 }),
556 );
557 });
558
516559 // Merges an older disjoint page below the currently loaded window. All four
517560 // windowed collections prepend; identity dedupe guards the (server-bug or
518561 // cursor-misuse) case of overlapping pages so a row never renders twice.
......@@ -773,7 +816,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
773816 retryExpectedFailureAfter: "250 millis",
774817 resubscribe: foregroundResubscriptions,
775818 },
776 ).pipe(Stream.runForEach(applyItem)),
819 ).pipe(
820 Stream.runForEachArray((items) =>
821 items.length === 1 ? applyItem(items[0]!) : applyItems(items),
822 ),
823 ),
777824 );
778825
779826 // Expose loadOlderTurns to UI actions through the request registry.