From 3228e1a59a1dff6f0992bd297d47bc287fdbc413 Mon Sep 17 00:00:00 2001 From: Anthony Ronning <101225832+AnthonyRonning@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:08:26 +0000 Subject: [PATCH] research: show compaction progress and history dividers --- .../scripts/disposable_db_tests.sh | 2 +- .../frontend/src/components/UnifiedChat.tsx | 262 +++++++++++++---- .../chat/ChatCompactionDivider.test.tsx | 27 ++ .../components/chat/ChatCompactionDivider.tsx | 23 ++ .../src/services/chatCompaction.test.ts | 266 ++++++++++++++++++ .../frontend/src/services/chatCompaction.ts | 210 ++++++++++++++ .../src/services/chatCompactionApi.test.ts | 88 ++++++ .../src/services/chatCompactionApi.ts | 20 ++ .../src/services/chatRuntimeStore.test.ts | 72 +++++ .../frontend/src/services/chatRuntimeStore.ts | 10 + services/opensecret/src/db.rs | 25 +- .../src/models/response_compaction.rs | 69 +++++ .../src/models/response_compaction/tests.rs | 138 +++++++++ .../src/web/responses/conversations.rs | 193 +++++++++++++ 14 files changed, 1343 insertions(+), 62 deletions(-) create mode 100644 apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.test.tsx create mode 100644 apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.tsx create mode 100644 apps/maple-research/frontend/src/services/chatCompaction.test.ts create mode 100644 apps/maple-research/frontend/src/services/chatCompaction.ts create mode 100644 apps/maple-research/frontend/src/services/chatCompactionApi.test.ts create mode 100644 apps/maple-research/frontend/src/services/chatCompactionApi.ts diff --git a/.agents/skills/validate-opensecret/scripts/disposable_db_tests.sh b/.agents/skills/validate-opensecret/scripts/disposable_db_tests.sh index 8a6565f16..bef9d9916 100755 --- a/.agents/skills/validate-opensecret/scripts/disposable_db_tests.sh +++ b/.agents/skills/validate-opensecret/scripts/disposable_db_tests.sh @@ -187,7 +187,7 @@ cargo test --locked --all-features models::response_compaction::tests::db_ \ compaction_count="$(awk '/^models::response_compaction::tests::db_.*: test$/ { count++ } END { print count + 0 }' "$workdir/compaction-tests.list")" # Keep the selected persistence contract explicit: a missing ignored test is a failure. -test "$compaction_count" -eq 20 +test "$compaction_count" -eq 21 cargo test --locked --all-features aead_db_tamper_tests \ -- --ignored --test-threads=1 --nocapture 2>&1 | tee "$workdir/aead-tests.log" diff --git a/apps/maple-research/frontend/src/components/UnifiedChat.tsx b/apps/maple-research/frontend/src/components/UnifiedChat.tsx index 401361b73..d9b74965c 100644 --- a/apps/maple-research/frontend/src/components/UnifiedChat.tsx +++ b/apps/maple-research/frontend/src/components/UnifiedChat.tsx @@ -67,6 +67,14 @@ import { QueuedComposerMessages } from "@/components/chat/QueuedComposerMessages"; import { ChatCopyButton } from "@/components/chat/ChatCopyButton"; +import { ChatCompactionDivider } from "@/components/chat/ChatCompactionDivider"; +import { + chatCompactionProgressUpdate, + mergeChatCompactionMarkers, + projectChatCompactions, + type ChatCompactionMarker +} from "@/services/chatCompaction"; +import { readChatCompactions } from "@/services/chatCompactionApi"; import { ToolActivityCard } from "@/components/ToolActivityCard"; import { continueChatComposerList, @@ -311,6 +319,7 @@ type ConversationContent = type MessageStatus = "completed" | "in_progress" | "incomplete" | "streaming" | "error"; type ExtendedMessage = Omit & { status?: MessageStatus; + response_id?: string; }; type ReasoningContentItem = { type: "reasoning_text"; text: string }; @@ -322,6 +331,7 @@ type ReasoningItem = Omit & { type ToolCallItem = { id: string; type: "tool_call"; + response_id?: string; call_id: string; name: string; arguments: string; @@ -331,6 +341,7 @@ type ToolCallItem = { type ToolOutputItem = { id: string; type: "tool_output"; + response_id?: string; call_id: string; output: string; status?: MessageStatus; @@ -497,6 +508,7 @@ function normalizeReasoningItem(item: ResponseReasoningItem | ReasoningItem): Re function normalizeToolCallItem(item: unknown): ToolCallItem | null { if (!item || typeof item !== "object" || !("type" in item)) return null; + const responseId = (item as { response_id?: unknown }).response_id; if ((item as { type?: string }).type === "tool_call") { const toolCall = item as Partial; @@ -505,6 +517,7 @@ function normalizeToolCallItem(item: unknown): ToolCallItem | null { return { id: toolCall.id, type: "tool_call", + ...(typeof responseId === "string" && { response_id: responseId }), call_id: toolCall.call_id, name: typeof toolCall.name === "string" ? toolCall.name : "function", arguments: @@ -522,6 +535,7 @@ function normalizeToolCallItem(item: unknown): ToolCallItem | null { return { id: toolCall.id, type: "tool_call", + ...(typeof responseId === "string" && { response_id: responseId }), call_id: toolCall.call_id, name: toolCall.name, arguments: toolCall.arguments, @@ -534,6 +548,7 @@ function normalizeToolCallItem(item: unknown): ToolCallItem | null { function normalizeToolOutputItem(item: unknown): ToolOutputItem | null { if (!item || typeof item !== "object" || !("type" in item)) return null; + const responseId = (item as { response_id?: unknown }).response_id; if ((item as { type?: string }).type === "tool_output") { const toolOutput = item as Partial; @@ -542,6 +557,7 @@ function normalizeToolOutputItem(item: unknown): ToolOutputItem | null { return { id: toolOutput.id, type: "tool_output", + ...(typeof responseId === "string" && { response_id: responseId }), call_id: toolOutput.call_id, output: typeof toolOutput.output === "string" @@ -558,6 +574,7 @@ function normalizeToolOutputItem(item: unknown): ToolOutputItem | null { return { id: toolOutput.id, type: "tool_output", + ...(typeof responseId === "string" && { response_id: responseId }), call_id: toolOutput.call_id, output: toolOutput.output, status: toolOutput.status @@ -983,6 +1000,11 @@ function mergePolledMessagesWithRuntime(cached: readonly Message[], polled: Mess // The API already returns items in the correct format (ConversationItem union) function convertItemsToMessages(items: Array): Message[] { return items.flatMap((item) => { + // Response retrieval may include opaque checkpoint output. Dividers use + // the bounded metadata API; never retain or log that opaque content here. + if (item && typeof item === "object" && (item as { type?: string }).type === "compaction") { + return []; + } const normalizedItem = normalizeConversationItem(item); if (!normalizedItem && item != null) { @@ -1187,7 +1209,8 @@ function ToolCallRenderer({ // Types for grouping messages into turns type MessageGroup = | { type: "user"; message: ExtendedMessage; id: string } - | { type: "assistant"; items: Message[]; id: string }; + | { type: "assistant"; items: Message[]; id: string } + | { type: "compaction"; marker: ChatCompactionMarker; id: string }; function getUserMessageText(message: ExtendedMessage): string { return ( @@ -1217,10 +1240,14 @@ const MessageList = memo( ({ messages, isGenerating, + compactions, + isCompacting, chatId }: { messages: Message[]; isGenerating: boolean; + compactions?: readonly ChatCompactionMarker[]; + isCompacting: boolean; chatId?: string; }) => { const toolCallsByCallId = useMemo(() => { @@ -1254,7 +1281,19 @@ const MessageList = memo( const groups: MessageGroup[] = []; let currentAssistantItems: Message[] = []; - for (const item of messages) { + for (const item of projectChatCompactions(messages, compactions ?? [])) { + if (item.type === "compaction") { + if (currentAssistantItems.length > 0) { + groups.push({ + type: "assistant", + items: [...currentAssistantItems], + id: `assistant-${currentAssistantItems[0].id}` + }); + currentAssistantItems = []; + } + groups.push({ type: "compaction", marker: item, id: item.id }); + continue; + } // Check if this is a user message if (item.type === "message" && (item as unknown as ExtendedMessage).role === "user") { // Flush any pending assistant items first @@ -1287,7 +1326,7 @@ const MessageList = memo( } return groups; - }, [messages]); + }, [messages, compactions]); const renderAssistantItems = (items: Message[]) => { const renderedItems: JSX.Element[] = []; @@ -1439,11 +1478,18 @@ const MessageList = memo( }; const shouldShowInitialAssistantLoader = - isGenerating && groupedMessages[groupedMessages.length - 1]?.type !== "assistant"; + isGenerating && + !isCompacting && + groupedMessages[groupedMessages.length - 1]?.type !== "assistant"; return ( <> {groupedMessages.map((group, groupIndex) => { + if (group.type === "compaction") { + return ( + + ); + } if (group.type === "user") { const message = group.message; if (!message.content || message.content.length === 0) return null; @@ -1529,6 +1575,8 @@ const MessageList = memo( return null; })} + {isCompacting && } + {/* Loading indicator - only show while waiting for the first assistant item (TTFT) */} {shouldShowInitialAssistantLoader && } @@ -2824,6 +2872,49 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { stoppingRuntimeRegistry ]); + // This optional metadata never enters canonical item pagination or model input. + const refreshChatCompactions = useCallback( + async (runtimeKey: ChatRuntimeKey, conversationId: string, older = false) => { + if (!openai) return; + const starting = runtimeStore.get(runtimeKey); + if (!starting || (older && !starting.hasMoreCompactionHistory)) return; + const after = older ? starting.compactionHistoryCursor : undefined; + try { + const page = await readChatCompactions(openai, conversationId, after); + const owned = runtimeStore.get(runtimeKey); + if ( + !owned || + isChatRuntimeDeletionPending(runtimeStore, runtimeKey) || + (owned.conversation?.id ?? + conversationIdFromChatRuntimeKey(runtimeStore.resolveKey(runtimeKey))) !== + conversationId + ) { + return; + } + runtimeStore.update(runtimeKey, (current) => { + // Concurrent refreshes merge by checkpoint identity. Only the owner of + // the oldest-page cursor may advance it; newest-page refreshes cannot + // reset a history pagination already in progress. + const advancesCursor = older + ? current.compactionHistoryCursor === after + : current.compactionHistoryCursor === undefined; + return { + ...current, + compactions: mergeChatCompactionMarkers(current.compactions ?? [], page.data), + ...(advancesCursor && { + compactionHistoryCursor: page.last_id ?? current.compactionHistoryCursor, + hasMoreCompactionHistory: page.has_more + }) + }; + }); + } catch { + // Older backends and unavailable presentation reads must not interrupt chat. + // Do not log raw encrypted API errors or private checkpoint payloads. + } + }, + [openai, runtimeStore] + ); + // Load conversation from API const loadConversation = useCallback( async (runtimeKey: ChatRuntimeKey, conversationId: string) => { @@ -2836,6 +2927,8 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { !runtimeStore.get(runtimeKey) || activeConversationLoadRef.current.get(runtimeStore.resolveKey(runtimeKey)) !== requestId; + void refreshChatCompactions(runtimeKey, conversationId); + try { // Start both fetches immediately in parallel const convPromise = openai.conversations.retrieve(conversationId).then( @@ -2957,6 +3050,7 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { activeConversationLoadRef, isRuntimeSelected, openai, + refreshChatCompactions, runtimeStore, selectFreshDraftRuntime, selectedProjectId, @@ -3001,6 +3095,7 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { order: "desc", after: requestOldestItemId }); + void refreshChatCompactions(runtimeKey, conversationId, true); // Convert items to messages, grouping tool calls with their messages const olderMessages = convertItemsToMessages( @@ -3094,6 +3189,7 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { historyPaginationLifecycle, openai, projectionScrollCoordinatorRef, + refreshChatCompactions, resolveScrollProjectionKey, runtimeStore, updateComposerForKey @@ -4703,6 +4799,22 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { streamEventCount += 1; logStreamEvent(runtimeKey, runToken, streamStartedAt, streamEventCount, eventType, event); + // Progress is advisory stream metadata, never a canonical item or a + // durable checkpoint. The run token fences late events after stop, + // account changes, or replacement by another request. + const compactionProgress = chatCompactionProgressUpdate(event); + if (compactionProgress !== undefined) { + const compactingMessageId = compactionProgress ? optimisticMessageId : undefined; + if (runtimeStore.get(runtimeKey)?.compactingMessageId !== compactingMessageId) { + deltaCoalescer.flush(); + runtimeStore.updateForRun(runtimeKey, runToken, (snapshot) => ({ + ...snapshot, + compactingMessageId + })); + } + if (compactionProgress) continue; + } + if ( eventType === "response.reasoning_text.delta" && (event as { delta?: string }).delta @@ -4741,6 +4853,14 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { if (responseId) { if (runtimeStore.setCurrentResponseId(runtimeKey, runToken, responseId)) { clearUnresolvedChatResponseMessage(runtimeStore, runToken, optimisticMessageId); + // Link the optimistic user turn to the durable Response so + // checkpoint metadata uses the same placement after reload. + updateRunMessages((messages) => + updateMessageById(messages, optimisticMessageId, (message) => ({ + ...message, + response_id: responseId + })) + ); } if (await onResponseCreated(responseId)) { terminalState = "cancelled"; @@ -4909,6 +5029,13 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { ); } else if (eventType === "response.output_item.done") { const doneEvent = event as ResponseOutputItemDoneEvent; + if ((doneEvent.item as { type?: string }).type === "compaction") { + const snapshot = runtimeStore.get(runtimeKey); + const conversationId = snapshot?.conversation?.id; + if (conversationId && runtimeStore.isRunCurrent(runtimeKey, runToken)) { + void refreshChatCompactions(runtimeKey, conversationId); + } + } const item = normalizeConversationItem(doneEvent.item); if (item) { @@ -4926,7 +5053,7 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { break; } else if (isTerminalChatStreamErrorEvent(eventType)) { terminalState = "error"; - console.error("Streaming error:", event); + console.error("Streaming error:", summarizeStreamEventForLog(eventType, event)); runtimeStore.updateForRun(runtimeKey, runToken, (snapshot) => ({ ...snapshot, messages: updateActiveItemStatuses( @@ -4969,7 +5096,7 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { return terminalState; }, - [logStreamEvent, runtimeStore] + [logStreamEvent, refreshChatCompactions, runtimeStore] ); // Every submit synchronously detaches its owning composer. During an active @@ -5231,62 +5358,72 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { return null; } registerChatOptimisticMessage(runtimeStore, run.token, localMessageId); - const responseParams: ResponseCreateParamsStreaming = { - conversation: conversationId!, - model: item.model, - input: [{ role: "user", content: messageContent }], - metadata: { internal_message_id: localMessageId }, - stream: true, - store: true, - ...(item.webSearchEnabled && { tools: [{ type: "web_search" }] }), - // The catalog decides which efforts a model accepts; openai 5.x's - // ReasoningEffort type predates `none`, `xhigh` and `max`. - ...(item.reasoningEffort && { - reasoning: { effort: item.reasoningEffort } as NonNullable< - ResponseCreateParamsStreaming["reasoning"] - > - }) - }; - const stream = await withInferenceCapacityRetry( - (maxInferenceSends) => { - registerUnresolvedChatResponseMessage(runtimeStore, run.token, localMessageId); - responseRequestStarted = true; - return openai.responses.create(responseParams, { - signal: run.signal, - headers: { - [OPEN_SECRET_INFERENCE_SEND_LIMIT_HEADER]: String(maxInferenceSends) - } - }); - }, - run.signal, - (capacityError) => { - // Capacity rejection proves this turn was not accepted. Stop or - // deletion can safely restore it and abort the replay delay. - responseRequestStarted = false; - clearUnresolvedChatResponseMessage(runtimeStore, run.token, localMessageId); - if ( - isChatRunQueueHaltRequested(runtimeStore, run.token) || - isChatRuntimeDeletionPending(runtimeStore, runtimeKey) - ) { - throw capacityError; - } - } - ); - - if (!runtimeStore.setAssistantStreaming(runtimeKey, run.token, true)) return null; try { - return await processStreamingResponse( - stream, - runtimeKey, - run.token, - localMessageId, - async (responseId) => { - if (!isRuntimeStopping(run.token)) return false; - return cancelKnownChatResponse(runtimeKey, run.token, responseId, localMessageId); + const responseParams: ResponseCreateParamsStreaming = { + conversation: conversationId!, + model: item.model, + input: [{ role: "user", content: messageContent }], + metadata: { internal_message_id: localMessageId }, + stream: true, + store: true, + ...(item.webSearchEnabled && { tools: [{ type: "web_search" }] }), + // The catalog decides which efforts a model accepts; openai 5.x's + // ReasoningEffort type predates `none`, `xhigh` and `max`. + ...(item.reasoningEffort && { + reasoning: { effort: item.reasoningEffort } as NonNullable< + ResponseCreateParamsStreaming["reasoning"] + > + }) + }; + const stream = await withInferenceCapacityRetry( + (maxInferenceSends) => { + registerUnresolvedChatResponseMessage(runtimeStore, run.token, localMessageId); + responseRequestStarted = true; + return openai.responses.create(responseParams, { + signal: run.signal, + headers: { + [OPEN_SECRET_INFERENCE_SEND_LIMIT_HEADER]: String(maxInferenceSends) + } + }); + }, + run.signal, + (capacityError) => { + // Capacity rejection proves this turn was not accepted. Stop or + // deletion can safely restore it and abort the replay delay. + responseRequestStarted = false; + clearUnresolvedChatResponseMessage(runtimeStore, run.token, localMessageId); + if ( + isChatRunQueueHaltRequested(runtimeStore, run.token) || + isChatRuntimeDeletionPending(runtimeStore, runtimeKey) + ) { + throw capacityError; + } } ); + + if (!runtimeStore.setAssistantStreaming(runtimeKey, run.token, true)) return null; + try { + return await processStreamingResponse( + stream, + runtimeKey, + run.token, + localMessageId, + async (responseId) => { + if (!isRuntimeStopping(run.token)) return false; + return cancelKnownChatResponse(runtimeKey, run.token, responseId, localMessageId); + } + ); + } finally { + runtimeStore.setAssistantStreaming(runtimeKey, run.token, false); + } } finally { - runtimeStore.setAssistantStreaming(runtimeKey, run.token, false); + runtimeStore.updateForRun(runtimeKey, run.token, (snapshot) => ({ + ...snapshot, + compactingMessageId: undefined + })); + // A checkpoint may have been saved even if the main response fails. + // Its response/anchor association decides whether it belongs in view. + void refreshChatCompactions(runtimeKey, conversationId!); } }; @@ -5640,6 +5777,7 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { openai, processStreamingResponse, queryClient, + refreshChatCompactions, runtimeStore, selectedProjectId, setErrorForKey, @@ -5814,7 +5952,13 @@ export function UnifiedChat({ isVisible = true }: { isVisible?: boolean }) { {/* Message list with modern ChatGPT/Claude style */}
- +
diff --git a/apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.test.tsx b/apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.test.tsx new file mode 100644 index 000000000..e9f6c9dde --- /dev/null +++ b/apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.test.tsx @@ -0,0 +1,27 @@ +import { describe, expect, test } from "bun:test"; +import { renderToStaticMarkup } from "react-dom/server"; +import { ChatCompactionDivider } from "./ChatCompactionDivider"; + +describe("conversation compaction divider", () => { + test("quietly announces actual compaction progress as a standalone line", () => { + const markup = renderToStaticMarkup(); + expect(markup).toContain('role="status"'); + expect(markup).toContain('aria-live="polite"'); + expect(markup).toContain(">compacting"); + expect(markup).toContain("italic"); + expect(markup).toContain("text-muted-foreground"); + expect(markup).not.toContain("animate-"); + expect(markup).not.toContain("m-avatar"); + expect(markup).not.toContain("button"); + }); + + test("retains a completed history anchor without repeatedly announcing old checkpoints", () => { + const markup = renderToStaticMarkup( + + ); + expect(markup).toContain(">compacted"); + expect(markup).toContain('data-history-anchor-ids="checkpoint-id"'); + expect(markup).not.toContain('role="status"'); + expect(markup).not.toContain("aria-live"); + }); +}); diff --git a/apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.tsx b/apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.tsx new file mode 100644 index 000000000..fca0d86b1 --- /dev/null +++ b/apps/maple-research/frontend/src/components/chat/ChatCompactionDivider.tsx @@ -0,0 +1,23 @@ +export function ChatCompactionDivider({ + status, + historyAnchorId +}: { + status: "compacting" | "completed"; + historyAnchorId?: string; +}) { + const compacting = status === "compacting"; + return ( +
+
+ ); +} diff --git a/apps/maple-research/frontend/src/services/chatCompaction.test.ts b/apps/maple-research/frontend/src/services/chatCompaction.test.ts new file mode 100644 index 000000000..1b34052ea --- /dev/null +++ b/apps/maple-research/frontend/src/services/chatCompaction.test.ts @@ -0,0 +1,266 @@ +import { describe, expect, test } from "bun:test"; +import { + chatCompactionProgressUpdate, + mergeChatCompactionMarkers, + parseCompactionPage, + projectChatCompactions, + type ChatCompactionMarker +} from "./chatCompaction"; + +const RESPONSE = "10000000-0000-4000-8000-000000000001"; +const NEXT_RESPONSE = "10000000-0000-4000-8000-000000000002"; +const USER = "20000000-0000-4000-8000-000000000001"; +const ASSISTANT = "20000000-0000-4000-8000-000000000002"; +const TOOL_OUTPUT = "20000000-0000-4000-8000-000000000003"; +const MARKER = "30000000-0000-4000-8000-000000000001"; +const NEXT_MARKER = "30000000-0000-4000-8000-000000000002"; + +function marker(overrides: Partial = {}): ChatCompactionMarker { + return { + id: MARKER, + type: "compaction", + status: "completed", + after_item_id: null, + response_id: RESPONSE, + created_at: 1792700000, + generation: 1, + ...overrides + }; +} + +function page(items: unknown[] = [marker()]) { + return { + object: "list", + data: items.map((item) => ({ ...(item as object), object: "conversation.compaction" })), + has_more: false, + first_id: items.length ? (items[0] as { id: string }).id : null, + last_id: items.length ? (items.at(-1) as { id: string }).id : null + }; +} + +describe("compaction metadata", () => { + test("keeps only presentation metadata, discarding opaque content and unknown fields", () => { + const parsed = parseCompactionPage( + page([{ ...marker(), encrypted_content: "private-checkpoint", summary: "private-summary" }]) + ); + expect(parsed.data).toEqual([marker()]); + expect(JSON.stringify(parsed)).not.toContain("private"); + }); + + test("accepts an empty terminal page", () => { + expect(parseCompactionPage(page([]))).toEqual({ + data: [], + has_more: false, + first_id: null, + last_id: null + }); + }); + + test("rejects malformed metadata with a static error that never reflects payload content", () => { + for (const value of [ + null, + [], + { ...page(), object: "not-a-list" }, + page([{ ...marker(), id: "private-invalid-id" }]), + page([{ ...marker(), after_item_id: "private-invalid-anchor" }]), + page([{ ...marker(), response_id: "private-invalid-response" }]), + page([{ ...marker(), status: "in_progress" }]), + page([{ ...marker(), created_at: -1 }]), + page([{ ...marker(), created_at: Number.POSITIVE_INFINITY }]), + page([{ ...marker(), generation: 0 }]), + page([{ ...marker(), generation: 1.5 }]) + ]) { + expect(() => parseCompactionPage(value)).toThrow("Invalid conversation compaction metadata"); + try { + parseCompactionPage(value); + } catch (error) { + expect(String(error)).not.toContain("private"); + } + } + }); + + test("rejects duplicate IDs, oversized pages, and cursor shapes that cannot progress", () => { + for (const value of [ + page([marker(), marker()]), + page(Array.from({ length: 101 }, () => marker())), + { ...page([]), has_more: true }, + { ...page(), last_id: NEXT_MARKER }, + { ...page(), first_id: null } + ]) { + expect(() => parseCompactionPage(value)).toThrow(); + } + }); + + test("deduplicates checkpoint refreshes while preserving known ordering and references", () => { + const first = marker(); + const refreshed = marker({ created_at: 1792700001 }); + const second = marker({ id: NEXT_MARKER, generation: 2 }); + const current = [first]; + const merged = mergeChatCompactionMarkers(current, [refreshed, second]); + expect(merged).toEqual([refreshed, second]); + expect(merged[0]).toBe(refreshed); + expect(current).toEqual([first]); + }); + + test("merges older metadata pages into generation order independent of timestamps", () => { + const first = marker(); + const second = marker({ id: NEXT_MARKER, generation: 2 }); + expect(mergeChatCompactionMarkers([second], [first])).toEqual([first, second]); + }); +}); + +describe("compaction timeline projection", () => { + const user = { id: USER, type: "message", role: "user", response_id: RESPONSE }; + const assistant = { + id: ASSISTANT, + type: "message", + role: "assistant", + response_id: RESPONSE + }; + + test("places initial preparation after its user turn without changing canonical items", () => { + const canonical = [user, assistant]; + const checkpoint = marker(); + const projected = projectChatCompactions(canonical, [checkpoint]); + expect(projected).toEqual([user, checkpoint, assistant]); + expect(projected[0]).toBe(user); + expect(projected[2]).toBe(assistant); + expect(canonical).toEqual([user, assistant]); + }); + + test("places an active-loop checkpoint after its exact completed tool output", () => { + const output = { id: TOOL_OUTPUT, type: "tool_output", response_id: RESPONSE }; + const checkpoint = marker({ after_item_id: TOOL_OUTPUT }); + expect(projectChatCompactions([user, output, assistant], [checkpoint])).toEqual([ + user, + output, + checkpoint, + assistant + ]); + }); + + test("omits an off-page tool anchor until older history is loaded", () => { + const checkpoint = marker({ after_item_id: TOOL_OUTPUT }); + expect(projectChatCompactions([assistant], [checkpoint])).toEqual([assistant]); + const output = { id: TOOL_OUTPUT, type: "tool_output", response_id: RESPONSE }; + expect(projectChatCompactions([output, assistant], [checkpoint])).toEqual([ + output, + checkpoint, + assistant + ]); + }); + + test("places initial checkpoint before its visible outputs when the user is off-page", () => { + const checkpoint = marker(); + expect(projectChatCompactions([assistant], [checkpoint])).toEqual([checkpoint, assistant]); + // Loading the user moves the same marker to its exact initial-turn location. + expect(projectChatCompactions([user, assistant], [checkpoint])).toEqual([ + user, + checkpoint, + assistant + ]); + }); + + test("does not guess positions for a different response or an empty history", () => { + const unrelated = marker({ response_id: NEXT_RESPONSE }); + expect(projectChatCompactions([user, assistant], [unrelated])).toEqual([user, assistant]); + expect(projectChatCompactions([], [marker()])).toEqual([]); + }); + + test("does not duplicate repeated stream/list metadata or already-projected markers", () => { + const checkpoint = marker(); + const projected = projectChatCompactions([user, assistant], [checkpoint, checkpoint]); + expect(projected).toEqual([user, checkpoint, assistant]); + expect(projectChatCompactions(projected, [checkpoint])).toEqual(projected); + }); + + test("preserves multiple checkpoint ordering and canonical item IDs for pagination", () => { + const first = marker(); + const second = marker({ id: NEXT_MARKER, generation: 2 }); + const canonical = [user, assistant]; + const projected = projectChatCompactions(canonical, [second, first]); + expect(projected).toEqual([user, first, second, assistant]); + expect(projected.filter((item) => item.type !== "compaction").map((item) => item.id)).toEqual([ + USER, + ASSISTANT + ]); + }); +}); + +describe("streamed compaction progress", () => { + const progress = { + type: "response.compaction.compacting", + sequence_number: 2, + output_index: 0, + item_id: MARKER + }; + + test("recognizes progress before any durable item without retaining private payloads", () => { + expect( + chatCompactionProgressUpdate({ + ...progress, + encrypted_content: "private-checkpoint", + summary: "private-summary" + }) + ).toBe(true); + expect(projectChatCompactions([{ id: USER, type: "message" }], [])).toEqual([ + { id: USER, type: "message" } + ]); + }); + + test("ignores malformed progress rather than reflecting or persisting its contents", () => { + for (const event of [ + { ...progress, item_id: "private-invalid-id" }, + { ...progress, item_id: undefined }, + { ...progress, sequence_number: -1 }, + { ...progress, sequence_number: 0.5 }, + { ...progress, sequence_number: Number.POSITIVE_INFINITY }, + { ...progress, output_index: -1 }, + { ...progress, output_index: undefined } + ]) { + expect(chatCompactionProgressUpdate(event)).toBeUndefined(); + } + }); + + test("does not require progress from old servers or clear it for unknown events", () => { + for (const event of [ + null, + {}, + { type: "response.in_progress" }, + { type: "response.future_event" }, + { type: "response.output_text.delta", delta: "" }, + { type: "response.reasoning_text.delta", delta: "" } + ]) { + expect(chatCompactionProgressUpdate(event)).toBeUndefined(); + } + }); + + test("clears advisory progress on main output, durable checkpoint, tools, or terminal", () => { + for (const event of [ + { type: "response.created" }, + { type: "response.output_item.added", item: { type: "message", id: ASSISTANT } }, + { type: "response.output_item.done", item: { type: "compaction", id: MARKER } }, + { type: "response.output_text.delta", delta: "answer" }, + { type: "response.reasoning_text.delta", delta: "reasoning" }, + { type: "response.content_part.added" }, + { type: "tool_call.created" }, + { type: "tool_output.created" }, + { type: "response.completed", response: { output: [] } }, + { type: "response.failed" }, + { type: "response.cancelled" }, + { type: "error" } + ]) { + expect(chatCompactionProgressUpdate(event)).toBe(false); + } + }); + + test("fallback can complete without manufacturing a durable compaction divider", () => { + let compacting = false; + const canonical = [{ id: USER, type: "message" }]; + for (const event of [progress, { type: "response.completed", response: { output: [] } }]) { + compacting = chatCompactionProgressUpdate(event) ?? compacting; + } + expect(compacting).toBe(false); + expect(projectChatCompactions(canonical, [])).toEqual(canonical); + }); +}); diff --git a/apps/maple-research/frontend/src/services/chatCompaction.ts b/apps/maple-research/frontend/src/services/chatCompaction.ts new file mode 100644 index 000000000..2f1f58ec1 --- /dev/null +++ b/apps/maple-research/frontend/src/services/chatCompaction.ts @@ -0,0 +1,210 @@ +/** Public compaction metadata only. Summary text and opaque checkpoint content never enter Chat. */ +export type ChatCompactionMarker = { + id: string; + type: "compaction"; + status: "completed"; + after_item_id: string | null; + response_id: string; + created_at: number; + generation: number; +}; + +export type ChatCompactionPage = { + data: ChatCompactionMarker[]; + has_more: boolean; + first_id: string | null; + last_id: string | null; +}; + +type TimelineItem = { + id: string; + type: string; + response_id?: string; + role?: string; +}; + +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const MAX_COMPACTION_PAGE_ITEMS = 100; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isUuid(value: unknown): value is string { + return typeof value === "string" && UUID.test(value); +} + +function invalidMetadata(): never { + // Do not include the rejected payload: it could contain private checkpoint data. + throw new Error("Invalid conversation compaction metadata"); +} + +export function parseCompactionPage(value: unknown): ChatCompactionPage { + if ( + !isRecord(value) || + value.object !== "list" || + !Array.isArray(value.data) || + value.data.length > MAX_COMPACTION_PAGE_ITEMS || + typeof value.has_more !== "boolean" + ) { + return invalidMetadata(); + } + + const data = value.data.map((item): ChatCompactionMarker => { + if ( + !isRecord(item) || + item.object !== "conversation.compaction" || + item.status !== "completed" || + !isUuid(item.id) || + !isUuid(item.response_id) || + (item.after_item_id !== null && !isUuid(item.after_item_id)) || + typeof item.created_at !== "number" || + !Number.isSafeInteger(item.created_at) || + item.created_at < 0 || + typeof item.generation !== "number" || + !Number.isSafeInteger(item.generation) || + item.generation <= 0 + ) { + return invalidMetadata(); + } + return { + id: item.id, + type: "compaction", + status: "completed", + after_item_id: item.after_item_id, + response_id: item.response_id, + created_at: item.created_at, + generation: item.generation + }; + }); + + const firstId = data[0]?.id ?? null; + const lastId = data.at(-1)?.id ?? null; + if ( + value.first_id !== firstId || + value.last_id !== lastId || + (value.has_more && !lastId) || + new Set(data.map((item) => item.id)).size !== data.length + ) { + return invalidMetadata(); + } + + return { data, has_more: value.has_more, first_id: firstId, last_id: lastId }; +} + +/** + * Standard Responses progress is advisory. Its item may not become a durable + * checkpoint (for example when optional compaction falls back to a fitting + * original prompt). Copy no summary/ciphertext, and never add it to history. + * Undefined leaves the current indicator unchanged, including on old servers + * and unknown future events. + */ +export function chatCompactionProgressUpdate(event: unknown): boolean | undefined { + if (!isRecord(event) || typeof event.type !== "string") return undefined; + if (event.type === "response.compaction.compacting") { + return isUuid(event.item_id) && + Number.isSafeInteger(event.sequence_number) && + (event.sequence_number as number) >= 0 && + Number.isSafeInteger(event.output_index) && + (event.output_index as number) >= 0 + ? true + : undefined; + } + + switch (event.type) { + case "response.output_text.delta": + case "response.reasoning_text.delta": + return typeof event.delta === "string" && event.delta.length > 0 ? false : undefined; + case "response.created": + case "response.output_item.added": + case "response.output_item.done": + case "response.content_part.added": + case "response.content_part.done": + case "response.output_text.done": + case "response.reasoning_text.done": + case "response.web_search_call.in_progress": + case "response.web_search_call.searching": + case "response.web_search_call.completed": + case "tool_call.created": + case "tool_output.created": + case "response.completed": + case "response.failed": + case "response.incomplete": + case "response.error": + case "response.cancelled": + case "error": + return false; + default: + return undefined; + } +} + +export function mergeChatCompactionMarkers( + current: readonly ChatCompactionMarker[], + incoming: readonly ChatCompactionMarker[] +): ChatCompactionMarker[] { + const markers = new Map(current.map((marker) => [marker.id, marker])); + for (const marker of incoming) markers.set(marker.id, marker); + return [...markers.values()].sort((left, right) => left.generation - right.generation); +} + +/** + * Add dividers to the loaded view without changing canonical history or its cursors. + * An absent tool-step anchor stays absent until that page of history is loaded. + */ +export function projectChatCompactions( + items: readonly T[], + markers: readonly ChatCompactionMarker[] +): Array { + const canonical = items.filter((item) => item.type !== "compaction"); + const canonicalIds = new Set(canonical.map((item) => item.id)); + const after = new Map(); + const before = new Map(); + const seen = new Set(); + const firstItemByResponse = new Map(); + const userByResponse = new Map(); + for (const item of canonical) { + if (!item.response_id) continue; + if (!firstItemByResponse.has(item.response_id)) firstItemByResponse.set(item.response_id, item); + if (item.type === "message" && item.role === "user" && !userByResponse.has(item.response_id)) { + userByResponse.set(item.response_id, item); + } + } + + const add = ( + destination: Map, + anchor: string, + marker: ChatCompactionMarker + ) => { + const bucket = destination.get(anchor) ?? []; + bucket.push(marker); + destination.set(anchor, bucket); + }; + + for (const marker of [...markers].sort((left, right) => left.generation - right.generation)) { + if (seen.has(marker.id) || canonicalIds.has(marker.id)) continue; + seen.add(marker.id); + if (marker.after_item_id !== null) { + if (canonicalIds.has(marker.after_item_id)) add(after, marker.after_item_id, marker); + continue; + } + + // Older checkpoints may have no exact item anchor. Their user turn is + // the natural place for the divider once that page is available. + const user = userByResponse.get(marker.response_id); + if (user) { + add(after, user.id, marker); + continue; + } + const firstOutput = firstItemByResponse.get(marker.response_id); + if (firstOutput) add(before, firstOutput.id, marker); + } + + return canonical.flatMap( + (item): Array => [ + ...(before.get(item.id) ?? []), + item, + ...(after.get(item.id) ?? []) + ] + ); +} diff --git a/apps/maple-research/frontend/src/services/chatCompactionApi.test.ts b/apps/maple-research/frontend/src/services/chatCompactionApi.test.ts new file mode 100644 index 000000000..eb7285bb9 --- /dev/null +++ b/apps/maple-research/frontend/src/services/chatCompactionApi.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, test } from "bun:test"; +import type OpenAI from "openai"; +import { readChatCompactions } from "./chatCompactionApi"; + +const RESPONSE = "10000000-0000-4000-8000-000000000001"; +const MARKER = "30000000-0000-4000-8000-000000000001"; + +function fakeClient(result: unknown) { + const calls: Array<{ path: string; options: unknown }> = []; + return { + calls, + client: { + get: async (path: string, options: unknown) => { + calls.push({ path, options }); + return result; + } + } as unknown as OpenAI + }; +} + +describe("encrypted Chat compaction metadata API", () => { + test("uses the configured transport timeout without adding metadata-specific auth or deadlines", async () => { + const { client, calls } = fakeClient({ + object: "list", + data: [], + has_more: false, + first_id: null, + last_id: null + }); + const controller = new AbortController(); + await readChatCompactions(client, "conversation/path?", undefined, controller.signal); + expect(calls).toEqual([ + { + path: "/conversations/conversation%2Fpath%3F/compactions", + options: { + query: { limit: 100, order: "desc" }, + signal: controller.signal + } + } + ]); + }); + + test("keeps metadata cursors separate and strips opaque checkpoint content", async () => { + const { client, calls } = fakeClient({ + object: "list", + data: [ + { + id: MARKER, + object: "conversation.compaction", + status: "completed", + response_id: RESPONSE, + after_item_id: null, + created_at: 1792700000, + generation: 1, + encrypted_content: "private-checkpoint", + summary: "private-summary" + } + ], + has_more: true, + first_id: MARKER, + last_id: MARKER + }); + const page = await readChatCompactions(client, "conversation", MARKER); + expect(calls[0].options).toMatchObject({ + query: { limit: 100, order: "desc", after: MARKER } + }); + expect(page.last_id).toBe(MARKER); + expect(page.data[0]).toMatchObject({ id: MARKER, type: "compaction" }); + expect(JSON.stringify(page)).not.toContain("private"); + }); + + test("surfaces old-server and transport failures once to the advisory caller", async () => { + const readers = [(client: OpenAI) => readChatCompactions(client, "conversation")]; + for (const read of readers) { + for (const failure of [{ status: 404 }, new Error("private transport details")]) { + let calls = 0; + const client = { + get: async () => { + calls += 1; + throw failure; + } + } as unknown as OpenAI; + await expect(read(client)).rejects.toBe(failure); + expect(calls).toBe(1); + } + } + }); +}); diff --git a/apps/maple-research/frontend/src/services/chatCompactionApi.ts b/apps/maple-research/frontend/src/services/chatCompactionApi.ts new file mode 100644 index 000000000..cbfc7d65e --- /dev/null +++ b/apps/maple-research/frontend/src/services/chatCompactionApi.ts @@ -0,0 +1,20 @@ +import type OpenAI from "openai"; +import { parseCompactionPage } from "./chatCompaction"; + +/** + * Additive OpenSecret metadata, over the same account-bound encrypted fetch as + * chat. Use that transport's configured timeout, not a short metadata deadline + * that could abort a healthy read and retire the shared encrypted session. + */ +export async function readChatCompactions( + client: OpenAI, + conversationId: string, + after?: string, + signal?: AbortSignal +) { + const result = await client.get( + `/conversations/${encodeURIComponent(conversationId)}/compactions`, + { query: { limit: 100, order: "desc", ...(after ? { after } : {}) }, signal } + ); + return parseCompactionPage(result); +} diff --git a/apps/maple-research/frontend/src/services/chatRuntimeStore.test.ts b/apps/maple-research/frontend/src/services/chatRuntimeStore.test.ts index f38774a1e..bf85f7e07 100644 --- a/apps/maple-research/frontend/src/services/chatRuntimeStore.test.ts +++ b/apps/maple-research/frontend/src/services/chatRuntimeStore.test.ts @@ -7,6 +7,7 @@ import { type ChatRuntimeSnapshot } from "./chatRuntimeStore"; import { chatCursorAfterSendFailure } from "./chatSendFailureRecovery"; +import { chatCompactionProgressUpdate } from "./chatCompaction"; type Conversation = { id: string; title: string }; type Message = { id: string; text: string; status: "streaming" | "completed" }; @@ -34,6 +35,77 @@ function streamingMessage(id: string, text: string): Message { } describe("ChatRuntimeStore", () => { + test("streamed progress remains owned by its run across navigation and cancellation", () => { + const store = createStore(); + const run = store.beginRun(A); + store.select(B); + const progress = chatCompactionProgressUpdate({ + type: "response.compaction.compacting", + sequence_number: 2, + output_index: 0, + item_id: "30000000-0000-4000-8000-000000000001" + }); + expect(progress).toBe(true); + store.updateForRun(A, run.token, (snapshot) => ({ + ...snapshot, + compactingMessageId: progress ? "a-user" : undefined + })); + expect(store.get(A)?.compactingMessageId).toBe("a-user"); + expect(store.get(B)?.compactingMessageId).toBeUndefined(); + expect(store.get(A)?.compactions).toBeUndefined(); + expect(store.get(A)?.messages).toEqual([]); + + store.cancelRun(A, run.token); + const replacement = store.beginRun(A); + expect( + store.updateForRun(A, run.token, (snapshot) => ({ + ...snapshot, + compactingMessageId: "stale-a-user" + })) + ).toBe(false); + expect(store.isRunCurrent(A, replacement.token)).toBe(true); + expect(store.get(A)?.compactingMessageId).toBeUndefined(); + }); + + test("clears advisory compaction on cancellation and cannot revive it in the next run", () => { + const store = createStore(); + const run = store.beginRun(A); + store.updateForRun(A, run.token, (snapshot) => ({ + ...snapshot, + compactingMessageId: "previous-message" + })); + store.cancelRun(A, run.token); + expect(store.get(A)?.compactingMessageId).toBeUndefined(); + expect( + store.updateForRun(A, run.token, (snapshot) => ({ + ...snapshot, + compactingMessageId: "late-message" + })) + ).toBe(false); + store.beginRun(A); + expect(store.get(A)?.compactingMessageId).toBeUndefined(); + }); + + test("settling a run clears advisory progress before its abort listeners run", () => { + for (const settle of ["completeRunAndAbort", "finishRun", "completeRun"] as const) { + const store = createStore(); + const run = store.beginRun(A); + store.updateForRun(A, run.token, (snapshot) => ({ + ...snapshot, + compactingMessageId: "active-message", + lastSeenItemId: "canonical-cursor" + })); + let observedDuringAbort: string | undefined; + run.signal.addEventListener("abort", () => { + observedDuringAbort = store.get(A)?.compactingMessageId; + }); + expect(store[settle](A, run.token)).toBe(true); + expect(store.get(A)?.compactingMessageId).toBeUndefined(); + expect(observedDuringAbort).toBeUndefined(); + expect(store.get(A)?.lastSeenItemId).toBe("canonical-cursor"); + } + }); + test("keeps concurrent A and B runs isolated while A receives offscreen updates", () => { const store = createStore(); store.select(A, { diff --git a/apps/maple-research/frontend/src/services/chatRuntimeStore.ts b/apps/maple-research/frontend/src/services/chatRuntimeStore.ts index 18cc2cdd7..a17b2d946 100644 --- a/apps/maple-research/frontend/src/services/chatRuntimeStore.ts +++ b/apps/maple-research/frontend/src/services/chatRuntimeStore.ts @@ -1,3 +1,5 @@ +import type { ChatCompactionMarker } from "./chatCompaction"; + export type ChatRuntimeKey = string; const EMPTY_CHAT_RUNTIME_KEYS: readonly ChatRuntimeKey[] = Object.freeze([]); @@ -34,6 +36,11 @@ export type ChatRuntimeSnapshot = Readonly<{ assistantStreaming: boolean; historyLoaded: boolean; runToken: number | null; + /** Presentation metadata is separate from canonical messages and their cursors. */ + compactions?: readonly ChatCompactionMarker[]; + compactionHistoryCursor?: string; + hasMoreCompactionHistory?: boolean; + compactingMessageId?: string; }>; export type ChatRuntimeInitialState = Partial< @@ -571,6 +578,7 @@ export class ChatRuntimeStore { ...entry.snapshot, isGenerating: true, currentResponseId: undefined, + compactingMessageId: undefined, error: null, assistantStreaming: false }, @@ -656,6 +664,7 @@ export class ChatRuntimeStore { ...completed, isGenerating: false, currentResponseId: undefined, + compactingMessageId: undefined, assistantStreaming: false }, { @@ -695,6 +704,7 @@ export class ChatRuntimeStore { ...entry.snapshot, isGenerating: false, currentResponseId: undefined, + compactingMessageId: undefined, assistantStreaming: false }, { diff --git a/services/opensecret/src/db.rs b/services/opensecret/src/db.rs index 19ca56a18..7ee70b102 100644 --- a/services/opensecret/src/db.rs +++ b/services/opensecret/src/db.rs @@ -32,8 +32,8 @@ use crate::models::project_settings::{ EmailSettings, NewProjectSetting, ProjectSetting, ProjectSettingError, SettingCategory, }; use crate::models::response_compaction::{ - CompactionCheckpoint, CompactionError, CompactionPublishResult, CompactionSnapshot, - ContextItemKey, NewCompactionCheckpoint, + CompactionCheckpoint, CompactionError, CompactionMarkerMetadata, CompactionPublishResult, + CompactionSnapshot, ContextItemKey, NewCompactionCheckpoint, }; use crate::models::responses::{ validate_conversation_project_limit, AssistantMessage, Conversation, ConversationProject, @@ -770,6 +770,14 @@ pub trait DBConnection { user_id: Uuid, response_uuid: Uuid, ) -> Result, DBError>; + fn list_conversation_compaction_markers( + &self, + conversation_id: i64, + user_id: Uuid, + limit: i64, + after: Option, + order: &str, + ) -> Result, DBError>; fn publish_compaction_checkpoint( &self, checkpoint: NewCompactionCheckpoint, @@ -3122,6 +3130,19 @@ impl DBConnection for PostgresConnection { CompactionCheckpoint::publish(conn, checkpoint).map_err(DBError::from) } + fn list_conversation_compaction_markers( + &self, + conversation_id: i64, + user_id: Uuid, + limit: i64, + after: Option, + order: &str, + ) -> Result, DBError> { + let conn = &mut self.db.get().map_err(|_| DBError::ConnectionError)?; + CompactionCheckpoint::list_markers(conn, conversation_id, user_id, limit, after, order) + .map_err(DBError::from) + } + // Optimized context reconstruction (metadata-based) fn get_conversation_context_metadata( &self, diff --git a/services/opensecret/src/models/response_compaction.rs b/services/opensecret/src/models/response_compaction.rs index 319475241..bf28cedce 100644 --- a/services/opensecret/src/models/response_compaction.rs +++ b/services/opensecret/src/models/response_compaction.rs @@ -175,6 +175,22 @@ pub struct CompactionCheckpoint { pub after_item_uuid: Option, } +/// Public marker projection. Its SQL deliberately never selects checkpoint +/// ciphertext, summaries, source fingerprints or covered conversation items. +#[derive(Debug, Clone, QueryableByName)] +pub struct CompactionMarkerMetadata { + #[diesel(sql_type = diesel::sql_types::Uuid)] + pub uuid: Uuid, + #[diesel(sql_type = diesel::sql_types::Uuid)] + pub origin_response_uuid: Uuid, + #[diesel(sql_type = Nullable)] + pub after_item_uuid: Option, + #[diesel(sql_type = Timestamptz)] + pub created_at: DateTime, + #[diesel(sql_type = BigInt)] + pub generation: i64, +} + /// The envelope is encrypted by the user-key owner, including the row/conversation /// identities and exact covered keys. Plaintext coverage is used only for this CAS. pub struct NewCompactionCheckpoint { @@ -536,6 +552,59 @@ impl CompactionSnapshot { } impl CompactionCheckpoint { + /// Generation is the stable cursor order; canonical item cursors remain + /// independent. `limit` includes the handler's single look-ahead row. + pub fn list_markers( + conn: &mut PgConnection, + conversation_id: i64, + user_id: Uuid, + limit: i64, + after: Option, + order: &str, + ) -> Result, CompactionError> { + if !(1..=101).contains(&limit) || !matches!(order, "asc" | "desc") { + return Err(CompactionError::InvalidSelection); + } + conversation_identity(conn, conversation_id, user_id, false)?; + #[derive(QueryableByName)] + struct Cursor { + #[diesel(sql_type = BigInt)] + generation: i64, + } + let generation = match after { + Some(uuid) => Some( + diesel::sql_query("SELECT generation FROM conversation_summaries s WHERE conversation_id=$1 AND user_id=$2 AND uuid=$3 AND format_version=1 AND EXISTS (SELECT 1 FROM responses r WHERE r.uuid=s.origin_response_uuid AND r.conversation_id=s.conversation_id AND r.user_id=s.user_id)") + .bind::(conversation_id) + .bind::(user_id) + .bind::(uuid) + .get_result::(conn) + .optional()? + .ok_or(CompactionError::InvalidSelection)? + .generation, + ), + None => None, + }; + // These SQL fragments come only from the validated order enum above. + let (comparison, direction) = if order == "asc" { + (">", "ASC") + } else { + ("<", "DESC") + }; + diesel::sql_query(format!( + "SELECT uuid, origin_response_uuid, after_item_uuid, created_at, generation FROM conversation_summaries s \ + WHERE conversation_id=$1 AND user_id=$2 AND format_version=1 \ + AND EXISTS (SELECT 1 FROM responses r WHERE r.uuid=s.origin_response_uuid \ + AND r.conversation_id=s.conversation_id AND r.user_id=s.user_id) \ + AND ($3::bigint IS NULL OR generation {comparison} $3) ORDER BY generation {direction} LIMIT $4" + )) + .bind::(conversation_id) + .bind::(user_id) + .bind::, _>(generation) + .bind::(limit) + .load(conn) + .map_err(CompactionError::from) + } + fn head( conn: &mut PgConnection, conversation_id: i64, diff --git a/services/opensecret/src/models/response_compaction/tests.rs b/services/opensecret/src/models/response_compaction/tests.rs index 71124f5f9..6e4e16de1 100644 --- a/services/opensecret/src/models/response_compaction/tests.rs +++ b/services/opensecret/src/models/response_compaction/tests.rs @@ -287,6 +287,144 @@ fn published(result: CompactionPublishResult) -> CompactionCheckpoint { } } +#[test] +#[ignore = "requires explicitly selected disposable migrated database"] +fn db_compaction_visibility_marker_pagination_is_bounded_stable_and_owner_scoped() { + let mut fixture = Fixture::new(); + let item = fixture.user_item(None, at(0)); + // Paid preparation can publish a usable checkpoint before the enclosing + // request is persisted. Its memory remains available to recall, but an + // unanchored attempt must not crowd actual history markers out of a page. + let source = fixture.snapshot(); + let orphan_candidate = fixture.candidate(&source, vec![item], None); + let orphan = published(fixture.publish(orphan_candidate)); + let response = fixture.response(ResponseStatus::Completed); + let mut published_rows = Vec::new(); + for index in 0..3 { + let source = fixture.snapshot(); + let mut candidate = fixture.candidate(&source, vec![item], Some(response.uuid)); + candidate.after_item_uuid = (index == 1).then_some(item.uuid); + published_rows.push(published(fixture.publish(candidate))); + } + let ascending = CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.user, + 2, + None, + "asc", + ) + .unwrap(); + assert_eq!( + ascending.iter().map(|row| row.uuid).collect::>(), + vec![published_rows[0].uuid, published_rows[1].uuid] + ); + assert_eq!(ascending[0].generation, 2); + assert!(ascending[0].after_item_uuid.is_none()); + assert_eq!(ascending[1].after_item_uuid, Some(item.uuid)); + let remaining = CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.user, + 2, + Some(ascending[1].uuid), + "asc", + ) + .unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].uuid, published_rows[2].uuid); + let descending = CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.user, + 2, + None, + "desc", + ) + .unwrap(); + assert_eq!( + descending.iter().map(|row| row.uuid).collect::>(), + vec![published_rows[2].uuid, published_rows[1].uuid] + ); + let remaining = CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.user, + 2, + Some(descending[1].uuid), + "desc", + ) + .unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].uuid, published_rows[0].uuid); + assert!(matches!( + CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.user, + 20, + Some(orphan.uuid), + "asc" + ), + Err(CompactionError::InvalidSelection) + )); + for (limit, order) in [(0, "asc"), (102, "desc"), (1, "unknown")] { + assert!(matches!( + CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.user, + limit, + None, + order + ), + Err(CompactionError::InvalidSelection) + )); + } + assert!(matches!( + CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.other_user, + 20, + None, + "asc" + ), + Err(CompactionError::ConversationNotFound) + )); + assert!(matches!( + CompactionCheckpoint::list_markers( + &mut fixture.conn, + fixture.conversation_id, + fixture.user, + 20, + Some(Uuid::new_v4()), + "asc" + ), + Err(CompactionError::InvalidSelection) + )); + let other_conversation = NewConversation { + uuid: Uuid::new_v4(), + user_id: fixture.user, + project_id: None, + is_pinned: false, + metadata_enc: None, + } + .insert(&mut fixture.conn) + .unwrap(); + assert!(matches!( + CompactionCheckpoint::list_markers( + &mut fixture.conn, + other_conversation.id, + fixture.user, + 20, + Some(published_rows[0].uuid), + "desc" + ), + Err(CompactionError::InvalidSelection) + )); +} + fn synthetic_snapshot() -> CompactionSnapshot { let call_uuid = Uuid::from_u128(102); let items = [ diff --git a/services/opensecret/src/web/responses/conversations.rs b/services/opensecret/src/web/responses/conversations.rs index 7f6d80284..43a5c8761 100644 --- a/services/opensecret/src/web/responses/conversations.rs +++ b/services/opensecret/src/web/responses/conversations.rs @@ -2,8 +2,10 @@ //! Provides server-side conversation state management compatible with OpenAI's Conversations API use crate::{ + db::DBError, encrypt::{decrypt_content, encrypt_with_key}, jwt::AuthContext, + models::response_compaction::{CompactionError, CompactionMarkerMetadata}, models::responses::{ConversationProjectFilter, NewConversation}, models::users::User, web::{ @@ -242,6 +244,76 @@ pub struct ConversationItemListResponse { pub last_id: Option, } +/// Additive visibility metadata. Canonical conversation items are unchanged. +#[derive(Debug, Serialize)] +pub struct ConversationCompactionMarker { + pub id: Uuid, + pub object: &'static str, + pub status: &'static str, + pub response_id: Uuid, + /// None for preparation before this response's user item is persisted. + /// Otherwise the exact canonical item preceding this checkpoint. + pub after_item_id: Option, + pub created_at: i64, + pub generation: i64, +} + +#[derive(Debug, Serialize)] +pub struct ConversationCompactionListResponse { + pub object: &'static str, + pub data: Vec, + pub has_more: bool, + pub first_id: Option, + pub last_id: Option, +} + +#[derive(Debug, Deserialize)] +pub struct ListCompactionsParams { + #[serde(default = "default_limit")] + pub limit: i64, + pub after: Option, + #[serde(default = "default_order")] + pub order: String, +} + +impl ListCompactionsParams { + fn validate(&self) -> Result<(), ApiError> { + if !(1..=MAX_PAGINATION_LIMIT).contains(&self.limit) + || !matches!(self.order.as_str(), "asc" | "desc") + { + return Err(ApiError::BadRequest); + } + Ok(()) + } +} + +fn compaction_marker_page( + mut rows: Vec, + limit: usize, +) -> ConversationCompactionListResponse { + let has_more = rows.len() > limit; + rows.truncate(limit); + let data: Vec<_> = rows + .into_iter() + .map(|row| ConversationCompactionMarker { + id: row.uuid, + object: "conversation.compaction", + status: "completed", + response_id: row.origin_response_uuid, + after_item_id: row.after_item_uuid, + created_at: row.created_at.timestamp(), + generation: row.generation, + }) + .collect(); + ConversationCompactionListResponse { + object: OBJECT_TYPE_LIST, + first_id: data.first().map(|item| item.id), + last_id: data.last().map(|item| item.id), + has_more, + data, + } +} + /// Response for listing conversations (custom OpenSecret extension) #[derive(Debug, Clone, Serialize)] pub struct ConversationListResponse { @@ -637,6 +709,37 @@ async fn list_conversation_items( encrypt_response(&state, &session_id, &response).await } +/// GET /v1/conversations/{id}/compactions - Paginated durable markers only. +async fn list_conversation_compactions( + State(state): State>, + Path(conversation_id): Path, + Query(params): Query, + Extension(session_id): Extension, + Extension(user): Extension, +) -> Result { + params.validate()?; + let conversation = state + .db + .get_conversation_by_uuid_and_user(conversation_id, user.uuid) + .map_err(error_mapping::map_conversation_error)?; + let rows = state + .db + .list_conversation_compaction_markers( + conversation.id, + user.uuid, + params.limit + 1, + params.after, + ¶ms.order, + ) + .map_err(|error| match error { + DBError::CompactionError(CompactionError::InvalidSelection) => ApiError::BadRequest, + DBError::CompactionError(CompactionError::ConversationNotFound) => ApiError::NotFound, + _ => ApiError::InternalServerError, + })?; + let response = compaction_marker_page(rows, params.limit as usize); + encrypt_response(&state, &session_id, &response).await +} + /// GET /v1/conversations/{conversation_id}/items/{item_id} - Retrieve a single item async fn get_conversation_item( State(state): State>, @@ -919,6 +1022,11 @@ pub fn router(state: Arc) -> Router { get(list_conversation_items) .layer(from_fn_with_state(state.clone(), decrypt_request::<()>)), ) + .route( + "/v1/conversations/:id/compactions", + get(list_conversation_compactions) + .layer(from_fn_with_state(state.clone(), decrypt_request::<()>)), + ) .route( "/v1/conversations/:id/items/:item_id", get(get_conversation_item) @@ -929,14 +1037,99 @@ pub fn router(state: Arc) -> Router { #[cfg(test)] mod tests { + use super::{compaction_marker_page, ListCompactionsParams}; use super::{ default_limit, default_order, BatchUpdateConversationProjectRequest, ListConversationsParams, UpdateConversationRequest, }; + use crate::models::response_compaction::CompactionMarkerMetadata; use crate::web::responses::NullableField; use crate::ApiError; use uuid::Uuid; + #[test] + fn compaction_visibility_rejects_unbounded_pages_and_unknown_order() { + for limit in [-1, 0, 101, i64::MAX] { + let params = ListCompactionsParams { + limit, + after: None, + order: "desc".into(), + }; + assert!(matches!(params.validate(), Err(ApiError::BadRequest))); + } + for order in ["asc", "desc"] { + for limit in [1, 20, 100] { + assert!(ListCompactionsParams { + limit, + after: None, + order: order.into() + } + .validate() + .is_ok()); + } + } + assert!(ListCompactionsParams { + limit: 20, + after: None, + order: "ASC; SELECT 1".into() + } + .validate() + .is_err()); + let defaults: ListCompactionsParams = serde_json::from_str("{}").unwrap(); + assert_eq!(defaults.limit, 20); + assert_eq!(defaults.order, "desc"); + } + + #[test] + fn compaction_marker_projection_is_bounded_and_contains_only_safe_metadata() { + let rows: Vec<_> = (1..=3) + .map(|generation| CompactionMarkerMetadata { + uuid: Uuid::new_v4(), + origin_response_uuid: Uuid::new_v4(), + after_item_uuid: (generation == 2).then(Uuid::new_v4), + created_at: chrono::Utc::now(), + generation, + }) + .collect(); + let first = rows[0].uuid; + let last = rows[1].uuid; + let page = compaction_marker_page(rows, 2); + assert!(page.has_more); + assert_eq!(page.first_id, Some(first)); + assert_eq!(page.last_id, Some(last)); + assert_eq!(page.data.len(), 2); + let wire = serde_json::to_value(page).unwrap(); + for marker in wire["data"].as_array().unwrap() { + let keys: std::collections::HashSet<_> = marker + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + assert_eq!( + keys, + [ + "id", + "object", + "status", + "response_id", + "after_item_id", + "created_at", + "generation" + ] + .into_iter() + .collect() + ); + assert_eq!(marker["object"], "conversation.compaction"); + assert_eq!(marker["status"], "completed"); + } + assert!(wire["data"][0]["after_item_id"].is_null()); + assert!(wire["data"][1]["after_item_id"].is_string()); + let empty = compaction_marker_page(Vec::new(), 20); + assert!(!empty.has_more); + assert!(empty.first_id.is_none() && empty.last_id.is_none()); + } + #[test] fn update_request_distinguishes_null_from_omitted_project_id() { let request: UpdateConversationRequest =