1import {
2 isCompactionCommandMessage,
3 type ThreadGuestMessage,
4 type ThreadGuestShare,
5} from "@t3tools/contracts";
6import type { ThreadFeedEntry } from "../../lib/threadActivity";
7import { scopedThreadKey } from "../../lib/scopedEntities";
8import { threadHasCaughtUpTo } from "@t3tools/client-runtime/state/threads";
9import { resolveUserMessageGrouping } from "@t3tools/client-runtime/user-message";
10import type {
11 AcknowledgedThreadMessage,
12 QueuedThreadMessage,
13} from "../../state/thread-outbox-model";
14
15/**
16 * A guest's message waiting for the owner's Send or Dismiss. It is a feed row, not the list
17 * footer, so the feed follows it arriving and leaving the way it follows any row.
18 */
19export interface GuestPendingFeedEntry {
20 readonly type: "guest-pending";
21 readonly id: string;
22 readonly createdAt: string;
23 readonly guestName: string;
24 readonly message: ThreadGuestMessage;
25}
26
27export type PendingThreadFeedEntry = ThreadFeedEntry & {
28 readonly pendingMessage?: QueuedThreadMessage;
29 readonly acknowledged?: boolean;
30};
31
32// An acknowledged row stops being pending once the thread catches up to it,
33// even when the feed hides that message (a rolled-back or cancelled run).
34export function selectPendingAcknowledgedMessages(
35 acknowledged: ReadonlyArray<AcknowledgedThreadMessage>,
36 thread: {
37 readonly threadKey: string | null;
38 readonly queuedMessages: ReadonlyArray<QueuedThreadMessage>;
39 readonly sequence: number;
40 readonly projectedMessages: ReadonlyArray<{ readonly id: string }> | undefined;
41 },
42): ReadonlyArray<AcknowledgedThreadMessage> {
43 return acknowledged.filter(
44 (message) =>
45 scopedThreadKey(message.environmentId, message.threadId) === thread.threadKey &&
46 !thread.queuedMessages.some((queued) => queued.messageId === message.messageId) &&
47 !threadHasCaughtUpTo(message, {
48 sequence: thread.sequence,
49 messages: thread.projectedMessages,
50 }),
51 );
52}
53
54/** A row of the thread feed's list. */
55export type ThreadFeedRow = PendingThreadFeedEntry | GuestPendingFeedEntry;
56
57export type GroupedThreadFeedRow = ThreadFeedRow & {
58 readonly userMessageGrouping: ReturnType<typeof resolveUserMessageGrouping>;
59};
60
61export function createThreadFeedGrouping() {
62 const rows = new WeakMap<ThreadFeedRow, GroupedThreadFeedRow>();
63 const ungrouped = resolveUserMessageGrouping(undefined, undefined, undefined);
64 return (
65 feed: ReadonlyArray<ThreadFeedRow>,
66 options: { readonly workflowOpeningMessageId?: string; readonly setupAnchorIndex?: number },
67 ): ReadonlyArray<GroupedThreadFeedRow> => {
68 const groupingMessage = (index: number) => {
69 const entry = feed[index];
70 return entry?.type === "message" &&
71 entry.message.role === "user" &&
72 !isCompactionCommandFeedEntry(entry) &&
73 entry.message.id !== options.workflowOpeningMessageId
74 ? entry.message
75 : undefined;
76 };
77 return feed.map((entry, index) => {
78 const message = groupingMessage(index);
79 const grouping =
80 message === undefined
81 ? ungrouped
82 : resolveUserMessageGrouping(
83 message,
84 index - 1 === options.setupAnchorIndex ? undefined : groupingMessage(index - 1),
85 index === options.setupAnchorIndex ? undefined : groupingMessage(index + 1),
86 );
87 const previous = rows.get(entry);
88 if (
89 previous?.userMessageGrouping.groupedWithPrevious === grouping.groupedWithPrevious &&
90 previous.userMessageGrouping.groupedWithNext === grouping.groupedWithNext &&
91 previous.userMessageGrouping.continuesSteer === grouping.continuesSteer
92 )
93 return previous;
94 const row = { ...entry, userMessageGrouping: grouping };
95 rows.set(entry, row);
96 return row;
97 });
98 };
99}
100
101/** Waiting guest messages go after the transcript, oldest first. */
102export function appendGuestPendingMessages(
103 presentedFeed: ReadonlyArray<PendingThreadFeedEntry>,
104 share: ThreadGuestShare | null,
105): ReadonlyArray<ThreadFeedRow> {
106 const pending = share?.messages.filter((message) => message.status === "pending") ?? [];
107 if (share === null || pending.length === 0) return presentedFeed;
108 const guestName = share.guestName ?? "Guest";
109 return [
110 ...presentedFeed,
111 ...pending.map((message): GuestPendingFeedEntry => ({
112 type: "guest-pending",
113 id: `guest-pending:${message.id}`,
114 createdAt: message.submittedAt,
115 guestName,
116 message,
117 })),
118 ];
119}
120
121export function appendPendingThreadMessages(
122 presentedFeed: ReadonlyArray<ThreadFeedEntry>,
123 feed: ReadonlyArray<ThreadFeedEntry>,
124 queuedMessages: ReadonlyArray<QueuedThreadMessage>,
125): ReadonlyArray<PendingThreadFeedEntry> {
126 if (queuedMessages.length === 0) return presentedFeed;
127 const deliveredIds = new Set(
128 feed.flatMap((entry) => (entry.type === "message" ? [entry.message.id] : [])),
129 );
130 return [
131 ...presentedFeed.map((entry): PendingThreadFeedEntry => {
132 const pendingMessage = queuedMessages.find(
133 (message) => message.creation !== undefined && message.messageId === entry.id,
134 );
135 return pendingMessage ? { ...entry, pendingMessage, acknowledged: false } : entry;
136 }),
137 ...queuedMessages
138 .filter((message) => !deliveredIds.has(message.messageId))
139 .map((pendingMessage): PendingThreadFeedEntry => ({
140 type: "message",
141 id: pendingMessage.messageId,
142 createdAt: pendingMessage.createdAt,
143 pendingMessage,
144 message: {
145 id: pendingMessage.messageId,
146 role: "user",
147 text: pendingMessage.text,
148 attachments: [],
149 context: pendingMessage.context,
150 createdAt: pendingMessage.createdAt,
151 updatedAt: pendingMessage.createdAt,
152 runId: null,
153 streaming: false,
154 visibility: "local",
155 sourceThreadId: pendingMessage.threadId,
156 },
157 })),
158 ];
159}
160
161export function isCompactionCommandFeedEntry(entry: PendingThreadFeedEntry): boolean {
162 return (
163 entry.type === "message" &&
164 entry.pendingMessage === undefined &&
165 isCompactionCommandMessage(entry.message)
166 );
167}