Carlos Florêncio
961 lines
29 KiB
TypeScript
961 lines
29 KiB
TypeScript
/* eslint-disable max-lines -- The facade hook centralizes ordered snapshot/live state coordination. */
|
|
"use client";
|
|
|
|
import * as React from "react";
|
|
import { useMessageFavoritesStore } from "@/lib/state/slices/message-favorites";
|
|
import {
|
|
compareConversationMessages,
|
|
eventMatchesTask,
|
|
eventString,
|
|
messageFromEvent,
|
|
turnFromEvent,
|
|
} from "./conversation-event-projection";
|
|
import type {
|
|
PluginConversationApi,
|
|
PluginConversationError,
|
|
PluginConversationMessage,
|
|
PluginConversationTurn,
|
|
PluginSessionMessagesQuery,
|
|
PluginSessionMessagesState,
|
|
PluginSessionTurnsState,
|
|
} from "./types";
|
|
import {
|
|
ConversationScopeContext,
|
|
parseConversationResponse,
|
|
pluginConversationUrl,
|
|
type ConversationScope,
|
|
type OrderedReady,
|
|
} from "./conversation-scope";
|
|
export { PluginConversationScopeProvider } from "./conversation-scope";
|
|
|
|
type MessagePage = {
|
|
messages: PluginConversationMessage[];
|
|
hasMore: boolean;
|
|
cursor: string | null;
|
|
epoch?: string;
|
|
revision?: string;
|
|
};
|
|
|
|
type TurnsPage = {
|
|
turns: PluginConversationTurn[];
|
|
hasMore?: boolean;
|
|
cursor?: string | null;
|
|
epoch?: string;
|
|
revision?: string;
|
|
};
|
|
|
|
const EMPTY_MESSAGES: PluginSessionMessagesState = {
|
|
messages: [],
|
|
loading: false,
|
|
hydrated: false,
|
|
loadingMore: false,
|
|
error: null,
|
|
hasMore: false,
|
|
removed: false,
|
|
loadMore: async () => 0,
|
|
retry: () => undefined,
|
|
};
|
|
|
|
const EMPTY_TURNS: PluginSessionTurnsState = {
|
|
turns: [],
|
|
loading: false,
|
|
hydrated: false,
|
|
error: null,
|
|
removed: false,
|
|
retry: () => undefined,
|
|
};
|
|
|
|
function conversationError(error: unknown): PluginConversationError {
|
|
if (typeof error === "object" && error !== null && "code" in error) {
|
|
return error as PluginConversationError;
|
|
}
|
|
return {
|
|
code: "upstream_failure",
|
|
// i18n-exempt: plugin-facing transport diagnostic; plugins render their own localized error UI.
|
|
message: error instanceof Error ? error.message : "Conversation request failed",
|
|
retryable: true,
|
|
};
|
|
}
|
|
|
|
function isRemovedConversationError(error: PluginConversationError): boolean {
|
|
return error.code === "not_found";
|
|
}
|
|
|
|
function resolveTaskId(
|
|
scope: ConversationScope,
|
|
taskId: string | null | undefined,
|
|
): { taskId: string | null; error: PluginConversationError | null } {
|
|
const resolved = taskId === undefined ? scope.taskId : taskId;
|
|
if (resolved === "" || (typeof resolved === "string" && resolved !== scope.taskId)) {
|
|
return {
|
|
taskId: null,
|
|
error: {
|
|
code: "invalid_query",
|
|
// i18n-exempt: plugin-facing programmer diagnostic; plugins render their own localized error UI.
|
|
message: "taskId must match the active task",
|
|
retryable: false,
|
|
},
|
|
};
|
|
}
|
|
return { taskId: resolved, error: null };
|
|
}
|
|
|
|
type MessagesState = Omit<PluginSessionMessagesState, "loadMore" | "retry">;
|
|
type MessagesSetter = React.Dispatch<React.SetStateAction<MessagesState>>;
|
|
|
|
// Session deletion emits child tombstones before its terminal marker. Keep
|
|
// rows removed by that ordered teardown so the terminal view can retain the
|
|
// committed conversation state.
|
|
function restoreDeletedMessages(
|
|
current: readonly PluginConversationMessage[],
|
|
deleted: ReadonlyMap<string, PluginConversationMessage>,
|
|
sort: "asc" | "desc",
|
|
): PluginConversationMessage[] {
|
|
const messages = [...current];
|
|
const currentIds = new Set(messages.map((message) => message.id));
|
|
for (const message of deleted.values()) {
|
|
if (currentIds.has(message.id)) continue;
|
|
messages.push(message);
|
|
}
|
|
messages.sort((left, right) => compareConversationMessages(left, right, sort));
|
|
return messages;
|
|
}
|
|
|
|
function useOrderedMessageEvents({
|
|
scope,
|
|
sessionId,
|
|
taskId,
|
|
authorsKey,
|
|
sort,
|
|
snapshotKey,
|
|
setState,
|
|
cursorRef,
|
|
messagesRef,
|
|
deletedMessagesRef,
|
|
}: {
|
|
scope: ConversationScope | null;
|
|
sessionId: string | null;
|
|
taskId: string | null;
|
|
authorsKey: string;
|
|
sort: "asc" | "desc";
|
|
snapshotKey: string;
|
|
setState: MessagesSetter;
|
|
cursorRef: React.MutableRefObject<string | null>;
|
|
messagesRef: React.MutableRefObject<readonly PluginConversationMessage[]>;
|
|
deletedMessagesRef: React.MutableRefObject<Map<string, PluginConversationMessage>>;
|
|
}) {
|
|
React.useEffect(() => {
|
|
if (!scope || !sessionId) return;
|
|
const allowedAuthorTypes = authorsKey ? new Set(authorsKey.split(",")) : null;
|
|
return scope.subscribe(
|
|
(event) => {
|
|
if (event.event_type === "session.removed") {
|
|
setState((current) => {
|
|
const messages = restoreDeletedMessages(
|
|
current.messages,
|
|
deletedMessagesRef.current,
|
|
sort,
|
|
);
|
|
messagesRef.current = messages;
|
|
return { ...current, messages, removed: true, hasMore: false, loading: false };
|
|
});
|
|
cursorRef.current = null;
|
|
return true;
|
|
}
|
|
if (!event.event_type.startsWith("message.") || !eventMatchesTask(event, taskId)) {
|
|
return true;
|
|
}
|
|
const messageId = eventString(event, "message_id");
|
|
if (!messageId) return false;
|
|
if (event.event_type === "message.deleted") {
|
|
setState((current) => {
|
|
const deleted = current.messages.find((message) => message.id === messageId);
|
|
if (deleted) deletedMessagesRef.current.set(messageId, deleted);
|
|
const messages = current.messages.filter((message) => message.id !== messageId);
|
|
messagesRef.current = messages;
|
|
return { ...current, messages };
|
|
});
|
|
return true;
|
|
}
|
|
const message = messageFromEvent(event);
|
|
if (!message) return false;
|
|
if (allowedAuthorTypes && !allowedAuthorTypes.has(message.authorType)) return true;
|
|
deletedMessagesRef.current.delete(message.id);
|
|
setState((current) => {
|
|
const messages = current.messages.filter((item) => item.id !== message.id);
|
|
messages.push(message);
|
|
messages.sort((left, right) => compareConversationMessages(left, right, sort));
|
|
messagesRef.current = messages;
|
|
return { ...current, messages };
|
|
});
|
|
return true;
|
|
},
|
|
"messages",
|
|
snapshotKey,
|
|
);
|
|
}, [
|
|
authorsKey,
|
|
cursorRef,
|
|
deletedMessagesRef,
|
|
messagesRef,
|
|
scope,
|
|
sessionId,
|
|
setState,
|
|
snapshotKey,
|
|
sort,
|
|
taskId,
|
|
]);
|
|
}
|
|
|
|
function useInitialMessagePage({
|
|
scope,
|
|
sessionId,
|
|
error,
|
|
revision,
|
|
loadPage,
|
|
setState,
|
|
cursorRef,
|
|
loadMoreRef,
|
|
messagesRef,
|
|
deletedMessagesRef,
|
|
requestRevisionRef,
|
|
snapshotKey,
|
|
}: {
|
|
scope: ConversationScope | null;
|
|
sessionId: string | null;
|
|
error: PluginConversationError | null;
|
|
revision: number;
|
|
loadPage: (cursor: string | null, append: boolean) => Promise<number>;
|
|
setState: MessagesSetter;
|
|
cursorRef: React.MutableRefObject<string | null>;
|
|
loadMoreRef: React.MutableRefObject<Promise<number> | null>;
|
|
messagesRef: React.MutableRefObject<readonly PluginConversationMessage[]>;
|
|
deletedMessagesRef: React.MutableRefObject<Map<string, PluginConversationMessage>>;
|
|
requestRevisionRef: React.MutableRefObject<number>;
|
|
snapshotKey: string;
|
|
}) {
|
|
React.useEffect(() => {
|
|
requestRevisionRef.current += 1;
|
|
cursorRef.current = null;
|
|
loadMoreRef.current = null;
|
|
messagesRef.current = [];
|
|
deletedMessagesRef.current.clear();
|
|
if (!scope || !sessionId) {
|
|
setState({ ...EMPTY_MESSAGES, messages: [] });
|
|
return;
|
|
}
|
|
if (error) {
|
|
setState({ ...EMPTY_MESSAGES, messages: [], error });
|
|
return;
|
|
}
|
|
setState({ ...EMPTY_MESSAGES, messages: [] });
|
|
void loadPage(null, false).catch((cause: unknown) => {
|
|
if (scope.signal.aborted || scope.isTerminal()) return;
|
|
const error = conversationError(cause);
|
|
if (isRemovedConversationError(error)) {
|
|
setState((current) => ({
|
|
...current,
|
|
loading: false,
|
|
loadingMore: false,
|
|
error: null,
|
|
hasMore: false,
|
|
removed: true,
|
|
}));
|
|
return;
|
|
}
|
|
setState({ ...EMPTY_MESSAGES, messages: [], error });
|
|
});
|
|
}, [
|
|
cursorRef,
|
|
deletedMessagesRef,
|
|
error,
|
|
loadMoreRef,
|
|
loadPage,
|
|
messagesRef,
|
|
requestRevisionRef,
|
|
revision,
|
|
scope,
|
|
sessionId,
|
|
setState,
|
|
snapshotKey,
|
|
]);
|
|
}
|
|
function invalidateFreshMessagesSnapshot(scope: ConversationScope, snapshotKey: string): void {
|
|
scope.invalidateSnapshot("messages", snapshotKey);
|
|
}
|
|
|
|
function isScopeRequestable(scope: ConversationScope): boolean {
|
|
return !scope.signal.aborted && !scope.isTerminal();
|
|
}
|
|
|
|
function canLoadMore(
|
|
scope: ConversationScope | null,
|
|
sessionId: string | null,
|
|
state: { removed: boolean; hasMore: boolean },
|
|
cursor: string | null,
|
|
): boolean {
|
|
return Boolean(
|
|
scope && sessionId && isScopeRequestable(scope) && !state.removed && state.hasMore && cursor,
|
|
);
|
|
}
|
|
|
|
type MessagePageLoaderOptions = {
|
|
scope: ConversationScope | null;
|
|
sessionId: string | null;
|
|
taskId: string | null;
|
|
error: PluginConversationError | null;
|
|
authorsKey: string;
|
|
sort: "asc" | "desc";
|
|
limit: number;
|
|
snapshotKey: string;
|
|
setState: MessagesSetter;
|
|
cursorRef: React.MutableRefObject<string | null>;
|
|
messagesRef: React.MutableRefObject<readonly PluginConversationMessage[]>;
|
|
requestRevisionRef: React.MutableRefObject<number>;
|
|
};
|
|
|
|
async function fetchMessagePage({
|
|
scope,
|
|
sessionId,
|
|
taskId,
|
|
authorsKey,
|
|
sort,
|
|
limit,
|
|
cursor,
|
|
binding,
|
|
}: {
|
|
scope: ConversationScope;
|
|
sessionId: string;
|
|
taskId: string | null;
|
|
authorsKey: string;
|
|
sort: "asc" | "desc";
|
|
limit: number;
|
|
cursor: string | null;
|
|
binding: Pick<OrderedReady, "bindingToken" | "snapshotToken" | "source" | "revision">;
|
|
}): Promise<MessagePage> {
|
|
const params = new URLSearchParams({ sort, limit: String(limit) });
|
|
if (taskId !== null) params.set("task_id", taskId);
|
|
for (const author of authorsKey ? authorsKey.split(",") : []) {
|
|
params.append("author_type", author);
|
|
}
|
|
if (cursor) params.set("cursor", cursor);
|
|
if (binding.source && binding.revision !== undefined) {
|
|
params.set("expected_revision", binding.revision);
|
|
}
|
|
const path = binding.source
|
|
? `/conversation/v2/task-sessions/${encodeURIComponent(sessionId)}/messages?${params}`
|
|
: `/conversation/task-sessions/${encodeURIComponent(sessionId)}/messages?${params}`;
|
|
const headers: Record<string, string> = {
|
|
"X-Kandev-Plugin-Binding": binding.bindingToken,
|
|
};
|
|
if (binding.snapshotToken) headers["X-Kandev-Snapshot-Token"] = binding.snapshotToken;
|
|
const response = await fetch(pluginConversationUrl(scope.pluginId, path), {
|
|
credentials: "include",
|
|
cache: "no-store",
|
|
headers,
|
|
signal: scope.signal,
|
|
});
|
|
return parseConversationResponse<MessagePage>(response);
|
|
}
|
|
|
|
async function fetchTurnPage({
|
|
scope,
|
|
sessionId,
|
|
taskId,
|
|
binding,
|
|
cursor,
|
|
}: {
|
|
cursor?: string;
|
|
scope: ConversationScope;
|
|
sessionId: string;
|
|
taskId: string | null;
|
|
binding: Pick<OrderedReady, "bindingToken" | "snapshotToken" | "source" | "revision">;
|
|
}): Promise<TurnsPage> {
|
|
const params = new URLSearchParams({ limit: "100" });
|
|
if (cursor) params.set("cursor", cursor);
|
|
if (taskId !== null) params.set("task_id", taskId);
|
|
if (binding.source && binding.revision !== undefined) {
|
|
params.set("expected_revision", binding.revision);
|
|
}
|
|
const queryString = params.size ? `?${params}` : "";
|
|
const path = binding.source
|
|
? `/conversation/v2/task-sessions/${encodeURIComponent(sessionId)}/turns${queryString}`
|
|
: `/conversation/task-sessions/${encodeURIComponent(sessionId)}/turns${queryString}`;
|
|
const headers: Record<string, string> = {
|
|
"X-Kandev-Plugin-Binding": binding.bindingToken,
|
|
};
|
|
if (binding.snapshotToken) headers["X-Kandev-Snapshot-Token"] = binding.snapshotToken;
|
|
const response = await fetch(pluginConversationUrl(scope.pluginId, path), {
|
|
credentials: "include",
|
|
cache: "no-store",
|
|
headers,
|
|
signal: scope.signal,
|
|
});
|
|
return parseConversationResponse<TurnsPage>(response);
|
|
}
|
|
|
|
async function fetchAllTurns(args: Parameters<typeof fetchTurnPage>[0]): Promise<TurnsPage> {
|
|
const first = await fetchTurnPage(args);
|
|
const turns = [...first.turns];
|
|
const cursors = new Set<string>();
|
|
let page = first;
|
|
while (page.hasMore) {
|
|
if (!page.cursor || cursors.has(page.cursor)) {
|
|
// i18n-exempt: internal protocol diagnostic rendered through structured API error handling.
|
|
throw new Error("Invalid conversation turn cursor");
|
|
}
|
|
cursors.add(page.cursor);
|
|
const binding = await args.scope.ready();
|
|
page = await fetchTurnPage({
|
|
...args,
|
|
cursor: page.cursor,
|
|
binding: { ...binding, revision: first.revision },
|
|
});
|
|
if (page.epoch !== first.epoch || page.revision !== first.revision) {
|
|
// i18n-exempt: internal protocol diagnostic rendered through structured API error handling.
|
|
throw Object.assign(new Error("Conversation changed during pagination"), {
|
|
code: "reconciliation_required",
|
|
retryable: true,
|
|
});
|
|
}
|
|
turns.push(...page.turns);
|
|
}
|
|
return { ...first, turns, hasMore: false, cursor: null };
|
|
}
|
|
|
|
function mergeMessagePage(
|
|
current: readonly PluginConversationMessage[],
|
|
page: readonly PluginConversationMessage[],
|
|
append: boolean,
|
|
sort: "asc" | "desc",
|
|
): { messages: PluginConversationMessage[]; additionCount: number } {
|
|
const existingIds = new Set(current.map((message) => message.id));
|
|
const messagesById = new Map<string, PluginConversationMessage>();
|
|
if (append) {
|
|
for (const message of current) messagesById.set(message.id, message);
|
|
}
|
|
for (const message of page) {
|
|
const previous = messagesById.get(message.id);
|
|
if (!previous || previous.updatedAt.localeCompare(message.updatedAt) < 0) {
|
|
messagesById.set(message.id, message);
|
|
}
|
|
}
|
|
const messages = Array.from(messagesById.values());
|
|
messages.sort((left, right) => compareConversationMessages(left, right, sort));
|
|
return {
|
|
messages,
|
|
additionCount: page.filter((message) => !existingIds.has(message.id)).length,
|
|
};
|
|
}
|
|
|
|
function useMessagePageLoader({
|
|
scope,
|
|
sessionId,
|
|
taskId,
|
|
error,
|
|
authorsKey,
|
|
sort,
|
|
limit,
|
|
snapshotKey,
|
|
setState,
|
|
cursorRef,
|
|
messagesRef,
|
|
requestRevisionRef,
|
|
}: MessagePageLoaderOptions) {
|
|
return React.useCallback(
|
|
async (cursor: string | null, append: boolean): Promise<number> => {
|
|
if (!scope || !sessionId || !isScopeRequestable(scope)) return 0;
|
|
if (error) throw error;
|
|
// A fresh full page or continuation must not let a concurrent live
|
|
// update project and then be overwritten by the stale page response.
|
|
// Continuations also invalidate their committed snapshot so deletions
|
|
// cannot be acknowledged while the page is in flight and resurrected
|
|
invalidateFreshMessagesSnapshot(scope, snapshotKey);
|
|
const continuationSnapshotInvalidated = append;
|
|
try {
|
|
let capturedRevision = requestRevisionRef.current;
|
|
let binding = await scope.ready();
|
|
let pageCursor = cursor;
|
|
if (pageCursor) {
|
|
const renewal = await scope.renewContinuation(pageCursor, snapshotKey);
|
|
if (renewal.recovered) {
|
|
capturedRevision = requestRevisionRef.current;
|
|
pageCursor = cursorRef.current;
|
|
if (!pageCursor) return 0;
|
|
} else {
|
|
pageCursor = renewal.cursor;
|
|
}
|
|
binding = renewal.binding;
|
|
}
|
|
const page = await fetchMessagePage({
|
|
scope,
|
|
sessionId,
|
|
taskId,
|
|
authorsKey,
|
|
sort,
|
|
limit,
|
|
cursor: pageCursor,
|
|
binding,
|
|
});
|
|
if (
|
|
requestRevisionRef.current !== capturedRevision ||
|
|
scope.signal.aborted ||
|
|
scope.isTerminal()
|
|
) {
|
|
if (continuationSnapshotInvalidated) scope.commitSnapshot("messages", snapshotKey);
|
|
return 0;
|
|
}
|
|
const { messages, additionCount } = mergeMessagePage(
|
|
messagesRef.current,
|
|
page.messages,
|
|
append,
|
|
sort,
|
|
);
|
|
messagesRef.current = messages;
|
|
setState((current) => ({
|
|
...current,
|
|
messages,
|
|
loading: false,
|
|
hydrated: true,
|
|
loadingMore: false,
|
|
error: null,
|
|
hasMore: page.hasMore,
|
|
}));
|
|
scope.setSourceSnapshot?.(page.epoch, page.revision);
|
|
scope.commitSnapshot("messages", snapshotKey, page.epoch, page.revision);
|
|
cursorRef.current = page.cursor;
|
|
return additionCount;
|
|
} catch (cause) {
|
|
if (continuationSnapshotInvalidated) scope.commitSnapshot("messages", snapshotKey);
|
|
throw cause;
|
|
}
|
|
},
|
|
[
|
|
authorsKey,
|
|
error,
|
|
limit,
|
|
messagesRef,
|
|
requestRevisionRef,
|
|
scope,
|
|
snapshotKey,
|
|
sessionId,
|
|
setState,
|
|
sort,
|
|
taskId,
|
|
],
|
|
);
|
|
}
|
|
|
|
function useMessageRebind({
|
|
scope,
|
|
sessionId,
|
|
error,
|
|
loadPage,
|
|
setState,
|
|
cursorRef,
|
|
deletedMessagesRef,
|
|
requestRevisionRef,
|
|
}: {
|
|
scope: ConversationScope | null;
|
|
sessionId: string | null;
|
|
error: PluginConversationError | null;
|
|
loadPage: (cursor: string | null, append: boolean) => Promise<number>;
|
|
setState: MessagesSetter;
|
|
cursorRef: React.MutableRefObject<string | null>;
|
|
deletedMessagesRef: React.MutableRefObject<Map<string, PluginConversationMessage>>;
|
|
requestRevisionRef: React.MutableRefObject<number>;
|
|
}) {
|
|
React.useEffect(() => {
|
|
if (!scope || !sessionId || error) return;
|
|
return scope.subscribeRebind(async () => {
|
|
requestRevisionRef.current += 1;
|
|
cursorRef.current = null;
|
|
deletedMessagesRef.current.clear();
|
|
setState((current) => ({ ...current, loading: true, loadingMore: false, error: null }));
|
|
try {
|
|
await loadPage(null, false);
|
|
} catch (cause) {
|
|
if (scope.signal.aborted || scope.isTerminal()) return;
|
|
setState((current) => ({
|
|
...current,
|
|
loading: false,
|
|
error: conversationError(cause),
|
|
}));
|
|
throw cause;
|
|
}
|
|
});
|
|
}, [
|
|
cursorRef,
|
|
deletedMessagesRef,
|
|
error,
|
|
loadPage,
|
|
requestRevisionRef,
|
|
scope,
|
|
sessionId,
|
|
setState,
|
|
]);
|
|
}
|
|
|
|
function useMessageControls({
|
|
scope,
|
|
sessionId,
|
|
state,
|
|
setState,
|
|
loadPage,
|
|
cursorRef,
|
|
loadMoreRef,
|
|
setRevision,
|
|
}: {
|
|
scope: ConversationScope | null;
|
|
sessionId: string | null;
|
|
state: MessagesState;
|
|
setState: MessagesSetter;
|
|
loadPage: (cursor: string | null, append: boolean) => Promise<number>;
|
|
cursorRef: React.MutableRefObject<string | null>;
|
|
loadMoreRef: React.MutableRefObject<Promise<number> | null>;
|
|
setRevision: React.Dispatch<React.SetStateAction<number>>;
|
|
}): PluginSessionMessagesState {
|
|
const loadMore = React.useCallback((): Promise<number> => {
|
|
if (!canLoadMore(scope, sessionId, state, cursorRef.current)) return Promise.resolve(0);
|
|
if (loadMoreRef.current) return loadMoreRef.current;
|
|
setState((current) => ({ ...current, loadingMore: true }));
|
|
const request = loadPage(cursorRef.current, true)
|
|
.catch((error: unknown) => {
|
|
setState((current) => ({
|
|
...current,
|
|
loadingMore: false,
|
|
error: conversationError(error),
|
|
}));
|
|
throw conversationError(error);
|
|
})
|
|
.finally(() => {
|
|
if (loadMoreRef.current === request) loadMoreRef.current = null;
|
|
});
|
|
loadMoreRef.current = request;
|
|
return request;
|
|
}, [cursorRef, loadMoreRef, loadPage, scope, sessionId, setState, state.hasMore, state.removed]);
|
|
|
|
const retry = React.useCallback(() => {
|
|
if (!state.error?.retryable || state.removed || !scope || scope.isTerminal() || !sessionId)
|
|
return;
|
|
setRevision((value) => value + 1);
|
|
}, [scope, sessionId, setRevision, state.error, state.removed]);
|
|
|
|
return React.useMemo(() => ({ ...state, loadMore, retry }), [loadMore, retry, state]);
|
|
}
|
|
|
|
function useSessionMessages(query: PluginSessionMessagesQuery): PluginSessionMessagesState {
|
|
const scope = React.useContext(ConversationScopeContext);
|
|
const [revision, setRevision] = React.useState(0);
|
|
const [state, setState] = React.useState<Omit<PluginSessionMessagesState, "loadMore" | "retry">>({
|
|
...EMPTY_MESSAGES,
|
|
});
|
|
const cursorRef = React.useRef<string | null>(null);
|
|
const loadMoreRef = React.useRef<Promise<number> | null>(null);
|
|
const messagesRef = React.useRef<readonly PluginConversationMessage[]>([]);
|
|
const deletedMessagesRef = React.useRef(new Map<string, PluginConversationMessage>());
|
|
const requestRevisionRef = React.useRef(0);
|
|
const resolved = React.useMemo(
|
|
() => (scope ? resolveTaskId(scope, query.taskId) : { taskId: null, error: null }),
|
|
[query.taskId, scope],
|
|
);
|
|
const authorsKey = [...(query.authorTypes ?? [])].join(",");
|
|
const sort = query.sort ?? "desc";
|
|
const limit = query.pageSize ?? 20;
|
|
const snapshotKey = JSON.stringify([query.sessionId, resolved.taskId, authorsKey, sort, limit]);
|
|
|
|
const loadPage = useMessagePageLoader({
|
|
scope,
|
|
sessionId: query.sessionId,
|
|
taskId: resolved.taskId,
|
|
error: resolved.error,
|
|
authorsKey,
|
|
sort,
|
|
limit,
|
|
snapshotKey,
|
|
setState,
|
|
cursorRef,
|
|
messagesRef,
|
|
requestRevisionRef,
|
|
});
|
|
useOrderedMessageEvents({
|
|
scope: resolved.error ? null : scope,
|
|
sessionId: query.sessionId,
|
|
taskId: resolved.taskId,
|
|
authorsKey,
|
|
sort,
|
|
snapshotKey,
|
|
setState,
|
|
cursorRef,
|
|
messagesRef,
|
|
deletedMessagesRef,
|
|
});
|
|
|
|
useInitialMessagePage({
|
|
scope,
|
|
sessionId: query.sessionId,
|
|
error: resolved.error,
|
|
revision,
|
|
loadPage,
|
|
setState,
|
|
cursorRef,
|
|
loadMoreRef,
|
|
messagesRef,
|
|
deletedMessagesRef,
|
|
requestRevisionRef,
|
|
snapshotKey,
|
|
});
|
|
|
|
useMessageRebind({
|
|
scope,
|
|
sessionId: query.sessionId,
|
|
error: resolved.error,
|
|
loadPage,
|
|
setState,
|
|
cursorRef,
|
|
deletedMessagesRef,
|
|
requestRevisionRef,
|
|
});
|
|
return useMessageControls({
|
|
scope,
|
|
sessionId: query.sessionId,
|
|
state,
|
|
setState,
|
|
loadPage,
|
|
cursorRef,
|
|
loadMoreRef,
|
|
setRevision,
|
|
});
|
|
}
|
|
|
|
type TurnsState = Omit<PluginSessionTurnsState, "retry">;
|
|
type TurnsSetter = React.Dispatch<React.SetStateAction<TurnsState>>;
|
|
|
|
function compareConversationTurns(
|
|
left: PluginConversationTurn,
|
|
right: PluginConversationTurn,
|
|
): number {
|
|
const started = left.startedAt.localeCompare(right.startedAt);
|
|
return started === 0 ? left.id.localeCompare(right.id) : started;
|
|
}
|
|
|
|
function restoreDeletedTurns(
|
|
current: readonly PluginConversationTurn[],
|
|
deleted: ReadonlyMap<string, PluginConversationTurn>,
|
|
): PluginConversationTurn[] {
|
|
const turns = [...current];
|
|
const currentIds = new Set(turns.map((turn) => turn.id));
|
|
for (const turn of deleted.values()) {
|
|
if (currentIds.has(turn.id)) continue;
|
|
turns.push(turn);
|
|
}
|
|
turns.sort(compareConversationTurns);
|
|
return turns;
|
|
}
|
|
|
|
function useOrderedTurnEvents({
|
|
scope,
|
|
sessionId,
|
|
taskId,
|
|
error,
|
|
snapshotKey,
|
|
setState,
|
|
deletedTurnsRef,
|
|
}: {
|
|
scope: ConversationScope | null;
|
|
sessionId: string | null;
|
|
taskId: string | null;
|
|
error: PluginConversationError | null;
|
|
snapshotKey: string;
|
|
setState: TurnsSetter;
|
|
deletedTurnsRef: React.MutableRefObject<Map<string, PluginConversationTurn>>;
|
|
}) {
|
|
React.useEffect(() => {
|
|
if (!scope || !sessionId || error) return;
|
|
return scope.subscribe(
|
|
(event) => {
|
|
if (event.event_type === "session.removed") {
|
|
setState((current) => ({
|
|
...current,
|
|
turns: restoreDeletedTurns(current.turns, deletedTurnsRef.current),
|
|
removed: true,
|
|
loading: false,
|
|
}));
|
|
return true;
|
|
}
|
|
if (!eventMatchesTask(event, taskId)) return true;
|
|
if (event.event_type === "session.turn.removed") {
|
|
const payload = event.payload;
|
|
const turnId =
|
|
payload &&
|
|
typeof payload === "object" &&
|
|
"id" in payload &&
|
|
typeof payload.id === "string"
|
|
? payload.id
|
|
: undefined;
|
|
if (!turnId) return false;
|
|
setState((current) => {
|
|
const deleted = current.turns.find((turn) => turn.id === turnId);
|
|
if (deleted) deletedTurnsRef.current.set(turnId, deleted);
|
|
return {
|
|
...current,
|
|
turns: current.turns.filter((item) => item.id !== turnId),
|
|
};
|
|
});
|
|
return true;
|
|
}
|
|
if (!event.event_type.startsWith("session.turn.")) return true;
|
|
const turn = turnFromEvent(event);
|
|
if (!turn) return false;
|
|
deletedTurnsRef.current.delete(turn.id);
|
|
setState((current) => {
|
|
const turns = current.turns.filter((item) => item.id !== turn.id);
|
|
turns.push(turn);
|
|
turns.sort(compareConversationTurns);
|
|
return { ...current, turns, hydrated: true };
|
|
});
|
|
return true;
|
|
},
|
|
"turns",
|
|
snapshotKey,
|
|
);
|
|
}, [deletedTurnsRef, error, scope, sessionId, setState, snapshotKey, taskId]);
|
|
}
|
|
// eslint-disable-next-line max-lines-per-function -- keeps turn snapshot lifecycle in one hook.
|
|
function useSessionTurns(
|
|
sessionId: string | null,
|
|
taskId?: string | null,
|
|
): PluginSessionTurnsState {
|
|
const scope = React.useContext(ConversationScopeContext);
|
|
const [revision, setRevision] = React.useState(0);
|
|
const [state, setState] = React.useState<Omit<PluginSessionTurnsState, "retry">>({
|
|
...EMPTY_TURNS,
|
|
turns: [],
|
|
});
|
|
const resolved = React.useMemo(
|
|
() => (scope ? resolveTaskId(scope, taskId) : { taskId: null, error: null }),
|
|
[scope, taskId],
|
|
);
|
|
const snapshotKey = JSON.stringify([sessionId, resolved.taskId]);
|
|
const turnsRequestRef = React.useRef(0);
|
|
const deletedTurnsRef = React.useRef(new Map<string, PluginConversationTurn>());
|
|
|
|
useOrderedTurnEvents({
|
|
scope,
|
|
sessionId,
|
|
taskId: resolved.taskId,
|
|
error: resolved.error,
|
|
snapshotKey,
|
|
setState,
|
|
deletedTurnsRef,
|
|
});
|
|
|
|
const loadTurns = React.useCallback(async (): Promise<void> => {
|
|
deletedTurnsRef.current.clear();
|
|
if (!scope || !sessionId) {
|
|
setState({ ...EMPTY_TURNS, turns: [] });
|
|
return;
|
|
}
|
|
if (resolved.error) {
|
|
setState({ ...EMPTY_TURNS, turns: [], error: resolved.error });
|
|
return;
|
|
}
|
|
const requestId = ++turnsRequestRef.current;
|
|
// A fresh turns page (initial load, retry, or binding refresh) must not
|
|
// let a concurrent live turn event project and then be overwritten by the
|
|
// stale page response: buffer live events until the new snapshot commits,
|
|
// then drain them on top.
|
|
scope.invalidateSnapshot("turns", snapshotKey);
|
|
setState((previous) =>
|
|
revision === 0
|
|
? { ...EMPTY_TURNS, turns: [], loading: true }
|
|
: { ...previous, loading: true, error: null },
|
|
);
|
|
try {
|
|
const binding = await scope.ready();
|
|
const page = await fetchAllTurns({
|
|
scope,
|
|
sessionId,
|
|
taskId: resolved.taskId,
|
|
binding,
|
|
});
|
|
if (turnsRequestRef.current !== requestId || scope.signal.aborted || scope.isTerminal()) {
|
|
return;
|
|
}
|
|
setState((previous) => ({
|
|
...previous,
|
|
turns: page.turns,
|
|
loading: false,
|
|
hydrated: true,
|
|
error: null,
|
|
}));
|
|
scope.setSourceSnapshot?.(page.epoch, page.revision);
|
|
scope.commitSnapshot("turns", snapshotKey, page.epoch, page.revision);
|
|
} catch (cause) {
|
|
if (turnsRequestRef.current !== requestId || scope.signal.aborted || scope.isTerminal()) {
|
|
return;
|
|
}
|
|
const error = conversationError(cause);
|
|
if (isRemovedConversationError(error)) {
|
|
setState((previous) => ({
|
|
...previous,
|
|
loading: false,
|
|
error: null,
|
|
removed: true,
|
|
}));
|
|
return;
|
|
}
|
|
setState((previous) => ({
|
|
...(revision === 0 ? { ...EMPTY_TURNS, turns: [] } : previous),
|
|
loading: false,
|
|
error,
|
|
}));
|
|
throw cause;
|
|
}
|
|
}, [
|
|
deletedTurnsRef,
|
|
resolved.error,
|
|
resolved.taskId,
|
|
revision,
|
|
scope,
|
|
sessionId,
|
|
setState,
|
|
snapshotKey,
|
|
]);
|
|
|
|
React.useEffect(() => {
|
|
void loadTurns().catch(() => {});
|
|
return () => {
|
|
turnsRequestRef.current += 1;
|
|
};
|
|
}, [loadTurns]);
|
|
|
|
React.useEffect(() => {
|
|
if (!scope || !sessionId || resolved.error) return;
|
|
return scope.subscribeRebind(() => loadTurns());
|
|
}, [loadTurns, resolved.error, scope, sessionId]);
|
|
|
|
const retry = React.useCallback(() => {
|
|
if (!state.error?.retryable || state.removed || !scope || !sessionId) return;
|
|
setRevision((value) => value + 1);
|
|
}, [scope, sessionId, state.error, state.removed]);
|
|
|
|
return React.useMemo(() => ({ ...state, retry }), [retry, state]);
|
|
}
|
|
|
|
function useMessageFavorite(sessionId: string | null, messageId: string): boolean {
|
|
const hydrateSession = useMessageFavoritesStore((state) => state.hydrateSession);
|
|
React.useEffect(() => {
|
|
if (sessionId) hydrateSession(sessionId);
|
|
}, [hydrateSession, sessionId]);
|
|
return useMessageFavoritesStore((state) =>
|
|
sessionId ? Boolean(state.bySession[sessionId]?.[messageId]) : false,
|
|
);
|
|
}
|
|
|
|
export const pluginConversationApi: PluginConversationApi = {
|
|
useSessionMessages,
|
|
useSessionTurns,
|
|
useMessageFavorite,
|
|
};
|