feat(conversations): per-turn process history (timeline refactor Phases 1–2, 4–5) (#4612)

This commit is contained in:
Steven Enamakel
2026-07-06 18:24:12 -07:00
committed by GitHub
parent 753349700b
commit 16f7bee42f
85 changed files with 2559 additions and 424 deletions
@@ -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';
+2 -2
View File
@@ -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.
@@ -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';
@@ -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';
@@ -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,
@@ -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,
@@ -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,
@@ -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';
@@ -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 }) {
@@ -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<string, true> = {};
// the same identity when the slice field is absent (narrow test stores).
const EMPTY_QUEUED_FOLLOWUPS: Record<string, QueuedFollowup[]> = {};
// 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<string, ToolTimelineEntry[]> = {};
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<string, ToolTimelineEntry[]> = {};
const seen = new Set<string>();
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.
<div className="h-full flex flex-col">
{/* Thread search — flush full-width input, mirrors the settings search. */}
<div className="relative border-b border-line-subtle">
<span className="pointer-events-none absolute inset-y-0 left-3 flex items-center text-content-faint">
<svg className="h-4 w-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M21 21l-4.35-4.35M11 19a8 8 0 100-16 8 8 0 000 16z"
/>
</svg>
</span>
<input
type="text"
value={threadSearch}
onChange={e => setThreadSearch(e.target.value)}
onKeyDown={e => {
if (e.key === 'Escape' && threadSearch) {
e.preventDefault();
setThreadSearch('');
<ThreadList
threads={visibleThreads}
selectedThreadId={selectedThreadId ?? null}
search={threadSearch}
onSearchChange={setThreadSearch}
onCreateThread={() => 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 && (
<button
type="button"
onClick={() => setThreadSearch('')}
aria-label={t('settings.settingsSearch.clear')}
data-testid="chat-thread-search-clear"
className="absolute inset-y-0 right-2 flex items-center px-1 text-content-faint hover:text-content-secondary">
<svg className="h-4 w-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M6 18L18 6M6 6l12 12"
/>
</svg>
</button>
)}
</div>
{/* New conversation a subtle, centered thread-style row (not a loud
button), below the search and above the thread list. */}
<button
type="button"
data-testid="new-thread-button"
data-analytics-id="chat-sidebar-new-thread"
onClick={() => void handleCreateNewThread()}
title={t('chat.newThreadShortcut')}
className="group w-full cursor-pointer border-b border-line-subtle/60 opacity-50 px-3 py-2 transition-colors hover:bg-surface-hover dark:border-line/60">
<div className="flex items-center justify-center gap-1.5">
<svg
className="h-3.5 w-3.5 flex-shrink-0 text-content-muted"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M12 4v16m8-8H4" />
</svg>
<span className="truncate text-xs text-content-secondary">
{t('chat.newConversation')}
</span>
</div>
</button>
<div className="flex-1 overflow-y-auto">
{visibleThreads.length === 0 ? (
<p className="px-4 py-6 text-xs text-content-faint text-center">{t('chat.noThreads')}</p>
) : (
visibleThreads.map(thread => (
<div
key={thread.id}
data-testid={`thread-row-${thread.id}`}
data-analytics-id="chat-sidebar-thread-row"
role="button"
tabIndex={0}
onClick={() => {
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'
}`}>
<div className="flex items-center justify-between">
{editingThreadId === thread.id ? (
<input
ref={editTitleInputRef}
value={editTitleValue}
onClick={e => 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
/>
) : (
<p
className={`text-xs truncate flex-1 ${
selectedThreadId === thread.id
? 'font-medium text-primary-700 dark:text-primary-200'
: 'text-content-secondary'
}`}>
{resolveThreadDisplayTitle(thread.id)}
</p>
)}
<button
type="button"
data-analytics-id="chat-sidebar-edit-thread-title"
onClick={e => {
e.stopPropagation();
handleStartEditTitle(thread.id);
}}
aria-label={t('chat.editThreadTitle')}
title={t('chat.editThreadTitle')}
className="ml-2 p-1 rounded opacity-0 group-hover:opacity-100 hover:bg-surface-strong dark:bg-surface-muted dark:hover:bg-surface-muted text-content-faint hover:text-primary-500 transition-all flex-shrink-0">
<svg className="w-3 h-3" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M15.232 5.232l3.536 3.536m-2.036-5.036a2.5 2.5 0 113.536 3.536L6.5 21.036H3v-3.572L16.732 3.732z"
/>
</svg>
</button>
<button
type="button"
data-analytics-id="chat-sidebar-delete-thread"
onClick={e => {
e.stopPropagation();
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 });
}
void dispatch(deleteThread(thread.id));
},
onCancel: () => {},
});
}}
className="ml-2 p-1 rounded opacity-0 group-hover:opacity-100 hover:bg-surface-strong dark:bg-surface-muted dark:hover:bg-surface-muted text-content-faint hover:text-coral-500 transition-all flex-shrink-0"
title={t('chat.deleteThread')}>
<svg className="w-3 h-3" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M6 18L18 6M6 6l12 12"
/>
</svg>
</button>
</div>
{/* <div className="flex items-center gap-2 mt-0.5">
<span className="text-[10px] text-content-faint">
{formatRelativeTime(thread.lastMessageAt)}
</span>
{thread.messageCount > 0 && (
<span className="text-[10px] text-content-faint">
{thread.messageCount} msg{thread.messageCount !== 1 ? 's' : ''}
</span>
)}
</div> */}
</div>
))
)}
</div>
</div>
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 (
<Fragment key={msg.id}>
{/* Past-turn process trail (Phase 5): each older settled turn's
tool timeline, collapsed, above the answer it produced. */}
{pastTurnEntries ? (
<div data-testid="past-turn-insights">
<ToolTimelineBlock entries={pastTurnEntries} />
</div>
) : null}
<div>
<div
className={`group/msg flex ${msg.sender === 'user' ? 'justify-end' : 'justify-start'}`}>
@@ -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 = () => <Conversations variant="page" />;
export const ConversationsPage = () => <Conversations variant="page" />;
+15
View File
@@ -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';
@@ -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<HTMLInputElement | null>;
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.
<div className="h-full flex flex-col">
{/* Thread search — flush full-width input, mirrors the settings search. */}
<div className="relative border-b border-line-subtle">
<span className="pointer-events-none absolute inset-y-0 left-3 flex items-center text-content-faint">
<svg className="h-4 w-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M21 21l-4.35-4.35M11 19a8 8 0 100-16 8 8 0 000 16z"
/>
</svg>
</span>
<input
type="text"
value={search}
onChange={e => 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 && (
<button
type="button"
onClick={() => onSearchChange('')}
aria-label={t('settings.settingsSearch.clear')}
data-testid="chat-thread-search-clear"
className="absolute inset-y-0 right-2 flex items-center px-1 text-content-faint hover:text-content-secondary">
<svg className="h-4 w-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M6 18L18 6M6 6l12 12"
/>
</svg>
</button>
)}
</div>
{/* New conversation — a subtle, centered thread-style row (not a loud
button), below the search and above the thread list. */}
<button
type="button"
data-testid="new-thread-button"
data-analytics-id="chat-sidebar-new-thread"
onClick={onCreateThread}
title={t('chat.newThreadShortcut')}
className="group w-full cursor-pointer border-b border-line-subtle/60 opacity-50 px-3 py-2 transition-colors hover:bg-surface-hover dark:border-line/60">
<div className="flex items-center justify-center gap-1.5">
<svg
className="h-3.5 w-3.5 flex-shrink-0 text-content-muted"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M12 4v16m8-8H4" />
</svg>
<span className="truncate text-xs text-content-secondary">
{t('chat.newConversation')}
</span>
</div>
</button>
<div className="flex-1 overflow-y-auto">
{threads.length === 0 ? (
<p className="px-4 py-6 text-xs text-content-faint text-center">{t('chat.noThreads')}</p>
) : (
threads.map(thread => (
<div
key={thread.id}
data-testid={`thread-row-${thread.id}`}
data-analytics-id="chat-sidebar-thread-row"
role="button"
tabIndex={0}
onClick={() => 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'
}`}>
<div className="flex items-center justify-between">
{editingThreadId === thread.id ? (
<input
ref={editTitleInputRef}
value={editTitleValue}
onClick={e => 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
/>
) : (
<p
className={`text-xs truncate flex-1 ${
selectedThreadId === thread.id
? 'font-medium text-primary-700 dark:text-primary-200'
: 'text-content-secondary'
}`}>
{resolveTitle(thread.id)}
</p>
)}
<button
type="button"
data-analytics-id="chat-sidebar-edit-thread-title"
onClick={e => {
e.stopPropagation();
onStartEditTitle(thread.id);
}}
aria-label={t('chat.editThreadTitle')}
title={t('chat.editThreadTitle')}
className="ml-2 p-1 rounded opacity-0 group-hover:opacity-100 hover:bg-surface-strong dark:bg-surface-muted dark:hover:bg-surface-muted text-content-faint hover:text-primary-500 transition-all flex-shrink-0">
<svg className="w-3 h-3" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M15.232 5.232l3.536 3.536m-2.036-5.036a2.5 2.5 0 113.536 3.536L6.5 21.036H3v-3.572L16.732 3.732z"
/>
</svg>
</button>
<button
type="button"
data-analytics-id="chat-sidebar-delete-thread"
onClick={e => {
e.stopPropagation();
onRequestDelete(thread);
}}
className="ml-2 p-1 rounded opacity-0 group-hover:opacity-100 hover:bg-surface-strong dark:bg-surface-muted dark:hover:bg-surface-muted text-content-faint hover:text-coral-500 transition-all flex-shrink-0"
title={t('chat.deleteThread')}>
<svg className="w-3 h-3" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M6 18L18 6M6 6l12 12"
/>
</svg>
</button>
</div>
</div>
))
)}
</div>
</div>
);
}
@@ -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(<Provider store={store}>{ui}</Provider>);
}
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<Parameters<typeof buildThreadTimeline>[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(<ConversationTimeline items={items} />);
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(<ConversationTimeline items={items} />);
// The two consecutive tool items coalesce into one ToolTimelineBlock (one
// `<details>` 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(<ConversationTimeline items={items} handlers={{ onOpenSubagent }} />);
// 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(<ConversationTimeline items={items} />);
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(<ConversationTimeline items={items} />);
expect(screen.getByTestId('stream-branch')).toBeTruthy();
expect(screen.getByText('branch answer')).toBeTruthy();
});
it('renders nothing but the container for an empty timeline', () => {
renderInStore(<ConversationTimeline items={[]} />);
const container = screen.getByTestId('conversation-timeline');
expect(container.children).toHaveLength(0);
});
});
@@ -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(
<ToolTimelineBlock
key={`process:${startId}`}
entries={run}
onViewSubagent={handlers.onOpenSubagent}
onViewWholeRun={handlers.onViewWholeRun}
/>
);
i = j - 1;
continue;
}
switch (item.kind) {
case 'userMessage':
rendered.push(<UserMessageItem key={item.id} content={item.message.content} />);
break;
case 'assistantMessage':
rendered.push(
<AssistantMessageItem
key={item.id}
content={item.message.content}
viewMode={agentMessageViewMode}
/>
);
break;
case 'streamingText':
rendered.push(
<StreamingTailItem
key={item.id}
text={item.text}
thinking={item.thinking}
branch={item.branch}
/>
);
break;
case 'reasoning':
// Reasoning is rendered inside the process/streaming affordances today;
// no standalone element yet.
break;
default:
break;
}
}
return (
<div data-testid="conversation-timeline" className="space-y-3">
{rendered}
</div>
);
}
@@ -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 <AgentMessageText content={content} />;
}
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 (
<Fragment key={index}>
<AgentMessageBubble content={segment} position={position} />
</Fragment>
);
})}
</>
);
}
@@ -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 (
<div className="flex justify-start" data-testid={branch ? 'stream-branch' : 'stream-primary'}>
<div className="max-w-[80%] space-y-1">
{thinking && thinking.length > 0 ? (
<details className="text-xs text-content-muted">
<summary className="cursor-pointer select-none">Thinking</summary>
<p className="whitespace-pre-wrap font-mono">
{thinking.slice(-STREAMING_PREVIEW_CHARS)}
</p>
</details>
) : null}
<div className="rounded-2xl bg-surface-subtle px-4 py-2 font-mono text-sm text-content-secondary">
{truncated ? '…' : ''}
{tail}
</div>
</div>
</div>
);
}
@@ -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 (
<div className="flex justify-end">
<div className="max-w-[80%] rounded-2xl bg-primary-500 px-4 py-2 text-sm text-white">
<BubbleMarkdown content={content} />
</div>
</div>
);
}
@@ -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<string, unknown> = {}
): 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<string, unknown> = {}
): 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<Parameters<typeof buildThreadTimeline>[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<RootState['chatRuntime']> = {},
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([]);
});
});
@@ -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,
})
);
@@ -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:');
}
+1 -1
View File
@@ -19,7 +19,7 @@ import HumanPage from './HumanPage';
// ── Heavy dependency stubs ────────────────────────────────────────────────
vi.mock('../../pages/Conversations', () => ({
vi.mock('../conversations/Conversations', () => ({
default: () => <div data-testid="conversations-stub" />,
}));
+1 -1
View File
@@ -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,
+2 -2
View File
@@ -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 = () => {
<main className="relative flex min-w-0 flex-1 flex-col overflow-hidden">
{isAgentSelected && (
<div className="min-h-0 flex-1 overflow-hidden">
<AgentChatPanel />
<ConversationsPage />
</div>
)}
</main>
@@ -48,9 +48,9 @@ vi.mock('../../store/mascotSlice', () => ({
selectCustomSecondaryColor: () => '#111111',
selectMascotColor: () => 'blue',
}));
vi.mock('../Conversations', () => ({
vi.mock('../../features/conversations/Conversations', () => ({
default: ({ variant }: { variant: string }) => <div data-testid="conversations">{variant}</div>,
AgentChatPanel: () => <div data-testid="agent-chat-panel" />,
ConversationsPage: () => <div data-testid="agent-chat-panel" />,
}));
describe('Accounts provider selection', () => {
@@ -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(
<Provider store={store}>
@@ -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(
<Provider store={store}>
@@ -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(
<Provider store={store}>
@@ -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(
<Provider store={store}>
@@ -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(
<Provider store={store}>
@@ -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> = {}): Thread {
async function renderConversations(preload: Record<string, unknown> = {}) {
const store = buildStore(preload);
const { default: Conversations } = await import('../Conversations');
const { default: Conversations } = await import('../../features/conversations/Conversations');
render(
<Provider store={store}>
@@ -241,7 +243,7 @@ async function renderConversations(preload: Record<string, unknown> = {}) {
async function renderConversationsRoute(route: string, preload: Record<string, unknown> = {}) {
const store = buildStore(preload);
const { default: Conversations } = await import('../Conversations');
const { default: Conversations } = await import('../../features/conversations/Conversations');
render(
<Provider store={store}>
@@ -272,7 +274,7 @@ async function renderEmbeddedConversationsRoute(
preload: Record<string, unknown> = {}
) {
const store = buildStore(preload);
const { default: Conversations } = await import('../Conversations');
const { default: Conversations } = await import('../../features/conversations/Conversations');
render(
<Provider store={store}>
@@ -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<typeof buildStore> | 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(
<Provider store={store}>
@@ -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<string, unknown> = {}) {
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 () => {
@@ -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', () => {
@@ -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
+2 -2
View File
@@ -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.
+32 -5
View File
@@ -207,7 +207,13 @@ function hasCompleteSegmentDelivery(
}
function chatDoneExtraMetadata(event: ChatDoneEvent): Record<string, unknown> | 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<string, unknown> = {};
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', {
@@ -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();
+27
View File
@@ -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<PersistedTurnState[]> => {
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<PersistedTurnState | null> => {
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<boolean> => {
const response = await callCoreRpc<{ data?: ClearTurnStateResponse }>({
method: 'openhuman.threads_turn_state_clear',
@@ -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();
});
});
+60
View File
@@ -565,6 +565,15 @@ interface ChatRuntimeState {
*/
parallelRequestThreads: Record<string, string>;
toolTimelineByThread: Record<string, ToolTimelineEntry[]>;
/**
* 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<string, Record<string, ToolTimelineEntry[]>>;
/**
* 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<string, ToolTimelineEntry[]> }>
) => {
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<string, ToolTimelineEntry[]> = {};
// 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;
@@ -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, L17751842).
**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`): `<ConversationsPage>`, `<ConversationsSidebar composer projectThreadList>`, and `<ConversationTimeline items onAction>` 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/<hex(thread_id)>/<request_id>.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 + `<ConversationTimeline>`
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 (~L24002700 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 13)
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<threadId, Record<requestId, rows>>` + `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 45 |
| 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)
+33 -2
View File
@@ -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<RpcOutcome<ApiEnvelope<ListTurnStatesResponse>>, 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<RpcOutcome<ApiEnvelope<GetTurnStateResponse>>, 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(
+70 -1
View File
@@ -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<ControllerSchema> {
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<RegisteredController> {
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<String, Value>) -> ControllerFuture {
Box::pin(async move { to_json(ops::turn_state_list(EmptyRequest {}).await?) })
}
fn handle_turn_state_history(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = parse::<GetTurnStateRequest>(params)?;
to_json(ops::turn_state_history(p).await?)
})
}
fn handle_turn_state_get_turn(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = parse::<GetTurnStateForRequestRequest>(params)?;
to_json(ops::turn_state_get_turn(p).await?)
})
}
fn handle_turn_state_clear(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = parse::<ClearTurnStateRequest>(params)?;
+2
View File
@@ -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",
+3 -3
View File
@@ -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,
};
+371 -77
View File
@@ -1,13 +1,27 @@
//! Filesystem-backed snapshot store for [`super::types::TurnState`].
//!
//! One JSON file per thread under
//! `<workspace>/memory/conversations/turn_states/<hex(thread_id)>.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:
//! `<workspace>/memory/conversations/turn_states/<hex(thread_id)>/<hex(request_id)>.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/<hex(thread_id)>.json`. They are migrated in place — read once,
//! rewritten under `<hex(thread_id)>/<hex(request_id)>.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<Mutex<()>> = 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<Option<TurnState>, 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<Option<TurnState>, 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<bool, String> {
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<Vec<TurnState>, 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<Vec<TurnState>, 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<usize, String> {
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<usize, String> {
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<PathBuf, String> {
let dir = self.dir();
// --- internals -------------------------------------------------------
fn ensure_thread_dir(&self, thread_id: &str) -> Result<PathBuf, String> {
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<Vec<TurnState>, 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<Vec<String>, 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<Vec<TurnState>, 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<TurnState> = 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<TurnState>) -> Option<TurnState> {
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<Option<TurnState>,
TurnStateStore::new(workspace_dir).get(thread_id)
}
pub fn get_turn(
workspace_dir: PathBuf,
thread_id: &str,
request_id: &str,
) -> Result<Option<TurnState>, String> {
TurnStateStore::new(workspace_dir).get_turn(thread_id, request_id)
}
pub fn delete(workspace_dir: PathBuf, thread_id: &str) -> Result<bool, String> {
TurnStateStore::new(workspace_dir).delete(thread_id)
}
@@ -258,6 +548,10 @@ pub fn list(workspace_dir: PathBuf) -> Result<Vec<TurnState>, String> {
TurnStateStore::new(workspace_dir).list()
}
pub fn list_thread(workspace_dir: PathBuf, thread_id: &str) -> Result<Vec<TurnState>, String> {
TurnStateStore::new(workspace_dir).list_thread(thread_id)
}
pub fn clear_all(workspace_dir: PathBuf) -> Result<usize, String> {
TurnStateStore::new(workspace_dir).clear_all()
}
@@ -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 `<hex(thread_id)>.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");
+11 -1
View File
@@ -335,7 +335,8 @@ pub struct GetTurnStateResponse {
pub turn_state: Option<TurnState>,
}
/// 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)]
+71
View File
@@ -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,