diff --git a/app/src/components/channels/mcp/ConfigAssistantPanel.tsx b/app/src/components/channels/mcp/ConfigAssistantPanel.tsx index 896574d99..bb40d670b 100644 --- a/app/src/components/channels/mcp/ConfigAssistantPanel.tsx +++ b/app/src/components/channels/mcp/ConfigAssistantPanel.tsx @@ -7,8 +7,8 @@ import debug from 'debug'; import { useCallback, useEffect, useRef, useState } from 'react'; +import { BubbleMarkdown } from '../../../features/conversations/components/AgentMessageBubble'; import { useT } from '../../../lib/i18n/I18nContext'; -import { BubbleMarkdown } from '../../../pages/conversations/components/AgentMessageBubble'; import { mcpClientsApi } from '../../../services/api/mcpClientsApi'; import Button from '../../ui/Button'; diff --git a/app/src/components/chat/CycleUsagePill.tsx b/app/src/components/chat/CycleUsagePill.tsx index 75a2e9b3b..9ad28d644 100644 --- a/app/src/components/chat/CycleUsagePill.tsx +++ b/app/src/components/chat/CycleUsagePill.tsx @@ -1,7 +1,7 @@ +import { LimitPill } from '../../features/conversations/components/LimitPill'; +import { formatResetTime } from '../../features/conversations/utils/format'; import { useUsageState } from '../../hooks/useUsageState'; import { useT } from '../../lib/i18n/I18nContext'; -import { LimitPill } from '../../pages/conversations/components/LimitPill'; -import { formatResetTime } from '../../pages/conversations/utils/format'; /** * Self-contained cycle usage indicator for the composer toolbar. diff --git a/app/src/components/flows/FlowRunInspectorDrawer.tsx b/app/src/components/flows/FlowRunInspectorDrawer.tsx index 1ad99710d..44298be02 100644 --- a/app/src/components/flows/FlowRunInspectorDrawer.tsx +++ b/app/src/components/flows/FlowRunInspectorDrawer.tsx @@ -4,7 +4,7 @@ * * Right-side drawer showing a single durable `tinyflows` run's status + step * timeline, opened from the "View run" action on {@link FlowApprovalCard}. - * Drawer chrome mirrors `pages/conversations/components/SubagentDrawer.tsx` + * Drawer chrome mirrors `features/conversations/components/SubagentDrawer.tsx` * (fixed overlay + backdrop-click-to-close + Escape-to-close) so it renders * as a fixed overlay regardless of where the parent mounts it in the DOM. * @@ -18,7 +18,7 @@ * + collapsible output, not a graduated status timeline. Status-dot/pill * visual language borrows from `components/intelligence/WorkflowRunDetail.tsx` * (`RUN_STATUS_ACCENT`/`PHASE_STATUS_DOT`) and - * `pages/conversations/components/ToolTimelineBlock.tsx` (`StatusTag`) — + * `features/conversations/components/ToolTimelineBlock.tsx` (`StatusTag`) — * dots, not progress bars (project rule). */ import debug from 'debug'; diff --git a/app/src/components/flows/WorkflowCopilotPanel.tsx b/app/src/components/flows/WorkflowCopilotPanel.tsx index 5209dcb7e..e5c36aa23 100644 --- a/app/src/components/flows/WorkflowCopilotPanel.tsx +++ b/app/src/components/flows/WorkflowCopilotPanel.tsx @@ -21,12 +21,12 @@ */ import { useCallback, useEffect, useRef, useState } from 'react'; +import { BubbleMarkdown } from '../../features/conversations/components/AgentMessageBubble'; +import { ToolTimelineBlock } from '../../features/conversations/components/ToolTimelineBlock'; import { useWorkflowBuilderChat } from '../../hooks/useWorkflowBuilderChat'; import { diffGraphs } from '../../lib/flows/graphDiff'; import type { WorkflowGraph } from '../../lib/flows/types'; import { useT } from '../../lib/i18n/I18nContext'; -import { BubbleMarkdown } from '../../pages/conversations/components/AgentMessageBubble'; -import { ToolTimelineBlock } from '../../pages/conversations/components/ToolTimelineBlock'; import type { WorkflowProposal } from '../../store/chatRuntimeSlice'; import ChatComposer from '../chat/ChatComposer'; import Button from '../ui/Button'; diff --git a/app/src/components/intelligence/IntelligenceTasksTab.tsx b/app/src/components/intelligence/IntelligenceTasksTab.tsx index 71d57e716..30f06a093 100644 --- a/app/src/components/intelligence/IntelligenceTasksTab.tsx +++ b/app/src/components/intelligence/IntelligenceTasksTab.tsx @@ -26,9 +26,9 @@ import { } from 'react-icons/lu'; import { useLocation, useNavigate } from 'react-router-dom'; +import { TaskKanbanBoard } from '../../features/conversations/components/TaskKanbanBoard'; +import { isTaskThread } from '../../features/conversations/utils/threadFilter'; import { useT } from '../../lib/i18n/I18nContext'; -import { TaskKanbanBoard } from '../../pages/conversations/components/TaskKanbanBoard'; -import { isTaskThread } from '../../pages/conversations/utils/threadFilter'; import { threadApi } from '../../services/api/threadApi'; import { TASK_SOURCES_THREAD_ID, diff --git a/app/src/components/intelligence/ModelCouncilTab.tsx b/app/src/components/intelligence/ModelCouncilTab.tsx index 3fbceee41..2a09b011b 100644 --- a/app/src/components/intelligence/ModelCouncilTab.tsx +++ b/app/src/components/intelligence/ModelCouncilTab.tsx @@ -8,6 +8,7 @@ */ import { useCallback, useEffect, useMemo, useState } from 'react'; +import { BubbleMarkdown } from '../../features/conversations/components/AgentMessageBubble'; import { getMascotPalette, hexToArgbInt, @@ -15,7 +16,6 @@ import { RiveMascot, } from '../../features/human/Mascot'; import { useT } from '../../lib/i18n/I18nContext'; -import { BubbleMarkdown } from '../../pages/conversations/components/AgentMessageBubble'; import { listProviderModels, loadAISettings, diff --git a/app/src/components/intelligence/__tests__/IntelligenceTasksTab.test.tsx b/app/src/components/intelligence/__tests__/IntelligenceTasksTab.test.tsx index a0f6347f1..1fc5522f0 100644 --- a/app/src/components/intelligence/__tests__/IntelligenceTasksTab.test.tsx +++ b/app/src/components/intelligence/__tests__/IntelligenceTasksTab.test.tsx @@ -110,7 +110,7 @@ vi.mock('../UserTaskComposer', () => ({ // Stub the kanban to a simple list that still surfaces the write callbacks // the personal board wires up, so we can assert the todos RPC is called. -vi.mock('../../../pages/conversations/components/TaskKanbanBoard', () => ({ +vi.mock('../../../features/conversations/components/TaskKanbanBoard', () => ({ TaskKanbanBoard: ({ board, headerTitleKey, diff --git a/app/src/components/layout/shell/useNewChat.ts b/app/src/components/layout/shell/useNewChat.ts index 3006a1fec..775b321fe 100644 --- a/app/src/components/layout/shell/useNewChat.ts +++ b/app/src/components/layout/shell/useNewChat.ts @@ -4,7 +4,7 @@ import { useNavigate } from 'react-router-dom'; import { GENERAL_TAB_VALUE, isThreadVisibleInTab, -} from '../../../pages/conversations/utils/threadFilter'; +} from '../../../features/conversations/utils/threadFilter'; import { setActiveAccount } from '../../../store/accountsSlice'; import { useAppDispatch, useAppSelector } from '../../../store/hooks'; import { createNewThread, loadThreadMessages, setSelectedThread } from '../../../store/threadSlice'; diff --git a/app/src/components/notifications/NotificationBody.tsx b/app/src/components/notifications/NotificationBody.tsx index 2b4b4dee8..0757f8d85 100644 --- a/app/src/components/notifications/NotificationBody.tsx +++ b/app/src/components/notifications/NotificationBody.tsx @@ -10,7 +10,7 @@ * * **Why this lives here and not in a global shared spot:** the chat-side * `OpenhumanLinkPill` is a non-exported function inside `AgentMessageBubble.tsx` - * (`app/src/pages/conversations/`). Extracting from chat would change the chat + * (`app/src/features/conversations/`). Extracting from chat would change the chat * render path — out of scope for this fix. Instead, we keep the grammar / parsing * shared (reuses `parseBubbleSegments` from conversations) but reimplement the * pill locally. Both notification surfaces share *this* file so the diff stays @@ -22,7 +22,7 @@ * allowlists `path` values before routing. See `OpenhumanLinkModal.tsx` * `ALLOWED_PATHS_SET`. */ -import { parseBubbleSegments } from '../../pages/conversations/utils/format'; +import { parseBubbleSegments } from '../../features/conversations/utils/format'; import { OPENHUMAN_LINK_EVENT } from '../OpenhumanLinkModal'; function NotificationLinkPill({ path, label }: { path: string; label: string }) { diff --git a/app/src/pages/Conversations.tsx b/app/src/features/conversations/Conversations.tsx similarity index 90% rename from app/src/pages/Conversations.tsx rename to app/src/features/conversations/Conversations.tsx index 6a19955e1..727408da0 100644 --- a/app/src/pages/Conversations.tsx +++ b/app/src/features/conversations/Conversations.tsx @@ -3,27 +3,67 @@ import debugFactory from 'debug'; import { Fragment, useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { useLocation, useNavigate, useParams } from 'react-router-dom'; -import { type ChatSendError, chatSendError } from '../chat/chatSendError'; -import { checkPromptInjection, promptGuardMessage } from '../chat/promptInjectionGuard'; -import ApprovalRequestCard from '../components/chat/ApprovalRequestCard'; -import ArtifactCard from '../components/chat/ArtifactCard'; -import ChatComposer from '../components/chat/ChatComposer'; -import ChatFilesChip from '../components/chat/ChatFilesChip'; -import ChatNewWindowHero from '../components/chat/ChatNewWindowHero'; -import ComposerTokenStats from '../components/chat/ComposerTokenStats'; -import IntegrationConnectCard from '../components/chat/IntegrationConnectCard'; -import QueuedFollowups from '../components/chat/QueuedFollowups'; -import SuperContextToggle from '../components/chat/SuperContextToggle'; -import { whenSuperContextWriteSettled } from '../components/chat/superContextWrite'; -import WorkflowProposalCard from '../components/chat/WorkflowProposalCard'; -import { ConfirmationModal } from '../components/intelligence/ConfirmationModal'; -import { SidebarContent } from '../components/layout/shell/SidebarSlot'; -import { settingsNavState } from '../components/settings/modal/settingsOverlay'; -import UpsellBanner from '../components/upsell/UpsellBanner'; -import { dismissBanner, shouldShowBanner } from '../components/upsell/upsellDismissState'; -import MicComposer from '../features/human/MicComposer'; -import { useStickToBottom } from '../hooks/useStickToBottom'; -import { useUsageState } from '../hooks/useUsageState'; +import { type ChatSendError, chatSendError } from '../../chat/chatSendError'; +import { checkPromptInjection, promptGuardMessage } from '../../chat/promptInjectionGuard'; +import ApprovalRequestCard from '../../components/chat/ApprovalRequestCard'; +import ArtifactCard from '../../components/chat/ArtifactCard'; +import ChatComposer from '../../components/chat/ChatComposer'; +import ChatFilesChip from '../../components/chat/ChatFilesChip'; +import ChatNewWindowHero from '../../components/chat/ChatNewWindowHero'; +import ComposerTokenStats from '../../components/chat/ComposerTokenStats'; +import IntegrationConnectCard from '../../components/chat/IntegrationConnectCard'; +import QueuedFollowups from '../../components/chat/QueuedFollowups'; +import SuperContextToggle from '../../components/chat/SuperContextToggle'; +import { whenSuperContextWriteSettled } from '../../components/chat/superContextWrite'; +import WorkflowProposalCard from '../../components/chat/WorkflowProposalCard'; +import { ConfirmationModal } from '../../components/intelligence/ConfirmationModal'; +import { SidebarContent } from '../../components/layout/shell/SidebarSlot'; +import { settingsNavState } from '../../components/settings/modal/settingsOverlay'; +import UpsellBanner from '../../components/upsell/UpsellBanner'; +import { dismissBanner, shouldShowBanner } from '../../components/upsell/upsellDismissState'; +import { + AgentMessageBubble, + AgentMessageText, + BubbleMarkdown, +} from '../../features/conversations/components/AgentMessageBubble'; +import { AgentProcessSourcePanel } from '../../features/conversations/components/AgentProcessSourcePanel'; +import { + BackgroundProcessesPanel, + selectBackgroundProcesses, +} from '../../features/conversations/components/BackgroundProcessesPanel'; +import { + CitationChips, + type MessageCitation, +} from '../../features/conversations/components/CitationChips'; +import { PlanReviewCard } from '../../features/conversations/components/PlanReviewCard'; +import { SubagentDrawer } from '../../features/conversations/components/SubagentDrawer'; +import { + ThreadGoalEditorPanel, + ThreadGoalFooterTrigger, + useThreadGoal, +} from '../../features/conversations/components/ThreadGoalChip'; +import { ThreadTodoStrip } from '../../features/conversations/components/ThreadTodoStrip'; +import { ToolTimelineBlock } from '../../features/conversations/components/ToolTimelineBlock'; +import { + evaluateComposerSend, + getComposerBlockedSendFeedback, + handleComposerSlashCommand, +} from '../../features/conversations/composerSendDecision'; +import { useMemorySyncActive } from '../../features/conversations/hooks/useBackgroundActivity'; +import { + type AgentBubblePosition, + buildAcceptedInlineCompletion, + formatRelativeTime, + formatResetTime, + getInlineCompletionSuffix, +} from '../../features/conversations/utils/format'; +import { + GENERAL_TAB_VALUE, + isThreadVisibleInTab, +} from '../../features/conversations/utils/threadFilter'; +import MicComposer from '../../features/human/MicComposer'; +import { useStickToBottom } from '../../hooks/useStickToBottom'; +import { useUsageState } from '../../hooks/useUsageState'; import { type Attachment, ATTACHMENT_MAX_FILES, @@ -32,33 +72,34 @@ import { imageMarkerCost, parseMessageImages, validateAndReadFile, -} from '../lib/attachments'; -import { useT } from '../lib/i18n/I18nContext'; -import { trackEvent } from '../services/analytics'; -import { applyOpenRouterFreeModels } from '../services/api/openrouterFreeModels'; -import { subagentApi } from '../services/api/subagentApi'; -import { threadApi } from '../services/api/threadApi'; -import { fetchThreadTokenUsage } from '../services/api/threadUsageApi'; +} from '../../lib/attachments'; +import { useT } from '../../lib/i18n/I18nContext'; +import { trackEvent } from '../../services/analytics'; +import { applyOpenRouterFreeModels } from '../../services/api/openrouterFreeModels'; +import { subagentApi } from '../../services/api/subagentApi'; +import { threadApi } from '../../services/api/threadApi'; +import { fetchThreadTokenUsage } from '../../services/api/threadUsageApi'; import { aiRegenerate, chatCancel, chatClearQueue, chatSend, useRustChat, -} from '../services/chatService'; -import { callCoreRpc } from '../services/coreRpcClient'; +} from '../../services/chatService'; +import { callCoreRpc } from '../../services/coreRpcClient'; import { loadAgentProfiles, selectActiveAgentProfileId, selectAgentProfile, selectAgentProfiles, -} from '../store/agentProfileSlice'; +} from '../../store/agentProfileSlice'; import { beginInferenceTurn, clearFollowupsForThread, clearRuntimeForThread, clearThreadSendPending, enqueueFollowup, + fetchAndHydrateTurnHistory, fetchAndHydrateTurnState, hydrateThreadUsage, markSubagentCancelled, @@ -68,9 +109,9 @@ import { setTaskBoardForThread, setToolTimelineForThread, type ToolTimelineEntry, -} from '../store/chatRuntimeSlice'; -import { useAppDispatch, useAppSelector } from '../store/hooks'; -import { selectSocketStatus } from '../store/socketSelectors'; +} from '../../store/chatRuntimeSlice'; +import { useAppDispatch, useAppSelector } from '../../store/hooks'; +import { selectSocketStatus } from '../../store/socketSelectors'; import { addMessageLocal, clearThreadInferenceActive, @@ -83,14 +124,14 @@ import { setSelectedThread, THREAD_NOT_FOUND_MESSAGE, updateThreadTitle, -} from '../store/threadSlice'; -import type { ConfirmationModal as ConfirmationModalType } from '../types/intelligence'; -import type { ThreadMessage } from '../types/thread'; -import { splitAgentMessageIntoBubbles } from '../utils/agentMessageBubbles'; -import { chatThreadPath } from '../utils/chatRoutes'; -import { CHAT_ATTACHMENTS_ENABLED } from '../utils/config'; -import { BILLING_DASHBOARD_URL } from '../utils/links'; -import { openUrl } from '../utils/openUrl'; +} from '../../store/threadSlice'; +import type { ConfirmationModal as ConfirmationModalType } from '../../types/intelligence'; +import type { ThreadMessage } from '../../types/thread'; +import { splitAgentMessageIntoBubbles } from '../../utils/agentMessageBubbles'; +import { chatThreadPath } from '../../utils/chatRoutes'; +import { CHAT_ATTACHMENTS_ENABLED } from '../../utils/config'; +import { BILLING_DASHBOARD_URL } from '../../utils/links'; +import { openUrl } from '../../utils/openUrl'; import { isTauri, notifyOverlaySttState, @@ -99,42 +140,10 @@ import { openhumanVoiceStatus, openhumanVoiceTranscribeBytes, openhumanVoiceTts, -} from '../utils/tauriCommands'; -import { formatTimelineEntry } from '../utils/toolTimelineFormatting'; -import { - AgentMessageBubble, - AgentMessageText, - BubbleMarkdown, -} from './conversations/components/AgentMessageBubble'; -import { AgentProcessSourcePanel } from './conversations/components/AgentProcessSourcePanel'; -import { - BackgroundProcessesPanel, - selectBackgroundProcesses, -} from './conversations/components/BackgroundProcessesPanel'; -import { CitationChips, type MessageCitation } from './conversations/components/CitationChips'; -import { PlanReviewCard } from './conversations/components/PlanReviewCard'; -import { SubagentDrawer } from './conversations/components/SubagentDrawer'; -import { - ThreadGoalEditorPanel, - ThreadGoalFooterTrigger, - useThreadGoal, -} from './conversations/components/ThreadGoalChip'; -import { ThreadTodoStrip } from './conversations/components/ThreadTodoStrip'; -import { ToolTimelineBlock } from './conversations/components/ToolTimelineBlock'; -import { - evaluateComposerSend, - getComposerBlockedSendFeedback, - handleComposerSlashCommand, -} from './conversations/composerSendDecision'; -import { useMemorySyncActive } from './conversations/hooks/useBackgroundActivity'; -import { - type AgentBubblePosition, - buildAcceptedInlineCompletion, - formatRelativeTime, - formatResetTime, - getInlineCompletionSuffix, -} from './conversations/utils/format'; -import { GENERAL_TAB_VALUE, isThreadVisibleInTab } from './conversations/utils/threadFilter'; +} from '../../utils/tauriCommands'; +import { formatTimelineEntry } from '../../utils/toolTimelineFormatting'; +import { ThreadList } from './threadList/ThreadList'; +import { buildThreadTimeline } from './timeline/selectors'; const CHAT_MODEL_HINT = 'hint:chat'; /** Maximum trailing characters rendered in the live-streaming assistant @@ -223,6 +232,10 @@ const EMPTY_ACTIVE_THREADS: Record = {}; // the same identity when the slice field is absent (narrow test stores). const EMPTY_QUEUED_FOLLOWUPS: Record = {}; +// Stable empty reference for the per-thread past-turn timelines map, so the +// derived value keeps the same identity when the slice field is absent. +const EMPTY_TURN_TIMELINES: Record = {}; + export function isComposerInteractionBlocked(args: { /** Whether the *currently selected* thread has an in-flight inference turn. */ selectedThreadActive: boolean; @@ -380,6 +393,7 @@ const Conversations = ({ // behaviour stays intact. const uiLocale = useAppSelector(state => state.locale?.current ?? 'en'); const toolTimelineByThread = useAppSelector(state => state.chatRuntime.toolTimelineByThread); + const turnTimelinesByThread = useAppSelector(state => state.chatRuntime.turnTimelinesByThread); const processingByThread = useAppSelector(state => state.chatRuntime.processingByThread); const taskBoardByThread = useAppSelector(state => state.chatRuntime.taskBoardByThread); const inferenceStatusByThread = useAppSelector( @@ -723,6 +737,8 @@ const Conversations = ({ if (selectedThreadId) { void dispatch(loadThreadMessages(selectedThreadId)); void dispatch(fetchAndHydrateTurnState(selectedThreadId)); + // Per-turn history: each past answer's own process trail (Phase 5). + void dispatch(fetchAndHydrateTurnHistory(selectedThreadId)); void threadApi .getTaskBoard(selectedThreadId) .then(board => { @@ -1667,6 +1683,50 @@ const Conversations = ({ const latestVisibleAgentMessage = [...visibleMessages] .reverse() .find(msg => msg.sender === 'agent'); + // Message list sourced from the unified timeline projection — the single + // source of render order (see `docs/plans/conversations-timeline-refactor.md` + // Phase 2). With the streaming/tool inputs omitted the projection yields + // exactly the visible messages in order; the tool-timeline block and streaming + // previews stay anchored inline below, so the rendered DOM is unchanged. This + // routes the live render loop through the projection ahead of the per-turn + // grouping in Phase 5. + const timelineMessages = useMemo( + () => + buildThreadTimeline({ + threadId: selectedThreadId ?? '', + messages: visibleMessages, + toolTimeline: [], + streaming: null, + parallelStreams: [], + hideAgentInsights: false, + }) + .map(item => ('message' in item ? item.message : null)) + .filter((message): message is ThreadMessage => message !== null), + [selectedThreadId, visibleMessages] + ); + // Past-turn tool timelines (Phase 5): map the first assistant message of each + // older settled turn to that turn's timeline, so each past answer renders its + // own collapsed process trail above it. The latest turn is excluded upstream + // (it renders as the live "agent insights" anchor), so there is no double + // render. Empty for legacy messages without a `requestId`. + const selectedThreadTurnTimelines = selectedThreadId + ? (turnTimelinesByThread[selectedThreadId] ?? EMPTY_TURN_TIMELINES) + : EMPTY_TURN_TIMELINES; + const pastTurnAnchors = useMemo(() => { + const anchors: Record = {}; + const seen = new Set(); + for (const msg of timelineMessages) { + if (msg.sender !== 'agent') continue; + const requestId = msg.extraMetadata?.requestId; + if (typeof requestId !== 'string' || seen.has(requestId)) continue; + const entries = selectedThreadTurnTimelines[requestId]; + if (entries && entries.length > 0) { + anchors[msg.id] = entries; + seen.add(requestId); + } + } + return anchors; + }, [timelineMessages, selectedThreadTurnTimelines]); const activeSubagentTimelineEntry = selectedThreadToolTimeline.find( entry => entry.status === 'running' && entry.name.startsWith('subagent:') ); @@ -1922,223 +1982,58 @@ const Conversations = ({ // Thread list (left pane). Rendered through `TwoPanelLayout` below in page // mode; the embedded `variant="sidebar"` mode shows no thread list at all. const threadSidebar = ( - // Card background / rounded corners come from TwoPanelLayout's pane styling. -
- {/* Thread search — flush full-width input, mirrors the settings search. */} -
- - - - - - setThreadSearch(e.target.value)} - onKeyDown={e => { - if (e.key === 'Escape' && threadSearch) { - e.preventDefault(); - setThreadSearch(''); + void handleCreateNewThread()} + onSelectThread={id => { + dispatch(setSelectedThread(id)); + void dispatch(loadThreadMessages(id)); + if (shouldSyncChatRoute) { + navigate(chatThreadPath(id)); + } + }} + resolveTitle={resolveThreadDisplayTitle} + onRequestDelete={thread => + setDeleteModal({ + isOpen: true, + title: t('chat.deleteThread'), + message: t('chat.deleteThreadConfirm').replace( + '{title}', + thread.title || t('chat.untitledThread') + ), + confirmText: t('common.delete'), + cancelText: t('common.cancel'), + destructive: true, + onConfirm: () => { + if (shouldSyncChatRoute && routeThreadId === thread.id) { + navigate('/chat', { replace: true }); } - }} - placeholder={t('chat.searchThreads')} - aria-label={t('chat.searchThreads')} - data-testid="chat-thread-search-input" - className="w-full border-0 bg-transparent py-2.5 pl-10 pr-10 text-sm text-content placeholder:text-stone-400 focus:outline-none focus:ring-0 dark:placeholder:text-neutral-500" - /> - {threadSearch && ( - - )} -
- {/* New conversation — a subtle, centered thread-style row (not a loud - button), below the search and above the thread list. */} - -
- {visibleThreads.length === 0 ? ( -

{t('chat.noThreads')}

- ) : ( - visibleThreads.map(thread => ( -
{ - dispatch(setSelectedThread(thread.id)); - void dispatch(loadThreadMessages(thread.id)); - if (shouldSyncChatRoute) { - navigate(chatThreadPath(thread.id)); - } - }} - onKeyDown={e => { - if (e.target !== e.currentTarget) return; - if (e.key === 'Enter' || e.key === ' ') { - e.preventDefault(); - dispatch(setSelectedThread(thread.id)); - void dispatch(loadThreadMessages(thread.id)); - if (shouldSyncChatRoute) { - navigate(chatThreadPath(thread.id)); - } - } - }} - className={`w-full text-left px-3 py-1.5 border-b border-line-subtle/60 dark:border-line/60 transition-colors group cursor-pointer ${ - selectedThreadId === thread.id - ? 'bg-primary-50 dark:bg-primary-900/30 border-l-2 border-l-primary-500' - : 'hover:bg-surface-hover' - }`}> -
- {editingThreadId === thread.id ? ( - e.stopPropagation()} - onChange={e => setEditTitleValue(e.target.value)} - onKeyDown={e => { - e.stopPropagation(); - // Ignore the Enter that confirms an IME composition - // candidate (CJK input) so it doesn't prematurely commit. - if (isImeCompositionKeyEvent(e)) return; - if (e.key === 'Enter') { - e.preventDefault(); - handleCommitTitle(thread.id); - } else if (e.key === 'Escape') { - // Escape is an explicit cancel — suppress the commit the - // ensuing blur would otherwise fire. - ignoreNextTitleBlurRef.current = true; - setEditingThreadId(null); - } - }} - onBlur={() => { - if (ignoreNextTitleBlurRef.current) { - ignoreNextTitleBlurRef.current = false; - return; - } - handleCommitTitle(thread.id); - }} - aria-label={t('chat.editThreadTitle')} - data-testid={`thread-title-input-${thread.id}`} - className="h-5 min-w-0 flex-1 border-b border-primary-400 bg-transparent py-0 text-xs font-medium leading-none text-content-secondary outline-none" - autoFocus - /> - ) : ( -

- {resolveThreadDisplayTitle(thread.id)} -

- )} - - -
- {/*
- - {formatRelativeTime(thread.lastMessageAt)} - - {thread.messageCount > 0 && ( - - {thread.messageCount} msg{thread.messageCount !== 1 ? 's' : ''} - - )} -
*/} -
- )) - )} -
-
+ void dispatch(deleteThread(thread.id)); + }, + onCancel: () => {}, + }) + } + editingThreadId={editingThreadId} + editTitleValue={editTitleValue} + editTitleInputRef={editTitleInputRef} + onEditTitleValueChange={setEditTitleValue} + onStartEditTitle={handleStartEditTitle} + onCommitTitle={handleCommitTitle} + onCancelEditTitle={() => { + ignoreNextTitleBlurRef.current = true; + setEditingThreadId(null); + }} + onBlurTitle={id => { + if (ignoreNextTitleBlurRef.current) { + ignoreNextTitleBlurRef.current = false; + return; + } + handleCommitTitle(id); + }} + /> ); // Main chat area (right pane): header, message list, composer. @@ -2208,7 +2103,7 @@ const Conversations = ({ // queued-followups panel and other dynamic footer content never // overlap the last message (#4268). style={!isSidebar ? { paddingBottom: composerFooterHeight + 16 } : undefined}> - {visibleMessages.map(msg => { + {timelineMessages.map(msg => { const isAgentTextMode = msg.sender === 'agent' && agentMessageViewMode === 'text'; // Parsed once per message: for current messages (extraMetadata // present, or agent messages) msg.content already has no markers, @@ -2217,8 +2112,16 @@ const Conversations = ({ // what keeps the marker text out of both the rendered bubble and // the copy-to-clipboard action. const parsedContent = parseMessageImages(msg.content ?? ''); + const pastTurnEntries = pastTurnAnchors[msg.id]; return ( + {/* Past-turn process trail (Phase 5): each older settled turn's + tool timeline, collapsed, above the answer it produced. */} + {pastTurnEntries ? ( +
+ +
+ ) : null}
@@ -3299,4 +3202,4 @@ export default Conversations; * Embeddable variant — same component, page layout (floating centered * card). Mounted inside /accounts when the Agent entry is selected. */ -export const AgentChatPanel = () => ; +export const ConversationsPage = () => ; diff --git a/app/src/pages/conversations/components/AgentMessageBubble.test.tsx b/app/src/features/conversations/components/AgentMessageBubble.test.tsx similarity index 100% rename from app/src/pages/conversations/components/AgentMessageBubble.test.tsx rename to app/src/features/conversations/components/AgentMessageBubble.test.tsx diff --git a/app/src/pages/conversations/components/AgentMessageBubble.tsx b/app/src/features/conversations/components/AgentMessageBubble.tsx similarity index 100% rename from app/src/pages/conversations/components/AgentMessageBubble.tsx rename to app/src/features/conversations/components/AgentMessageBubble.tsx diff --git a/app/src/pages/conversations/components/AgentProcessSourcePanel.tsx b/app/src/features/conversations/components/AgentProcessSourcePanel.tsx similarity index 100% rename from app/src/pages/conversations/components/AgentProcessSourcePanel.tsx rename to app/src/features/conversations/components/AgentProcessSourcePanel.tsx diff --git a/app/src/pages/conversations/components/AgentTimelineRail.tsx b/app/src/features/conversations/components/AgentTimelineRail.tsx similarity index 100% rename from app/src/pages/conversations/components/AgentTimelineRail.tsx rename to app/src/features/conversations/components/AgentTimelineRail.tsx diff --git a/app/src/pages/conversations/components/BackgroundActivityRows.tsx b/app/src/features/conversations/components/BackgroundActivityRows.tsx similarity index 100% rename from app/src/pages/conversations/components/BackgroundActivityRows.tsx rename to app/src/features/conversations/components/BackgroundActivityRows.tsx diff --git a/app/src/pages/conversations/components/BackgroundProcessesPanel.tsx b/app/src/features/conversations/components/BackgroundProcessesPanel.tsx similarity index 100% rename from app/src/pages/conversations/components/BackgroundProcessesPanel.tsx rename to app/src/features/conversations/components/BackgroundProcessesPanel.tsx diff --git a/app/src/pages/conversations/components/CitationChips.tsx b/app/src/features/conversations/components/CitationChips.tsx similarity index 100% rename from app/src/pages/conversations/components/CitationChips.tsx rename to app/src/features/conversations/components/CitationChips.tsx diff --git a/app/src/pages/conversations/components/LimitPill.tsx b/app/src/features/conversations/components/LimitPill.tsx similarity index 100% rename from app/src/pages/conversations/components/LimitPill.tsx rename to app/src/features/conversations/components/LimitPill.tsx diff --git a/app/src/pages/conversations/components/PlanReviewCard.test.tsx b/app/src/features/conversations/components/PlanReviewCard.test.tsx similarity index 100% rename from app/src/pages/conversations/components/PlanReviewCard.test.tsx rename to app/src/features/conversations/components/PlanReviewCard.test.tsx diff --git a/app/src/pages/conversations/components/PlanReviewCard.tsx b/app/src/features/conversations/components/PlanReviewCard.tsx similarity index 100% rename from app/src/pages/conversations/components/PlanReviewCard.tsx rename to app/src/features/conversations/components/PlanReviewCard.tsx diff --git a/app/src/pages/conversations/components/ProcessingTranscriptView.test.tsx b/app/src/features/conversations/components/ProcessingTranscriptView.test.tsx similarity index 100% rename from app/src/pages/conversations/components/ProcessingTranscriptView.test.tsx rename to app/src/features/conversations/components/ProcessingTranscriptView.test.tsx diff --git a/app/src/pages/conversations/components/ProcessingTranscriptView.tsx b/app/src/features/conversations/components/ProcessingTranscriptView.tsx similarity index 100% rename from app/src/pages/conversations/components/ProcessingTranscriptView.tsx rename to app/src/features/conversations/components/ProcessingTranscriptView.tsx diff --git a/app/src/pages/conversations/components/SubagentDrawer.tsx b/app/src/features/conversations/components/SubagentDrawer.tsx similarity index 100% rename from app/src/pages/conversations/components/SubagentDrawer.tsx rename to app/src/features/conversations/components/SubagentDrawer.tsx diff --git a/app/src/pages/conversations/components/TaskKanbanBoard.test.tsx b/app/src/features/conversations/components/TaskKanbanBoard.test.tsx similarity index 100% rename from app/src/pages/conversations/components/TaskKanbanBoard.test.tsx rename to app/src/features/conversations/components/TaskKanbanBoard.test.tsx diff --git a/app/src/pages/conversations/components/TaskKanbanBoard.tsx b/app/src/features/conversations/components/TaskKanbanBoard.tsx similarity index 100% rename from app/src/pages/conversations/components/TaskKanbanBoard.tsx rename to app/src/features/conversations/components/TaskKanbanBoard.tsx diff --git a/app/src/pages/conversations/components/ThreadGoalChip.test.tsx b/app/src/features/conversations/components/ThreadGoalChip.test.tsx similarity index 100% rename from app/src/pages/conversations/components/ThreadGoalChip.test.tsx rename to app/src/features/conversations/components/ThreadGoalChip.test.tsx diff --git a/app/src/pages/conversations/components/ThreadGoalChip.tsx b/app/src/features/conversations/components/ThreadGoalChip.tsx similarity index 100% rename from app/src/pages/conversations/components/ThreadGoalChip.tsx rename to app/src/features/conversations/components/ThreadGoalChip.tsx diff --git a/app/src/pages/conversations/components/ThreadTodoStrip.test.tsx b/app/src/features/conversations/components/ThreadTodoStrip.test.tsx similarity index 100% rename from app/src/pages/conversations/components/ThreadTodoStrip.test.tsx rename to app/src/features/conversations/components/ThreadTodoStrip.test.tsx diff --git a/app/src/pages/conversations/components/ThreadTodoStrip.tsx b/app/src/features/conversations/components/ThreadTodoStrip.tsx similarity index 100% rename from app/src/pages/conversations/components/ThreadTodoStrip.tsx rename to app/src/features/conversations/components/ThreadTodoStrip.tsx diff --git a/app/src/pages/conversations/components/ToolFailureLines.tsx b/app/src/features/conversations/components/ToolFailureLines.tsx similarity index 100% rename from app/src/pages/conversations/components/ToolFailureLines.tsx rename to app/src/features/conversations/components/ToolFailureLines.tsx diff --git a/app/src/pages/conversations/components/ToolTimelineBlock.tsx b/app/src/features/conversations/components/ToolTimelineBlock.tsx similarity index 100% rename from app/src/pages/conversations/components/ToolTimelineBlock.tsx rename to app/src/features/conversations/components/ToolTimelineBlock.tsx diff --git a/app/src/pages/conversations/components/WorkerThreadRefCard.tsx b/app/src/features/conversations/components/WorkerThreadRefCard.tsx similarity index 100% rename from app/src/pages/conversations/components/WorkerThreadRefCard.tsx rename to app/src/features/conversations/components/WorkerThreadRefCard.tsx diff --git a/app/src/pages/conversations/components/__tests__/AgentProcessSourcePanel.test.tsx b/app/src/features/conversations/components/__tests__/AgentProcessSourcePanel.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/AgentProcessSourcePanel.test.tsx rename to app/src/features/conversations/components/__tests__/AgentProcessSourcePanel.test.tsx diff --git a/app/src/pages/conversations/components/__tests__/AgentTimelineRail.test.tsx b/app/src/features/conversations/components/__tests__/AgentTimelineRail.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/AgentTimelineRail.test.tsx rename to app/src/features/conversations/components/__tests__/AgentTimelineRail.test.tsx diff --git a/app/src/pages/conversations/components/__tests__/BackgroundActivityRows.test.tsx b/app/src/features/conversations/components/__tests__/BackgroundActivityRows.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/BackgroundActivityRows.test.tsx rename to app/src/features/conversations/components/__tests__/BackgroundActivityRows.test.tsx diff --git a/app/src/pages/conversations/components/__tests__/BackgroundProcessesPanel.test.tsx b/app/src/features/conversations/components/__tests__/BackgroundProcessesPanel.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/BackgroundProcessesPanel.test.tsx rename to app/src/features/conversations/components/__tests__/BackgroundProcessesPanel.test.tsx diff --git a/app/src/pages/conversations/components/__tests__/SubagentDrawer.test.tsx b/app/src/features/conversations/components/__tests__/SubagentDrawer.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/SubagentDrawer.test.tsx rename to app/src/features/conversations/components/__tests__/SubagentDrawer.test.tsx diff --git a/app/src/pages/conversations/components/__tests__/TaskKanbanBoard.test.tsx b/app/src/features/conversations/components/__tests__/TaskKanbanBoard.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/TaskKanbanBoard.test.tsx rename to app/src/features/conversations/components/__tests__/TaskKanbanBoard.test.tsx diff --git a/app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx b/app/src/features/conversations/components/__tests__/ToolTimelineBlock.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx rename to app/src/features/conversations/components/__tests__/ToolTimelineBlock.test.tsx diff --git a/app/src/pages/conversations/components/__tests__/WorkerThreadRefCard.test.tsx b/app/src/features/conversations/components/__tests__/WorkerThreadRefCard.test.tsx similarity index 100% rename from app/src/pages/conversations/components/__tests__/WorkerThreadRefCard.test.tsx rename to app/src/features/conversations/components/__tests__/WorkerThreadRefCard.test.tsx diff --git a/app/src/pages/conversations/composerSendDecision.test.ts b/app/src/features/conversations/composerSendDecision.test.ts similarity index 100% rename from app/src/pages/conversations/composerSendDecision.test.ts rename to app/src/features/conversations/composerSendDecision.test.ts diff --git a/app/src/pages/conversations/composerSendDecision.ts b/app/src/features/conversations/composerSendDecision.ts similarity index 100% rename from app/src/pages/conversations/composerSendDecision.ts rename to app/src/features/conversations/composerSendDecision.ts diff --git a/app/src/pages/conversations/hooks/useBackgroundActivity.test.ts b/app/src/features/conversations/hooks/useBackgroundActivity.test.ts similarity index 100% rename from app/src/pages/conversations/hooks/useBackgroundActivity.test.ts rename to app/src/features/conversations/hooks/useBackgroundActivity.test.ts diff --git a/app/src/pages/conversations/hooks/useBackgroundActivity.ts b/app/src/features/conversations/hooks/useBackgroundActivity.ts similarity index 100% rename from app/src/pages/conversations/hooks/useBackgroundActivity.ts rename to app/src/features/conversations/hooks/useBackgroundActivity.ts diff --git a/app/src/features/conversations/index.ts b/app/src/features/conversations/index.ts new file mode 100644 index 000000000..cdb3d5560 --- /dev/null +++ b/app/src/features/conversations/index.ts @@ -0,0 +1,15 @@ +/** + * Public surface for the conversations feature — the reusability contract other + * parts of the app import against (see `docs/plans/conversations-timeline-refactor.md`). + * + * Today this is the monolithic `Conversations` panel plus its page-variant + * wrapper and pure helpers. Later phases add the split shells + * (`ConversationsSidebar`) and the `ConversationTimeline` renderer here. + */ +export { + default as Conversations, + ConversationsPage, + isComposerInteractionBlocked, + isImeCompositionKeyEvent, + formatThreadLoadError, +} from './Conversations'; diff --git a/app/src/pages/conversations/taskPlanActions.test.ts b/app/src/features/conversations/taskPlanActions.test.ts similarity index 100% rename from app/src/pages/conversations/taskPlanActions.test.ts rename to app/src/features/conversations/taskPlanActions.test.ts diff --git a/app/src/pages/conversations/taskPlanActions.ts b/app/src/features/conversations/taskPlanActions.ts similarity index 100% rename from app/src/pages/conversations/taskPlanActions.ts rename to app/src/features/conversations/taskPlanActions.ts diff --git a/app/src/features/conversations/threadList/ThreadList.tsx b/app/src/features/conversations/threadList/ThreadList.tsx new file mode 100644 index 000000000..ee6dc2d34 --- /dev/null +++ b/app/src/features/conversations/threadList/ThreadList.tsx @@ -0,0 +1,232 @@ +import type { RefObject } from 'react'; + +import { useT } from '../../../lib/i18n/I18nContext'; +import type { Thread } from '../../../types/thread'; +import { isImeCompositionKeyEvent } from '../Conversations'; + +export interface ThreadListProps { + /** Threads visible after the sidebar's search/tab filtering. */ + threads: Thread[]; + selectedThreadId: string | null; + /** Free-text thread-title search. */ + search: string; + onSearchChange: (value: string) => void; + onCreateThread: () => void; + /** Select a thread (owns dispatch + message load + route sync). */ + onSelectThread: (threadId: string) => void; + /** Stable, human-readable title for a thread id. */ + resolveTitle: (threadId: string) => string; + onRequestDelete: (thread: Thread) => void; + // Inline title rename — controlled by the parent so the edit state stays + // co-located with the rest of the panel's thread state. + editingThreadId: string | null; + editTitleValue: string; + editTitleInputRef: RefObject; + onEditTitleValueChange: (value: string) => void; + onStartEditTitle: (threadId: string) => void; + onCommitTitle: (threadId: string) => void; + onCancelEditTitle: () => void; + onBlurTitle: (threadId: string) => void; +} + +/** + * The conversations left rail: thread-title search, a "new conversation" row, + * and the scrollable thread list with inline rename + delete affordances. + * Extracted verbatim from the panel (Phase 1 shell split) — presentational, + * driven entirely by props so it can be reused by the page and sidebar shells. + */ +export function ThreadList({ + threads, + selectedThreadId, + search, + onSearchChange, + onCreateThread, + onSelectThread, + resolveTitle, + onRequestDelete, + editingThreadId, + editTitleValue, + editTitleInputRef, + onEditTitleValueChange, + onStartEditTitle, + onCommitTitle, + onCancelEditTitle, + onBlurTitle, +}: ThreadListProps) { + const { t } = useT(); + return ( + // Card background / rounded corners come from TwoPanelLayout's pane styling. +
+ {/* Thread search — flush full-width input, mirrors the settings search. */} +
+ + + + + + onSearchChange(e.target.value)} + onKeyDown={e => { + if (e.key === 'Escape' && search) { + e.preventDefault(); + onSearchChange(''); + } + }} + placeholder={t('chat.searchThreads')} + aria-label={t('chat.searchThreads')} + data-testid="chat-thread-search-input" + className="w-full border-0 bg-transparent py-2.5 pl-10 pr-10 text-sm text-content placeholder:text-stone-400 focus:outline-none focus:ring-0 dark:placeholder:text-neutral-500" + /> + {search && ( + + )} +
+ {/* New conversation — a subtle, centered thread-style row (not a loud + button), below the search and above the thread list. */} + +
+ {threads.length === 0 ? ( +

{t('chat.noThreads')}

+ ) : ( + threads.map(thread => ( +
onSelectThread(thread.id)} + onKeyDown={e => { + if (e.target !== e.currentTarget) return; + if (e.key === 'Enter' || e.key === ' ') { + e.preventDefault(); + onSelectThread(thread.id); + } + }} + className={`w-full text-left px-3 py-1.5 border-b border-line-subtle/60 dark:border-line/60 transition-colors group cursor-pointer ${ + selectedThreadId === thread.id + ? 'bg-primary-50 dark:bg-primary-900/30 border-l-2 border-l-primary-500' + : 'hover:bg-surface-hover' + }`}> +
+ {editingThreadId === thread.id ? ( + e.stopPropagation()} + onChange={e => onEditTitleValueChange(e.target.value)} + onKeyDown={e => { + e.stopPropagation(); + // Ignore the Enter that confirms an IME composition + // candidate (CJK input) so it doesn't prematurely commit. + if (isImeCompositionKeyEvent(e)) return; + if (e.key === 'Enter') { + e.preventDefault(); + onCommitTitle(thread.id); + } else if (e.key === 'Escape') { + // Escape is an explicit cancel — suppress the commit the + // ensuing blur would otherwise fire. + onCancelEditTitle(); + } + }} + onBlur={() => onBlurTitle(thread.id)} + aria-label={t('chat.editThreadTitle')} + data-testid={`thread-title-input-${thread.id}`} + className="h-5 min-w-0 flex-1 border-b border-primary-400 bg-transparent py-0 text-xs font-medium leading-none text-content-secondary outline-none" + autoFocus + /> + ) : ( +

+ {resolveTitle(thread.id)} +

+ )} + + +
+
+ )) + )} +
+
+ ); +} diff --git a/app/src/features/conversations/timeline/ConversationTimeline.test.tsx b/app/src/features/conversations/timeline/ConversationTimeline.test.tsx new file mode 100644 index 000000000..96fb5a504 --- /dev/null +++ b/app/src/features/conversations/timeline/ConversationTimeline.test.tsx @@ -0,0 +1,125 @@ +import { render, screen } from '@testing-library/react'; +import { Provider } from 'react-redux'; +import { describe, expect, it, vi } from 'vitest'; + +import { store } from '../../../store'; +import type { StreamingAssistantState, ToolTimelineEntry } from '../../../store/chatRuntimeSlice'; +import type { ThreadMessage } from '../../../types/thread'; +import { ConversationTimeline } from './ConversationTimeline'; +import { buildThreadTimeline } from './selectors'; + +function renderInStore(ui: React.ReactNode) { + return render({ui}); +} + +function userMsg(id: string, content: string): ThreadMessage { + return { + id, + content, + type: 'text', + extraMetadata: {}, + sender: 'user', + createdAt: '2026-01-01T00:00:00Z', + }; +} +function agentMsg(id: string, content: string): ThreadMessage { + return { + id, + content, + type: 'text', + extraMetadata: {}, + sender: 'agent', + createdAt: '2026-01-01T00:00:01Z', + }; +} +function tool(id: string, name: string, round = 0): ToolTimelineEntry { + return { id, name, round, status: 'success' }; +} +function subagentRow(id: string, taskId: string): ToolTimelineEntry { + return { + id, + name: 'subagent:researcher', + round: 0, + status: 'running', + subagent: { taskId, agentId: 'researcher', toolCalls: [] }, + }; +} +function stream(requestId: string, content: string, thinking = ''): StreamingAssistantState { + return { requestId, content, thinking }; +} + +function timeline(over: Partial[0]>) { + return buildThreadTimeline({ + threadId: 't1', + messages: [], + toolTimeline: [], + streaming: null, + parallelStreams: [], + hideAgentInsights: false, + ...over, + }); +} + +describe('ConversationTimeline', () => { + it('renders user + assistant messages in order', () => { + const items = timeline({ messages: [userMsg('u1', 'hello there'), agentMsg('a1', 'hi back')] }); + renderInStore(); + expect(screen.getByTestId('conversation-timeline')).toBeTruthy(); + expect(screen.getByText('hello there')).toBeTruthy(); + expect(screen.getByText('hi back')).toBeTruthy(); + }); + + it('coalesces a run of process items into a single ToolTimelineBlock', () => { + const items = timeline({ + messages: [userMsg('u1', 'go')], + toolTimeline: [tool('c1', 'read_file', 0), tool('c2', 'write_file', 1)], + }); + const { container } = renderInStore(); + // The two consecutive tool items coalesce into one ToolTimelineBlock (one + // `
` group), not two separate blocks. + expect(container.querySelectorAll('details')).toHaveLength(1); + }); + + it('invokes onOpenSubagent when a subagent row requests the drawer', () => { + const onOpenSubagent = vi.fn(); + const items = timeline({ + messages: [userMsg('u1', 'go')], + toolTimeline: [subagentRow('row1', 'sub-1')], + }); + renderInStore(); + // The subagent row renders within the block; the handler is wired through. + expect(screen.getByTestId('conversation-timeline')).toBeTruthy(); + }); + + it('renders a streaming tail with the 120-char slice and thinking details', () => { + const long = 'x'.repeat(200); + const items = timeline({ + messages: [userMsg('u1', 'go')], + streaming: stream('req1', long, 'deliberating'), + }); + renderInStore(); + const primary = screen.getByTestId('stream-primary'); + // Truncation ellipsis is present and the thinking summary renders. + expect(primary.textContent).toContain('…'); + expect(primary.textContent).toContain('deliberating'); + // Only the last 120 chars of the 200-char body render in the content bubble. + const contentBubble = primary.querySelector('.font-mono.text-sm'); + expect(contentBubble?.textContent).toBe(`…${'x'.repeat(120)}`); + }); + + it('renders forked branch streams as branch tails', () => { + const items = timeline({ + messages: [userMsg('u1', 'go')], + parallelStreams: [stream('b1', 'branch answer')], + }); + renderInStore(); + expect(screen.getByTestId('stream-branch')).toBeTruthy(); + expect(screen.getByText('branch answer')).toBeTruthy(); + }); + + it('renders nothing but the container for an empty timeline', () => { + renderInStore(); + const container = screen.getByTestId('conversation-timeline'); + expect(container.children).toHaveLength(0); + }); +}); diff --git a/app/src/features/conversations/timeline/ConversationTimeline.tsx b/app/src/features/conversations/timeline/ConversationTimeline.tsx new file mode 100644 index 000000000..b83b3b3c1 --- /dev/null +++ b/app/src/features/conversations/timeline/ConversationTimeline.tsx @@ -0,0 +1,102 @@ +import type { SubagentActivity, ToolTimelineEntry } from '../../../store/chatRuntimeSlice'; +import { ToolTimelineBlock } from '../components/ToolTimelineBlock'; +import { AssistantMessageItem } from './items/AssistantMessageItem'; +import { StreamingTailItem } from './items/StreamingTailItem'; +import { UserMessageItem } from './items/UserMessageItem'; +import type { TimelineItem } from './types'; + +export interface ConversationTimelineHandlers { + /** Opens the full-transcript drawer for a subagent row. */ + onOpenSubagent?: (subagent: SubagentActivity) => void; + /** Opens the whole-run "Agent Process Source" panel. */ + onViewWholeRun?: () => void; +} + +/** True for the process kinds that coalesce into one `ToolTimelineBlock`. */ +function isProcessItem(item: TimelineItem): item is TimelineItem & { entry: ToolTimelineEntry } { + return item.kind === 'toolCall' || item.kind === 'subagentActivity'; +} + +/** + * Pure renderer: `TimelineItem[]` → one element per kind, wrapping the existing + * components (see `docs/plans/conversations-timeline-refactor.md` Phase 2). + * + * Consecutive process items (`toolCall`/`subagentActivity`) are coalesced into a + * single `ToolTimelineBlock` so the timeline reads as one grouped block, + * matching today's `agentInsights` anchor. Ordering, anchoring, and + * `hideAgentInsights` filtering live upstream in the selector — this component + * only maps items to UI. + */ +export function ConversationTimeline({ + items, + agentMessageViewMode = 'bubbles', + handlers = {}, +}: { + items: TimelineItem[]; + agentMessageViewMode?: 'text' | 'bubbles'; + handlers?: ConversationTimelineHandlers; +}) { + const rendered: React.ReactNode[] = []; + + for (let i = 0; i < items.length; i += 1) { + const item = items[i]; + + // Coalesce a contiguous run of process items into one ToolTimelineBlock. + if (isProcessItem(item)) { + const run: ToolTimelineEntry[] = []; + const startId = item.id; + let j = i; + while (j < items.length && isProcessItem(items[j])) { + run.push((items[j] as TimelineItem & { entry: ToolTimelineEntry }).entry); + j += 1; + } + rendered.push( + + ); + i = j - 1; + continue; + } + + switch (item.kind) { + case 'userMessage': + rendered.push(); + break; + case 'assistantMessage': + rendered.push( + + ); + break; + case 'streamingText': + rendered.push( + + ); + break; + case 'reasoning': + // Reasoning is rendered inside the process/streaming affordances today; + // no standalone element yet. + break; + default: + break; + } + } + + return ( +
+ {rendered} +
+ ); +} diff --git a/app/src/features/conversations/timeline/items/AssistantMessageItem.tsx b/app/src/features/conversations/timeline/items/AssistantMessageItem.tsx new file mode 100644 index 000000000..84ed15f1d --- /dev/null +++ b/app/src/features/conversations/timeline/items/AssistantMessageItem.tsx @@ -0,0 +1,42 @@ +import { Fragment } from 'react'; + +import { splitAgentMessageIntoBubbles } from '../../../../utils/agentMessageBubbles'; +import { AgentMessageBubble, AgentMessageText } from '../../components/AgentMessageBubble'; +import type { AgentBubblePosition } from '../../utils/format'; + +/** + * Renders one assistant message. In `text` view mode the whole message is one + * markdown block; otherwise it is split into positioned bubbles via + * `splitAgentMessageIntoBubbles` (matching the current render loop). + */ +export function AssistantMessageItem({ + content, + viewMode = 'bubbles', +}: { + content: string; + viewMode?: 'text' | 'bubbles'; +}) { + if (viewMode === 'text') { + return ; + } + const parts = splitAgentMessageIntoBubbles(content); + return ( + <> + {parts.map((segment, index) => { + const position: AgentBubblePosition = + parts.length === 1 + ? 'single' + : index === 0 + ? 'first' + : index === parts.length - 1 + ? 'last' + : 'middle'; + return ( + + + + ); + })} + + ); +} diff --git a/app/src/features/conversations/timeline/items/StreamingTailItem.tsx b/app/src/features/conversations/timeline/items/StreamingTailItem.tsx new file mode 100644 index 000000000..59186e7f2 --- /dev/null +++ b/app/src/features/conversations/timeline/items/StreamingTailItem.tsx @@ -0,0 +1,37 @@ +/** + * Ephemeral streaming preview (primary stream or a forked branch). Mirrors the + * current 120-char tail-slice ticker: the full answer arrives as a durable + * message on completion, this only signals progress without jumping scroll. + */ +const STREAMING_PREVIEW_CHARS = 120; + +export function StreamingTailItem({ + text, + thinking, + branch = false, +}: { + text: string; + thinking?: string; + branch?: boolean; +}) { + const tail = text.slice(-STREAMING_PREVIEW_CHARS); + const truncated = text.length > STREAMING_PREVIEW_CHARS; + return ( +
+
+ {thinking && thinking.length > 0 ? ( +
+ Thinking… +

+ {thinking.slice(-STREAMING_PREVIEW_CHARS)} +

+
+ ) : null} +
+ {truncated ? '…' : ''} + {tail} +
+
+
+ ); +} diff --git a/app/src/features/conversations/timeline/items/UserMessageItem.tsx b/app/src/features/conversations/timeline/items/UserMessageItem.tsx new file mode 100644 index 000000000..0fd0a025d --- /dev/null +++ b/app/src/features/conversations/timeline/items/UserMessageItem.tsx @@ -0,0 +1,16 @@ +import { BubbleMarkdown } from '../../components/AgentMessageBubble'; + +/** + * Renders one user message bubble (text only). Attachment/reaction chrome is + * layered on by the panel at swap time; this keeps the pure renderer focused on + * the message text. + */ +export function UserMessageItem({ content }: { content: string }) { + return ( +
+
+ +
+
+ ); +} diff --git a/app/src/features/conversations/timeline/selectors.test.ts b/app/src/features/conversations/timeline/selectors.test.ts new file mode 100644 index 000000000..108cb2af0 --- /dev/null +++ b/app/src/features/conversations/timeline/selectors.test.ts @@ -0,0 +1,298 @@ +import { describe, expect, it } from 'vitest'; + +import type { StreamingAssistantState, ToolTimelineEntry } from '../../../store/chatRuntimeSlice'; +import type { RootState } from '../../../store/index'; +import type { ThreadMessage } from '../../../types/thread'; +import { buildThreadTimeline, groupTimelineIntoTurns, selectTimelineForThread } from './selectors'; +import { LEGACY_TURN_ID, type TimelineItem } from './types'; + +const THREAD = 't1'; + +function userMsg( + id: string, + content = `u-${id}`, + extra: Record = {} +): ThreadMessage { + return { + id, + content, + type: 'text', + extraMetadata: extra, + sender: 'user', + createdAt: '2026-01-01T00:00:00Z', + }; +} +function agentMsg( + id: string, + content = `a-${id}`, + extra: Record = {} +): ThreadMessage { + return { + id, + content, + type: 'text', + extraMetadata: extra, + sender: 'agent', + createdAt: '2026-01-01T00:00:01Z', + }; +} +function tool( + id: string, + name: string, + round = 0, + status: ToolTimelineEntry['status'] = 'success' +): ToolTimelineEntry { + return { id, name, round, status }; +} +function subagentRow(id: string, taskId: string, round = 0): ToolTimelineEntry { + return { + id, + name: 'subagent:researcher', + round, + status: 'running', + subagent: { taskId, agentId: 'researcher', toolCalls: [] }, + }; +} +function stream(requestId: string, content: string, thinking = ''): StreamingAssistantState { + return { requestId, content, thinking }; +} + +function build(partial: Partial[0]>): TimelineItem[] { + return buildThreadTimeline({ + threadId: THREAD, + messages: [], + toolTimeline: [], + streaming: null, + parallelStreams: [], + hideAgentInsights: false, + ...partial, + }); +} + +const kinds = (items: TimelineItem[]) => items.map(i => i.kind); +const ids = (items: TimelineItem[]) => items.map(i => i.id); + +describe('buildThreadTimeline — ordering', () => { + it('emits messages in stored order with monotonic seq, all in the legacy turn', () => { + const items = build({ messages: [userMsg('u1'), agentMsg('a1')] }); + expect(kinds(items)).toEqual(['userMessage', 'assistantMessage']); + expect(items.map(i => i.seq)).toEqual([0, 1]); + expect(items.every(i => i.turnId === LEGACY_TURN_ID)).toBe(true); + expect(items.every(i => i.threadId === THREAD)).toBe(true); + }); + + it('anchors the tool timeline immediately after the last user message', () => { + // Thread: u1, a1, u2, a2 — anchor is u2, so process rows sit between u2 and a2. + const items = build({ + messages: [userMsg('u1'), agentMsg('a1'), userMsg('u2'), agentMsg('a2')], + toolTimeline: [tool('call1', 'read_file', 0), tool('call2', 'write_file', 1)], + }); + expect(kinds(items)).toEqual([ + 'userMessage', // u1 + 'assistantMessage', // a1 + 'userMessage', // u2 (anchor) + 'toolCall', // call1 + 'toolCall', // call2 + 'assistantMessage', // a2 + ]); + expect(ids(items)).toEqual(['u1', 'a1', 'u2', 'call1', 'call2', 'a2']); + // seq is monotonic across the spliced process block. + expect(items.map(i => i.seq)).toEqual([0, 1, 2, 3, 4, 5]); + }); + + it('preserves tool-timeline array order and maps success→ok', () => { + const items = build({ + messages: [userMsg('u1')], + toolTimeline: [ + tool('c1', 'a', 0, 'success'), + tool('c2', 'b', 0, 'error'), + tool('c3', 'c', 1, 'running'), + ], + }); + const calls = items.filter(i => i.kind === 'toolCall'); + expect(calls.map(c => (c.kind === 'toolCall' ? c.status : null))).toEqual([ + 'ok', + 'error', + 'running', + ]); + expect(calls.map(c => c.id)).toEqual(['c1', 'c2', 'c3']); + }); + + it('classifies subagent:* rows as subagentActivity with the taskId', () => { + const items = build({ + messages: [userMsg('u1')], + toolTimeline: [subagentRow('row1', 'sub-42', 0)], + }); + const sub = items.find(i => i.kind === 'subagentActivity'); + expect(sub).toBeDefined(); + expect(sub && sub.kind === 'subagentActivity' && sub.taskId).toBe('sub-42'); + }); +}); + +describe('buildThreadTimeline — legacy / proactive fallbacks', () => { + it('empty thread yields no items', () => { + expect(build({})).toEqual([]); + }); + + it('proactive thread with no user message trails process items after the messages (L2669 fallback)', () => { + const items = build({ + messages: [agentMsg('a1'), agentMsg('a2')], + toolTimeline: [tool('c1', 'read_file', 0)], + }); + expect(kinds(items)).toEqual(['assistantMessage', 'assistantMessage', 'toolCall']); + expect(ids(items)).toEqual(['a1', 'a2', 'c1']); + }); + + it('drops hidden messages before anchoring', () => { + const items = build({ + messages: [userMsg('u1'), userMsg('u2', 'hidden', { hidden: true }), agentMsg('a1')], + toolTimeline: [tool('c1', 'x', 0)], + }); + // u2 is hidden, so the anchor is u1 — process rows sit between u1 and a1. + expect(ids(items)).toEqual(['u1', 'c1', 'a1']); + }); +}); + +describe('buildThreadTimeline — hideAgentInsights', () => { + it('omits tool/subagent process items when hidden but keeps messages', () => { + const items = build({ + messages: [userMsg('u1'), agentMsg('a1')], + toolTimeline: [tool('c1', 'x', 0), subagentRow('row1', 'sub-1', 0)], + hideAgentInsights: true, + }); + expect(kinds(items)).toEqual(['userMessage', 'assistantMessage']); + }); + + it('keeps the streaming preview (answer) even when insights are hidden', () => { + const items = build({ + messages: [userMsg('u1')], + toolTimeline: [tool('c1', 'x', 0)], + streaming: stream('req1', 'partial answer', 'private thoughts'), + hideAgentInsights: true, + }); + expect(kinds(items)).toEqual(['userMessage', 'streamingText']); + }); +}); + +describe('buildThreadTimeline — streaming previews', () => { + it('trails the primary streaming item after durable items, carrying thinking', () => { + const items = build({ + messages: [userMsg('u1'), agentMsg('a1')], + streaming: stream('req1', 'hello', 'thinking...'), + }); + const last = items[items.length - 1]; + expect(last.kind).toBe('streamingText'); + if (last.kind === 'streamingText') { + expect(last.text).toBe('hello'); + expect(last.thinking).toBe('thinking...'); + expect(last.branch).toBe(false); + expect(last.streamId).toBe('req1'); + } + }); + + it('skips an empty primary stream but renders forked branches (content only)', () => { + const items = build({ + messages: [userMsg('u1')], + streaming: stream('req1', '', ''), + parallelStreams: [stream('b1', 'branch one'), stream('b2', '', '')], + }); + const streams = items.filter(i => i.kind === 'streamingText'); + expect(streams).toHaveLength(1); + const branch = streams[0]; + expect(branch.kind === 'streamingText' && branch.branch).toBe(true); + expect(branch.kind === 'streamingText' && branch.text).toBe('branch one'); + // Branch thinking is not carried (rendered content-only today). + expect(branch.kind === 'streamingText' && branch.thinking).toBeUndefined(); + }); +}); + +describe('buildThreadTimeline — requestId grouping (Phase 4 anchoring)', () => { + it('derives turnId from extraMetadata.requestId when present', () => { + const items = build({ + messages: [ + userMsg('u1', 'q1', { requestId: 'req-1' }), + agentMsg('a1', 'ans1', { requestId: 'req-1' }), + userMsg('u2', 'q2', { requestId: 'req-2' }), + agentMsg('a2', 'ans2', { requestId: 'req-2' }), + ], + }); + expect(items.map(i => i.turnId)).toEqual(['req-1', 'req-1', 'req-2', 'req-2']); + }); + + it('falls back to the legacy turn for messages without a requestId', () => { + const items = build({ + messages: [userMsg('u1'), agentMsg('a1', 'ans', { requestId: 'req-9' })], + }); + expect(items.map(i => i.turnId)).toEqual([LEGACY_TURN_ID, 'req-9']); + }); + + it('groups a mixed thread into per-request turns', () => { + const items = build({ + messages: [ + userMsg('u1', 'q1', { requestId: 'req-1' }), + agentMsg('a1', 'ans1', { requestId: 'req-1' }), + userMsg('u2', 'q2', { requestId: 'req-2' }), + agentMsg('a2', 'ans2', { requestId: 'req-2' }), + ], + }); + const turns = groupTimelineIntoTurns(items); + expect(turns.map(t => t.turnId)).toEqual(['req-1', 'req-2']); + expect(turns.map(t => t.items.length)).toEqual([2, 2]); + }); +}); + +describe('groupTimelineIntoTurns', () => { + it('groups a single legacy turn into one group', () => { + const items = build({ + messages: [userMsg('u1'), agentMsg('a1')], + toolTimeline: [tool('c1', 'x')], + }); + const turns = groupTimelineIntoTurns(items); + expect(turns).toHaveLength(1); + expect(turns[0].turnId).toBe(LEGACY_TURN_ID); + expect(turns[0].items).toHaveLength(3); + }); + + it('empty input yields no turns', () => { + expect(groupTimelineIntoTurns([])).toEqual([]); + }); +}); + +describe('selectTimelineForThread (memoized)', () => { + function stateWith( + over: Partial = {}, + theme = { hideAgentInsights: false } + ): RootState { + return { + thread: { messagesByThreadId: { [THREAD]: [userMsg('u1'), agentMsg('a1')] } }, + chatRuntime: { + toolTimelineByThread: { [THREAD]: [tool('c1', 'read_file', 0)] }, + streamingAssistantByThread: {}, + parallelStreamsByThread: {}, + ...over, + }, + theme, + } as unknown as RootState; + } + + it('projects from state and anchors the tool row after the user message', () => { + const items = selectTimelineForThread(stateWith(), THREAD); + expect(ids(items)).toEqual(['u1', 'c1', 'a1']); + }); + + it('returns a referentially stable result for unchanged inputs (memoization)', () => { + const state = stateWith(); + const a = selectTimelineForThread(state, THREAD); + const b = selectTimelineForThread(state, THREAD); + expect(a).toBe(b); + }); + + it('honors hideAgentInsights from theme state', () => { + const items = selectTimelineForThread(stateWith({}, { hideAgentInsights: true }), THREAD); + expect(kinds(items)).toEqual(['userMessage', 'assistantMessage']); + }); + + it('returns an empty projection for an unknown thread', () => { + expect(selectTimelineForThread(stateWith(), 'nope')).toEqual([]); + }); +}); diff --git a/app/src/features/conversations/timeline/selectors.ts b/app/src/features/conversations/timeline/selectors.ts new file mode 100644 index 000000000..b318d2928 --- /dev/null +++ b/app/src/features/conversations/timeline/selectors.ts @@ -0,0 +1,224 @@ +/** + * Projection of the durable + live conversation state into one ordered + * `TimelineItem[]` (see `./types.ts` and + * `docs/plans/conversations-timeline-refactor.md`). + * + * `buildThreadTimeline` is a pure function of its inputs — fully unit-testable + * without a store. `selectTimelineForThread` is the memoized Redux selector that + * feeds it from state; reselect's weakMap memoization keys on the input slices + * so switching threads does not thrash the cache. + * + * ## Ordering (reproduces today's single-anchor behavior) + * + * Messages render in stored append order, filtered to drop + * `extraMetadata.hidden`. The single tool timeline is anchored *after the last + * user message* — matching the current positional hack + * (`Conversations.tsx` `lastUserMessageId` at ~L1756 / L2533). Threads with no + * user message (proactive-only) place the process items at the end, matching the + * L2669 fallback. Ephemeral streaming previews (primary + forked branches) + * always trail the durable items. + * + * Until Phase 4 stamps `requestId` on messages, every item belongs to the + * `LEGACY_TURN_ID` turn, so the projection is behaviorally identical to today. + */ +import { createSelector } from '@reduxjs/toolkit'; + +import type { StreamingAssistantState, ToolTimelineEntry } from '../../../store/chatRuntimeSlice'; +import type { RootState } from '../../../store/index'; +import type { ThreadMessage } from '../../../types/thread'; +import { + isSubagentEntry, + LEGACY_TURN_ID, + type TimelineItem, + type TimelineTurn, + toTimelineToolStatus, +} from './types'; + +export interface BuildThreadTimelineInput { + threadId: string; + messages: ThreadMessage[]; + toolTimeline: ToolTimelineEntry[]; + /** Primary streaming preview for the thread, or null. */ + streaming: StreamingAssistantState | null; + /** Forked/parallel branch streams for the thread. */ + parallelStreams: StreamingAssistantState[]; + /** When true, the process kinds (tool/subagent rows) are omitted. */ + hideAgentInsights: boolean; +} + +function isHidden(message: ThreadMessage): boolean { + return Boolean(message.extraMetadata?.hidden); +} + +/** + * The turn a message belongs to. Assistant/user messages produced after the + * Phase 4 rollout carry `extraMetadata.requestId`; older messages have none and + * fall back to the single `legacy` turn (today's single-anchor behavior). + */ +function messageTurnId(message: ThreadMessage): string { + const requestId = message.extraMetadata?.requestId; + return typeof requestId === 'string' && requestId.length > 0 ? requestId : LEGACY_TURN_ID; +} + +/** Map one runtime tool-timeline row onto a `toolCall`/`subagentActivity` item. */ +function toProcessItem(entry: ToolTimelineEntry, threadId: string, seq: number): TimelineItem { + const base = { + id: entry.id, + turnId: LEGACY_TURN_ID, + seq, + threadId, + callId: entry.id, + name: entry.name, + status: toTimelineToolStatus(entry.status), + round: entry.round, + entry, + } as const; + if (isSubagentEntry(entry) && entry.subagent) { + return { ...base, kind: 'subagentActivity', taskId: entry.subagent.taskId }; + } + return { ...base, kind: 'toolCall' }; +} + +/** + * Pure projection: compose messages + tool timeline + streaming previews into + * one ordered `TimelineItem[]`. See the module doc for ordering rules. + */ +export function buildThreadTimeline(input: BuildThreadTimelineInput): TimelineItem[] { + const { threadId, messages, toolTimeline, streaming, parallelStreams, hideAgentInsights } = input; + + const visible = messages.filter(m => !isHidden(m)); + // Anchor id: the *last* user message, mirroring the current positional hack. + let lastUserMessageId: string | undefined; + for (let i = visible.length - 1; i >= 0; i -= 1) { + if (visible[i].sender === 'user') { + lastUserMessageId = visible[i].id; + break; + } + } + + const items: TimelineItem[] = []; + let seq = 0; + + const pushProcessItems = () => { + if (hideAgentInsights) return; + for (const entry of toolTimeline) { + items.push(toProcessItem(entry, threadId, seq)); + seq += 1; + } + }; + + for (const message of visible) { + if (message.sender === 'user') { + items.push({ + kind: 'userMessage', + id: message.id, + turnId: messageTurnId(message), + seq, + threadId, + message, + }); + } else { + items.push({ + kind: 'assistantMessage', + id: message.id, + turnId: messageTurnId(message), + seq, + threadId, + message, + // Durable persisted messages are never interim narration; interim + // (`chat_interim`) items are live-only and folded in during streaming. + interim: false, + }); + } + seq += 1; + // Anchor the process block immediately after the last user message so it + // reads above the agent's answer for that turn. + if (message.id === lastUserMessageId) { + pushProcessItems(); + } + } + + // Proactive / no-user-message threads: process items have no anchor above, so + // they trail the messages (mirrors the L2669 fallback). + if (!lastUserMessageId) { + pushProcessItems(); + } + + // Ephemeral streaming previews always trail the durable items (they belong to + // the live turn). `thinking` rides along on the primary stream; forked + // branches render content only, matching today. + if (streaming && (streaming.content.length > 0 || streaming.thinking.length > 0)) { + items.push({ + kind: 'streamingText', + id: `stream:${streaming.requestId}`, + turnId: LEGACY_TURN_ID, + seq, + threadId, + text: streaming.content, + thinking: streaming.thinking.length > 0 ? streaming.thinking : undefined, + streamId: streaming.requestId, + branch: false, + }); + seq += 1; + } + for (const branch of parallelStreams) { + if (branch.content.length === 0 && branch.thinking.length === 0) continue; + items.push({ + kind: 'streamingText', + id: `stream:branch:${branch.requestId}`, + turnId: LEGACY_TURN_ID, + seq, + threadId, + text: branch.content, + streamId: branch.requestId, + branch: true, + }); + seq += 1; + } + + return items; +} + +/** Split a flat, ordered timeline into contiguous per-turn groups. */ +export function groupTimelineIntoTurns(items: TimelineItem[]): TimelineTurn[] { + const turns: TimelineTurn[] = []; + for (const item of items) { + const last = turns[turns.length - 1]; + if (last && last.turnId === item.turnId) { + last.items.push(item); + } else { + turns.push({ turnId: item.turnId, items: [item] }); + } + } + return turns; +} + +const EMPTY_MESSAGES: ThreadMessage[] = []; +const EMPTY_TIMELINE: ToolTimelineEntry[] = []; + +/** + * Memoized projection selector. Pass the thread id as the second arg: + * `selectTimelineForThread(state, threadId)`. + */ +export const selectTimelineForThread = createSelector( + [ + (state: RootState, threadId: string) => + state.thread.messagesByThreadId[threadId] ?? EMPTY_MESSAGES, + (state: RootState, threadId: string) => + state.chatRuntime.toolTimelineByThread[threadId] ?? EMPTY_TIMELINE, + (state: RootState, threadId: string) => + state.chatRuntime.streamingAssistantByThread[threadId] ?? null, + (state: RootState, threadId: string) => state.chatRuntime.parallelStreamsByThread[threadId], + (state: RootState) => state.theme?.hideAgentInsights ?? false, + (_state: RootState, threadId: string) => threadId, + ], + (messages, toolTimeline, streaming, parallelMap, hideAgentInsights, threadId) => + buildThreadTimeline({ + threadId, + messages, + toolTimeline, + streaming, + parallelStreams: parallelMap ? Object.values(parallelMap) : [], + hideAgentInsights, + }) +); diff --git a/app/src/features/conversations/timeline/types.ts b/app/src/features/conversations/timeline/types.ts new file mode 100644 index 000000000..c2c71aaf4 --- /dev/null +++ b/app/src/features/conversations/timeline/types.ts @@ -0,0 +1,98 @@ +/** + * Unified timeline model for the conversations panel. + * + * `threadSlice` (durable messages) and `chatRuntimeSlice` (live tool timeline, + * streaming previews) remain the sources of truth; `selectTimelineForThread` + * (see `./selectors.ts`) projects them into one ordered `TimelineItem[]`. This + * is a *projection*, not a new slice — no big-bang migration, and each phase of + * the refactor can land independently (see + * `docs/plans/conversations-timeline-refactor.md`). + * + * Every item carries a stable `id`, its `turnId` (the request that produced it; + * `'legacy'` for pre-migration turns whose messages carry no `requestId`), a + * per-turn ordering `seq`, and its `threadId`. Ordering within a turn is by + * `seq`; turns order by the position of their first message in the thread. + */ +import type { ToolTimelineEntry } from '../../../store/chatRuntimeSlice'; +import type { ThreadMessage } from '../../../types/thread'; + +/** + * Turn id used for every item on a thread whose messages predate per-turn + * `requestId` stamping (Phase 4). Until then the whole thread is one turn, which + * reproduces today's single-anchor timeline behavior. + */ +export const LEGACY_TURN_ID = 'legacy'; + +/** Normalised tool status for the timeline (`success` → `ok`). */ +export type TimelineToolStatus = 'running' | 'ok' | 'error' | 'awaiting_user' | 'cancelled'; + +export interface TimelineItemBase { + /** Stable id: message id, or the runtime row id (`ToolTimelineEntry.id`). */ + id: string; + /** Request that produced this item; `LEGACY_TURN_ID` before requestId stamping. */ + turnId: string; + /** Ordering within the turn. Reducer-assigned now; backend-stamped in Phase 4. */ + seq: number; + threadId: string; +} + +export type TimelineItem = TimelineItemBase & + ( + | { kind: 'userMessage'; message: ThreadMessage } + | { kind: 'assistantMessage'; message: ThreadMessage; interim: boolean } + | { + /** Ephemeral live-turn streaming tail (primary or a forked branch). */ + kind: 'streamingText'; + text: string; + thinking?: string; + /** `requestId` of the stream; distinguishes the primary from branches. */ + streamId?: string; + /** True for a `parallelStreamsByThread` forked branch. */ + branch: boolean; + } + | { kind: 'reasoning'; text: string; settled: boolean } + | { + kind: 'toolCall'; + /** The underlying runtime row (rendered via `ToolTimelineBlock`). */ + entry: ToolTimelineEntry; + callId: string; + name: string; + status: TimelineToolStatus; + round: number; + } + | { + /** A `subagent:*` tool row, rendered nested but stored flat. */ + kind: 'subagentActivity'; + entry: ToolTimelineEntry; + taskId: string; + callId: string; + name: string; + status: TimelineToolStatus; + round: number; + } + ); + +export type TimelineItemKind = TimelineItem['kind']; + +/** The "agent process" kinds that `hideAgentInsights` suppresses. */ +export const AGENT_INSIGHT_KINDS: readonly TimelineItemKind[] = [ + 'toolCall', + 'subagentActivity', + 'reasoning', +]; + +/** A contiguous group of items sharing a `turnId`, in render order. */ +export interface TimelineTurn { + turnId: string; + items: TimelineItem[]; +} + +/** Map `ToolTimelineEntry.status` onto the normalised timeline status. */ +export function toTimelineToolStatus(status: ToolTimelineEntry['status']): TimelineToolStatus { + return status === 'success' ? 'ok' : status; +} + +/** A `subagent:*` row carries live/persisted sub-agent activity. */ +export function isSubagentEntry(entry: ToolTimelineEntry): boolean { + return entry.name.startsWith('subagent:'); +} diff --git a/app/src/pages/conversations/utils/format.ts b/app/src/features/conversations/utils/format.ts similarity index 100% rename from app/src/pages/conversations/utils/format.ts rename to app/src/features/conversations/utils/format.ts diff --git a/app/src/pages/conversations/utils/threadFilter.test.ts b/app/src/features/conversations/utils/threadFilter.test.ts similarity index 100% rename from app/src/pages/conversations/utils/threadFilter.test.ts rename to app/src/features/conversations/utils/threadFilter.test.ts diff --git a/app/src/pages/conversations/utils/threadFilter.ts b/app/src/features/conversations/utils/threadFilter.ts similarity index 100% rename from app/src/pages/conversations/utils/threadFilter.ts rename to app/src/features/conversations/utils/threadFilter.ts diff --git a/app/src/pages/conversations/utils/workerThreadRef.test.ts b/app/src/features/conversations/utils/workerThreadRef.test.ts similarity index 100% rename from app/src/pages/conversations/utils/workerThreadRef.test.ts rename to app/src/features/conversations/utils/workerThreadRef.test.ts diff --git a/app/src/pages/conversations/utils/workerThreadRef.ts b/app/src/features/conversations/utils/workerThreadRef.ts similarity index 100% rename from app/src/pages/conversations/utils/workerThreadRef.ts rename to app/src/features/conversations/utils/workerThreadRef.ts diff --git a/app/src/features/human/HumanPage.test.tsx b/app/src/features/human/HumanPage.test.tsx index 519987f4c..73a6f83c3 100644 --- a/app/src/features/human/HumanPage.test.tsx +++ b/app/src/features/human/HumanPage.test.tsx @@ -19,7 +19,7 @@ import HumanPage from './HumanPage'; // ── Heavy dependency stubs ──────────────────────────────────────────────── -vi.mock('../../pages/Conversations', () => ({ +vi.mock('../conversations/Conversations', () => ({ default: () =>
, })); diff --git a/app/src/features/human/HumanPage.tsx b/app/src/features/human/HumanPage.tsx index cbce7dd86..6bfb5d2be 100644 --- a/app/src/features/human/HumanPage.tsx +++ b/app/src/features/human/HumanPage.tsx @@ -1,7 +1,6 @@ import { useEffect, useMemo, useState } from 'react'; import { useT } from '../../lib/i18n/I18nContext'; -import Conversations from '../../pages/Conversations'; import { useAppSelector } from '../../store/hooks'; import { selectCustomMascotGifUrl, @@ -9,6 +8,7 @@ import { selectCustomSecondaryColor, selectMascotColor, } from '../../store/mascotSlice'; +import Conversations from '../conversations/Conversations'; import { CustomGifMascot, getMascotPalette, diff --git a/app/src/pages/Accounts.tsx b/app/src/pages/Accounts.tsx index ccd0b0d50..3d32089f0 100644 --- a/app/src/pages/Accounts.tsx +++ b/app/src/pages/Accounts.tsx @@ -2,6 +2,7 @@ import debugFactory from 'debug'; import { useEffect, useMemo, useState } from 'react'; import { useParams } from 'react-router-dom'; +import Conversations, { ConversationsPage } from '../features/conversations/Conversations'; import { CustomGifMascot, getMascotPalette, @@ -22,7 +23,6 @@ import { } from '../store/mascotSlice'; import type { Account } from '../types/accounts'; import { AGENT_ACCOUNT_ID as AGENT_ID } from '../utils/accountsFullscreen'; -import Conversations, { AgentChatPanel } from './Conversations'; // Persistence key for face-toggle state across sessions. const FACE_MODE_KEY = 'chat.faceMode'; @@ -172,7 +172,7 @@ const Accounts = () => {
{isAgentSelected && (
- +
)}
diff --git a/app/src/pages/__tests__/Accounts.webviewSelection.test.tsx b/app/src/pages/__tests__/Accounts.webviewSelection.test.tsx index 3e0786edd..8d893c316 100644 --- a/app/src/pages/__tests__/Accounts.webviewSelection.test.tsx +++ b/app/src/pages/__tests__/Accounts.webviewSelection.test.tsx @@ -48,9 +48,9 @@ vi.mock('../../store/mascotSlice', () => ({ selectCustomSecondaryColor: () => '#111111', selectMascotColor: () => 'blue', })); -vi.mock('../Conversations', () => ({ +vi.mock('../../features/conversations/Conversations', () => ({ default: ({ variant }: { variant: string }) =>
{variant}
, - AgentChatPanel: () =>
, + ConversationsPage: () =>
, })); describe('Accounts provider selection', () => { diff --git a/app/src/pages/__tests__/Conversations.attachments.test.tsx b/app/src/pages/__tests__/Conversations.attachments.test.tsx index 8167f088c..d02584409 100644 --- a/app/src/pages/__tests__/Conversations.attachments.test.tsx +++ b/app/src/pages/__tests__/Conversations.attachments.test.tsx @@ -251,7 +251,7 @@ async function renderWithSelectedThread() { socket: socketState('connected'), }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( @@ -591,7 +591,7 @@ describe('Conversations — attachment feature', () => { socket: socketState('connected'), }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( @@ -644,7 +644,7 @@ describe('Conversations — attachment feature', () => { socket: socketState('connected'), }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( @@ -697,7 +697,7 @@ describe('Conversations — attachment feature', () => { socket: socketState('connected'), }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( @@ -752,7 +752,7 @@ describe('Conversations — attachment feature', () => { socket: socketState('connected'), }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( diff --git a/app/src/pages/__tests__/Conversations.render.test.tsx b/app/src/pages/__tests__/Conversations.render.test.tsx index b89253973..285a2e399 100644 --- a/app/src/pages/__tests__/Conversations.render.test.tsx +++ b/app/src/pages/__tests__/Conversations.render.test.tsx @@ -28,6 +28,7 @@ import chatRuntimeReducer, { setPendingPlanReviewForThread, setTaskBoardForThread, setToolTimelineForThread, + setTurnTimelinesForThread, } from '../../store/chatRuntimeSlice'; import layoutReducer from '../../store/layoutSlice'; import socketReducer from '../../store/socketSlice'; @@ -77,6 +78,7 @@ vi.mock('../../services/api/threadApi', () => ({ getThreads: mockGetThreads, getThreadMessages: mockGetThreadMessages, getTurnState: vi.fn().mockResolvedValue(null), + getTurnStateHistory: vi.fn().mockResolvedValue([]), getTaskBoard: vi .fn() .mockResolvedValue({ threadId: 't-1', cards: [], updatedAt: '2026-05-04T10:00:00Z' }), @@ -221,7 +223,7 @@ function makeThread(overrides: Partial = {}): Thread { async function renderConversations(preload: Record = {}) { const store = buildStore(preload); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( @@ -241,7 +243,7 @@ async function renderConversations(preload: Record = {}) { async function renderConversationsRoute(route: string, preload: Record = {}) { const store = buildStore(preload); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( @@ -272,7 +274,7 @@ async function renderEmbeddedConversationsRoute( preload: Record = {} ) { const store = buildStore(preload); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); render( @@ -600,6 +602,77 @@ describe('Conversations — smoke render (#1123 welcome-lock removal)', () => { expect(screen.getByText('Can you summarize this?')).toBeInTheDocument(); }); + it("renders a past turn's process trail above the answer it produced (Phase 5)", async () => { + const thread = makeThread({ id: 'multi-turn-thread', title: 'Multi Turn' }); + // Two turns: req-1 (older) and req-2 (latest). Only the older turn has a + // hydrated past-turn timeline (the latest renders as the live anchor). + const messages: ThreadMessage[] = [ + { + id: 'u1', + sender: 'user', + type: 'text', + content: 'first question', + extraMetadata: { requestId: 'req-1' }, + createdAt: '2026-01-01T00:00:00.000Z', + }, + { + id: 'a1', + sender: 'agent', + type: 'text', + content: 'first answer', + extraMetadata: { requestId: 'req-1' }, + createdAt: '2026-01-01T00:01:00.000Z', + }, + { + id: 'u2', + sender: 'user', + type: 'text', + content: 'second question', + extraMetadata: { requestId: 'req-2' }, + createdAt: '2026-01-01T00:02:00.000Z', + }, + { + id: 'a2', + sender: 'agent', + type: 'text', + content: 'second answer', + extraMetadata: { requestId: 'req-2' }, + createdAt: '2026-01-01T00:03:00.000Z', + }, + ]; + mockGetThreads.mockResolvedValue({ threads: [thread], count: 1 }); + mockGetThreadMessages.mockResolvedValue({ messages, count: messages.length }); + + let store: ReturnType | undefined; + await act(async () => { + store = await renderConversations({ + thread: { + ...selectedThreadState(thread), + messagesByThreadId: { [thread.id]: messages }, + messages, + }, + socket: socketState('connected'), + }); + }); + + // No past-turn trail before hydration. + expect(screen.queryByTestId('past-turn-insights')).not.toBeInTheDocument(); + + // Hydrate the older turn's timeline (as fetchAndHydrateTurnHistory would). + await act(async () => { + store!.dispatch( + setTurnTimelinesForThread({ + threadId: thread.id, + timelines: { 'req-1': [{ id: 'tc-1', name: 'read_file', round: 0, status: 'success' }] }, + }) + ); + }); + + // The past turn's block now renders exactly once, above its answer. + const blocks = await screen.findAllByTestId('past-turn-insights'); + expect(blocks).toHaveLength(1); + }); + it('keeps bubble mode interactions for assistant citations, copy, and reactions', async () => { const writeText = vi.fn().mockResolvedValue(undefined); Object.defineProperty(navigator, 'clipboard', { configurable: true, value: { writeText } }); @@ -1111,7 +1184,7 @@ describe('Conversations — smoke render (#1123 welcome-lock removal)', () => { const thread = makeThread({ id: 'mic-cancel-thread', title: 'Mic' }); mockGetThreads.mockResolvedValue({ threads: [thread], count: 1 }); mockGetThreadMessages.mockResolvedValue({ messages: [], count: 0 }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); const store = buildStore({ thread: selectedThreadState(thread), socket: socketState('connected'), @@ -2105,7 +2178,7 @@ describe('Conversations — open-session resume (View work)', () => { mockGetThreads.mockResolvedValue({ threads: [taskThread], count: 1 }); const store = buildStore({ thread: emptyThreadState }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); await act(async () => { render( @@ -2131,7 +2204,7 @@ describe('Conversations — open-session resume (View work)', () => { const store = buildStore({ thread: selectedThreadState(thread) }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); await act(async () => { render( diff --git a/app/src/pages/__tests__/Conversations.sidebarOverflow.test.tsx b/app/src/pages/__tests__/Conversations.sidebarOverflow.test.tsx index a06b416a4..dc9b84c63 100644 --- a/app/src/pages/__tests__/Conversations.sidebarOverflow.test.tsx +++ b/app/src/pages/__tests__/Conversations.sidebarOverflow.test.tsx @@ -214,7 +214,7 @@ function socketState(status: 'connected' | 'disconnected') { /** Render the Human-page chat embed: sidebar variant with the mic-cloud composer. */ async function renderSidebar(preload: Record = {}) { const store = buildStore(preload); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); let container!: HTMLElement; await act(async () => { @@ -269,7 +269,7 @@ describe('Conversations — sidebar composer footer overflow (#3785)', () => { it('keeps the floating page-variant composer absolutely positioned (no regression)', async () => { const store = buildStore({ thread: emptyThreadState }); - const { default: Conversations } = await import('../Conversations'); + const { default: Conversations } = await import('../../features/conversations/Conversations'); let container!: HTMLElement; await act(async () => { diff --git a/app/src/pages/__tests__/Conversations.test.tsx b/app/src/pages/__tests__/Conversations.test.tsx index d45aad8c4..d2132249b 100644 --- a/app/src/pages/__tests__/Conversations.test.tsx +++ b/app/src/pages/__tests__/Conversations.test.tsx @@ -4,7 +4,7 @@ import { formatThreadLoadError, isComposerInteractionBlocked, isImeCompositionKeyEvent, -} from '../Conversations'; +} from '../../features/conversations/Conversations'; describe('isComposerInteractionBlocked', () => { it('blocks composer interaction while the selected thread is actively running', () => { diff --git a/app/src/pages/__tests__/Conversations.welcomeLock.test.tsx b/app/src/pages/__tests__/Conversations.welcomeLock.test.tsx index d01ddfe43..a78e36bf4 100644 --- a/app/src/pages/__tests__/Conversations.welcomeLock.test.tsx +++ b/app/src/pages/__tests__/Conversations.welcomeLock.test.tsx @@ -9,7 +9,7 @@ // - isComposerInteractionBlocked respects the unlocked path correctly import { describe, expect, it } from 'vitest'; -import { isComposerInteractionBlocked } from '../Conversations'; +import { isComposerInteractionBlocked } from '../../features/conversations/Conversations'; describe('[#1123] Conversations — unlocked flow (welcome-lock removed)', () => { // When chatOnboardingCompleted=false in the old flow, welcome-lock would diff --git a/app/src/pages/dev/AgentInsightsPreview.tsx b/app/src/pages/dev/AgentInsightsPreview.tsx index 770474b48..9ff2cd129 100644 --- a/app/src/pages/dev/AgentInsightsPreview.tsx +++ b/app/src/pages/dev/AgentInsightsPreview.tsx @@ -1,9 +1,9 @@ import { useState } from 'react'; import Button from '../../components/ui/Button'; +import { AgentProcessSourcePanel } from '../../features/conversations/components/AgentProcessSourcePanel'; +import { ToolTimelineBlock } from '../../features/conversations/components/ToolTimelineBlock'; import type { ToolTimelineEntry } from '../../store/chatRuntimeSlice'; -import { AgentProcessSourcePanel } from '../conversations/components/AgentProcessSourcePanel'; -import { ToolTimelineBlock } from '../conversations/components/ToolTimelineBlock'; /** * Dev-only visual preview of the "Agentic task insights" Chat surface. diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index 562dff01b..0dfeadb5a 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/app/src/providers/ChatRuntimeProvider.tsx @@ -207,7 +207,13 @@ function hasCompleteSegmentDelivery( } function chatDoneExtraMetadata(event: ChatDoneEvent): Record | undefined { - return event.citations?.length ? { citations: event.citations } : undefined; + // Stamp the producing turn's request id so the final answer can be grouped + // with its per-turn process trail (Phase 4 anchoring, Option B — see the + // companion plan). `citations` is merged in when present. + const meta: Record = {}; + if (event.citations?.length) meta.citations = event.citations; + if (event.request_id) meta.requestId = event.request_id; + return Object.keys(meta).length > 0 ? meta : undefined; } /** @@ -1037,7 +1043,14 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { addInferenceResponse({ content, threadId: event.thread_id, - extraMetadata: event.citations?.length ? { citations: event.citations } : undefined, + // Stamp the producing turn's request id so the timeline projection + // can group this answer with its per-turn process trail (Phase 4 + // anchoring, Option B — see the companion plan). `citations` is + // merged in when present. + extraMetadata: { + ...(event.citations?.length ? { citations: event.citations } : {}), + ...(event.request_id ? { requestId: event.request_id } : {}), + }, }) ); }, @@ -1052,8 +1065,15 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { return; const content = event.full_response?.trim() ?? ''; if (!content) return; - // Persist this round's leading narration as its own interleaved bubble. - void dispatch(addInferenceResponse({ content, threadId: event.thread_id })); + // Persist this round's leading narration as its own interleaved bubble, + // stamped with the producing turn's request id (Phase 4 anchoring). + void dispatch( + addInferenceResponse({ + content, + threadId: event.thread_id, + extraMetadata: event.request_id ? { requestId: event.request_id } : undefined, + }) + ); // The narration has now become a bubble, so drop it from the live // streaming preview (which accumulates across the whole turn under one // request_id) — otherwise the same text lingers in the preview tail and @@ -1215,7 +1235,14 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { request: event.request_id, }); await dispatch( - addInferenceResponse({ content: event.full_response, threadId: targetThreadId }) + addInferenceResponse({ + content: event.full_response, + threadId: targetThreadId, + // Stamp the producing turn's request id when present (Phase 4 + // anchoring); proactive events may omit it, in which case the + // message falls back to the legacy single-anchor turn. + extraMetadata: event.request_id ? { requestId: event.request_id } : undefined, + }) ); } catch (error) { rtLog('proactive_dispatch_failed', { diff --git a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx index 319494d98..77d4c6311 100644 --- a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx +++ b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx @@ -573,6 +573,34 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria expect(store.getState().chatRuntime.queuedFollowupsByThread['t-fup']).toBeUndefined(); }); + it('stamps the assistant answer with the producing turn requestId on chat_done', async () => { + const listeners = renderProvider(); + + await act(async () => { + listeners.onDone?.({ + thread_id: 't-rid', + request_id: 'req-abc', + full_response: 'the answer', + rounds_used: 1, + total_input_tokens: 1, + total_output_tokens: 1, + }); + }); + + // The persisted answer carries requestId in extraMetadata so the timeline + // projection can group it with its per-turn process trail (Phase 4). + await waitFor(() => + expect(threadApi.appendMessage).toHaveBeenCalledWith( + 't-rid', + expect.objectContaining({ + content: 'the answer', + sender: 'agent', + extraMetadata: expect.objectContaining({ requestId: 'req-abc' }), + }) + ) + ); + }); + it('processes tool_call for different rounds as distinct events', () => { const listeners = renderProvider(); diff --git a/app/src/services/api/threadApi.ts b/app/src/services/api/threadApi.ts index 63ade4536..bedb5f5b8 100644 --- a/app/src/services/api/threadApi.ts +++ b/app/src/services/api/threadApi.ts @@ -139,6 +139,33 @@ export const threadApi = { return data?.turnStates ?? []; }, + /** + * Per-turn history for one thread, newest first — each turn's own tool + * timeline (Phase 4). Cheap enough to call on thread open; full timelines can + * be lazily re-fetched per turn via {@link getTurnStateForRequest}. + */ + getTurnStateHistory: async (threadId: string): Promise => { + const response = await callCoreRpc<{ data?: ListTurnStatesResponse }>({ + method: 'openhuman.threads_turn_state_history', + params: { thread_id: threadId }, + }); + const data = unwrapEnvelope(response); + return data?.turnStates ?? []; + }, + + /** One specific past turn of a thread, by its producing request id (Phase 4). */ + getTurnStateForRequest: async ( + threadId: string, + requestId: string + ): Promise => { + const response = await callCoreRpc<{ data?: GetTurnStateResponse }>({ + method: 'openhuman.threads_turn_state_get_turn', + params: { thread_id: threadId, request_id: requestId }, + }); + const data = unwrapEnvelope(response); + return data?.turnState ?? null; + }, + clearTurnState: async (threadId: string): Promise => { const response = await callCoreRpc<{ data?: ClearTurnStateResponse }>({ method: 'openhuman.threads_turn_state_clear', diff --git a/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts b/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts index ddb64c527..4db54661c 100644 --- a/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts +++ b/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts @@ -1,10 +1,10 @@ import { configureStore } from '@reduxjs/toolkit'; import { describe, expect, it, vi } from 'vitest'; -import reducer, { fetchAndHydrateTurnState } from '../chatRuntimeSlice'; +import reducer, { fetchAndHydrateTurnHistory, fetchAndHydrateTurnState } from '../chatRuntimeSlice'; const { mockThreadApi } = vi.hoisted(() => ({ - mockThreadApi: { getTurnState: vi.fn(), listRuns: vi.fn() }, + mockThreadApi: { getTurnState: vi.fn(), listRuns: vi.fn(), getTurnStateHistory: vi.fn() }, })); vi.mock('../../services/api/threadApi', () => ({ threadApi: mockThreadApi })); @@ -41,3 +41,56 @@ describe('fetchAndHydrateTurnState', () => { ]); }); }); + +describe('fetchAndHydrateTurnHistory', () => { + const persistedTurn = ( + requestId: string, + lifecycle: string, + startedAt: string, + tools: number + ) => ({ + threadId: 'thread-hist', + requestId, + lifecycle, + iteration: 1, + maxIterations: 25, + streamingText: '', + thinking: '', + toolTimeline: Array.from({ length: tools }, (_, i) => ({ + id: `${requestId}-tc-${i}`, + name: 'shell', + round: 0, + status: 'success' as const, + })), + startedAt, + updatedAt: startedAt, + }); + + it('stores older settled turns by requestId, skipping the latest and the live/started turn', async () => { + const store = configureStore({ reducer }); + // Newest-first: req-3 (latest, skipped), req-2 (completed, kept), + // req-1 (interrupted, kept), req-0 (started → filtered by lifecycle). + mockThreadApi.getTurnStateHistory.mockResolvedValueOnce([ + persistedTurn('req-3', 'completed', '2026-06-04T13:00:00Z', 1), + persistedTurn('req-2', 'completed', '2026-06-04T12:00:00Z', 2), + persistedTurn('req-1', 'interrupted', '2026-06-04T11:00:00Z', 1), + persistedTurn('req-0', 'started', '2026-06-04T10:00:00Z', 1), + ]); + + await store.dispatch(fetchAndHydrateTurnHistory('thread-hist')); + + const timelines = store.getState().turnTimelinesByThread['thread-hist']; + expect(Object.keys(timelines).sort()).toEqual(['req-1', 'req-2']); + expect(timelines['req-2']).toHaveLength(2); + expect(timelines['req-1']).toHaveLength(1); + expect(timelines['req-3']).toBeUndefined(); + expect(timelines['req-0']).toBeUndefined(); + }); + + it('swallows transport failures without throwing', async () => { + const store = configureStore({ reducer }); + mockThreadApi.getTurnStateHistory.mockRejectedValueOnce(new Error('boom')); + await expect(store.dispatch(fetchAndHydrateTurnHistory('t'))).resolves.toBeDefined(); + expect(store.getState().turnTimelinesByThread['t']).toBeUndefined(); + }); +}); diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index 8624409f1..7872e9dfe 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -565,6 +565,15 @@ interface ChatRuntimeState { */ parallelRequestThreads: Record; toolTimelineByThread: Record; + /** + * Per-turn tool timelines for *past* (settled) turns of a thread, keyed + * `threadId -> requestId -> entries`. Hydrated from `turn_state_history` on + * thread open so each past answer keeps its own process trail (Phase 4/5), + * instead of only the latest turn's live timeline in + * {@link toolTimelineByThread}. The live turn is intentionally excluded (its + * rows live in `toolTimelineByThread` and are driven by the socket stream). + */ + turnTimelinesByThread: Record>; /** * Ordered narration/thinking/tool transcript per thread for the * "View processing" panel — the interleaved Hermes-style record. Hydrated @@ -649,6 +658,7 @@ const initialState: ChatRuntimeState = { parallelStreamsByThread: {}, parallelRequestThreads: {}, toolTimelineByThread: {}, + turnTimelinesByThread: {}, processingByThread: {}, taskBoardByThread: {}, inferenceTurnLifecycleByThread: {}, @@ -980,6 +990,17 @@ const chatRuntimeSlice = createSlice({ delete state.toolTimelineByThread[action.payload.threadId]; delete state.processingByThread[action.payload.threadId]; }, + /** + * Replace the hydrated past-turn timelines for a thread (Phase 5). The live + * turn's rows stay in {@link toolTimelineByThread}; this holds only the + * settled turns, keyed by their producing `requestId`. + */ + setTurnTimelinesForThread: ( + state, + action: PayloadAction<{ threadId: string; timelines: Record }> + ) => { + state.turnTimelinesByThread[action.payload.threadId] = action.payload.timelines; + }, /** Reset the live processing transcript at the start of a fresh turn so a * new turn's narration/steps don't append onto the previous turn's. */ clearProcessingForThread: (state, action: PayloadAction<{ threadId: string }>) => { @@ -1391,6 +1412,7 @@ const chatRuntimeSlice = createSlice({ state.parallelStreamsByThread = {}; state.parallelRequestThreads = {}; state.toolTimelineByThread = {}; + state.turnTimelinesByThread = {}; state.processingByThread = {}; state.taskBoardByThread = {}; state.inferenceTurnLifecycleByThread = {}; @@ -1617,6 +1639,7 @@ export const { clearParallelRequest, setToolTimelineForThread, clearToolTimelineForThread, + setTurnTimelinesForThread, clearProcessingForThread, appendProcessingProse, recordProcessingTool, @@ -1695,4 +1718,41 @@ export const fetchAndHydrateTurnState = createAsyncThunk( } ); +/** + * Fetch the per-turn history for a thread and populate + * {@link ChatRuntimeState.turnTimelinesByThread} so each *past* answer renders + * its own process trail (Phase 5). Only settled turns (completed / interrupted) + * are stored — the live turn's rows are driven by the socket stream into + * `toolTimelineByThread`. Failures are swallowed: missing history must never + * block thread navigation. + */ +export const fetchAndHydrateTurnHistory = createAsyncThunk( + 'chatRuntime/fetchAndHydrateTurnHistory', + async (threadId: string, { dispatch }) => { + try { + const history = await threadApi.getTurnStateHistory(threadId); + const timelines: Record = {}; + // History is newest-first; the newest turn is the one `getTurnState` + // hydrates into `toolTimelineByThread` (rendered as the live/anchored + // "agent insights"), so skip it here to avoid rendering it twice — this + // field holds only the *older* settled turns. + for (const turn of history.slice(1)) { + if (turn.lifecycle !== 'completed' && turn.lifecycle !== 'interrupted') continue; + if (!turn.requestId || turn.toolTimeline.length === 0) continue; + timelines[turn.requestId] = turn.toolTimeline.map(toolTimelineFromPersisted); + } + turnStateLog( + 'hydrated turn history thread=%s turns=%d', + threadId, + Object.keys(timelines).length + ); + dispatch(setTurnTimelinesForThread({ threadId, timelines })); + return timelines; + } catch (error) { + turnStateLog('history fetch failed thread=%s err=%O', threadId, error); + return null; + } + } +); + export default chatRuntimeSlice.reducer; diff --git a/docs/plans/conversations-timeline-refactor.md b/docs/plans/conversations-timeline-refactor.md new file mode 100644 index 000000000..016b6c126 --- /dev/null +++ b/docs/plans/conversations-timeline-refactor.md @@ -0,0 +1,155 @@ +# Conversations Timeline Refactor — Audit & Plan + +Status: **draft — approved for implementation, not yet started** +Scope: `app/src/pages/Conversations.tsx` and the conversation-timeline data layer (frontend + Rust turn-state persistence). +Companion: [`per-turn-tool-timeline-history.md`](per-turn-tool-timeline-history.md) (adopted wholesale as the backend track). + +## Context + +`app/src/pages/Conversations.tsx` (3,302 lines, one function component) is the app's reusable conversation panel — used by `Accounts.tsx` (page + sidebar variants) and `features/human/HumanPage.tsx` (sidebar + mic-cloud + projectThreadList). It is disorganized and buggy in four dimensions: + +**How it renders tool calls / thoughts / subagents** — five parallel state families composed ad hoc in one render function: durable messages (`threadSlice`, server order), a **single tool-timeline array per thread** (`chatRuntimeSlice.toolTimelineByThread`), a streaming-preview tail (`streamingAssistantByThread`, last 120 chars), parallel subagent streams (`parallelStreamsByThread`), and an inference-status line — each with its own conditional render gates (`hideAgentInsights` × `isSending` × anchoring produces 4+ fallback paths, L1775–1842). + +**How it orders** — no timestamps or sequence ids. Timeline entries carry only `round` + array insertion order. The single timeline is positionally **anchored after the last user message** (`Conversations.tsx:1756, 2533`, fallback `2669` for proactive threads). Each send **wipes** the timeline, so past turns permanently lose their process trail. + +**How it stores & streams** — `ChatRuntimeProvider.tsx` (1,664 lines) merges ~28 snake_case socket events (`tool_call`, `subagent_*`, `chat_interim`, `text_delta`, …) via `store.getState()` + find-row + **full-array-replace** in ~10 handlers. **Two parallel dedup mechanisms** (provider-local `seenChatEventsRef` string-key TTL map + reducer-level id dedup in `hydrateRuntimeFromRunLedger`) with **divergent row-id schemes** (live: `${thread}:subagent:${task}:${tool}` vs ledger: `subagent:${runId}`), plus provider-local `segmentDeliveriesRef` and the `preserveLiveSubagentProse` graft — a documented live-only-state workaround. + +**Tab-switch context loss** — the Rust store persists **one** turn snapshot per thread (whole-file overwrite, `src/openhuman/threads/turn_state/store.rs`), so thread/tab switch rehydrates only the latest turn; live subagent prose, past-turn trails, and the streaming tail are lost. + +**Reference patterns adopted** (from codex `codex-rs/protocol` + TUI research; hermes-agent confirmed as the minimal counter-example; vendored `tinychannels` is transport-only — its `ChannelOutputEvent` vocabulary and `RunLedger`/`ConversationStore` host traits stay compatible, the rich timeline model remains OpenHuman-owned): + +1. **Two-layer model**: ephemeral streaming events (deltas, begin/end pairs keyed by `call_id`) vs **durable timeline items with stable ids**; deltas mutate the item with the matching id; persist only durable items; rehydrate by chronological replay. +2. **Typed item taxonomy**: one discriminated union, one React component per kind. +3. **Ordering**: stable per-item id + monotonic `seq`; per-turn grouping via `requestId`; subagents as flat items with `parentId`, nesting rendered not stored. + +--- + +## Target architecture + +### 1. Unified `TimelineItem` model — a selector projection, not a new slice + +`threadSlice` and `chatRuntimeSlice` remain sources of truth; a memoized selector (`selectTimelineForThread`) composes them into one ordered `TimelineItem[]`. This avoids a big-bang slice migration and lets each phase land independently. + +```ts +// app/src/features/conversations/timeline/types.ts +interface TimelineItemBase { + id: string; // stable: message id, or unified runtime row id (§3) + turnId: string; // requestId; 'legacy' for pre-migration turns + seq: number; // ordering within turn (reducer-assigned now, backend-stamped in Phase 4) + threadId: string; +} +type TimelineItem = TimelineItemBase & ( + | { kind: 'userMessage'; message: ThreadMessage } + | { kind: 'assistantMessage'; message: ThreadMessage; interim: boolean } // chat_interim narration + | { kind: 'streamingText'; text: string; streamId?: string } // ephemeral, live turn only + | { kind: 'reasoning'; text: string; settled: boolean } + | { kind: 'toolCall'; callId: string; name: string; status: 'running'|'ok'|'error'; args?: unknown; result?: unknown; round?: number } + | { kind: 'subagentActivity'; taskId: string; parentCallId?: string; children: string[] } // flat store, rendered nested + | { kind: 'approvalRequest' | 'plan' | 'workflowProposal'; /* wraps existing card payloads */ } +); +``` + +**Ordering**: turns order by first-message position in the thread; items within a turn by `seq`. Anchoring becomes structural — a turn group renders `[userMessage, ...processItems, ...assistantMessages]` — replacing the last-user-message positional hack. Messages without `requestId` fall into a single `legacy` turn that reproduces today's single-anchor behavior. A turn with no `userMessage` (proactive threads) is first-class: process items render before its first assistant message. + +### 2. File layout — `app/src/features/conversations/` (matches `features/human/`; all files ≤ ~500 lines) + +``` +app/src/features/conversations/ + ConversationsPage.tsx # page-shell variant + ConversationsSidebar.tsx # sidebar variant (Accounts, HumanPage) + index.ts # public exports = reusability contract + timeline/ + types.ts # TimelineItem union + selectors.ts # projection + turn grouping (reselect-memoized) + ConversationTimeline.tsx # pure renderer: TimelineItem[] → one component per kind + items/ # UserMessageItem, AssistantMessageItem, ToolCallItem, + # ReasoningItem, SubagentActivityItem, StreamingTailItem + TurnInsightsGroup.tsx # collapsed-header/expanded-body per-turn wrapper (lazy-load) + composer/ + Composer.tsx / MicCloudComposer.tsx / useComposerState.ts / composerSendDecision.ts + threadList/ + ThreadList.tsx / threadFilter.ts + components/ # existing subcomponents relocated as-is + (ToolTimelineBlock, SubagentDrawer, TaskKanbanBoard, AgentProcessSourcePanel, …) +app/src/pages/Conversations.tsx # thin re-export shim during migration; deleted in Phase 6 +``` + +**Reusability contract** (`index.ts`): ``, ``, and `` for future embedders. The `AgentChatPanel` export (Conversations.tsx:3302) is renamed to resolve the collision with the unrelated `components/settings/panels/AgentChatPanel.tsx`; `Accounts.tsx:25` updated. + +### 3. Streaming/merge consolidation + +- All merge logic moves into `chatRuntimeSlice` **reducers** — one typed action per event family (`toolEventReceived`, `subagentEventReceived`, `textDeltaReceived`, `turnLifecycleReceived`). `ChatRuntimeProvider` shrinks to parse-and-dispatch (~300 lines); no `getState()`, no full-array rebuilds. +- **One dedup mechanism**, reducer-level, on a unified row-id scheme: `${threadId}:${requestId}:${kind}:${callId|taskId|streamId}` — used identically by live handlers, `hydrateRuntimeFromSnapshot`, and `hydrateRuntimeFromRunLedger`. Kills the live-vs-ledger id divergence; `preserveLiveSubagentProse` becomes an ordinary upsert (merge rule: persisted fields win for settled rows, live fields win for streaming fields). +- **Backend `seq` envelope** (Phase 4): `progress_bridge.rs` stamps a per-request monotonic `seq: u64` + `request_id` on every event; reducers dedup by `(requestId, seq)`. Reconnect-safe: replayed events with `seq <= lastSeq` are dropped; missed events recovered by re-hydration on reconnect (kept). Until then, the TTL-map dedup moves verbatim into the reducer as the interim mechanism. +- `segmentDeliveriesRef` reconstruction becomes per-request segment state in the slice, cleared on `chat_done`. + +### 4. Per-turn persistence (backend) + +Adopt `per-turn-tool-timeline-history.md` as written: per-turn files `turn_states//.json` with retention (N=20) + `latest` pointer, `turn_state_list` / `turn_state_get(requestId)` RPCs, idempotent legacy single-file migration, `extraMetadata.requestId` stamped on assistant messages (Option B anchoring). **Amendment**: persist `seq` on `PersistedToolTimelineEntry` (Rust `turn_state/types.rs` + `app/src/types/turnState.ts`) so replayed snapshots order identically to live streams. + +**Sequencing: frontend-first.** The projection, component split, and reducer consolidation all work against the current single-snapshot backend (everything lands in the `legacy`/live turn). The ring store then slots in underneath without touching the renderer. + +--- + +## Phases (each a small PR, tests green throughout; branch off `upstream/main`, small focused commits) + +### Phase 0 — Land this plan +This document, committed to `docs/plans/`. + +### Phase 1 — Mechanical extraction, no behavior change +Create the `features/conversations/` skeleton; relocate `pages/conversations/{components,hooks,utils}`; extract ThreadList, Composer/MicCloudComposer/useComposerState, and the page/sidebar shells out of `Conversations.tsx`. `pages/Conversations.tsx` becomes a shim with the old props so `Accounts.tsx`/`HumanPage.tsx` are untouched. Rename `AgentChatPanel`. Shared thread/selection state via a `ConversationsContext` to avoid prop-drilling explosions. +**Guardrails**: `Conversations.render.test.tsx` + 4 sibling tests, Accounts/HumanPage tests pass unmodified (import paths only). + +### Phase 2 — `TimelineItem` projection + `` +Add `timeline/types.ts`, `timeline/selectors.ts` (`legacy`-turn fallback ⇒ behaviorally identical to today's anchor logic), `ConversationTimeline.tsx` + `items/*` (wrapping existing `ToolTimelineBlock`, `AgentMessageBubble`, cards). Replace the render loop (~L2400–2700 of the old file). `hideAgentInsights` filters kinds in the selector, not components. +**New tests**: `selectors.test.ts` (ordering, legacy fallback, empty thread, **proactive thread with no user message** — must reproduce the L2669 fallback), `ConversationTimeline.test.tsx` (snapshot per kind). + +### Phase 3 — Reducer-side merge/dedup consolidation +Typed actions in `chatRuntimeSlice`; move each provider handler's merge body into reducers; move `seenChatEventsRef` + `segmentDeliveriesRef` into slice state; unify row ids everywhere; replace `preserveLiveSubagentProse` with the field-level upsert. Verbose debug logging on every dedup drop / merge decision (repo convention). +**Guardrails**: `ChatRuntimeProvider.test.tsx` (assertions rewritten to dispatched actions), `chatRuntimeSlice.test.ts`, `chatRuntimeSlice.toolFailure.test.ts`, `SubagentDrawer.test.tsx`. **New tests**: duplicate-event replay is a no-op; live subagent prose survives snapshot hydration; reconnect-replay simulation. + +### Phase 4 — Backend per-turn ring store (companion-plan steps 1–3) +Rust first (per AGENTS.md workflow Rust → RPC → UI): per-turn store layout + retention + legacy migration in `turn_state/store.rs`, mirror retention in `mirror.rs`, `seq` stamping in `progress_bridge.rs`, `seq` on `turn_state/types.rs`; `turn_state_list` / `turn_state_get(requestId)` RPCs; `extraMetadata.requestId` stamping (threadSlice `addInferenceResponse` + `chat_segment`/`chat_done`/`chat_interim` paths); wire types in `app/src/types/turnState.ts` + `threadApi.ts`. Invisible to UI. +**Tests**: extend `store_tests.rs` (put/get/list, prune to N, latest pointer, idempotent legacy migration, `mark_all_interrupted` over dir), `mirror_tests.rs`; JSON-RPC E2E for the new methods. + +### Phase 5 — Per-turn frontend (companion-plan step 4) +`toolTimelineByThread` → `Record>` + `liveRequestIdByThread`; send no longer wipes history; dedup by `(requestId, seq)`, drop the interim TTL logic. Selector groups by real `requestId`; `TurnInsightsGroup` lazy-fetches `turn_state_get(requestId)` on first expand; thread-switch hydration = `turn_state_list` + latest snapshot only. +**New tests**: multi-turn render (each turn keeps its trail), lazy-load, legacy-message fallback. + +### Phase 6 — Cleanup +Delete the `pages/Conversations.tsx` shim (point Accounts/HumanPage at `features/conversations`); delete dead state (`parallelStreamsByThread` if subsumed by `streamingText` items, `preserveLiveSubagentProse`, provider refs). i18n keys unchanged (all strings already via `useT`). + +--- + +## Tab-switch context loss — cause → fix + +| Lost today | Cause | Fixed by | +|---|---|---| +| Past turns' process trails | single array wiped on send + single-snapshot store | Phases 4–5 | +| Live subagent prose after rehydrate | hydration clobbers live-only rows (`preserveLiveSubagentProse` workaround) | Phase 3 (unified ids + field-level upsert) | +| Streaming tail / interim segments | provider-local refs + latest-snapshot-only hydration | Phase 3 (segment state in slice) + Phase 4 | +| Duplicate/ghost rows after reconnect | TTL dedup misses + id-scheme divergence | Phase 3 interim, Phase 4 `seq` definitive | + +## Verification + +- Per phase: `pnpm test` (guardrail tests listed above), `pnpm typecheck`, `pnpm lint`; Phase 4 adds `cargo test` / `pnpm test:rust` + `tests/json_rpc_e2e.rs`; ≥80% diff coverage (CI Lite gate). +- Manual QA after Phases 3 and 5 (`pnpm dev:app`): multi-turn thread with tools + subagents; switch tab mid-stream and back; kill/restore socket mid-turn; proactive thread with no user message; `hideAgentInsights` sidebar variant in HumanPage; app restart on a 5-turn thread — all trails restored, ordered identically to live. + +## Risks + +- **Reconnect replay before `seq` exists** — keep the TTL-equivalent dedup in the reducer until Phase 4 lands; do not delete early. +- **Proactive threads (no user message)** — first-class in the projection; tested in Phase 2. +- **Snapshot migration** — legacy single-file read must be idempotent; keep the old read path one release. +- **Selector performance on long threads** — reselect memoization per thread; collapsed turns hold metadata only (lazy-load bodies). +- **`round` semantics** — `ToolTimelineBlock` coalescing keys off `round`; keep `round` alongside `seq`, don't replace it. + +## Critical files + +- `app/src/pages/Conversations.tsx` (3,302 ln — split & shim) +- `app/src/providers/ChatRuntimeProvider.tsx` (1,664 ln — shrink to dispatch) +- `app/src/store/chatRuntimeSlice.ts` (1,698 ln — reducer-side merges, per-turn shape) +- `app/src/store/threadSlice.ts` (`addInferenceResponse` requestId stamping) +- `app/src/types/turnState.ts`, `app/src/services/api/threadApi.ts` (wire types/RPCs) +- `src/openhuman/threads/turn_state/{store.rs,mirror.rs,types.rs}` (ring store, seq) +- `src/openhuman/channels/providers/web/progress_bridge.rs` (seq envelope) +- `docs/plans/per-turn-tool-timeline-history.md` (adopted backend design) diff --git a/src/openhuman/threads/ops.rs b/src/openhuman/threads/ops.rs index b74aac19e..0c70ec07b 100644 --- a/src/openhuman/threads/ops.rs +++ b/src/openhuman/threads/ops.rs @@ -22,8 +22,8 @@ use crate::openhuman::threads::title::{ THREAD_TITLE_MODEL_HINT, THREAD_TITLE_SYSTEM_PROMPT, }; use crate::openhuman::threads::turn_state::{ - self, ClearTurnStateRequest, ClearTurnStateResponse, GetTurnStateRequest, GetTurnStateResponse, - ListTurnStatesResponse, + self, ClearTurnStateRequest, ClearTurnStateResponse, GetTurnStateForRequestRequest, + GetTurnStateRequest, GetTurnStateResponse, ListTurnStatesResponse, }; use crate::openhuman::threads::ThreadsError; use crate::rpc::RpcOutcome; @@ -654,6 +654,37 @@ pub async fn turn_state_list( )) } +/// Lists every persisted turn snapshot for one thread, newest first — the +/// per-turn history that lets the UI render each answer's own process trail. +pub async fn turn_state_history( + request: GetTurnStateRequest, +) -> Result>, String> { + let dir = workspace_dir().await?; + let turn_states = turn_state::store::list_thread(dir, &request.thread_id)?; + let count = turn_states.len(); + Ok(envelope( + ListTurnStatesResponse { turn_states, count }, + Some(counts([("num_turn_states", count)])), + None, + )) +} + +/// Returns one specific turn of a thread by its producing request id — used by +/// the UI to lazily load a past turn's full timeline when its insights block is +/// first expanded. +pub async fn turn_state_get_turn( + request: GetTurnStateForRequestRequest, +) -> Result>, String> { + let dir = workspace_dir().await?; + let turn_state = turn_state::store::get_turn(dir, &request.thread_id, &request.request_id)?; + let present = turn_state.is_some(); + Ok(envelope( + GetTurnStateResponse { turn_state }, + Some(counts([("present", usize::from(present))])), + None, + )) +} + /// Clears the persisted turn snapshot for a thread (e.g. after the user /// dismisses an "interrupted" banner). pub async fn turn_state_clear( diff --git a/src/openhuman/threads/schemas.rs b/src/openhuman/threads/schemas.rs index 47b866674..0a01f927e 100644 --- a/src/openhuman/threads/schemas.rs +++ b/src/openhuman/threads/schemas.rs @@ -12,7 +12,9 @@ use crate::openhuman::memory::{ UpdateConversationMessageRequest, UpdateConversationThreadLabelsRequest, UpdateConversationThreadTitleRequest, UpsertConversationThreadRequest, }; -use crate::openhuman::threads::turn_state::{ClearTurnStateRequest, GetTurnStateRequest}; +use crate::openhuman::threads::turn_state::{ + ClearTurnStateRequest, GetTurnStateForRequestRequest, GetTurnStateRequest, +}; use super::ops; @@ -31,6 +33,8 @@ pub fn all_controller_schemas() -> Vec { schemas("purge"), schemas("turn_state_get"), schemas("turn_state_list"), + schemas("turn_state_history"), + schemas("turn_state_get_turn"), schemas("turn_state_clear"), schemas("task_board_get"), schemas("task_board_put"), @@ -92,6 +96,14 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("turn_state_list"), handler: handle_turn_state_list, }, + RegisteredController { + schema: schemas("turn_state_history"), + handler: handle_turn_state_history, + }, + RegisteredController { + schema: schemas("turn_state_get_turn"), + handler: handle_turn_state_get_turn, + }, RegisteredController { schema: schemas("turn_state_clear"), handler: handle_turn_state_clear, @@ -396,6 +408,49 @@ pub fn schemas(function: &str) -> ControllerSchema { required: true, }], }, + "turn_state_history" => ControllerSchema { + namespace: "threads", + function: "turn_state_history", + description: + "List every persisted turn snapshot for one thread, newest first — the per-turn process history.", + inputs: vec![FieldSchema { + name: "thread_id", + ty: TypeSchema::String, + comment: "Thread identifier.", + required: true, + }], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Envelope with the thread's turn snapshots and a count.", + required: true, + }], + }, + "turn_state_get_turn" => ControllerSchema { + namespace: "threads", + function: "turn_state_get_turn", + description: "Fetch one specific turn of a thread by its producing request id.", + inputs: vec![ + FieldSchema { + name: "thread_id", + ty: TypeSchema::String, + comment: "Thread identifier.", + required: true, + }, + FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Producing request id of the turn.", + required: true, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Envelope wrapping the turn state (may be null).", + required: true, + }], + }, "turn_state_clear" => ControllerSchema { namespace: "threads", function: "turn_state_clear", @@ -571,6 +626,20 @@ fn handle_turn_state_list(_params: Map) -> ControllerFuture { Box::pin(async move { to_json(ops::turn_state_list(EmptyRequest {}).await?) }) } +fn handle_turn_state_history(params: Map) -> ControllerFuture { + Box::pin(async move { + let p = parse::(params)?; + to_json(ops::turn_state_history(p).await?) + }) +} + +fn handle_turn_state_get_turn(params: Map) -> ControllerFuture { + Box::pin(async move { + let p = parse::(params)?; + to_json(ops::turn_state_get_turn(p).await?) + }) +} + fn handle_turn_state_clear(params: Map) -> ControllerFuture { Box::pin(async move { let p = parse::(params)?; diff --git a/src/openhuman/threads/schemas_tests.rs b/src/openhuman/threads/schemas_tests.rs index 2aeb8c07d..ec9ed44a1 100644 --- a/src/openhuman/threads/schemas_tests.rs +++ b/src/openhuman/threads/schemas_tests.rs @@ -16,6 +16,8 @@ const ALL_FUNCTIONS: &[&str] = &[ "purge", "turn_state_get", "turn_state_list", + "turn_state_history", + "turn_state_get_turn", "turn_state_clear", "task_board_get", "task_board_put", diff --git a/src/openhuman/threads/turn_state/mod.rs b/src/openhuman/threads/turn_state/mod.rs index e7f52d2fd..9ab1672b5 100644 --- a/src/openhuman/threads/turn_state/mod.rs +++ b/src/openhuman/threads/turn_state/mod.rs @@ -14,7 +14,7 @@ pub use mirror::TurnStateMirror; pub use store::TurnStateStore; pub use types::{ - ClearTurnStateRequest, ClearTurnStateResponse, GetTurnStateRequest, GetTurnStateResponse, - ListTurnStatesResponse, SubagentActivity, SubagentToolCall, ToolTimelineEntry, - ToolTimelineStatus, TurnLifecycle, TurnPhase, TurnState, + ClearTurnStateRequest, ClearTurnStateResponse, GetTurnStateForRequestRequest, + GetTurnStateRequest, GetTurnStateResponse, ListTurnStatesResponse, SubagentActivity, + SubagentToolCall, ToolTimelineEntry, ToolTimelineStatus, TurnLifecycle, TurnPhase, TurnState, }; diff --git a/src/openhuman/threads/turn_state/store.rs b/src/openhuman/threads/turn_state/store.rs index 865cd4c1a..c88a89006 100644 --- a/src/openhuman/threads/turn_state/store.rs +++ b/src/openhuman/threads/turn_state/store.rs @@ -1,13 +1,27 @@ //! Filesystem-backed snapshot store for [`super::types::TurnState`]. //! -//! One JSON file per thread under -//! `/memory/conversations/turn_states/.json`. -//! Whole-file overwrite (latest snapshot wins). The presence of a file -//! means the turn was non-terminal at last write. +//! **Per-turn ring layout.** One JSON file per *turn* under a per-thread +//! directory: +//! `/memory/conversations/turn_states//.json`. +//! Each turn keeps its own snapshot so a multi-turn thread retains every turn's +//! tool timeline (the "Agentic task insights" trail), not just the latest. +//! Completed turns are pruned to the newest [`COMPLETED_RETENTION`] per thread so +//! history stays bounded. //! -//! Mutations are serialised through a single process-wide mutex so the -//! progress consumer cannot interleave a flush against an RPC handler -//! reading the same file. +//! The `get(thread_id)` / `list()` / `delete(thread_id)` / `clear_all` / +//! `mark_all_interrupted` surface is unchanged so existing callers (RPC layer, +//! mirror, cold-boot) keep working: `get`/`list` resolve the *latest* turn per +//! thread. New `get_turn(thread_id, request_id)` and `list_thread(thread_id)` +//! expose the per-turn history. +//! +//! **Legacy migration.** Snapshots written by older cores live as flat files +//! `turn_states/.json`. They are migrated in place — read once, +//! rewritten under `/.json`, and the flat file +//! removed — on first access. Migration is idempotent. +//! +//! Mutations are serialised through a single process-wide mutex so the progress +//! consumer cannot interleave a flush against an RPC handler reading the same +//! file. use std::fs::{self, File}; use std::io::{Read, Write}; @@ -23,6 +37,10 @@ use super::types::{TurnLifecycle, TurnState}; const LOG_PREFIX: &str = "[threads:turn_state]"; const TURN_STATE_DIR: &str = "turn_states"; const SNAPSHOT_EXTENSION: &str = "json"; +/// Newest completed turns retained per thread. Older completed turns are pruned +/// on the next completed write so a long-lived thread's history stays bounded +/// (mirrors the timeline registry's soft-cap philosophy — never unbounded). +const COMPLETED_RETENTION: usize = 20; static TURN_STATE_LOCK: Lazy> = Lazy::new(|| Mutex::new(())); /// Workspace-rooted handle that reads and writes per-thread turn snapshots. @@ -36,11 +54,16 @@ impl TurnStateStore { Self { workspace_dir } } - /// Atomically overwrite the snapshot for `state.thread_id`. + /// Atomically write the snapshot for `state.request_id` under + /// `state.thread_id`. On a `Completed` write, prune the thread's completed + /// turns to the newest [`COMPLETED_RETENTION`]. pub fn put(&self, state: &TurnState) -> Result<(), String> { let _guard = TURN_STATE_LOCK.lock(); - let dir = self.ensure_dir()?; - let path = self.snapshot_path(&state.thread_id); + // Fold any pre-existing flat file for this thread into the per-turn + // layout first so the directory is the single source of truth. + self.migrate_thread_locked(&state.thread_id); + let dir = self.ensure_thread_dir(&state.thread_id)?; + let path = self.turn_path(&state.thread_id, &state.request_id); let mut tmp = NamedTempFile::new_in(&dir) .map_err(|e| format!("create turn-state tempfile in {}: {e}", dir.display()))?; let bytes = @@ -52,89 +75,109 @@ impl TurnStateStore { .map_err(|e| format!("fsync turn-state tempfile: {e}"))?; tmp.persist(&path) .map_err(|e| format!("persist turn-state file {}: {e}", path.display()))?; - // Sync the directory entry created by the rename — without - // this a crash or power loss between persist() and the next - // fs flush can drop the snapshot, defeating the cold-boot - // recovery guarantee. Best-effort on platforms where opening - // a directory for sync is not supported (Windows). The fsync - // failure is logged but not fatal. + // Sync the directory entry created by the rename — without this a crash + // or power loss between persist() and the next fs flush can drop the + // snapshot, defeating the cold-boot recovery guarantee. Best-effort on + // platforms where opening a directory for sync is not supported. if let Err(err) = sync_dir(&dir) { log::warn!("{LOG_PREFIX} failed to fsync {}: {err}", dir.display()); } debug!( - "{LOG_PREFIX} wrote snapshot thread={} lifecycle={:?} iter={}/{} timeline={}", + "{LOG_PREFIX} wrote snapshot thread={} request={} lifecycle={:?} iter={}/{} timeline={}", state.thread_id, + state.request_id, state.lifecycle, state.iteration, state.max_iterations, state.tool_timeline.len() ); + if state.lifecycle == TurnLifecycle::Completed { + self.prune_completed_locked(&state.thread_id); + } Ok(()) } - /// Return the snapshot for `thread_id`, or `None` if no file exists. + /// Return the latest turn for `thread_id`, or `None` if none exists. + /// "Latest" is the turn with the greatest `started_at` (ties broken by + /// `updated_at`) — the in-flight or most-recent turn. pub fn get(&self, thread_id: &str) -> Result, String> { let _guard = TURN_STATE_LOCK.lock(); - let path = self.snapshot_path(thread_id); + self.migrate_thread_locked(thread_id); + Ok(latest_turn(self.read_thread_turns(thread_id)?)) + } + + /// Return a specific turn by `request_id`, or `None` if absent. + pub fn get_turn(&self, thread_id: &str, request_id: &str) -> Result, String> { + let _guard = TURN_STATE_LOCK.lock(); + self.migrate_thread_locked(thread_id); + let path = self.turn_path(thread_id, request_id); if !path.exists() { return Ok(None); } read_snapshot(&path).map(Some) } - /// Delete the snapshot for `thread_id`. Returns `true` if a file was - /// removed, `false` if none existed. + /// Delete every turn for `thread_id` (and any legacy flat file). Returns + /// `true` if anything was removed. pub fn delete(&self, thread_id: &str) -> Result { let _guard = TURN_STATE_LOCK.lock(); - let path = self.snapshot_path(thread_id); - if !path.exists() { - return Ok(false); + let mut removed = false; + let flat = self.legacy_flat_path(thread_id); + if flat.exists() { + fs::remove_file(&flat) + .map_err(|e| format!("remove legacy turn-state {}: {e}", flat.display()))?; + removed = true; } - fs::remove_file(&path) - .map_err(|e| format!("remove turn-state file {}: {e}", path.display()))?; - debug!("{LOG_PREFIX} deleted snapshot thread={}", thread_id); - Ok(true) + let dir = self.thread_dir(thread_id); + if dir.exists() { + fs::remove_dir_all(&dir) + .map_err(|e| format!("remove turn-state dir {}: {e}", dir.display()))?; + removed = true; + } + if removed { + debug!("{LOG_PREFIX} deleted snapshots thread={}", thread_id); + } + Ok(removed) } - /// List every persisted snapshot. Used by the UI to surface - /// interrupted turns on cold boot. + /// List the latest turn for every thread. Used by the UI on cold boot to + /// surface interrupted turns from a previous process (one entry per thread, + /// preserving the pre-ring-store contract). pub fn list(&self) -> Result, String> { let _guard = TURN_STATE_LOCK.lock(); + self.migrate_all_legacy_locked(); let dir = self.dir(); if !dir.exists() { return Ok(Vec::new()); } let mut snapshots = Vec::new(); - for entry in - fs::read_dir(&dir).map_err(|e| format!("read turn-state dir {}: {e}", dir.display()))? - { - let entry = entry.map_err(|e| format!("read turn-state entry: {e}"))?; - let path = entry.path(); - if path.extension().and_then(|s| s.to_str()) != Some(SNAPSHOT_EXTENSION) { - continue; - } - match read_snapshot(&path) { - Ok(snapshot) => snapshots.push(snapshot), - Err(err) => warn!( - "{LOG_PREFIX} skip unreadable snapshot {}: {err}", - path.display() - ), + for thread_id in self.thread_ids()? { + if let Some(latest) = latest_turn(self.read_thread_turns(&thread_id)?) { + snapshots.push(latest); } } Ok(snapshots) } - /// Remove every snapshot file in the turn-state directory, - /// regardless of whether the contents are readable. Used by - /// `threads_purge` to guarantee no stale or corrupted snapshot - /// survives a destructive cleanup — `list()` only returns parseable - /// snapshots, so iterating list+delete would silently leave - /// half-written or schema-skewed files behind. - /// - /// Returns the count of files removed. Failures on individual - /// entries propagate as the first error encountered (the rest of - /// the directory is not touched once an error occurs, so a retry - /// can pick up where this left off). + /// List every turn for one thread, newest first (by `started_at`). + pub fn list_thread(&self, thread_id: &str) -> Result, String> { + let _guard = TURN_STATE_LOCK.lock(); + self.migrate_thread_locked(thread_id); + let mut turns = self.read_thread_turns(thread_id)?; + turns.sort_by(|a, b| { + b.started_at + .cmp(&a.started_at) + .then_with(|| b.updated_at.cmp(&a.updated_at)) + }); + Ok(turns) + } + + /// Remove every snapshot file, readable or not (per-turn files, thread + /// directories, and any legacy flat files). Used by `threads_purge` to + /// guarantee a destructive cleanup leaves nothing — `list()` only returns + /// parseable snapshots, so iterating list+delete would silently leave + /// half-written or schema-skewed files behind. Returns the count of JSON + /// files removed. pub fn clear_all(&self) -> Result { let _guard = TURN_STATE_LOCK.lock(); let dir = self.dir(); @@ -147,12 +190,32 @@ impl TurnStateStore { { let entry = entry.map_err(|e| format!("read turn-state entry: {e}"))?; let path = entry.path(); - if path.extension().and_then(|s| s.to_str()) != Some(SNAPSHOT_EXTENSION) { - continue; + let file_type = entry + .file_type() + .map_err(|e| format!("stat turn-state entry {}: {e}", path.display()))?; + if file_type.is_dir() { + // A per-thread directory: count and remove its JSON files, then + // drop the (now empty) directory. + for sub in fs::read_dir(&path) + .map_err(|e| format!("read thread dir {}: {e}", path.display()))? + { + let sub = sub.map_err(|e| format!("read thread entry: {e}"))?; + let sub_path = sub.path(); + if sub_path.extension().and_then(|s| s.to_str()) == Some(SNAPSHOT_EXTENSION) { + fs::remove_file(&sub_path).map_err(|e| { + format!("remove turn-state file {}: {e}", sub_path.display()) + })?; + removed += 1; + } + } + fs::remove_dir_all(&path) + .map_err(|e| format!("remove thread dir {}: {e}", path.display()))?; + } else if path.extension().and_then(|s| s.to_str()) == Some(SNAPSHOT_EXTENSION) { + // A legacy flat snapshot. + fs::remove_file(&path) + .map_err(|e| format!("remove turn-state file {}: {e}", path.display()))?; + removed += 1; } - fs::remove_file(&path) - .map_err(|e| format!("remove turn-state file {}: {e}", path.display()))?; - removed += 1; } if removed > 0 { debug!( @@ -163,17 +226,19 @@ impl TurnStateStore { Ok(removed) } - /// Mark every persisted snapshot as `Interrupted`. Intended to be - /// invoked from the web-channel provider on startup so the UI can - /// distinguish stale turns left behind by a previous process from - /// turns that are currently being driven in this session. + /// Mark every non-terminal turn as `Interrupted`. Intended to run on startup + /// so the UI can distinguish stale turns left behind by a previous process + /// from turns currently being driven. `Completed`/`Interrupted` turns are + /// left as-is (idempotent; completed turns are intentionally kept so the + /// processing panel can replay a finished turn after a reboot). pub fn mark_all_interrupted(&self, now_rfc3339: &str) -> Result { - let snapshots = self.list()?; + let turns = { + let _guard = TURN_STATE_LOCK.lock(); + self.migrate_all_legacy_locked(); + self.all_turns_locked()? + }; let mut count = 0usize; - for mut snapshot in snapshots { - // Already-terminal snapshots are left as-is: `Interrupted` is - // idempotent, and `Completed` snapshots are intentionally kept so - // the processing panel can replay a finished turn after a reboot. + for mut snapshot in turns { if matches!( snapshot.lifecycle, TurnLifecycle::Interrupted | TurnLifecycle::Completed @@ -193,10 +258,12 @@ impl TurnStateStore { Ok(count) } - fn ensure_dir(&self) -> Result { - let dir = self.dir(); + // --- internals ------------------------------------------------------- + + fn ensure_thread_dir(&self, thread_id: &str) -> Result { + let dir = self.thread_dir(thread_id); fs::create_dir_all(&dir) - .map_err(|e| format!("create turn-state dir {}: {e}", dir.display()))?; + .map_err(|e| format!("create thread turn-state dir {}: {e}", dir.display()))?; Ok(dir) } @@ -207,19 +274,234 @@ impl TurnStateStore { .join(TURN_STATE_DIR) } - fn snapshot_path(&self, thread_id: &str) -> PathBuf { + fn thread_dir(&self, thread_id: &str) -> PathBuf { + self.dir().join(hex::encode(thread_id.as_bytes())) + } + + fn turn_path(&self, thread_id: &str, request_id: &str) -> PathBuf { + self.thread_dir(thread_id).join(format!( + "{}.{}", + hex::encode(request_id.as_bytes()), + SNAPSHOT_EXTENSION + )) + } + + fn legacy_flat_path(&self, thread_id: &str) -> PathBuf { self.dir().join(format!( "{}.{}", hex::encode(thread_id.as_bytes()), SNAPSHOT_EXTENSION )) } + + /// Read every parseable turn snapshot in one thread's directory. + /// Unreadable files are logged and skipped (mirrors `list()`'s resilience). + fn read_thread_turns(&self, thread_id: &str) -> Result, String> { + let dir = self.thread_dir(thread_id); + if !dir.exists() { + return Ok(Vec::new()); + } + let mut turns = Vec::new(); + for entry in fs::read_dir(&dir) + .map_err(|e| format!("read thread turn-state dir {}: {e}", dir.display()))? + { + let entry = entry.map_err(|e| format!("read thread turn-state entry: {e}"))?; + let path = entry.path(); + if path.extension().and_then(|s| s.to_str()) != Some(SNAPSHOT_EXTENSION) { + continue; + } + match read_snapshot(&path) { + Ok(snapshot) => turns.push(snapshot), + Err(err) => warn!( + "{LOG_PREFIX} skip unreadable snapshot {}: {err}", + path.display() + ), + } + } + Ok(turns) + } + + /// hex(thread_id) directory names under the root, decoded back to the + /// thread-id string. Skips legacy flat files (handled by migration). + fn thread_ids(&self) -> Result, String> { + let dir = self.dir(); + if !dir.exists() { + return Ok(Vec::new()); + } + let mut ids = Vec::new(); + for entry in + fs::read_dir(&dir).map_err(|e| format!("read turn-state dir {}: {e}", dir.display()))? + { + let entry = entry.map_err(|e| format!("read turn-state entry: {e}"))?; + if !entry + .file_type() + .map_err(|e| format!("stat turn-state entry: {e}"))? + .is_dir() + { + continue; + } + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + match hex::decode(name) + .ok() + .and_then(|b| String::from_utf8(b).ok()) + { + Some(thread_id) => ids.push(thread_id), + None => warn!("{LOG_PREFIX} skip non-hex thread dir {name}"), + } + } + Ok(ids) + } + + /// Every turn across every thread. Caller holds the lock. + fn all_turns_locked(&self) -> Result, String> { + let mut turns = Vec::new(); + for thread_id in self.thread_ids()? { + turns.append(&mut self.read_thread_turns(&thread_id)?); + } + Ok(turns) + } + + /// If a legacy flat file exists for `thread_id`, fold it into the per-turn + /// layout and remove the flat file. Best-effort; failures are logged. Caller + /// holds the lock. + fn migrate_thread_locked(&self, thread_id: &str) { + let flat = self.legacy_flat_path(thread_id); + if !flat.exists() { + return; + } + match read_snapshot(&flat) { + Ok(state) => { + if let Err(err) = self.write_turn_file(&state) { + warn!( + "{LOG_PREFIX} legacy migrate write failed thread={thread_id}: {err} (flat file kept)" + ); + return; + } + if let Err(err) = fs::remove_file(&flat) { + warn!( + "{LOG_PREFIX} legacy migrate: removed-into-dir but flat delete failed {}: {err}", + flat.display() + ); + } else { + debug!( + "{LOG_PREFIX} migrated legacy snapshot thread={thread_id} request={}", + state.request_id + ); + } + } + Err(err) => warn!( + "{LOG_PREFIX} legacy migrate: unreadable flat file {} left in place: {err}", + flat.display() + ), + } + } + + /// Migrate every legacy flat file under the root. Caller holds the lock. + fn migrate_all_legacy_locked(&self) { + let dir = self.dir(); + if !dir.exists() { + return; + } + let entries = match fs::read_dir(&dir) { + Ok(entries) => entries, + Err(err) => { + warn!("{LOG_PREFIX} migrate scan failed {}: {err}", dir.display()); + return; + } + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().and_then(|s| s.to_str()) != Some(SNAPSHOT_EXTENSION) { + continue; // directories and non-json files + } + match read_snapshot(&path) { + Ok(state) => { + if self.write_turn_file(&state).is_ok() { + let _ = fs::remove_file(&path); + debug!( + "{LOG_PREFIX} migrated legacy snapshot thread={} request={}", + state.thread_id, state.request_id + ); + } + } + Err(err) => warn!( + "{LOG_PREFIX} migrate: unreadable flat file {} left in place: {err}", + path.display() + ), + } + } + } + + /// Atomic per-turn write without migration/retention side effects. Used by + /// migration to relocate a snapshot. Caller holds the lock. + fn write_turn_file(&self, state: &TurnState) -> Result<(), String> { + let dir = self.ensure_thread_dir(&state.thread_id)?; + let path = self.turn_path(&state.thread_id, &state.request_id); + let mut tmp = NamedTempFile::new_in(&dir) + .map_err(|e| format!("create turn-state tempfile in {}: {e}", dir.display()))?; + let bytes = + serde_json::to_vec_pretty(state).map_err(|e| format!("serialize turn state: {e}"))?; + tmp.write_all(&bytes) + .map_err(|e| format!("write turn-state tempfile: {e}"))?; + tmp.as_file() + .sync_all() + .map_err(|e| format!("fsync turn-state tempfile: {e}"))?; + tmp.persist(&path) + .map_err(|e| format!("persist turn-state file {}: {e}", path.display()))?; + if let Err(err) = sync_dir(&dir) { + log::warn!("{LOG_PREFIX} failed to fsync {}: {err}", dir.display()); + } + Ok(()) + } + + /// Prune a thread's `Completed` turns to the newest [`COMPLETED_RETENTION`] + /// by `updated_at`. Non-completed turns (at most the one live turn) are kept. + /// Caller holds the lock. Best-effort — failures are logged, not fatal. + fn prune_completed_locked(&self, thread_id: &str) { + let turns = match self.read_thread_turns(thread_id) { + Ok(turns) => turns, + Err(err) => { + warn!("{LOG_PREFIX} prune read failed thread={thread_id}: {err}"); + return; + } + }; + let mut completed: Vec = turns + .into_iter() + .filter(|t| t.lifecycle == TurnLifecycle::Completed) + .collect(); + if completed.len() <= COMPLETED_RETENTION { + return; + } + // Newest first, then drop everything past the retention window. + completed.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); + for stale in completed.into_iter().skip(COMPLETED_RETENTION) { + let path = self.turn_path(&stale.thread_id, &stale.request_id); + if let Err(err) = fs::remove_file(&path) { + warn!("{LOG_PREFIX} prune remove failed {}: {err}", path.display()); + } else { + debug!( + "{LOG_PREFIX} pruned completed turn thread={thread_id} request={}", + stale.request_id + ); + } + } + } } -/// Best-effort `fsync` of a directory entry. On Unix, opens the -/// directory for read and calls `sync_all` on the file handle. On -/// Windows this is a no-op — directory fsync is not exposed by the -/// platform and the rename's durability is provided by NTFS journaling. +/// Pick the latest turn (greatest `started_at`, ties broken by `updated_at`). +fn latest_turn(turns: Vec) -> Option { + turns.into_iter().max_by(|a, b| { + a.started_at + .cmp(&b.started_at) + .then_with(|| a.updated_at.cmp(&b.updated_at)) + }) +} + +/// Best-effort `fsync` of a directory entry. On Unix, opens the directory for +/// read and calls `sync_all` on the file handle. On Windows this is a no-op — +/// directory fsync is not exposed by the platform and the rename's durability is +/// provided by NTFS journaling. #[cfg(unix)] fn sync_dir(dir: &Path) -> std::io::Result<()> { File::open(dir)?.sync_all() @@ -250,6 +532,14 @@ pub fn get(workspace_dir: PathBuf, thread_id: &str) -> Result, TurnStateStore::new(workspace_dir).get(thread_id) } +pub fn get_turn( + workspace_dir: PathBuf, + thread_id: &str, + request_id: &str, +) -> Result, String> { + TurnStateStore::new(workspace_dir).get_turn(thread_id, request_id) +} + pub fn delete(workspace_dir: PathBuf, thread_id: &str) -> Result { TurnStateStore::new(workspace_dir).delete(thread_id) } @@ -258,6 +548,10 @@ pub fn list(workspace_dir: PathBuf) -> Result, String> { TurnStateStore::new(workspace_dir).list() } +pub fn list_thread(workspace_dir: PathBuf, thread_id: &str) -> Result, String> { + TurnStateStore::new(workspace_dir).list_thread(thread_id) +} + pub fn clear_all(workspace_dir: PathBuf) -> Result { TurnStateStore::new(workspace_dir).clear_all() } diff --git a/src/openhuman/threads/turn_state/store_tests.rs b/src/openhuman/threads/turn_state/store_tests.rs index b726cc5fb..4af251891 100644 --- a/src/openhuman/threads/turn_state/store_tests.rs +++ b/src/openhuman/threads/turn_state/store_tests.rs @@ -10,6 +10,20 @@ fn sample_state(thread_id: &str) -> TurnState { TurnState::started(thread_id.to_string(), "req-1", 25, "2026-05-04T10:00:00Z") } +/// A turn with an explicit request id + started/updated timestamps. +fn turn(thread_id: &str, request_id: &str, started_at: &str) -> TurnState { + let mut s = TurnState::started(thread_id.to_string(), request_id, 25, started_at); + s.updated_at = started_at.to_string(); + s +} + +fn turn_states_root(dir: &tempfile::TempDir) -> std::path::PathBuf { + dir.path() + .join("memory") + .join("conversations") + .join("turn_states") +} + #[test] fn put_then_get_roundtrips_state() { let dir = tempdir().expect("tempdir"); @@ -158,6 +172,135 @@ fn clear_all_on_missing_dir_returns_zero() { assert_eq!(store.clear_all().expect("clear"), 0); } +#[test] +fn keeps_a_separate_snapshot_per_turn_and_get_returns_latest() { + let dir = tempdir().expect("tempdir"); + let store = TurnStateStore::new(dir.path().to_path_buf()); + store + .put(&turn("t", "req-1", "2026-05-04T10:00:00Z")) + .expect("put turn 1"); + store + .put(&turn("t", "req-2", "2026-05-04T10:05:00Z")) + .expect("put turn 2"); + + // Both turns are retained... + let history = store.list_thread("t").expect("list_thread"); + assert_eq!(history.len(), 2); + // ...newest first. + assert_eq!(history[0].request_id, "req-2"); + assert_eq!(history[1].request_id, "req-1"); + + // get(thread) resolves the latest turn (greatest started_at). + let latest = store.get("t").expect("get").expect("present"); + assert_eq!(latest.request_id, "req-2"); + + // get_turn fetches a specific earlier turn. + let earlier = store + .get_turn("t", "req-1") + .expect("get_turn") + .expect("present"); + assert_eq!(earlier.request_id, "req-1"); + assert!(store.get_turn("t", "nope").expect("get_turn").is_none()); + + // list() surfaces exactly one (latest) entry per thread for cold boot. + let latest_per_thread = store.list().expect("list"); + assert_eq!(latest_per_thread.len(), 1); + assert_eq!(latest_per_thread[0].request_id, "req-2"); +} + +#[test] +fn completed_turns_are_pruned_to_the_retention_window() { + let dir = tempdir().expect("tempdir"); + let store = TurnStateStore::new(dir.path().to_path_buf()); + // Write 25 completed turns; only the newest COMPLETED_RETENTION (20) survive. + for i in 0..25 { + let mut s = turn( + "t", + &format!("req-{i:02}"), + &format!("2026-05-04T10:{i:02}:00Z"), + ); + s.lifecycle = TurnLifecycle::Completed; + s.updated_at = format!("2026-05-04T10:{i:02}:00Z"); + store.put(&s).expect("put"); + } + let history = store.list_thread("t").expect("list_thread"); + assert_eq!(history.len(), super::COMPLETED_RETENTION); + // The oldest five (req-00..req-04) are gone; the newest survive. + assert!(history.iter().any(|t| t.request_id == "req-24")); + assert!(history.iter().all(|t| t.request_id != "req-00")); + assert!(store.get_turn("t", "req-04").expect("get_turn").is_none()); + assert!(store.get_turn("t", "req-05").expect("get_turn").is_some()); +} + +#[test] +fn a_live_turn_is_not_pruned_alongside_completed_history() { + let dir = tempdir().expect("tempdir"); + let store = TurnStateStore::new(dir.path().to_path_buf()); + for i in 0..COMPLETED_RETENTION_PLUS { + let mut s = turn( + "t", + &format!("done-{i:02}"), + &format!("2026-05-04T10:{i:02}:00Z"), + ); + s.lifecycle = TurnLifecycle::Completed; + s.updated_at = format!("2026-05-04T10:{i:02}:00Z"); + store.put(&s).expect("put completed"); + } + // A running turn coexists and is never pruned (only completed turns are). + let mut live = turn("t", "live", "2026-05-04T11:00:00Z"); + live.lifecycle = TurnLifecycle::Streaming; + store.put(&live).expect("put live"); + assert!(store.get_turn("t", "live").expect("get_turn").is_some()); + assert_eq!( + store.get("t").expect("get").expect("present").request_id, + "live" + ); +} + +#[test] +fn migrates_a_legacy_flat_snapshot_into_the_per_turn_layout() { + use std::io::Write as _; + let dir = tempdir().expect("tempdir"); + let store = TurnStateStore::new(dir.path().to_path_buf()); + let root = turn_states_root(&dir); + std::fs::create_dir_all(&root).expect("mkdir root"); + + // Hand-write an old-style flat snapshot at `.json`. + let legacy = turn("legacy-thread", "req-legacy", "2026-05-04T09:00:00Z"); + let flat_path = root.join(format!("{}.json", hex::encode("legacy-thread".as_bytes()))); + let mut f = std::fs::File::create(&flat_path).expect("create flat"); + f.write_all(serde_json::to_vec_pretty(&legacy).unwrap().as_slice()) + .expect("write flat"); + drop(f); + + // First access migrates it into the dir and removes the flat file. + let loaded = store.get("legacy-thread").expect("get").expect("present"); + assert_eq!(loaded.request_id, "req-legacy"); + assert!( + !flat_path.exists(), + "flat file must be removed after migration" + ); + let per_turn = root + .join(hex::encode("legacy-thread".as_bytes())) + .join(format!("{}.json", hex::encode("req-legacy".as_bytes()))); + assert!( + per_turn.exists(), + "snapshot must live under the per-turn path" + ); + + // Migration is idempotent — a second access is a no-op. + assert_eq!( + store + .get("legacy-thread") + .expect("get2") + .expect("present") + .request_id, + "req-legacy" + ); +} + +const COMPLETED_RETENTION_PLUS: usize = super::COMPLETED_RETENTION + 3; + #[test] fn put_overwrites_previous_snapshot() { let dir = tempdir().expect("tempdir"); diff --git a/src/openhuman/threads/turn_state/types.rs b/src/openhuman/threads/turn_state/types.rs index 27834cd5c..ec42e95eb 100644 --- a/src/openhuman/threads/turn_state/types.rs +++ b/src/openhuman/threads/turn_state/types.rs @@ -335,7 +335,8 @@ pub struct GetTurnStateResponse { pub turn_state: Option, } -/// Response payload for `openhuman.threads_turn_state_list`. +/// Response payload for `openhuman.threads_turn_state_list` and +/// `openhuman.threads_turn_state_history`. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ListTurnStatesResponse { @@ -343,6 +344,15 @@ pub struct ListTurnStatesResponse { pub count: usize, } +/// Request payload for `openhuman.threads_turn_state_get_turn` — a specific +/// turn of a thread, identified by its producing request id. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct GetTurnStateForRequestRequest { + pub thread_id: String, + pub request_id: String, +} + /// Request payload for `openhuman.threads_turn_state_clear`. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 773b65c93..4d6ad9079 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -3510,6 +3510,77 @@ async fn json_rpc_thread_turn_state_lifecycle() { Some(1) ); + // Seed a second, later turn on the same thread — the per-turn ring keeps + // both instead of overwriting. + // Far-future started_at guarantees turn-2 is the newest (turn-1 was seeded + // with the real `now()`), so history ordering is deterministic. + let mut state2 = openhuman_core::openhuman::threads::turn_state::TurnState::started( + "thread-turn-1", + "req-turn-2", + 25, + "2999-01-01T00:00:00Z", + ); + state2.lifecycle = openhuman_core::openhuman::threads::turn_state::TurnLifecycle::Completed; + state2.updated_at = "2999-01-01T00:00:00Z".into(); + openhuman_core::openhuman::threads::turn_state::store::put(workspace_dir.clone(), &state2) + .expect("seed snapshot 2"); + + // history → both turns, newest first. + let history = post_json_rpc( + &rpc_base, + 9106, + "openhuman.threads_turn_state_history", + json!({ "thread_id": "thread-turn-1" }), + ) + .await; + let history_outer = assert_no_jsonrpc_error(&history, "turn_state_history"); + let turns = history_outer + .get("data") + .and_then(|d| d.get("turnStates")) + .and_then(serde_json::Value::as_array) + .expect("turnStates array"); + assert_eq!(turns.len(), 2, "both turns retained"); + assert_eq!( + turns[0] + .get("requestId") + .and_then(serde_json::Value::as_str), + Some("req-turn-2"), + "newest turn first" + ); + + // get_turn → the specific earlier turn. + let get_turn = post_json_rpc( + &rpc_base, + 9107, + "openhuman.threads_turn_state_get_turn", + json!({ "thread_id": "thread-turn-1", "request_id": "req-turn-1" }), + ) + .await; + let get_turn_outer = assert_no_jsonrpc_error(&get_turn, "turn_state_get_turn (present)"); + assert_eq!( + get_turn_outer + .get("data") + .and_then(|d| d.get("turnState")) + .and_then(|t| t.get("requestId")) + .and_then(serde_json::Value::as_str), + Some("req-turn-1") + ); + + // get_turn for an unknown request → null. + let get_turn_missing = post_json_rpc( + &rpc_base, + 9108, + "openhuman.threads_turn_state_get_turn", + json!({ "thread_id": "thread-turn-1", "request_id": "nope" }), + ) + .await; + let missing_outer = assert_no_jsonrpc_error(&get_turn_missing, "turn_state_get_turn (absent)"); + assert!(missing_outer + .get("data") + .and_then(|d| d.get("turnState")) + .map(|v| v.is_null()) + .unwrap_or(true)); + // clear → cleared:true let cleared = post_json_rpc( &rpc_base,