authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-01 18:07:26-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 18:07:26-07:00
logc2283ce14628127ef1758974ddab4f5b1f89d47a
treea875787f661606d6f973b866bff89423f0ff872e
parentfeb3ea7ebf83de2e301adb1f20300b262773ec54
signature Signed by PGP key B5690EEEBB952194

perf: make streaming projection and activity appends incremental (#9152)

Takes over #6608 by @x1xhlol. Server: thread.message-sent updates the projection row directly and folds latestUserMessageAt in as a running maximum instead of running refreshThreadShellSummary. Client: thread.activity-appended takes an O(1) append path for in-order, unseen activities on reducer-produced arrays, with a WeakMap id index, and falls back to filter and sort for out-of-order, re-delivered, context-window, and snapshot-loaded cases. Reducer benchmark: 10,000 in-order appends went from 1,180 ms to 70 ms. Change authored by Claude Fable 5.1 running in Claude Code, based on work by @x1xhlol. Co-authored-by: Lucas Valbuena <lucknitelol@proton.me>

4 files changed, 464 insertions(+), 31 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+227-25
......@@ -2422,44 +2422,35 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
24222422 json_object('requestId', 'user-input-active'),
24232423 NULL,
24242424 '2026-02-26T12:35:07.000Z'
2425 ),
2426 (
2427 'activity-user-input-active-failed',
2428 'thread-stale-user-input',
2429 NULL,
2430 'error',
2431 'provider.user-input.respond.failed',
2432 'Provider user input response failed',
2433 json_object(
2434 'requestId',
2435 'user-input-active',
2436 'detail',
2437 'Provider is temporarily unavailable'
2438 ),
2439 NULL,
2440 '2026-02-26T12:35:08.000Z'
24412425 )
24422426 `;
24432427
2428 // A user-input lifecycle activity is one of the events that still
2429 // refreshes the shell summary, so it forces the read under test.
24442430 yield* appendAndProject({
2445 type: "thread.message-sent",
2431 type: "thread.activity-appended",
24462432 eventId: EventId.make("evt-stale-user-input-3"),
24472433 aggregateKind: "thread",
24482434 aggregateId: ThreadId.make("thread-stale-user-input"),
2449 occurredAt: "2026-02-26T12:35:09.000Z",
2435 occurredAt: "2026-02-26T12:35:08.000Z",
24502436 commandId: CommandId.make("cmd-stale-user-input-3"),
24512437 causationEventId: null,
24522438 correlationId: CorrelationId.make("cmd-stale-user-input-3"),
24532439 metadata: {},
24542440 payload: {
24552441 threadId: ThreadId.make("thread-stale-user-input"),
2456 messageId: MessageId.make("message-stale-user-input"),
2457 role: "user",
2458 text: "Continue",
2459 turnId: null,
2460 streaming: false,
2461 createdAt: "2026-02-26T12:35:09.000Z",
2462 updatedAt: "2026-02-26T12:35:09.000Z",
2442 activity: {
2443 id: EventId.make("activity-user-input-active-failed"),
2444 tone: "error",
2445 kind: "provider.user-input.respond.failed",
2446 summary: "Provider user input response failed",
2447 payload: {
2448 requestId: "user-input-active",
2449 detail: "Provider is temporarily unavailable",
2450 },
2451 turnId: null,
2452 createdAt: "2026-02-26T12:35:08.000Z",
2453 },
24632454 },
24642455 });
24652456
......@@ -2474,6 +2465,217 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
24742465 }),
24752466 );
24762467
2468 it.effect("maintains shell summary fields across message and activity streams", () =>
2469 Effect.gen(function* () {
2470 const projectionPipeline = yield* OrchestrationProjectionPipeline;
2471 const eventStore = yield* OrchestrationEventStore;
2472 const sql = yield* SqlClient.SqlClient;
2473 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
2474 eventStore
2475 .append(event)
2476 .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
2477
2478 yield* appendAndProject({
2479 type: "project.created",
2480 eventId: EventId.make("evt-shell-summary-1"),
2481 aggregateKind: "project",
2482 aggregateId: ProjectId.make("project-shell-summary"),
2483 occurredAt: "2026-03-01T08:00:00.000Z",
2484 commandId: CommandId.make("cmd-shell-summary-1"),
2485 causationEventId: null,
2486 correlationId: CorrelationId.make("cmd-shell-summary-1"),
2487 metadata: {},
2488 payload: {
2489 projectId: ProjectId.make("project-shell-summary"),
2490 title: "Project Shell Summary",
2491 workspaceRoot: "/tmp/project-shell-summary",
2492 defaultModelSelection: null,
2493 scripts: [],
2494 createdAt: "2026-03-01T08:00:00.000Z",
2495 updatedAt: "2026-03-01T08:00:00.000Z",
2496 },
2497 });
2498
2499 yield* appendAndProject({
2500 type: "thread.created",
2501 eventId: EventId.make("evt-shell-summary-2"),
2502 aggregateKind: "thread",
2503 aggregateId: ThreadId.make("thread-shell-summary"),
2504 occurredAt: "2026-03-01T08:00:01.000Z",
2505 commandId: CommandId.make("cmd-shell-summary-2"),
2506 causationEventId: null,
2507 correlationId: CorrelationId.make("cmd-shell-summary-2"),
2508 metadata: {},
2509 payload: {
2510 threadId: ThreadId.make("thread-shell-summary"),
2511 projectId: ProjectId.make("project-shell-summary"),
2512 title: "Thread Shell Summary",
2513 modelSelection: {
2514 instanceId: ProviderInstanceId.make("codex"),
2515 model: "gpt-5-codex",
2516 },
2517 runtimeMode: "approval-required",
2518 interactionMode: "default",
2519 branch: null,
2520 worktreePath: null,
2521 createdAt: "2026-03-01T08:00:01.000Z",
2522 updatedAt: "2026-03-01T08:00:01.000Z",
2523 },
2524 });
2525
2526 const readSummary = sql<{
2527 readonly latestUserMessageAt: string | null;
2528 readonly pendingUserInputCount: number;
2529 readonly updatedAt: string;
2530 }>`
2531 SELECT
2532 latest_user_message_at AS "latestUserMessageAt",
2533 pending_user_input_count AS "pendingUserInputCount",
2534 updated_at AS "updatedAt"
2535 FROM projection_threads
2536 WHERE thread_id = 'thread-shell-summary'
2537 `;
2538
2539 yield* appendAndProject({
2540 type: "thread.message-sent",
2541 eventId: EventId.make("evt-shell-summary-3"),
2542 aggregateKind: "thread",
2543 aggregateId: ThreadId.make("thread-shell-summary"),
2544 occurredAt: "2026-03-01T08:00:02.000Z",
2545 commandId: CommandId.make("cmd-shell-summary-3"),
2546 causationEventId: null,
2547 correlationId: CorrelationId.make("cmd-shell-summary-3"),
2548 metadata: {},
2549 payload: {
2550 threadId: ThreadId.make("thread-shell-summary"),
2551 messageId: MessageId.make("message-shell-summary-user"),
2552 role: "user",
2553 text: "please do the thing",
2554 turnId: TurnId.make("turn-shell-summary-1"),
2555 streaming: false,
2556 createdAt: "2026-03-01T08:00:02.000Z",
2557 updatedAt: "2026-03-01T08:00:02.000Z",
2558 },
2559 });
2560
2561 assert.deepEqual(yield* readSummary, [
2562 {
2563 latestUserMessageAt: "2026-03-01T08:00:02.000Z",
2564 pendingUserInputCount: 0,
2565 updatedAt: "2026-03-01T08:00:02.000Z",
2566 },
2567 ]);
2568
2569 // Streaming assistant deltas bump updatedAt but must not disturb
2570 // latestUserMessageAt or the pending counters.
2571 yield* appendAndProject({
2572 type: "thread.message-sent",
2573 eventId: EventId.make("evt-shell-summary-4"),
2574 aggregateKind: "thread",
2575 aggregateId: ThreadId.make("thread-shell-summary"),
2576 occurredAt: "2026-03-01T08:00:03.000Z",
2577 commandId: CommandId.make("cmd-shell-summary-4"),
2578 causationEventId: null,
2579 correlationId: CorrelationId.make("cmd-shell-summary-4"),
2580 metadata: {},
2581 payload: {
2582 threadId: ThreadId.make("thread-shell-summary"),
2583 messageId: MessageId.make("message-shell-summary-assistant"),
2584 role: "assistant",
2585 text: "working on it",
2586 turnId: TurnId.make("turn-shell-summary-1"),
2587 streaming: true,
2588 createdAt: "2026-03-01T08:00:03.000Z",
2589 updatedAt: "2026-03-01T08:00:03.000Z",
2590 },
2591 });
2592
2593 assert.deepEqual(yield* readSummary, [
2594 {
2595 latestUserMessageAt: "2026-03-01T08:00:02.000Z",
2596 pendingUserInputCount: 0,
2597 updatedAt: "2026-03-01T08:00:03.000Z",
2598 },
2599 ]);
2600
2601 // Ordinary tool activities bump updatedAt without touching the
2602 // user-input counter; user-input lifecycle activities update it.
2603 yield* appendAndProject({
2604 type: "thread.activity-appended",
2605 eventId: EventId.make("evt-shell-summary-5"),
2606 aggregateKind: "thread",
2607 aggregateId: ThreadId.make("thread-shell-summary"),
2608 occurredAt: "2026-03-01T08:00:04.000Z",
2609 commandId: CommandId.make("cmd-shell-summary-5"),
2610 causationEventId: null,
2611 correlationId: CorrelationId.make("cmd-shell-summary-5"),
2612 metadata: {},
2613 payload: {
2614 threadId: ThreadId.make("thread-shell-summary"),
2615 activity: {
2616 id: EventId.make("activity-shell-summary-command"),
2617 tone: "tool",
2618 kind: "command",
2619 summary: "Ran a command",
2620 payload: {},
2621 turnId: TurnId.make("turn-shell-summary-1"),
2622 createdAt: "2026-03-01T08:00:04.000Z",
2623 },
2624 },
2625 });
2626
2627 assert.deepEqual(yield* readSummary, [
2628 {
2629 latestUserMessageAt: "2026-03-01T08:00:02.000Z",
2630 pendingUserInputCount: 0,
2631 updatedAt: "2026-03-01T08:00:04.000Z",
2632 },
2633 ]);
2634
2635 yield* appendAndProject({
2636 type: "thread.activity-appended",
2637 eventId: EventId.make("evt-shell-summary-6"),
2638 aggregateKind: "thread",
2639 aggregateId: ThreadId.make("thread-shell-summary"),
2640 occurredAt: "2026-03-01T08:00:05.000Z",
2641 commandId: CommandId.make("cmd-shell-summary-6"),
2642 causationEventId: null,
2643 correlationId: CorrelationId.make("cmd-shell-summary-6"),
2644 metadata: {},
2645 payload: {
2646 threadId: ThreadId.make("thread-shell-summary"),
2647 activity: {
2648 id: EventId.make("activity-shell-summary-user-input"),
2649 tone: "info",
2650 kind: "user-input.requested",
2651 summary: "User input requested",
2652 payload: {
2653 requestId: "user-input-request-shell-summary-1",
2654 questions: [
2655 {
2656 id: "confirm",
2657 header: "Confirm",
2658 question: "Proceed?",
2659 options: [{ label: "yes", description: "Proceed" }],
2660 },
2661 ],
2662 },
2663 turnId: TurnId.make("turn-shell-summary-1"),
2664 createdAt: "2026-03-01T08:00:05.000Z",
2665 },
2666 },
2667 });
2668
2669 assert.deepEqual(yield* readSummary, [
2670 {
2671 latestUserMessageAt: "2026-03-01T08:00:02.000Z",
2672 pendingUserInputCount: 1,
2673 updatedAt: "2026-03-01T08:00:05.000Z",
2674 },
2675 ]);
2676 }),
2677 );
2678
24772679 it.effect("ignores non-stale provider approval response failures", () =>
24782680 Effect.gen(function* () {
24792681 const projectionPipeline = yield* OrchestrationProjectionPipeline;
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+24-6
......@@ -130,12 +130,8 @@ function isStalePendingApprovalFailureDetail(detail: string | null): boolean {
130130 );
131131}
132132
133// A refresh reads each persisted summary source, so skip events that cannot change the result.
133// A refresh reads each persisted summary source, so skip activities that cannot change the result.
134134function shouldRefreshThreadShellSummary(event: OrchestrationEvent): boolean {
135 if (event.type === "thread.message-sent") {
136 return event.payload.role === "user";
137 }
138
139135 if (event.type !== "thread.activity-appended") {
140136 return true;
141137 }
......@@ -897,7 +893,29 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
897893 return;
898894 }
899895
900 case "thread.message-sent":
896 // A message cannot change any summary field except latestUserMessageAt,
897 // which is a monotonic maximum that folds in directly. The full refresh
898 // would re-read every message body in the thread per user message.
899 case "thread.message-sent": {
900 const existingRow = yield* projectionThreadRepository.getById({
901 threadId: event.payload.threadId,
902 });
903 if (Option.isNone(existingRow)) {
904 return;
905 }
906 const previousLatest = existingRow.value.latestUserMessageAt;
907 yield* projectionThreadRepository.upsert({
908 ...existingRow.value,
909 updatedAt: event.occurredAt,
910 latestUserMessageAt:
911 event.payload.role === "user" &&
912 (previousLatest === null || event.payload.createdAt > previousLatest)
913 ? event.payload.createdAt
914 : previousLatest,
915 });
916 return;
917 }
918
901919 case "thread.proposed-plan-upserted":
902920 case "thread.activity-appended":
903921 case "thread.approval-response-requested":
packages/client-runtime/src/state/threadReducer.test.ts+178
......@@ -728,6 +728,184 @@ describe("applyThreadDetailEvent", () => {
728728 }
729729 });
730730
731 it("re-sorts when an activity arrives out of order", () => {
732 const makeActivity = (id: string, sequence: number) => ({
733 id: EventId.make(id),
734 tone: "tool" as const,
735 kind: "command",
736 summary: `Ran command ${sequence}`,
737 payload: {},
738 turnId: TurnId.make("turn-1"),
739 sequence,
740 createdAt: "2026-04-01T11:00:00.000Z",
741 });
742 const result = applyThreadDetailEvent(
743 {
744 ...baseThread,
745 activities: [makeActivity("activity-a", 1), makeActivity("activity-c", 3)],
746 },
747 {
748 ...baseEventFields,
749 sequence: 131,
750 occurredAt: "2026-04-01T11:01:00.000Z",
751 aggregateKind: "thread",
752 aggregateId: ThreadId.make("thread-1"),
753 type: "thread.activity-appended",
754 payload: {
755 threadId: ThreadId.make("thread-1"),
756 activity: makeActivity("activity-b", 2),
757 },
758 },
759 );
760
761 expect(result.kind).toBe("updated");
762 if (result.kind === "updated") {
763 expect(result.thread.activities.map((activity) => activity.id)).toEqual([
764 "activity-a",
765 "activity-b",
766 "activity-c",
767 ]);
768 }
769 });
770
771 it("repairs snapshot ordering before fast-path appends engage", () => {
772 const makeActivity = (id: string, sequence: number | null) => ({
773 id: EventId.make(id),
774 tone: "tool" as const,
775 kind: "command",
776 summary: `Ran ${id}`,
777 payload: {},
778 turnId: TurnId.make("turn-1"),
779 ...(sequence === null ? {} : { sequence }),
780 createdAt: "2026-04-01T11:00:00.000Z",
781 });
782 // Snapshot loads deliver null-sequence rows first (DB order), which
783 // activityOrder sorts last; an in-order live append must not freeze
784 // that prefix.
785 const result = applyThreadDetailEvent(
786 {
787 ...baseThread,
788 activities: [makeActivity("activity-null", null), makeActivity("activity-a", 1)],
789 },
790 {
791 ...baseEventFields,
792 sequence: 135,
793 occurredAt: "2026-04-01T11:01:00.000Z",
794 aggregateKind: "thread",
795 aggregateId: ThreadId.make("thread-1"),
796 type: "thread.activity-appended",
797 payload: {
798 threadId: ThreadId.make("thread-1"),
799 activity: makeActivity("activity-b", 2),
800 },
801 },
802 );
803
804 expect(result.kind).toBe("updated");
805 if (result.kind === "updated") {
806 expect(result.thread.activities.map((activity) => activity.id)).toEqual([
807 "activity-a",
808 "activity-b",
809 "activity-null",
810 ]);
811 }
812 });
813
814 it("dedupes a re-delivery arriving right after an in-order append", () => {
815 const makeActivity = (id: string, sequence: number, summary: string) => ({
816 id: EventId.make(id),
817 tone: "tool" as const,
818 kind: "command",
819 summary,
820 payload: {},
821 turnId: TurnId.make("turn-1"),
822 sequence,
823 createdAt: "2026-04-01T11:00:00.000Z",
824 });
825 const makeEvent = (sequence: number, activity: ReturnType<typeof makeActivity>) =>
826 ({
827 ...baseEventFields,
828 sequence,
829 occurredAt: "2026-04-01T11:01:00.000Z",
830 aggregateKind: "thread",
831 aggregateId: ThreadId.make("thread-1"),
832 type: "thread.activity-appended",
833 payload: { threadId: ThreadId.make("thread-1"), activity },
834 }) as const;
835 const first = applyThreadDetailEvent(
836 { ...baseThread, activities: [makeActivity("activity-a", 1, "first")] },
837 makeEvent(133, makeActivity("activity-b", 2, "second")),
838 );
839 expect(first.kind).toBe("updated");
840 if (first.kind !== "updated") {
841 return;
842 }
843 const second = applyThreadDetailEvent(
844 first.thread,
845 makeEvent(134, makeActivity("activity-c", 3, "third")),
846 );
847 expect(second.kind).toBe("updated");
848 if (second.kind !== "updated") {
849 return;
850 }
851 const third = applyThreadDetailEvent(
852 second.thread,
853 makeEvent(135, makeActivity("activity-c", 4, "third (redelivered)")),
854 );
855 expect(third.kind).toBe("updated");
856 if (third.kind === "updated") {
857 expect(third.thread.activities.map((activity) => activity.id)).toEqual([
858 "activity-a",
859 "activity-b",
860 "activity-c",
861 ]);
862 expect(third.thread.activities[2]?.summary).toBe("third (redelivered)");
863 }
864 });
865
866 it("replaces a re-delivered activity instead of duplicating it", () => {
867 const makeActivity = (id: string, sequence: number, summary: string) => ({
868 id: EventId.make(id),
869 tone: "tool" as const,
870 kind: "command",
871 summary,
872 payload: {},
873 turnId: TurnId.make("turn-1"),
874 sequence,
875 createdAt: "2026-04-01T11:00:00.000Z",
876 });
877 const result = applyThreadDetailEvent(
878 {
879 ...baseThread,
880 activities: [
881 makeActivity("activity-a", 1, "first"),
882 makeActivity("activity-b", 2, "second"),
883 ],
884 },
885 {
886 ...baseEventFields,
887 sequence: 132,
888 occurredAt: "2026-04-01T11:01:00.000Z",
889 aggregateKind: "thread",
890 aggregateId: ThreadId.make("thread-1"),
891 type: "thread.activity-appended",
892 payload: {
893 threadId: ThreadId.make("thread-1"),
894 activity: makeActivity("activity-b", 2, "second (redelivered)"),
895 },
896 },
897 );
898
899 expect(result.kind).toBe("updated");
900 if (result.kind === "updated") {
901 expect(result.thread.activities.map((activity) => activity.id)).toEqual([
902 "activity-a",
903 "activity-b",
904 ]);
905 expect(result.thread.activities[1]?.summary).toBe("second (redelivered)");
906 }
907 });
908
731909 it("replaces earlier resolvable context-window updates for the same turn", () => {
732910 const contextWindowActivity = (id: string, sequence: number, usedTokens: unknown) => ({
733911 id: EventId.make(id),
packages/client-runtime/src/state/threadReducer.ts+35
......@@ -35,6 +35,15 @@ const activityOrder = O.combineAll<OrchestrationThreadActivity>([
3535 O.mapInput(O.String, (a) => a.id),
3636]);
3737
38// Per-array id index so the streaming append path can reject a re-delivered
39// id without rescanning the history. Only arrays this reducer produced are
40// indexed: presence also proves the array is activityOrder-sorted, which
41// snapshot-loaded arrays (DB order, null sequences first) are not.
42const activityIdIndex = new WeakMap<
43 ReadonlyArray<OrchestrationThreadActivity>,
44 Set<OrchestrationThreadActivity["id"]>
45>();
46
3847/**
3948 * Matches the validity rule in `deriveLatestContextWindowSnapshot` (and the
4049 * server's snapshot-side `dropStaleContextWindowActivities`): rows without a
......@@ -582,6 +591,31 @@ export function applyThreadDetailEvent(
582591 // thread.reverted that discards turns can still resolve a value from
583592 // the turns that survive.
584593 const supersedesContextWindow = isResolvableContextWindowActivity(activity);
594 // Live streams append in order: an unseen id sorting at/after the tail
595 // of a known-sorted array appends without re-filtering and re-sorting
596 // the whole history on every event. The id set moves forward to the new
597 // array; a superseded array falls back to the sorting path.
598 const ids = activityIdIndex.get(thread.activities);
599 const lastActivity = thread.activities.at(-1);
600 if (
601 !supersedesContextWindow &&
602 ids !== undefined &&
603 (lastActivity === undefined || activityOrder(lastActivity, activity) <= 0) &&
604 !ids.has(activity.id)
605 ) {
606 const activities = Arr.append(thread.activities, activity);
607 activityIdIndex.delete(thread.activities);
608 ids.add(activity.id);
609 activityIdIndex.set(activities, ids);
610 return {
611 kind: "updated",
612 thread: {
613 ...thread,
614 activities,
615 updatedAt: event.occurredAt,
616 },
617 };
618 }
585619 const activities = pipe(
586620 thread.activities,
587621 Arr.filter(
......@@ -596,6 +630,7 @@ export function applyThreadDetailEvent(
596630 Arr.append(activity),
597631 Arr.sort(activityOrder),
598632 );
633 activityIdIndex.set(activities, new Set(activities.map((entry) => entry.id)));
599634
600635 return {
601636 kind: "updated",