diff --git a/app/src/features/conversations/Conversations.tsx b/app/src/features/conversations/Conversations.tsx index 2f42ed9427..243f234dc9 100644 --- a/app/src/features/conversations/Conversations.tsx +++ b/app/src/features/conversations/Conversations.tsx @@ -46,6 +46,8 @@ import { } from '../../features/conversations/components/ThreadGoalChip'; import { ThreadTodoStrip } from '../../features/conversations/components/ThreadTodoStrip'; import { ToolTimelineBlock } from '../../features/conversations/components/ToolTimelineBlock'; +import { InterruptedAnswer } from '../../features/conversations/components/InterruptedAnswer'; +import { PastTurnInsights } from '../../features/conversations/components/PastTurnInsights'; import { evaluateComposerSend, getComposerBlockedSendFeedback, @@ -106,6 +108,7 @@ import { hydrateThreadUsage, markSubagentCancelled, markThreadSendPending, + type ProcessingTranscriptItem, type QueuedFollowup, registerParallelRequest, setTaskBoardForThread, @@ -239,6 +242,16 @@ const EMPTY_QUEUED_FOLLOWUPS: Record = {}; // Stable empty reference for the per-thread past-turn timelines map, so the // derived value keeps the same identity when the slice field is absent. const EMPTY_TURN_TIMELINES: Record = {}; +// Sibling stable empty for the per-thread past-turn processing transcripts map +// (restore-fidelity fix 1). +const EMPTY_TURN_TRANSCRIPTS: Record = {}; +// Stable empty transcript for a past turn that has tool rows but no persisted +// reasoning/narration trail (legacy snapshot), so `PastTurnInsights` falls back +// to the tool-only view without allocating a fresh array each render. +const EMPTY_TRANSCRIPT: ProcessingTranscriptItem[] = []; +// Stable empty tool-row list for a transcript-only past turn (agent thought / +// narrated but ran no tools). +const EMPTY_TRANSCRIPT_ENTRIES: ToolTimelineEntry[] = []; export function isComposerInteractionBlocked(args: { /** Whether the *currently selected* thread has an in-flight inference turn. */ @@ -398,6 +411,12 @@ const Conversations = ({ const uiLocale = useAppSelector(state => state.locale?.current ?? 'en'); const toolTimelineByThread = useAppSelector(state => state.chatRuntime.toolTimelineByThread); const turnTimelinesByThread = useAppSelector(state => state.chatRuntime.turnTimelinesByThread); + const turnTranscriptsByThread = useAppSelector( + state => state.chatRuntime.turnTranscriptsByThread + ); + const interruptedAssistantByThread = useAppSelector( + state => state.chatRuntime.interruptedAssistantByThread + ); const processingByThread = useAppSelector(state => state.chatRuntime.processingByThread); const taskBoardByThread = useAppSelector(state => state.chatRuntime.taskBoardByThread); const inferenceStatusByThread = useAppSelector( @@ -1834,21 +1853,33 @@ const Conversations = ({ const selectedThreadTurnTimelines = selectedThreadId ? (turnTimelinesByThread[selectedThreadId] ?? EMPTY_TURN_TIMELINES) : EMPTY_TURN_TIMELINES; + // Sibling map: each past turn's persisted reasoning/narration trail, so a + // reopened turn replays its thoughts, not just its tool cards (fix 1). + const selectedThreadTurnTranscripts = selectedThreadId + ? (turnTranscriptsByThread[selectedThreadId] ?? EMPTY_TURN_TRANSCRIPTS) + : EMPTY_TURN_TRANSCRIPTS; const pastTurnAnchors = useMemo(() => { - const anchors: Record = {}; + const anchors: Record< + string, + { entries: ToolTimelineEntry[]; transcript: ProcessingTranscriptItem[] } + > = {}; const seen = new Set(); for (const msg of timelineMessages) { if (msg.sender !== 'agent') continue; const requestId = msg.extraMetadata?.requestId; if (typeof requestId !== 'string' || seen.has(requestId)) continue; - const entries = selectedThreadTurnTimelines[requestId]; - if (entries && entries.length > 0) { - anchors[msg.id] = entries; + const entries = selectedThreadTurnTimelines[requestId] ?? EMPTY_TRANSCRIPT_ENTRIES; + const transcript = selectedThreadTurnTranscripts[requestId] ?? EMPTY_TRANSCRIPT; + // Anchor the turn when it has EITHER tool rows OR a reasoning/narration + // trail — a tool-less turn (agent only thought/narrated) must still + // render its restored thoughts above its answer (fix 1). + if (entries.length > 0 || transcript.length > 0) { + anchors[msg.id] = { entries, transcript }; seen.add(requestId); } } return anchors; - }, [timelineMessages, selectedThreadTurnTimelines]); + }, [timelineMessages, selectedThreadTurnTimelines, selectedThreadTurnTranscripts]); const activeSubagentTimelineEntry = selectedThreadToolTimeline.find( entry => entry.status === 'running' && entry.name.startsWith('subagent:') ); @@ -1861,6 +1892,12 @@ const Conversations = ({ const selectedStreamingAssistant = selectedThreadId ? (streamingAssistantByThread[selectedThreadId] ?? null) : null; + // The partial reply an interrupted turn left behind (restore-fidelity fix 2): + // surfaced as a settled, marked-interrupted bubble on restore so a turn that + // crashed mid-answer keeps its visible work instead of rendering blank. + const selectedInterruptedAssistant = selectedThreadId + ? (interruptedAssistantByThread[selectedThreadId] ?? null) + : null; // Live streams for concurrent parallel (forked) turns on the selected thread, // rendered as separate interleaved branch bubbles. const selectedParallelStreams = selectedThreadId @@ -1928,7 +1965,10 @@ const Conversations = ({ isSending || selectedThreadToolTimeline.length > 0 || selectedThreadProcessing.length > 0 || - Boolean(selectedStreamingAssistant); + Boolean(selectedStreamingAssistant) || + // An interrupted turn's restored partial answer must surface too, even + // before the durable message history loads (restore-fidelity fix 2). + Boolean(selectedInterruptedAssistant); // Anchor the "Agentic task insights" panel right after the latest turn's user // message — processing happens *before* the answer, so it reads above the @@ -2259,14 +2299,20 @@ 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]; + const pastTurn = pastTurnAnchors[msg.id]; return ( - {/* Past-turn process trail (Phase 5): each older settled turn's - tool timeline, collapsed, above the answer it produced. */} - {pastTurnEntries ? ( + {/* Past-turn process trail (Phase 5 + restore-fidelity fix 1): + each older settled turn's interleaved reasoning/narration + + tool steps (and restored sub-agent transcripts), collapsed, + above the answer it produced. Falls back to tool-cards-only + for legacy snapshots with no persisted transcript. */} + {pastTurn ? (
- +
) : null}
@@ -2661,6 +2707,16 @@ const Conversations = ({
)} + {/* Interrupted turn's partial answer (restore-fidelity fix 2): + a settled, marked-interrupted bubble surfaced on restore. Only + when NOT streaming live (the buffer is cleared by any live turn + in the slice; this guard is belt-and-braces). */} + {!isSending && selectedInterruptedAssistant ? ( + + ) : null} {/* Parallel (forked) branch streams — concurrent turns on this thread, each its own labeled bubble so they don't collide with the primary stream above. */} diff --git a/app/src/features/conversations/components/InterruptedAnswer.test.tsx b/app/src/features/conversations/components/InterruptedAnswer.test.tsx new file mode 100644 index 0000000000..ccd450197a --- /dev/null +++ b/app/src/features/conversations/components/InterruptedAnswer.test.tsx @@ -0,0 +1,30 @@ +import { render, screen } from '@testing-library/react'; +import { Provider } from 'react-redux'; +import { describe, expect, it } from 'vitest'; + +import { store } from '../../../store'; +import { InterruptedAnswer } from './InterruptedAnswer'; + +function renderInStore(ui: React.ReactNode) { + return render({ui}); +} + +describe('InterruptedAnswer', () => { + it('surfaces the partial reply with an interrupted marker', () => { + renderInStore( + + ); + + const block = screen.getByTestId('interrupted-answer'); + expect(block.textContent).toContain('Here is the partial answer'); + // The reasoning it had streamed is kept in a collapsed block. + expect(block.textContent).toContain('was reasoning about it'); + // Marked interrupted rather than presented as a finished answer. + expect(screen.getByTestId('interrupted-answer-marker')).toBeTruthy(); + }); + + it('renders nothing when neither content nor thinking has any text', () => { + const { container } = renderInStore(); + expect(container.querySelector('[data-testid="interrupted-answer"]')).toBeNull(); + }); +}); diff --git a/app/src/features/conversations/components/InterruptedAnswer.tsx b/app/src/features/conversations/components/InterruptedAnswer.tsx new file mode 100644 index 0000000000..5989bf81b2 --- /dev/null +++ b/app/src/features/conversations/components/InterruptedAnswer.tsx @@ -0,0 +1,59 @@ +import { useT } from '../../../lib/i18n/I18nContext'; +import { BubbleMarkdown } from './AgentMessageBubble'; + +/** + * The partial assistant reply left behind by an INTERRUPTED turn — the core + * process that was streaming it exited before `chat_done`, so no final message + * was ever committed to the durable thread log (restore-fidelity fix 2). + * + * Unlike the live streaming preview, this is a SETTLED buffer: it renders as a + * static agent bubble (no pulsing cursor, full Markdown like a finished answer) + * carrying an "Interrupted" marker, plus the hidden reasoning it had streamed in + * a collapsed block. It is deliberately NOT written into the durable message + * list — it is a restore-time surfacing of what the agent had produced, so the + * user sees the partial work instead of a blank turn. + */ +export function InterruptedAnswer({ + content, + thinking, +}: { + content: string; + thinking: string; +}) { + const { t } = useT(); + const trimmedContent = content.trim(); + const trimmedThinking = thinking.trim(); + // Nothing persisted to show — render nothing rather than an empty marked bubble. + if (!trimmedContent && !trimmedThinking) return null; + + return ( +
+
+ {trimmedThinking ? ( +
+ + + 💭 + + {t('chat.thinking')} + + 
+              {trimmedThinking}
+
+
+ ) : null} +
+
+ + ⚠ + + {t('intelligence.agentWork.status.interrupted')} +
+ {trimmedContent ? : null} +
+
+
+ ); +} diff --git a/app/src/features/conversations/components/PastTurnInsights.test.tsx b/app/src/features/conversations/components/PastTurnInsights.test.tsx new file mode 100644 index 0000000000..ef0314b83b --- /dev/null +++ b/app/src/features/conversations/components/PastTurnInsights.test.tsx @@ -0,0 +1,70 @@ +import { render, screen } from '@testing-library/react'; +import { Provider } from 'react-redux'; +import { describe, expect, it } from 'vitest'; + +import { store } from '../../../store'; +import type { ProcessingTranscriptItem, ToolTimelineEntry } from '../../../store/chatRuntimeSlice'; +import { PastTurnInsights } from './PastTurnInsights'; + +function renderInStore(ui: React.ReactNode) { + return render({ui}); +} + +describe('PastTurnInsights', () => { + it('replays a restored turn as interleaved thoughts + tool rows, not just tool cards', () => { + // A reopened past turn carries both its reasoning trail and its tools. + const entries: ToolTimelineEntry[] = [ + { id: 'c1', name: 'read_file', round: 0, seq: 0, status: 'success' }, + ]; + const transcript: ProcessingTranscriptItem[] = [ + { kind: 'thinking', round: 0, seq: 1, text: 'planning the search' }, + { kind: 'toolCall', round: 0, seq: 2, callId: 'c1' }, + ]; + + renderInStore(); + + // The hidden reasoning replays (fix 1) — a restored turn is no longer + // tool-cards-only. + expect(screen.getByTestId('processing-thinking').textContent).toContain('planning the search'); + // And its tool step still renders in the interleaved view. + expect(screen.getAllByTestId('processing-tool-row').length).toBeGreaterThan(0); + }); + + it('renders restored sub-agent transcripts beneath the trail', () => { + const entries: ToolTimelineEntry[] = [ + { + id: 'subagent:task-y', + name: 'subagent:researcher', + round: 0, + seq: 0, + status: 'success', + subagent: { + taskId: 'task-y', + agentId: 'researcher', + toolCalls: [], + transcript: [{ kind: 'thinking', iteration: 1, text: 'child reasoning trail' }], + }, + }, + ]; + const transcript: ProcessingTranscriptItem[] = [ + { kind: 'narration', round: 0, seq: 0, text: 'delegating to a researcher' }, + ]; + + renderInStore(); + + const subagents = screen.getByTestId('past-turn-subagents'); + expect(subagents.textContent).toContain('child reasoning trail'); + }); + + it('falls back to the tool-only timeline for a legacy turn with no transcript', () => { + const entries: ToolTimelineEntry[] = [ + { id: 'c1', name: 'read_file', round: 0, seq: 0, status: 'success' }, + ]; + + renderInStore(); + + // The interleaved transcript view is absent; the tool-only block renders. + expect(screen.queryByTestId('processing-transcript')).toBeNull(); + expect(screen.getByTestId('agent-task-insights')).toBeTruthy(); + }); +}); diff --git a/app/src/features/conversations/components/PastTurnInsights.tsx b/app/src/features/conversations/components/PastTurnInsights.tsx new file mode 100644 index 0000000000..28d2e5c6a2 --- /dev/null +++ b/app/src/features/conversations/components/PastTurnInsights.tsx @@ -0,0 +1,60 @@ +import type { ProcessingTranscriptItem, ToolTimelineEntry } from '../../../store/chatRuntimeSlice'; +import { formatTimelineEntry } from '../../../utils/toolTimelineFormatting'; +import { ProcessingTranscriptView } from './ProcessingTranscriptView'; +import { SubagentActivityBlock, ToolTimelineBlock } from './ToolTimelineBlock'; + +/** + * The collapsed process trail rendered above a PAST (settled) turn's answer on a + * reopened thread — restore-fidelity fix 1. + * + * A restored turn must show the *same* interleaved thoughts + tools view a live + * turn does, not just its tool cards: the persisted `transcript` + * (narration/thinking/tool pointers, hydrated into + * `turnTranscriptsByThread`) drives {@link ProcessingTranscriptView}, so the + * reasoning and narration replay inline exactly where they streamed. Restored + * sub-agent transcripts (fix 4) render beneath as their own activity blocks, + * mirroring the whole-run {@link AgentProcessSourcePanel} body. + * + * Legacy turns persisted before the transcript field existed have no + * `transcript` — they fall back to the tool-only {@link ToolTimelineBlock}, so + * older threads keep rendering unchanged. + */ +export function PastTurnInsights({ + entries, + transcript, +}: { + entries: ToolTimelineEntry[]; + transcript: ProcessingTranscriptItem[]; +}) { + // No reasoning/narration trail persisted (legacy snapshot): render the + // tool-only timeline, which already nests each sub-agent's activity inline. + if (transcript.length === 0) { + return ; + } + + const subagentEntries = entries.filter(entry => entry.subagent); + + return ( +
+ {/* Interleaved narration + hidden reasoning + grouped tool steps. */} + + + {/* Sub-agents — each delegated agent's restored transcript (thoughts + + tool rows), so a reopened turn keeps its sub-agent reasoning, not just + a flat tool row. `ProcessingTranscriptView` shows the spawn as a tool + row but does not nest the child's activity, so surface it here. */} + {subagentEntries.length > 0 ? ( +
+ {subagentEntries.map(entry => ( +
+

+ {formatTimelineEntry(entry).title} +

+ +
+ ))} +
+ ) : null} +
+ ); +} diff --git a/app/src/features/conversations/timeline/ConversationTimeline.tsx b/app/src/features/conversations/timeline/ConversationTimeline.tsx index b83b3b3c1a..1014172f1a 100644 --- a/app/src/features/conversations/timeline/ConversationTimeline.tsx +++ b/app/src/features/conversations/timeline/ConversationTimeline.tsx @@ -85,10 +85,6 @@ export function ConversationTimeline({ /> ); break; - case 'reasoning': - // Reasoning is rendered inside the process/streaming affordances today; - // no standalone element yet. - break; default: break; } diff --git a/app/src/features/conversations/timeline/types.ts b/app/src/features/conversations/timeline/types.ts index c2c71aaf4e..c495b55859 100644 --- a/app/src/features/conversations/timeline/types.ts +++ b/app/src/features/conversations/timeline/types.ts @@ -50,7 +50,6 @@ export type TimelineItem = TimelineItemBase & /** True for a `parallelStreamsByThread` forked branch. */ branch: boolean; } - | { kind: 'reasoning'; text: string; settled: boolean } | { kind: 'toolCall'; /** The underlying runtime row (rendered via `ToolTimelineBlock`). */ @@ -75,11 +74,7 @@ export type TimelineItem = TimelineItemBase & export type TimelineItemKind = TimelineItem['kind']; /** The "agent process" kinds that `hideAgentInsights` suppresses. */ -export const AGENT_INSIGHT_KINDS: readonly TimelineItemKind[] = [ - 'toolCall', - 'subagentActivity', - 'reasoning', -]; +export const AGENT_INSIGHT_KINDS: readonly TimelineItemKind[] = ['toolCall', 'subagentActivity']; /** A contiguous group of items sharing a `turnId`, in render order. */ export interface TimelineTurn { diff --git a/app/src/services/chatService.ts b/app/src/services/chatService.ts index f6b25e34d6..dc3571a94a 100644 --- a/app/src/services/chatService.ts +++ b/app/src/services/chatService.ts @@ -17,6 +17,8 @@ const chatLog = debug('realtime:chat'); export interface ChatToolCallEvent { thread_id: string; request_id?: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; tool_name: string; skill_id: string; args: Record; @@ -42,6 +44,8 @@ export interface ChatToolCallEvent { export interface ChatToolResultEvent { thread_id: string; request_id?: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; tool_name: string; skill_id: string; output: string; @@ -85,6 +89,8 @@ export interface TurnUsageWire { export interface ChatDoneEvent { thread_id: string; request_id?: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; full_response: string; rounds_used: number; /** @@ -126,6 +132,8 @@ export interface ChatSegmentEvent { */ full_response: string; request_id: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; segment_index: number; segment_total: number; reaction_emoji?: string | null; @@ -147,6 +155,8 @@ export function segmentText(event: ChatSegmentEvent): string { export interface ChatInterimEvent { thread_id: string; request_id: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; /** Wire name is `full_response`; carries only this round's narration text. */ full_response: string; round: number; @@ -463,6 +473,8 @@ export interface ChatSubagentToolResultEvent { export interface ChatSubagentTextDeltaEvent { thread_id: string; request_id: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; /** Parent iteration index (inherited from the parent context). */ round: number; /** Text fragment from the sub-agent. */ @@ -478,6 +490,8 @@ export interface ChatSubagentTextDeltaEvent { export interface ChatSubagentThinkingDeltaEvent { thread_id: string; request_id: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; round: number; delta: string; subagent?: SubagentProgressDetail; @@ -491,6 +505,8 @@ export interface ChatSubagentThinkingDeltaEvent { export interface ChatTextDeltaEvent { thread_id: string; request_id: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; /** 1-based iteration index the chunk belongs to. */ round: number; /** Text fragment; may be a single token or a few characters. */ @@ -506,6 +522,8 @@ export interface ChatTextDeltaEvent { export interface ChatThinkingDeltaEvent { thread_id: string; request_id: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; round: number; delta: string; } @@ -519,6 +537,8 @@ export interface ChatThinkingDeltaEvent { export interface ChatToolArgsDeltaEvent { thread_id: string; request_id: string; + /** Per-request monotonic ordering key stamped by the core progress bridge. */ + seq?: number; round: number; tool_call_id: string; tool_name: string; diff --git a/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts b/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts index 4db54661ca..fb1cbc4495 100644 --- a/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts +++ b/app/src/store/__tests__/chatRuntimeSlice.thunk.test.ts @@ -47,7 +47,8 @@ describe('fetchAndHydrateTurnHistory', () => { requestId: string, lifecycle: string, startedAt: string, - tools: number + tools: number, + transcript?: Array> ) => ({ threadId: 'thread-hist', requestId, @@ -62,6 +63,7 @@ describe('fetchAndHydrateTurnHistory', () => { round: 0, status: 'success' as const, })), + ...(transcript ? { transcript } : {}), startedAt, updatedAt: startedAt, }); @@ -87,10 +89,43 @@ describe('fetchAndHydrateTurnHistory', () => { expect(timelines['req-0']).toBeUndefined(); }); + it('keeps each past turn\'s reasoning/narration transcript (fix 1), ordered by seq (fix 5)', async () => { + const store = configureStore({ reducer }); + mockThreadApi.getTurnStateHistory.mockResolvedValueOnce([ + // req-latest is skipped (index 0). + persistedTurn('req-latest', 'completed', '2026-06-04T13:00:00Z', 1), + // req-a: has tools AND a transcript, deliberately out of seq order. + persistedTurn('req-a', 'completed', '2026-06-04T12:00:00Z', 1, [ + { kind: 'narration', round: 0, seq: 1, text: 'second' }, + { kind: 'thinking', round: 0, seq: 0, text: 'first' }, + ]), + // req-b: a tool-LESS turn that only thought/narrated — must still be kept + // for its transcript rather than dropped. + persistedTurn('req-b', 'completed', '2026-06-04T11:00:00Z', 0, [ + { kind: 'thinking', round: 0, seq: 0, text: 'only thinking, no tools' }, + ]), + ]); + + await store.dispatch(fetchAndHydrateTurnHistory('thread-hist')); + + const transcripts = store.getState().turnTranscriptsByThread['thread-hist']; + expect(Object.keys(transcripts).sort()).toEqual(['req-a', 'req-b']); + // Ordered by persisted `seq`, not wire order. + expect(transcripts['req-a'].map(i => ('text' in i ? i.text : undefined))).toEqual([ + 'first', + 'second', + ]); + // The tool-less turn is retained purely for its transcript. + const timelines = store.getState().turnTimelinesByThread['thread-hist']; + expect(timelines['req-b']).toBeUndefined(); + expect(transcripts['req-b']).toHaveLength(1); + }); + 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(); + expect(store.getState().turnTranscriptsByThread['t']).toBeUndefined(); }); }); diff --git a/app/src/store/chatRuntimeSlice.test.ts b/app/src/store/chatRuntimeSlice.test.ts index 316c7da7fd..396ab8d67f 100644 --- a/app/src/store/chatRuntimeSlice.test.ts +++ b/app/src/store/chatRuntimeSlice.test.ts @@ -909,3 +909,150 @@ describe('hydrateRuntimeFromSnapshot — persisted tool result output', () => { expect(timeline[1].subagent?.toolCalls[0]?.result).toBe('file body'); }); }); + +describe('hydrateRuntimeFromSnapshot — interrupted partial answer (fix 2)', () => { + function makeInterruptedPartialSnapshot( + threadId: string, + over: Partial = {} + ): PersistedTurnState { + return { + threadId, + requestId: 'req-int', + lifecycle: 'interrupted', + iteration: 2, + maxIterations: 10, + streamingText: 'Here is the partial ans', + thinking: 'was still reasoning', + toolTimeline: [], + startedAt: '2026-06-23T00:00:00Z', + updatedAt: '2026-06-23T00:00:00Z', + ...over, + }; + } + + it('surfaces the persisted partial reply + thinking as a settled buffer', () => { + const store = makeStore(); + store.dispatch(hydrateRuntimeFromSnapshot({ snapshot: makeInterruptedPartialSnapshot('t-int') })); + + const state = store.getState().chatRuntime; + expect(state.interruptedAssistantByThread['t-int']).toEqual({ + requestId: 'req-int', + content: 'Here is the partial ans', + thinking: 'was still reasoning', + }); + // It is NOT resurrected as a live streaming buffer (would pulse). + expect(state.streamingAssistantByThread['t-int']).toBeUndefined(); + // The lifecycle is recorded as interrupted, not a fake in-flight status. + expect(state.inferenceTurnLifecycleByThread['t-int']).toBe('interrupted'); + }); + + it('keeps an interrupted turn that only produced thinking', () => { + const store = makeStore(); + store.dispatch( + hydrateRuntimeFromSnapshot({ + snapshot: makeInterruptedPartialSnapshot('t-think', { streamingText: '' }), + }) + ); + expect(store.getState().chatRuntime.interruptedAssistantByThread['t-think']).toMatchObject({ + content: '', + thinking: 'was still reasoning', + }); + }); + + it('does not surface a partial for an interrupted turn with no persisted text', () => { + const store = makeStore(); + store.dispatch( + hydrateRuntimeFromSnapshot({ + snapshot: makeInterruptedPartialSnapshot('t-empty', { streamingText: '', thinking: '' }), + }) + ); + expect(store.getState().chatRuntime.interruptedAssistantByThread['t-empty']).toBeUndefined(); + }); + + it('clears a stale interrupted partial when a completed snapshot lands', () => { + const store = makeStore(); + store.dispatch(hydrateRuntimeFromSnapshot({ snapshot: makeInterruptedPartialSnapshot('t-c') })); + expect(store.getState().chatRuntime.interruptedAssistantByThread['t-c']).toBeDefined(); + + store.dispatch( + hydrateRuntimeFromSnapshot({ + snapshot: makeInterruptedPartialSnapshot('t-c', { + lifecycle: 'completed', + streamingText: '', + thinking: '', + }), + }) + ); + expect(store.getState().chatRuntime.interruptedAssistantByThread['t-c']).toBeUndefined(); + }); +}); + +describe('hydrateRuntimeFromSnapshot — transcript seq ordering (fix 5)', () => { + it('orders the processing transcript by persisted seq, not array order', () => { + const store = makeStore(); + const snapshot: PersistedTurnState = { + threadId: 't-seq', + requestId: 'req-1', + lifecycle: 'completed', + iteration: 1, + maxIterations: 10, + streamingText: '', + thinking: '', + toolTimeline: [], + // Deliberately out of order on the wire. + transcript: [ + { kind: 'narration', round: 0, seq: 2, text: 'second' }, + { kind: 'thinking', round: 0, seq: 0, text: 'first' }, + { kind: 'narration', round: 0, seq: 1, text: 'middle' }, + ], + startedAt: '2026-06-23T00:00:00Z', + updatedAt: '2026-06-23T00:00:00Z', + }; + + store.dispatch(hydrateRuntimeFromSnapshot({ snapshot })); + + const items = store.getState().chatRuntime.processingByThread['t-seq']; + expect(items.map(i => ('text' in i ? i.text : i.kind))).toEqual(['first', 'middle', 'second']); + }); +}); + +describe('hydrateRuntimeFromSnapshot — sub-agent transcript fallback (fix 4)', () => { + it('rebuilds a tool-only transcript when no persisted prose field is present', () => { + const store = makeStore(); + const snapshot: PersistedTurnState = { + threadId: 't-fallback', + requestId: 'req-1', + lifecycle: 'completed', + iteration: 1, + maxIterations: 10, + streamingText: '', + thinking: '', + toolTimeline: [ + { + id: 'subagent:task-old', + name: 'subagent:researcher', + round: 1, + status: 'success', + subagent: { + taskId: 'task-old', + agentId: 'researcher', + // No `transcript` field (old snapshot) — only tool calls. + toolCalls: [{ callId: 'c1', toolName: 'web_search', status: 'success' }], + }, + }, + ], + startedAt: '2026-06-23T00:00:00Z', + updatedAt: '2026-06-23T00:00:00Z', + }; + + store.dispatch(hydrateRuntimeFromSnapshot({ snapshot })); + + const row = store + .getState() + .chatRuntime.toolTimelineByThread['t-fallback'].find(e => e.subagent?.taskId === 'task-old'); + const transcript = row?.subagent?.transcript ?? []; + // Falls back to tool-only items so an old snapshot still shows the sequence. + expect(transcript).toHaveLength(1); + expect(transcript[0].kind).toBe('tool'); + }); +}); diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index c06fb27336..40122ac0a3 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -647,6 +647,31 @@ interface ChatRuntimeState { * rows live in `toolTimelineByThread` and are driven by the socket stream). */ turnTimelinesByThread: Record>; + /** + * Per-turn processing transcripts (narration / thinking / tool pointers) for + * *past* (settled) turns of a thread, keyed `threadId -> requestId -> items`. + * Sibling of {@link turnTimelinesByThread}: that map holds the past turn's + * tool rows, this one its interleaved reasoning/narration trail so a reopened + * thread replays each past answer's thoughts — not just its tool cards + * (restore-fidelity fix 1). Hydrated from `turn_state_history`; the live turn + * is excluded (its transcript lives in {@link processingByThread}). Absent for + * legacy snapshots written before the transcript field existed. + */ + turnTranscriptsByThread: Record>; + /** + * The partial assistant answer left behind by an INTERRUPTED turn (the core + * process that was streaming it is gone), keyed by thread. Surfaced on restore + * so a turn that crashed mid-answer keeps its visible partial reply + hidden + * reasoning instead of dropping them (restore-fidelity fix 2). Unlike + * {@link streamingAssistantByThread} this is a SETTLED, non-live buffer: it is + * rendered statically (no pulsing cursor) and marked interrupted. Populated + * only when an interrupted snapshot carries `streamingText`/`thinking`; + * cleared on any live turn, a completed snapshot, or a thread reset. + */ + interruptedAssistantByThread: Record< + string, + { requestId: string; content: string; thinking: string } + >; /** * Ordered narration/thinking/tool transcript per thread for the * "View processing" panel — the interleaved Hermes-style record. Hydrated @@ -733,6 +758,8 @@ const initialState: ChatRuntimeState = { toolTimelineByThread: {}, toolTimelineSeqByThread: {}, turnTimelinesByThread: {}, + turnTranscriptsByThread: {}, + interruptedAssistantByThread: {}, processingByThread: {}, taskBoardByThread: {}, inferenceTurnLifecycleByThread: {}, @@ -868,6 +895,26 @@ function subagentActivityFromPersisted(activity: PersistedSubagentActivity): Sub }; } +/** + * Order a persisted processing transcript by its per-item `seq` when every item + * carries one, falling back to the array (arrival) order otherwise + * (restore-fidelity fix 5: prefer `seq` for replay ordering when present). The + * core already writes items in `seq` order, so this is a defensive stable sort + * that also tolerates a snapshot whose items were reordered in transit. A stable + * sort preserves arrival order for any items that happen to share a `seq`. + */ +function orderTranscriptBySeq(items: ProcessingTranscriptItem[]): ProcessingTranscriptItem[] { + if (items.length < 2) return items; + const allHaveSeq = items.every(item => typeof item.seq === 'number'); + if (!allHaveSeq) return items; + // `.sort` is stable in modern engines; map to (item, index) to make the + // tie-break on equal `seq` explicit rather than engine-dependent. + return items + .map((item, index) => ({ item, index })) + .sort((a, b) => a.item.seq - b.item.seq || a.index - b.index) + .map(({ item }) => item); +} + /** * `seq` defaults to the array index the caller maps over — persisted * `toolTimeline` order IS issue order (the core appends rows as it issues @@ -1085,9 +1132,22 @@ const chatRuntimeSlice = createSlice({ */ setTurnTimelinesForThread: ( state, - action: PayloadAction<{ threadId: string; timelines: Record }> + action: PayloadAction<{ + threadId: string; + timelines: Record; + transcripts?: Record; + }> ) => { - state.turnTimelinesByThread[action.payload.threadId] = action.payload.timelines; + const { threadId, timelines, transcripts } = action.payload; + state.turnTimelinesByThread[threadId] = timelines; + if (transcripts) { + state.turnTranscriptsByThread[threadId] = transcripts; + turnStateLog( + 'past-turn transcripts set thread=%s turns=%d', + threadId, + Object.keys(transcripts).length + ); + } }, /** 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. */ @@ -1864,6 +1924,7 @@ const chatRuntimeSlice = createSlice({ clearRuntimeForThread: (state, action: PayloadAction<{ threadId: string }>) => { delete state.inferenceStatusByThread[action.payload.threadId]; delete state.streamingAssistantByThread[action.payload.threadId]; + delete state.interruptedAssistantByThread[action.payload.threadId]; delete state.inferenceHeartbeatByThread[action.payload.threadId]; // Drop any parallel (forked) streams for this thread and their // request→thread mappings — a hard per-thread reset covers every branch. @@ -1900,6 +1961,8 @@ const chatRuntimeSlice = createSlice({ state.toolTimelineByThread = {}; state.toolTimelineSeqByThread = {}; state.turnTimelinesByThread = {}; + state.turnTranscriptsByThread = {}; + state.interruptedAssistantByThread = {}; state.processingByThread = {}; state.taskBoardByThread = {}; state.inferenceTurnLifecycleByThread = {}; @@ -2009,6 +2072,9 @@ const chatRuntimeSlice = createSlice({ if (snapshot.taskBoard) { state.taskBoardByThread[threadId] = snapshot.taskBoard; } + // A live turn is driving the thread — any interrupted partial from a + // prior crashed turn is superseded and must not linger under it. + delete state.interruptedAssistantByThread[threadId]; return; } @@ -2077,7 +2143,32 @@ const chatRuntimeSlice = createSlice({ // up rather than restarting at 0 and colliding with existing seqs. state.toolTimelineSeqByThread[threadId] = snapshot.toolTimeline.length; } - state.processingByThread[threadId] = snapshot.transcript ?? []; + // An interrupted turn was killed mid-answer (its core process is gone, + // so no `chat_done` will ever complete it). The partial reply + + // reasoning it had already streamed are persisted — surface them as a + // SETTLED buffer (rendered static + marked interrupted, not as a live + // pulsing stream) instead of dropping them (restore-fidelity fix 2). A + // `completed` turn's answer is the durable message, so it has no partial + // to keep — clear any stale interrupted buffer for the thread instead. + if ( + snapshot.lifecycle === 'interrupted' && + (snapshot.streamingText.length > 0 || snapshot.thinking.length > 0) + ) { + state.interruptedAssistantByThread[threadId] = { + requestId: snapshot.requestId, + content: snapshot.streamingText, + thinking: snapshot.thinking, + }; + turnStateLog( + 'interrupted partial kept thread=%s chars=%d thinkingChars=%d', + threadId, + snapshot.streamingText.length, + snapshot.thinking.length + ); + } else { + delete state.interruptedAssistantByThread[threadId]; + } + state.processingByThread[threadId] = orderTranscriptBySeq(snapshot.transcript ?? []); return; } @@ -2102,6 +2193,9 @@ const chatRuntimeSlice = createSlice({ } else { delete state.streamingAssistantByThread[threadId]; } + // This snapshot is in-flight (a live driver may be resuming it), not a + // settled interruption — drop any stale interrupted partial for the thread. + delete state.interruptedAssistantByThread[threadId]; state.toolTimelineByThread[threadId] = preserveLiveSubagentProse( state.toolTimelineByThread[threadId], @@ -2110,7 +2204,7 @@ const chatRuntimeSlice = createSlice({ // Persisted order is issue order — seed the live counter with the row // count so events arriving after this hydration keep counting up. state.toolTimelineSeqByThread[threadId] = snapshot.toolTimeline.length; - state.processingByThread[threadId] = snapshot.transcript ?? []; + state.processingByThread[threadId] = orderTranscriptBySeq(snapshot.transcript ?? []); }, /** * Rebuild durable historical subagent rows from the run ledger. This is @@ -2266,23 +2360,41 @@ export const fetchAndHydrateTurnHistory = createAsyncThunk( try { const history = await threadApi.getTurnStateHistory(threadId); const timelines: Record = {}; + const transcripts: Record = {}; // History is newest-first; the newest turn is the one `getTurnState` // hydrates into `toolTimelineByThread` (rendered as the live/anchored // "agent insights"), so skip it here to avoid rendering it twice — this // field holds only the *older* settled turns. for (const turn of history.slice(1)) { if (turn.lifecycle !== 'completed' && turn.lifecycle !== 'interrupted') continue; - if (!turn.requestId || turn.toolTimeline.length === 0) continue; - timelines[turn.requestId] = turn.toolTimeline.map((e, seq) => - toolTimelineFromPersisted(e, seq) - ); + if (!turn.requestId) continue; + // A past turn can have a reasoning/narration trail with NO tool calls + // (the agent only thought/narrated). Keep the turn whenever it has + // either a tool timeline OR a transcript so a tool-less answer still + // replays its thoughts (restore-fidelity fix 1) — the old + // `toolTimeline.length === 0` skip dropped those turns entirely. + const hasTools = turn.toolTimeline.length > 0; + const persistedTranscript = turn.transcript ?? []; + const hasTranscript = persistedTranscript.length > 0; + if (!hasTools && !hasTranscript) continue; + if (hasTools) { + timelines[turn.requestId] = turn.toolTimeline.map((e, seq) => + toolTimelineFromPersisted(e, seq) + ); + } + if (hasTranscript) { + // Prefer persisted `seq` for replay order, falling back to array + // order (restore-fidelity fix 5). + transcripts[turn.requestId] = orderTranscriptBySeq(persistedTranscript); + } } turnStateLog( - 'hydrated turn history thread=%s turns=%d', + 'hydrated turn history thread=%s timelines=%d transcripts=%d', threadId, - Object.keys(timelines).length + Object.keys(timelines).length, + Object.keys(transcripts).length ); - dispatch(setTurnTimelinesForThread({ threadId, timelines })); + dispatch(setTurnTimelinesForThread({ threadId, timelines, transcripts })); return timelines; } catch (error) { turnStateLog('history fetch failed thread=%s err=%O', threadId, error); diff --git a/app/src/types/turnState.ts b/app/src/types/turnState.ts index 69c81d34a1..92ea84bfa9 100644 --- a/app/src/types/turnState.ts +++ b/app/src/types/turnState.ts @@ -156,6 +156,11 @@ export interface PersistedToolTimelineEntry { /** Size-capped tool result text. Absent while running and on snapshots * written before this field. */ output?: string; + /** Per-turn monotonic ordering key stamped when the row is first created, so + * a rehydrated timeline orders rows identically to the live stream (shares + * the per-turn ordering space with {@link PersistedTranscriptItem.seq}). + * Absent on snapshots written before this field. */ + seq?: number; } export interface PersistedTurnState { diff --git a/src/core/socketio.rs b/src/core/socketio.rs index 9459c21ddf..cc0397943b 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -223,6 +223,15 @@ pub struct WebChannelEvent { /// for synthetic done events that never ran a real turn. #[serde(skip_serializing_if = "Option::is_none")] pub usage: Option, + /// Additive per-request monotonic ordering key stamped by the web-channel + /// progress bridge on every event it emits (conversations-timeline-refactor, + /// Phase 4). Together with the always-present `request_id`, the frontend + /// dedups replayed vs live events by `(request_id, seq)` and orders them + /// identically to the persisted turn-state snapshot. `None` on events not + /// emitted through the stamping bridge and on older cores — older frontends + /// simply ignore it. + #[serde(skip_serializing_if = "Option::is_none")] + pub seq: Option, } /// Token/cost/context totals for one completed turn, attached to `chat_done`. diff --git a/src/openhuman/agent/harness/session/runtime.rs b/src/openhuman/agent/harness/session/runtime.rs index e75bcf4a7d..074c756612 100644 --- a/src/openhuman/agent/harness/session/runtime.rs +++ b/src/openhuman/agent/harness/session/runtime.rs @@ -386,6 +386,105 @@ impl Agent { Ok(()) } + /// Cold-boot resume for the web-chat path: pre-populate this session's + /// LLM context from the **full-fidelity** `session_raw/{stem}.jsonl` + /// transcript for `thread_id`. + /// + /// This is the high-fidelity counterpart to + /// [`Self::seed_resume_from_messages`]. That fallback sources lossy + /// `(sender, content)` prose from the conversation log, so it drops every + /// tool call, tool-role result, and reasoning block — after an app restart + /// the model then "forgets" all its tool interactions. This path instead + /// routes thread → transcript via + /// [`transcript::find_root_transcript_for_thread`] and reuses the exact + /// [`transcript::read_transcript`] + + /// [`Self::bound_cached_transcript_messages`] machinery as + /// [`Self::try_load_session_transcript`], so `tool_calls`, `role:"tool"` + /// messages, and `reasoning_content` all survive the round-trip. The only + /// difference from `try_load_session_transcript` is the lookup key (thread + /// id vs. per-thread agent name), so a thread whose transcript was written + /// under a differently-scoped agent name still resumes. + /// + /// Returns `true` when a transcript was found, loaded, and seeded into + /// `cached_transcript_messages`; `false` (a no-op) when the agent is already + /// warm, no root transcript exists for the thread, the transcript is empty, + /// or it fails to parse — the caller then falls back to prose-pair seeding. + /// + /// Best-effort like `try_load_session_transcript`: read/parse failures are + /// logged and reported as `false` rather than propagated. The current turn's + /// user message is appended later by [`Self::run_single`] / `turn`, so it is + /// intentionally absent from the loaded prefix — no dedup is needed here (the + /// on-disk transcript ends at the previous completed turn). + pub fn seed_resume_from_thread_transcript(&mut self, thread_id: &str) -> bool { + if !self.history.is_empty() || self.cached_transcript_messages.is_some() { + log::debug!( + "[web-channel] seed_resume_from_thread_transcript no-op — agent already warm \ + (history_len={}, cached={}) thread={thread_id}", + self.history.len(), + self.cached_transcript_messages.is_some() + ); + return false; + } + + let Some(path) = + super::transcript::find_root_transcript_for_thread(&self.workspace_dir, thread_id) + else { + log::debug!( + "[web-channel] no root session_raw transcript for thread={thread_id} — \ + falling back to conversation-log prose seeding" + ); + return false; + }; + + log::info!( + "[web-channel] cold-boot resume — loading full-fidelity transcript for \ + thread={thread_id} path={}", + path.display() + ); + + match super::transcript::read_transcript(&path) { + Ok(session) => { + if session.messages.is_empty() { + log::debug!( + "[web-channel] root transcript for thread={thread_id} is empty — \ + falling back to prose seeding" + ); + return false; + } + let loaded_count = session.messages.len(); + // Count the tool-role results carried into the resumed prefix — + // the fidelity the prose fallback would have silently dropped. + let tool_result_msgs = session.messages.iter().filter(|m| m.role == "tool").count(); + let bounded = self.bound_cached_transcript_messages(session.messages); + if bounded.len() < loaded_count { + log::warn!( + "[web-channel] resume prefix trimmed from {} to {} messages \ + (max_history_messages={}) for thread={thread_id}", + loaded_count, + bounded.len(), + self.config.max_history_messages + ); + } + log::info!( + "[web-channel] cold-boot resume — primed {} transcript message(s) \ + ({} tool-role result(s) preserved) for thread={thread_id}", + bounded.len(), + tool_result_msgs + ); + self.cached_transcript_messages = Some(bounded); + true + } + Err(err) => { + log::warn!( + "[web-channel] failed to parse root transcript {} for thread={thread_id}: \ + {err} — falling back to prose seeding", + path.display() + ); + false + } + } + } + /// Drain and return memory citations collected for the latest completed turn. pub fn take_last_turn_citations( &mut self, diff --git a/src/openhuman/agent/harness/session/tests.rs b/src/openhuman/agent/harness/session/tests.rs index da62b719cf..fba8213a77 100644 --- a/src/openhuman/agent/harness/session/tests.rs +++ b/src/openhuman/agent/harness/session/tests.rs @@ -1342,6 +1342,171 @@ fn bound_cached_transcript_messages_snaps_past_leading_orphan_tool() { ); } +/// Cold-boot web-chat resume must prefer the full-fidelity `session_raw` +/// transcript over lossy conversation-log prose. This is the regression for the +/// "model forgets its tool interactions after an app restart" bug: once the +/// in-memory agent is dropped and a fresh agent cold-boots for the same thread, +/// the resumed context must still carry the tool call, the tool-role result, and +/// the reasoning that prose seeding (`seed_resume_from_messages`) discards. +#[test] +fn seed_resume_from_thread_transcript_preserves_tool_calls_and_reasoning() { + use super::transcript::{self, MessageUsage, TranscriptMeta, TurnUsage}; + use crate::openhuman::inference::provider::{ChatMessage, ToolCall}; + + let ws = tempfile::TempDir::new().expect("temp workspace"); + let wsp = ws.path().to_path_buf(); + let thread_id = "thr_resume_fidelity"; + + // ── Simulate a prior session persisted to session_raw carrying a tool + // call + reasoning on the tool-calling assistant turn and a tool-role + // result — exactly the fidelity the prose fallback drops. ── + let mut assistant_toolcall = ChatMessage::assistant("Let me look that up."); + transcript::attach_turn_usage_metadata( + &mut assistant_toolcall, + &TurnUsage { + provider: "openai".to_string(), + model: "gpt-x".to_string(), + usage: MessageUsage { + input: 10, + output: 5, + cached_input: 0, + context_window: 0, + cost_usd: 0.0, + }, + ts: "2026-01-01T00:00:00Z".to_string(), + reasoning_content: Some("I should search the web for the price.".to_string()), + tool_calls: vec![ToolCall { + id: "call_1".to_string(), + name: "web_search".to_string(), + arguments: r#"{"query":"btc price"}"#.to_string(), + extra_content: None, + }], + iteration: 1, + }, + ); + + let messages = vec![ + ChatMessage::system("system prompt"), + ChatMessage::user("what is btc price"), + assistant_toolcall, + ChatMessage::tool(r#"{"tool_call_id":"call_1","content":"$80,000"}"#), + ChatMessage::assistant("BTC is around $80,000."), + ]; + let meta = TranscriptMeta { + agent_name: "orchestrator_thread-resume".to_string(), + agent_id: Some("orchestrator".to_string()), + agent_type: Some("root".to_string()), + dispatcher: "native".to_string(), + provider: None, + model: None, + created: "2026-01-01T00:00:00Z".to_string(), + updated: "2026-01-01T00:00:00Z".to_string(), + turn_count: 1, + input_tokens: 10, + output_tokens: 5, + cached_input_tokens: 0, + charged_amount_usd: 0.0, + thread_id: Some(thread_id.to_string()), + task_id: None, + }; + // Root stem: no `__`, so `find_root_transcript_for_thread` accepts it. + let path = transcript::resolve_keyed_transcript_path(&wsp, "1700000000_orchestrator") + .expect("resolve transcript path"); + transcript::write_transcript(&path, &messages, &meta, None).expect("write transcript"); + + // ── Cold boot: a brand-new agent for the same thread whose agent + // definition name deliberately does NOT match the transcript stem — the + // resume must route purely by thread id, not by agent name. ── + let memory_cfg = crate::openhuman::config::MemoryConfig { + backend: "none".into(), + ..crate::openhuman::config::MemoryConfig::default() + }; + let mem: Arc = + Arc::from(crate::openhuman::memory_store::create_memory(&memory_cfg, &wsp).unwrap()); + let mut agent = Agent::builder() + .provider(Box::new(MockProvider { + responses: Mutex::new(vec![]), + })) + .tools(vec![Box::new(MockTool)]) + .memory(mem) + .tool_dispatcher(Box::new(NativeToolDispatcher)) + .agent_definition_name("some_other_agent_name") + .workspace_dir(wsp.clone()) + .build() + .expect("agent build should succeed"); + + let loaded = agent.seed_resume_from_thread_transcript(thread_id); + assert!( + loaded, + "cold-boot resume must load the thread's root transcript" + ); + + let cached = agent + .cached_transcript_messages + .as_ref() + .expect("cached transcript populated"); + + // The tool-role result must survive — prose seeding would have dropped it. + assert!( + cached.iter().any(|m| m.role == "tool"), + "resumed context must include the tool-role result message" + ); + + // The assistant tool call + reasoning survive, carried in metadata. + let tool_call_carrier = cached + .iter() + .find(|m| { + m.role == "assistant" + && m.extra_metadata + .as_ref() + .and_then(|v| v.get("openhuman_turn_usage")) + .is_some() + }) + .expect("resumed context must include the assistant tool-call turn"); + let usage_value = tool_call_carrier + .extra_metadata + .as_ref() + .and_then(|v| v.get("openhuman_turn_usage")) + .cloned() + .expect("turn usage metadata present"); + let parsed: TurnUsage = serde_json::from_value(usage_value).expect("turn usage deserializes"); + assert!( + parsed.tool_calls.iter().any(|c| c.name == "web_search"), + "the persisted tool call must round-trip into the resumed context" + ); + assert_eq!( + parsed.reasoning_content.as_deref(), + Some("I should search the web for the price."), + "reasoning content must be preserved on resume" + ); +} + +/// When no root transcript exists for the thread, the transcript resume is a +/// no-op returning `false` so the caller falls back to prose-pair seeding. +#[test] +fn seed_resume_from_thread_transcript_returns_false_without_transcript() { + let mut agent = build_minimal_agent_with_definition_name(Some("orchestrator")); + assert!(!agent.seed_resume_from_thread_transcript("thr_missing")); + assert!(agent.cached_transcript_messages.is_none()); +} + +/// Transcript resume must not stomp an already-warm agent (in-process session +/// cache hit) — mirrors the `seed_resume_from_messages` warm-agent guard. +#[test] +fn seed_resume_from_thread_transcript_is_noop_on_warm_agent() { + let mut agent = build_minimal_agent_with_definition_name(Some("orchestrator")); + agent.cached_transcript_messages = Some(vec![ + crate::openhuman::inference::provider::ChatMessage::system("warm prefix"), + ]); + assert!(!agent.seed_resume_from_thread_transcript("thr_x")); + let cached = agent + .cached_transcript_messages + .as_ref() + .expect("still populated"); + assert_eq!(cached.len(), 1); + assert_eq!(cached[0].content, "warm prefix"); +} + /// `hide_tools` on an agent that already has a visible-tool filter must drop /// only the named tools and leave the rest of the belt intact. #[test] diff --git a/src/openhuman/channels/proactive.rs b/src/openhuman/channels/proactive.rs index fe9f762d42..c7f7cfcc74 100644 --- a/src/openhuman/channels/proactive.rs +++ b/src/openhuman/channels/proactive.rs @@ -220,6 +220,9 @@ impl EventHandler for ProactiveMessageSubscriber { tool_display_label: None, tool_display_detail: None, usage: None, + // Proactive delivery is emitted outside the seq-stamping progress + // bridge; leave `seq` unset (older clients ignore it). + seq: None, }); // 2. If an active external channel is configured, deliver there too. diff --git a/src/openhuman/threads/turn_state/mirror.rs b/src/openhuman/threads/turn_state/mirror.rs index b36faa121a..cb0bc0c3b2 100644 --- a/src/openhuman/threads/turn_state/mirror.rs +++ b/src/openhuman/threads/turn_state/mirror.rs @@ -36,6 +36,47 @@ const MAX_PERSISTED_TOOL_OUTPUT: usize = 64 * 1024; /// [`MAX_PERSISTED_TOOL_OUTPUT`]. const TRUNCATION_MARKER_BUDGET: usize = 80; +/// Upper bound on a single persisted transcript prose item (one coalesced +/// narration or reasoning block, parent or sub-agent). A runaway reasoning +/// stream would otherwise grow one item without bound and bloat every +/// full-file snapshot rewrite. Tighter than [`MAX_PERSISTED_TOOL_OUTPUT`] +/// because a turn can accumulate many prose items. +const MAX_PERSISTED_TRANSCRIPT_ITEM: usize = 16 * 1024; + +/// Marker appended once when a transcript prose item is truncated at its cap. +const TRANSCRIPT_TRUNCATION_MARKER: &str = "\n…[truncated]"; + +/// Append `delta` to a coalescing transcript prose buffer, enforcing +/// [`MAX_PERSISTED_TRANSCRIPT_ITEM`] on a char boundary and stamping a one-time +/// truncation marker the first time the cap is hit. Once at the cap, further +/// deltas are dropped (the marker is already present). Used by both the parent +/// and sub-agent transcript coalescers so a single streamed block stays bounded. +fn append_capped_transcript_text(text: &mut String, delta: &str) { + if delta.is_empty() { + return; + } + if text.len() >= MAX_PERSISTED_TRANSCRIPT_ITEM { + // Already at the cap but more content is arriving — ensure the marker is + // present exactly once so the truncation is visible even when deltas + // land exactly on the boundary (never straddling it). + if !text.ends_with(TRANSCRIPT_TRUNCATION_MARKER) { + text.push_str(TRANSCRIPT_TRUNCATION_MARKER); + } + return; + } + let remaining = MAX_PERSISTED_TRANSCRIPT_ITEM - text.len(); + if delta.len() <= remaining { + text.push_str(delta); + return; + } + let mut end = remaining; + while end > 0 && !delta.is_char_boundary(end) { + end -= 1; + } + text.push_str(&delta[..end]); + text.push_str(TRANSCRIPT_TRUNCATION_MARKER); +} + /// Cap `output` for snapshot persistence, slicing on a char boundary and /// appending a truncation marker when content was dropped. Returns `None` /// for empty output (payload capture off) so the field serializes away. @@ -69,6 +110,11 @@ pub struct TurnStateMirror { /// order narration vs thinking vs tool calls *within* one iteration, so /// every transcript push stamps and increments this. next_seq: u32, + /// Separate monotonic ordering key for [`ToolTimelineEntry::seq`] — the flat + /// timeline is an independent projection from the interleaved transcript, so + /// it gets its own space (sharing `next_seq` would leave gaps in the + /// transcript's contiguous ordering). + next_tool_seq: u64, } impl TurnStateMirror { @@ -87,6 +133,7 @@ impl TurnStateMirror { state, turn_completed: false, next_seq: 0, + next_tool_seq: 0, }; mirror.flush(); mirror @@ -153,6 +200,7 @@ impl TurnStateMirror { existing.detail = display_detail.clone(); } } else { + let seq = self.next_tool_seq(); self.state.tool_timeline.push(ToolTimelineEntry { id: call_id.clone(), name: tool_name.clone(), @@ -165,6 +213,7 @@ impl TurnStateMirror { subagent: None, failure: None, output: None, + seq: Some(seq), }); } self.flush(); @@ -216,6 +265,7 @@ impl TurnStateMirror { } => { self.state.phase = Some(TurnPhase::Subagent); self.state.active_subagent = Some(agent_id.clone()); + let seq = self.next_tool_seq(); self.state.tool_timeline.push(ToolTimelineEntry { id: format!("subagent:{task_id}"), name: format!("subagent:{agent_id}"), @@ -242,6 +292,7 @@ impl TurnStateMirror { }), failure: None, output: None, + seq: Some(seq), }); self.flush(); true @@ -447,6 +498,7 @@ impl TurnStateMirror { // No matching entry yet — `ToolCallArgsDelta` may // arrive before `ToolCallStarted` so synthesise a // placeholder we can update once the start event lands. + let seq = self.next_tool_seq(); self.state.tool_timeline.push(ToolTimelineEntry { id: call_id.clone(), name: tool_name.clone(), @@ -459,6 +511,7 @@ impl TurnStateMirror { subagent: None, failure: None, output: None, + seq: Some(seq), }); } false @@ -526,16 +579,16 @@ impl TurnStateMirror { self.state.transcript.last_mut() { if *r == round { - text.push_str(delta); + append_capped_transcript_text(text, delta); return; } } let seq = self.next_seq(); - self.state.transcript.push(TranscriptItem::Narration { - round, - seq, - text: delta.to_string(), - }); + let mut text = String::new(); + append_capped_transcript_text(&mut text, delta); + self.state + .transcript + .push(TranscriptItem::Narration { round, seq, text }); } /// Append a hidden-reasoning delta to the transcript, with the same @@ -545,16 +598,16 @@ impl TurnStateMirror { self.state.transcript.last_mut() { if *r == round { - text.push_str(delta); + append_capped_transcript_text(text, delta); return; } } let seq = self.next_seq(); - self.state.transcript.push(TranscriptItem::Thinking { - round, - seq, - text: delta.to_string(), - }); + let mut text = String::new(); + append_capped_transcript_text(&mut text, delta); + self.state + .transcript + .push(TranscriptItem::Thinking { round, seq, text }); } /// Record a tool call in the transcript at the point it occurred, as a @@ -583,6 +636,13 @@ impl TurnStateMirror { seq } + /// Return the next monotonic tool-timeline ordering key and advance it. + fn next_tool_seq(&mut self) -> u64 { + let seq = self.next_tool_seq; + self.next_tool_seq = self.next_tool_seq.saturating_add(1); + seq + } + fn find_subagent_entry_mut(&mut self, task_id: &str) -> Option<&mut ToolTimelineEntry> { let needle = format!("subagent:{task_id}"); self.state @@ -616,27 +676,29 @@ impl TurnStateMirror { iteration: it, text, }) if is_thinking && *it == Some(iteration) => { - text.push_str(delta); + append_capped_transcript_text(text, delta); return; } Some(SubagentTranscriptItem::Text { iteration: it, text, }) if !is_thinking && *it == Some(iteration) => { - text.push_str(delta); + append_capped_transcript_text(text, delta); return; } _ => {} } + let mut text = String::new(); + append_capped_transcript_text(&mut text, delta); activity.transcript.push(if is_thinking { SubagentTranscriptItem::Thinking { iteration: Some(iteration), - text: delta.to_string(), + text, } } else { SubagentTranscriptItem::Text { iteration: Some(iteration), - text: delta.to_string(), + text, } }); } diff --git a/src/openhuman/threads/turn_state/mirror_tests.rs b/src/openhuman/threads/turn_state/mirror_tests.rs index 2ca9ab0c97..2e46bd6691 100644 --- a/src/openhuman/threads/turn_state/mirror_tests.rs +++ b/src/openhuman/threads/turn_state/mirror_tests.rs @@ -194,6 +194,160 @@ fn tool_call_completed_persists_capped_output() { assert!(persisted.contains("truncated")); } +#[test] +fn tool_timeline_entries_carry_monotonic_seq() { + // Every timeline row is stamped with a per-turn monotonic `seq` at creation + // so a rehydrated snapshot can order rows identically to the live stream + // (conversations-timeline-refactor, Phase 4 amendment). Cover the three + // creation paths: a normal tool start, an args-delta placeholder, and a + // sub-agent spawn. + let (_d, mut m) = fresh("t"); + m.observe(&AgentProgress::IterationStarted { + iteration: 1, + max_iterations: 25, + }); + m.observe(&AgentProgress::ToolCallStarted { + call_id: "tc-1".into(), + tool_name: "shell".into(), + arguments: serde_json::json!({}), + iteration: 1, + display_label: None, + display_detail: None, + }); + // Placeholder-first path (args delta before start) for a second call. + m.observe(&AgentProgress::ToolCallArgsDelta { + call_id: "tc-2".into(), + tool_name: "shell".into(), + delta: "{".into(), + iteration: 1, + }); + m.observe(&AgentProgress::SubagentSpawned { + agent_id: "researcher".into(), + task_id: "sub-1".into(), + mode: "typed".into(), + dedicated_thread: false, + prompt_chars: 10, + prompt: String::new(), + worker_thread_id: None, + display_name: None, + }); + + let seqs: Vec = m + .snapshot() + .tool_timeline + .iter() + .map(|e| e.seq.expect("every row is seq-stamped")) + .collect(); + assert_eq!(seqs.len(), 3); + // Strictly increasing in creation order. + assert!( + seqs.windows(2).all(|w| w[0] < w[1]), + "tool timeline seqs must be strictly increasing, got {seqs:?}" + ); + + // A later `ToolCallStarted` that reuses the args-delta placeholder must NOT + // restamp the row's seq (the row keeps its original creation order). + let tc2_seq_before = m + .snapshot() + .tool_timeline + .iter() + .find(|e| e.id == "tc-2") + .and_then(|e| e.seq) + .expect("placeholder seq"); + m.observe(&AgentProgress::ToolCallStarted { + call_id: "tc-2".into(), + tool_name: "shell".into(), + arguments: serde_json::json!({}), + iteration: 1, + display_label: None, + display_detail: None, + }); + let tc2_seq_after = m + .snapshot() + .tool_timeline + .iter() + .find(|e| e.id == "tc-2") + .and_then(|e| e.seq) + .expect("placeholder seq after reuse"); + assert_eq!( + tc2_seq_before, tc2_seq_after, + "reusing a placeholder must not restamp its seq" + ); +} + +#[test] +fn subagent_prose_item_is_size_capped() { + // A runaway reasoning stream must not grow a single transcript item without + // bound (the snapshot is rewritten in full at every flush). Streaming far + // past the per-item cap coalesces into one item that stays bounded and + // carries a truncation marker. + let (_d, mut m) = fresh("t"); + m.observe(&AgentProgress::IterationStarted { + iteration: 1, + max_iterations: 25, + }); + m.observe(&AgentProgress::SubagentSpawned { + agent_id: "researcher".into(), + task_id: "sub-1".into(), + mode: "typed".into(), + dedicated_thread: false, + prompt_chars: 10, + prompt: String::new(), + worker_thread_id: None, + display_name: None, + }); + // 40 KiB of reasoning in same-iteration chunks — must coalesce and cap. + for _ in 0..40 { + m.observe(&AgentProgress::SubagentThinkingDelta { + agent_id: "researcher".into(), + task_id: "sub-1".into(), + delta: "x".repeat(1024), + iteration: 1, + }); + } + let activity = m.snapshot().tool_timeline[0] + .subagent + .as_ref() + .expect("activity") + .clone(); + assert_eq!( + activity.transcript.len(), + 1, + "same-iteration prose coalesces" + ); + match &activity.transcript[0] { + SubagentTranscriptItem::Thinking { text, .. } => { + assert!( + text.len() <= super::MAX_PERSISTED_TRANSCRIPT_ITEM + 64, + "capped prose item stays bounded, got {} bytes", + text.len() + ); + assert!(text.contains("truncated"), "capped item carries a marker"); + } + other => panic!("expected thinking, got {other:?}"), + } +} + +#[test] +fn parent_transcript_prose_item_is_size_capped() { + let (_d, mut m) = fresh("t"); + for _ in 0..40 { + m.observe(&AgentProgress::ThinkingDelta { + delta: "y".repeat(1024), + iteration: 1, + }); + } + let s = m.snapshot(); + assert_eq!(s.transcript.len(), 1); + match &s.transcript[0] { + TranscriptItem::Thinking { text, .. } => { + assert!(text.len() <= super::MAX_PERSISTED_TRANSCRIPT_ITEM + 64); + assert!(text.contains("truncated")); + } + other => panic!("expected thinking, got {other:?}"), + } +} + #[test] fn args_delta_arriving_before_start_creates_placeholder() { let (_d, mut m) = fresh("t"); diff --git a/src/openhuman/threads/turn_state/store_tests.rs b/src/openhuman/threads/turn_state/store_tests.rs index 4af251891e..6f48527786 100644 --- a/src/openhuman/threads/turn_state/store_tests.rs +++ b/src/openhuman/threads/turn_state/store_tests.rs @@ -2,7 +2,8 @@ use super::*; use crate::openhuman::threads::turn_state::types::{ - ToolTimelineEntry, ToolTimelineStatus, TurnLifecycle, TurnState, + SubagentActivity, SubagentToolCall, SubagentTranscriptItem, ToolTimelineEntry, + ToolTimelineStatus, TurnLifecycle, TurnState, }; use tempfile::tempdir; @@ -44,6 +45,7 @@ fn put_then_get_roundtrips_state() { subagent: None, failure: None, output: None, + seq: Some(0), }); store.put(&state).expect("put"); @@ -51,6 +53,107 @@ fn put_then_get_roundtrips_state() { assert_eq!(loaded, state); } +#[test] +fn roundtrips_subagent_interleaved_transcript_with_full_fidelity() { + // A settled turn whose subagent streamed reasoning, called a tool, then + // narrated — the interleaved transcript (not just the flat tool rows) plus + // the per-row `seq` ordering keys must survive a disk round-trip verbatim, + // so a reopened transcript rehydrates without losing the subagent's + // reasoning (the gap SubagentDrawer/chatRuntimeSlice documented). + let dir = tempdir().expect("tempdir"); + let store = TurnStateStore::new(dir.path().to_path_buf()); + let mut state = sample_state("thread-sub"); + state.lifecycle = TurnLifecycle::Completed; + + let activity = SubagentActivity { + task_id: "sub-1".into(), + agent_id: "researcher".into(), + status: Some("completed".into()), + mode: Some("typed".into()), + dedicated_thread: Some(false), + child_iteration: Some(1), + child_max_iterations: Some(8), + iterations: Some(2), + elapsed_ms: Some(1234), + output_chars: Some(42), + worker_thread_id: Some("worker-thread-9".into()), + tool_calls: vec![SubagentToolCall { + call_id: "c1".into(), + tool_name: "search".into(), + status: ToolTimelineStatus::Success, + iteration: Some(1), + elapsed_ms: Some(12), + output_chars: Some(6), + display_name: Some("Searching".into()), + detail: None, + failure: None, + output: Some("3 hits".into()), + }], + transcript: vec![ + SubagentTranscriptItem::Thinking { + iteration: Some(1), + text: "let me search.".into(), + }, + SubagentTranscriptItem::Tool { + iteration: Some(1), + call_id: "c1".into(), + tool_name: "search".into(), + status: ToolTimelineStatus::Success, + elapsed_ms: Some(12), + output_chars: Some(6), + display_name: Some("Searching".into()), + detail: None, + }, + SubagentTranscriptItem::Text { + iteration: Some(1), + text: "Found it.".into(), + }, + ], + }; + + state.tool_timeline.push(ToolTimelineEntry { + id: "subagent:sub-1".into(), + name: "subagent:researcher".into(), + round: 1, + status: ToolTimelineStatus::Success, + args_buffer: None, + display_name: Some("Researcher".into()), + detail: None, + source_tool_name: Some("spawn_subagent".into()), + subagent: Some(activity), + failure: None, + output: None, + seq: Some(3), + }); + + store.put(&state).expect("put"); + let loaded = store.get("thread-sub").expect("get").expect("present"); + // Structural equality proves nothing in the interleaved transcript, + // subagent activity, or the `seq` ordering keys was dropped or reordered. + assert_eq!(loaded, state); + + // Spot-check the interleaving explicitly so a future regression that keeps + // the fields but loses the ordering still fails here. + let activity = loaded.tool_timeline[0] + .subagent + .as_ref() + .expect("subagent activity restored"); + assert_eq!(activity.transcript.len(), 3); + assert!(matches!( + activity.transcript[0], + SubagentTranscriptItem::Thinking { .. } + )); + assert!(matches!( + activity.transcript[1], + SubagentTranscriptItem::Tool { .. } + )); + assert!(matches!( + activity.transcript[2], + SubagentTranscriptItem::Text { .. } + )); + assert_eq!(loaded.tool_timeline[0].seq, Some(3)); +} + #[test] fn get_returns_none_when_absent() { let dir = tempdir().expect("tempdir"); diff --git a/src/openhuman/threads/turn_state/types.rs b/src/openhuman/threads/turn_state/types.rs index ec42e95ebb..054a588547 100644 --- a/src/openhuman/threads/turn_state/types.rs +++ b/src/openhuman/threads/turn_state/types.rs @@ -125,6 +125,13 @@ pub struct ToolTimelineEntry { /// `tool_result`. `None` while running and on legacy snapshots. #[serde(default, skip_serializing_if = "Option::is_none")] pub output: Option, + /// Per-turn monotonic ordering key stamped at the moment the row is first + /// created, so a rehydrated timeline can order rows identically to the live + /// stream (conversations-timeline-refactor, Phase 4 amendment). Shares the + /// per-turn ordering space with [`TranscriptItem::seq`]. `None` on snapshots + /// written before this field. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub seq: Option, } /// Live sub-agent activity nested under a `subagent:*` timeline row. diff --git a/src/openhuman/web_chat/presentation.rs b/src/openhuman/web_chat/presentation.rs index ef40227b0a..dfb117477d 100644 --- a/src/openhuman/web_chat/presentation.rs +++ b/src/openhuman/web_chat/presentation.rs @@ -110,6 +110,9 @@ pub(crate) async fn deliver_response( Some(serde_json::json!(citations)) }, usage: usage_payload, + // Terminal delivery events are emitted outside the seq-stamping + // progress bridge; leave `seq` unset (older clients ignore it). + seq: None, }); return; } @@ -161,6 +164,7 @@ pub(crate) async fn deliver_response( }, // Usage is attached only to the terminal `chat_done`, never segments. usage: None, + seq: None, }); } @@ -201,6 +205,9 @@ pub(crate) async fn deliver_response( Some(serde_json::json!(citations)) }, usage: usage_payload, + // Terminal delivery events are emitted outside the seq-stamping + // progress bridge; leave `seq` unset (older clients ignore it). + seq: None, }); } diff --git a/src/openhuman/web_chat/progress_bridge.rs b/src/openhuman/web_chat/progress_bridge.rs index fec232ae61..d3fd5fc111 100644 --- a/src/openhuman/web_chat/progress_bridge.rs +++ b/src/openhuman/web_chat/progress_bridge.rs @@ -34,6 +34,7 @@ fn flush_interim_narration( client_id: &str, thread_id: &str, request_id: &str, + emit_seq: &mut u64, ) { let text = std::mem::take(buffer); let Some(narration) = interim_narration_text(&text) else { @@ -45,15 +46,31 @@ fn flush_interim_narration( narration.chars().count(), request_id, ); - publish_web_channel_event(WebChannelEvent { - event: "chat_interim".to_string(), - client_id: client_id.to_string(), - thread_id: thread_id.to_string(), - request_id: request_id.to_string(), - full_response: Some(narration), - round: Some(round), - ..Default::default() - }); + publish_seq_stamped( + emit_seq, + WebChannelEvent { + event: "chat_interim".to_string(), + client_id: client_id.to_string(), + thread_id: thread_id.to_string(), + request_id: request_id.to_string(), + full_response: Some(narration), + round: Some(round), + ..Default::default() + }, + ); +} + +/// Stamp a per-request monotonic sequence number on an outgoing web-channel +/// event and publish it. `request_id` is already carried on every +/// [`WebChannelEvent`]; `seq` is the additive ordering key the frontend uses to +/// dedup replayed vs live events by `(request_id, seq)` and to order them +/// identically to the persisted turn-state snapshot +/// (conversations-timeline-refactor, Phase 4). The counter advances once per +/// emitted event so each `(request_id, seq)` pair is unique within a turn. +fn publish_seq_stamped(next_seq: &mut u64, mut event: WebChannelEvent) { + event.seq = Some(*next_seq); + *next_seq = next_seq.saturating_add(1); + publish_web_channel_event(event); } /// The trimmed narration to surface as an interim bubble, or `None` when it is @@ -340,6 +357,10 @@ pub(crate) fn spawn_progress_bridge( // (it belongs to the terminal round, which ends with no tool call). let mut pending_narration = String::new(); let mut events_seen: u64 = 0; + // Per-request monotonic ordering key stamped on every emitted + // web-channel event (see `publish_seq_stamped`). Unique per emission so + // the frontend can dedup by `(request_id, seq)`. + let mut emit_seq: u64 = 0; let mut parent_completed = false; let mut parent_tool_count: u64 = 0; let mut child_tool_counts: HashMap = HashMap::new(); @@ -444,7 +465,7 @@ pub(crate) fn spawn_progress_bridge( thread_id, request_id, ); - publish_web_channel_event(WebChannelEvent { + publish_seq_stamped(&mut emit_seq, WebChannelEvent { event: "inference_heartbeat".to_string(), client_id: client_id.clone(), thread_id: thread_id.clone(), @@ -580,13 +601,16 @@ pub(crate) fn spawn_progress_bridge( payload: json!({ "threadId": thread_id, "clientId": client_id }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "inference_start".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "inference_start".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + ..Default::default() + }, + ); } AgentProgress::IterationStarted { iteration, @@ -594,15 +618,18 @@ pub(crate) fn spawn_progress_bridge( } => { round = iteration; parent_max_iterations = max_iterations; - publish_web_channel_event(WebChannelEvent { - event: "iteration_start".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - message: Some(format!("Iteration {iteration}/{max_iterations}")), - round: Some(iteration), - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "iteration_start".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + message: Some(format!("Iteration {iteration}/{max_iterations}")), + round: Some(iteration), + ..Default::default() + }, + ); } AgentProgress::ToolCallStarted { call_id, @@ -621,6 +648,7 @@ pub(crate) fn spawn_progress_bridge( &client_id, &thread_id, &request_id, + &mut emit_seq, ); parent_tool_count += 1; ledger_append_event( @@ -643,20 +671,23 @@ pub(crate) fn spawn_progress_bridge( ..Default::default() }, ); - publish_web_channel_event(WebChannelEvent { - event: "tool_call".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - tool_name: Some(tool_name), - skill_id: Some("web_channel".to_string()), - args: Some(arguments), - round: Some(iteration), - tool_call_id: Some(call_id), - tool_display_label: display_label, - tool_display_detail: display_detail, - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "tool_call".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + tool_name: Some(tool_name), + skill_id: Some("web_channel".to_string()), + args: Some(arguments), + round: Some(iteration), + tool_call_id: Some(call_id), + tool_display_label: display_label, + tool_display_detail: display_detail, + ..Default::default() + }, + ); } AgentProgress::ToolCallCompleted { call_id, @@ -687,24 +718,27 @@ pub(crate) fn spawn_progress_bridge( }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "tool_result".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - tool_name: Some(tool_name), - skill_id: Some("web_channel".to_string()), - // Forward the real tool result (size-capped) so the UI - // can render tool output — mirrors the subagent - // `subagent_tool_result` path. Frontends that only - // need size/timing read the ledger telemetry instead. - output: Some(cap_wire_output(output)), - success: Some(success), - round: Some(iteration), - tool_call_id: Some(call_id), - failure: failure_json, - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "tool_result".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + tool_name: Some(tool_name), + skill_id: Some("web_channel".to_string()), + // Forward the real tool result (size-capped) so the UI + // can render tool output — mirrors the subagent + // `subagent_tool_result` path. Frontends that only + // need size/timing read the ledger telemetry instead. + output: Some(cap_wire_output(output)), + success: Some(success), + round: Some(iteration), + tool_call_id: Some(call_id), + failure: failure_json, + ..Default::default() + }, + ); } AgentProgress::SubagentSpawned { agent_id, @@ -770,25 +804,28 @@ pub(crate) fn spawn_progress_bridge( }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "subagent_spawned".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - message: Some(format!("Sub-agent '{label}' spawned")), - tool_name: Some(agent_id), - skill_id: Some(task_id), - round: Some(round), - subagent: Some(SubagentProgressDetail { - mode: Some(mode), - dedicated_thread: Some(dedicated_thread), - prompt_chars: Some(prompt_chars as u64), - worker_thread_id, - display_name, + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_spawned".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + message: Some(format!("Sub-agent '{label}' spawned")), + tool_name: Some(agent_id), + skill_id: Some(task_id), + round: Some(round), + subagent: Some(SubagentProgressDetail { + mode: Some(mode), + dedicated_thread: Some(dedicated_thread), + prompt_chars: Some(prompt_chars as u64), + worker_thread_id, + display_name, + ..Default::default() + }), ..Default::default() - }), - ..Default::default() - }); + }, + ); } AgentProgress::SubagentCompleted { agent_id, @@ -851,29 +888,36 @@ pub(crate) fn spawn_progress_bridge( }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "subagent_completed".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - message: Some(format!( - "Sub-agent '{agent_id}' completed in {elapsed_ms}ms" - )), - tool_name: Some(agent_id), - skill_id: Some(task_id), - success: Some(true), - round: Some(round), - subagent: Some(SubagentProgressDetail { - elapsed_ms: Some(elapsed_ms), - iterations: Some(iterations), - output_chars: Some(output_chars as u64), - // Worktree isolation metadata (#3376) — drives the - // inline subagent worktree row's open/diff/remove - // actions. All `None`/absent for non-isolated workers. - ..subagent_worktree_detail(worktree_path, changed_files, dirty_status) - }), - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_completed".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + message: Some(format!( + "Sub-agent '{agent_id}' completed in {elapsed_ms}ms" + )), + tool_name: Some(agent_id), + skill_id: Some(task_id), + success: Some(true), + round: Some(round), + subagent: Some(SubagentProgressDetail { + elapsed_ms: Some(elapsed_ms), + iterations: Some(iterations), + output_chars: Some(output_chars as u64), + // Worktree isolation metadata (#3376) — drives the + // inline subagent worktree row's open/diff/remove + // actions. All `None`/absent for non-isolated workers. + ..subagent_worktree_detail( + worktree_path, + changed_files, + dirty_status, + ) + }), + ..Default::default() + }, + ); } AgentProgress::SubagentFailed { agent_id, @@ -920,18 +964,21 @@ pub(crate) fn spawn_progress_bridge( payload: json!({ "agentId": agent_id, "error": error }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "subagent_failed".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - message: Some(error), - tool_name: Some(agent_id), - skill_id: Some(task_id), - success: Some(false), - round: Some(round), - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_failed".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + message: Some(error), + tool_name: Some(agent_id), + skill_id: Some(task_id), + success: Some(false), + round: Some(round), + ..Default::default() + }, + ); } AgentProgress::SubagentAwaitingUser { agent_id, @@ -995,22 +1042,25 @@ pub(crate) fn spawn_progress_bridge( }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "subagent_awaiting_user".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - message: Some(question), - tool_name: Some(agent_id), - skill_id: Some(task_id), - success: Some(true), - round: Some(round), - subagent: Some(SubagentProgressDetail { - worker_thread_id, + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_awaiting_user".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + message: Some(question), + tool_name: Some(agent_id), + skill_id: Some(task_id), + success: Some(true), + round: Some(round), + subagent: Some(SubagentProgressDetail { + worker_thread_id, + ..Default::default() + }), ..Default::default() - }), - ..Default::default() - }); + }, + ); } AgentProgress::SubagentIterationStarted { agent_id, @@ -1019,30 +1069,35 @@ pub(crate) fn spawn_progress_bridge( max_iterations, extended_policy, } => { - publish_web_channel_event(WebChannelEvent { - event: "subagent_iteration_start".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - message: Some(if extended_policy { - format!("Sub-agent '{agent_id}' step {iteration}") - } else { - format!("Sub-agent '{agent_id}' iteration {iteration}/{max_iterations}") - }), - tool_name: Some(agent_id), - skill_id: Some(task_id), - round: Some(round), - subagent: Some(SubagentProgressDetail { - child_iteration: Some(iteration), - child_max_iterations: if extended_policy { - None + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_iteration_start".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + message: Some(if extended_policy { + format!("Sub-agent '{agent_id}' step {iteration}") } else { - Some(max_iterations) - }, + format!( + "Sub-agent '{agent_id}' iteration {iteration}/{max_iterations}" + ) + }), + tool_name: Some(agent_id), + skill_id: Some(task_id), + round: Some(round), + subagent: Some(SubagentProgressDetail { + child_iteration: Some(iteration), + child_max_iterations: if extended_policy { + None + } else { + Some(max_iterations) + }, + ..Default::default() + }), ..Default::default() - }), - ..Default::default() - }); + }, + ); } AgentProgress::SubagentToolCallStarted { agent_id, @@ -1077,33 +1132,36 @@ pub(crate) fn spawn_progress_bridge( }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "subagent_tool_call".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - tool_name: Some(tool_name), - skill_id: Some(task_id.clone()), - // The child's tool arguments, so the UI can show what - // the sub-agent actually did (issue: subagent drawer - // detail). Skipped from the wire when `null`. - args: if arguments.is_null() { - None - } else { - Some(arguments) - }, - round: Some(round), - tool_call_id: Some(call_id), - tool_display_label: display_label, - tool_display_detail: display_detail, - subagent: Some(SubagentProgressDetail { - child_iteration: Some(iteration), - agent_id: Some(agent_id), - task_id: Some(task_id), + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_tool_call".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + tool_name: Some(tool_name), + skill_id: Some(task_id.clone()), + // The child's tool arguments, so the UI can show what + // the sub-agent actually did (issue: subagent drawer + // detail). Skipped from the wire when `null`. + args: if arguments.is_null() { + None + } else { + Some(arguments) + }, + round: Some(round), + tool_call_id: Some(call_id), + tool_display_label: display_label, + tool_display_detail: display_detail, + subagent: Some(SubagentProgressDetail { + child_iteration: Some(iteration), + agent_id: Some(agent_id), + task_id: Some(task_id), + ..Default::default() + }), ..Default::default() - }), - ..Default::default() - }); + }, + ); } AgentProgress::SubagentToolCallCompleted { agent_id, @@ -1139,32 +1197,35 @@ pub(crate) fn spawn_progress_bridge( }), }, ); - publish_web_channel_event(WebChannelEvent { - event: "subagent_tool_result".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - tool_name: Some(tool_name), - skill_id: Some(task_id.clone()), - success: Some(success), - round: Some(round), - tool_call_id: Some(call_id), - // The child's actual tool output, so the drawer can show - // *what came back* (not just a char count). Capped to a - // bounded size for the wire (#4007); `output_chars` + - // `elapsed_ms` still ride along in `subagent` below. - output: Some(cap_wire_output(output)), - failure: failure_json, - subagent: Some(SubagentProgressDetail { - child_iteration: Some(iteration), - agent_id: Some(agent_id), - task_id: Some(task_id), - elapsed_ms: Some(elapsed_ms), - output_chars: Some(output_chars as u64), + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_tool_result".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + tool_name: Some(tool_name), + skill_id: Some(task_id.clone()), + success: Some(success), + round: Some(round), + tool_call_id: Some(call_id), + // The child's actual tool output, so the drawer can show + // *what came back* (not just a char count). Capped to a + // bounded size for the wire (#4007); `output_chars` + + // `elapsed_ms` still ride along in `subagent` below. + output: Some(cap_wire_output(output)), + failure: failure_json, + subagent: Some(SubagentProgressDetail { + child_iteration: Some(iteration), + agent_id: Some(agent_id), + task_id: Some(task_id), + elapsed_ms: Some(elapsed_ms), + output_chars: Some(output_chars as u64), + ..Default::default() + }), ..Default::default() - }), - ..Default::default() - }); + }, + ); } AgentProgress::SubagentTextDelta { agent_id, @@ -1172,23 +1233,26 @@ pub(crate) fn spawn_progress_bridge( delta, iteration, } => { - publish_web_channel_event(WebChannelEvent { - event: "subagent_text_delta".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - round: Some(round), - delta: Some(delta), - delta_kind: Some("text".to_string()), - skill_id: Some(task_id.clone()), - subagent: Some(SubagentProgressDetail { - child_iteration: Some(iteration), - agent_id: Some(agent_id), - task_id: Some(task_id), + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_text_delta".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + round: Some(round), + delta: Some(delta), + delta_kind: Some("text".to_string()), + skill_id: Some(task_id.clone()), + subagent: Some(SubagentProgressDetail { + child_iteration: Some(iteration), + agent_id: Some(agent_id), + task_id: Some(task_id), + ..Default::default() + }), ..Default::default() - }), - ..Default::default() - }); + }, + ); } AgentProgress::SubagentThinkingDelta { agent_id, @@ -1196,23 +1260,26 @@ pub(crate) fn spawn_progress_bridge( delta, iteration, } => { - publish_web_channel_event(WebChannelEvent { - event: "subagent_thinking_delta".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - round: Some(round), - delta: Some(delta), - delta_kind: Some("thinking".to_string()), - skill_id: Some(task_id.clone()), - subagent: Some(SubagentProgressDetail { - child_iteration: Some(iteration), - agent_id: Some(agent_id), - task_id: Some(task_id), + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "subagent_thinking_delta".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + round: Some(round), + delta: Some(delta), + delta_kind: Some("thinking".to_string()), + skill_id: Some(task_id.clone()), + subagent: Some(SubagentProgressDetail { + child_iteration: Some(iteration), + agent_id: Some(agent_id), + task_id: Some(task_id), + ..Default::default() + }), ..Default::default() - }), - ..Default::default() - }); + }, + ); } AgentProgress::TaskBoardUpdated { board } => { log::debug!( @@ -1222,43 +1289,52 @@ pub(crate) fn spawn_progress_bridge( request_id, board.cards.len() ); - publish_web_channel_event(WebChannelEvent { - event: "task_board_updated".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - task_board: Some(serde_json::to_value(board).unwrap_or_else( - |_| serde_json::json!({ "threadId": thread_id, "cards": [] }), - )), - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "task_board_updated".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + task_board: Some(serde_json::to_value(board).unwrap_or_else( + |_| serde_json::json!({ "threadId": thread_id, "cards": [] }), + )), + ..Default::default() + }, + ); } AgentProgress::TextDelta { delta, iteration } => { // Buffer the round's narration so it can be flushed as an // interim bubble if a tool call closes this round. pending_narration.push_str(&delta); - publish_web_channel_event(WebChannelEvent { - event: "text_delta".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - round: Some(iteration), - delta: Some(delta), - delta_kind: Some("text".to_string()), - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "text_delta".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + round: Some(iteration), + delta: Some(delta), + delta_kind: Some("text".to_string()), + ..Default::default() + }, + ); } AgentProgress::ThinkingDelta { delta, iteration } => { - publish_web_channel_event(WebChannelEvent { - event: "thinking_delta".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - round: Some(iteration), - delta: Some(delta), - delta_kind: Some("thinking".to_string()), - ..Default::default() - }); + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "thinking_delta".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + round: Some(iteration), + delta: Some(delta), + delta_kind: Some("thinking".to_string()), + ..Default::default() + }, + ); } AgentProgress::ToolCallArgsDelta { call_id, @@ -1266,23 +1342,26 @@ pub(crate) fn spawn_progress_bridge( delta, iteration, } => { - publish_web_channel_event(WebChannelEvent { - event: "tool_args_delta".to_string(), - client_id: client_id.clone(), - thread_id: thread_id.clone(), - request_id: request_id.clone(), - tool_name: if tool_name.is_empty() { - None - } else { - Some(tool_name) + publish_seq_stamped( + &mut emit_seq, + WebChannelEvent { + event: "tool_args_delta".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + tool_name: if tool_name.is_empty() { + None + } else { + Some(tool_name) + }, + skill_id: Some("web_channel".to_string()), + round: Some(iteration), + delta: Some(delta), + delta_kind: Some("tool_args".to_string()), + tool_call_id: Some(call_id), + ..Default::default() }, - skill_id: Some("web_channel".to_string()), - round: Some(iteration), - delta: Some(delta), - delta_kind: Some("tool_args".to_string()), - tool_call_id: Some(call_id), - ..Default::default() - }); + ); } AgentProgress::TurnCompleted { iterations } => { parent_completed = true; @@ -1790,4 +1869,59 @@ mod tests { drop(tx); } + + /// Every event the bridge emits carries an additive per-request monotonic + /// `seq` (conversations-timeline-refactor, Phase 4), so the frontend can + /// dedup replayed vs live events by `(request_id, seq)` and order them + /// identically to the persisted snapshot. Drive a short deterministic + /// sequence and assert the emitted seqs are present and strictly increasing. + #[tokio::test] + async fn stamps_monotonic_seq_on_emitted_events() { + let mut events = super::super::event_bus::subscribe_web_channel_events(); + let thread_id = "thread-seq-stamp"; + let request_id = "req-seq-stamp"; + let tx = spawn_test_bridge(thread_id, request_id); + + // Each of these emits exactly one web-channel event: inference_start, + // tool_call, tool_result (the fast test never trips the 20s heartbeat). + tx.send(AgentProgress::TurnStarted).await.unwrap(); + tx.send(AgentProgress::ToolCallStarted { + call_id: "tc-1".into(), + tool_name: "shell".into(), + arguments: serde_json::json!({}), + iteration: 1, + display_label: None, + display_detail: None, + }) + .await + .unwrap(); + tx.send(AgentProgress::ToolCallCompleted { + call_id: "tc-1".into(), + tool_name: "shell".into(), + success: true, + output_chars: 0, + output: String::new(), + arguments: None, + elapsed_ms: 5, + iteration: 1, + failure: None, + }) + .await + .unwrap(); + + let mut seqs = Vec::new(); + for _ in 0..3 { + let ev = recv_for_thread(&mut events, thread_id).await; + assert_eq!(ev.request_id, request_id); + seqs.push(ev.seq.expect("every emitted event carries a seq")); + } + // The very first emitted event starts the per-request counter at 0. + assert_eq!(seqs[0], 0, "seq counter starts at 0 for the request"); + assert!( + seqs.windows(2).all(|w| w[0] < w[1]), + "emitted seqs must be strictly increasing, got {seqs:?}" + ); + + drop(tx); + } } diff --git a/src/openhuman/web_chat/run_task.rs b/src/openhuman/web_chat/run_task.rs index a5b6c34114..cb3f874393 100644 --- a/src/openhuman/web_chat/run_task.rs +++ b/src/openhuman/web_chat/run_task.rs @@ -163,39 +163,57 @@ pub(crate) async fn run_chat_task( ), }; - // Cold-boot resume from the conversation JSONL. + // Cold-boot resume. Prefer the full-fidelity `session_raw/{stem}.jsonl` + // transcript (tool calls, tool-role results, reasoning) routed by thread + // id — the model must not "forget" its tool interactions across an app + // restart. Only fall back to the lossy conversation-log prose pairs when + // no root transcript exists for the thread or it fails to load; the two + // sources overlap (user prompts + final assistant text), so we take one + // or the other, never both, to avoid duplicated context. if was_built_fresh { - match crate::openhuman::memory_conversations::get_messages( - config.workspace_dir.clone(), - thread_id, - ) { - Ok(prior_messages) if !prior_messages.is_empty() => { - let pairs: Vec<(String, String)> = prior_messages - .into_iter() - .map(|m| (m.sender, m.content)) - .collect(); - if let Err(err) = agent.seed_resume_from_messages(pairs, message) { + if agent.seed_resume_from_thread_transcript(thread_id) { + log::info!( + "[web-channel] cold-boot resumed thread={} from full-fidelity session transcript", + thread_id + ); + } else { + log::debug!( + "[web-channel] no usable session transcript for thread={} — seeding resume \ + from conversation-log prose", + thread_id + ); + match crate::openhuman::memory_conversations::get_messages( + config.workspace_dir.clone(), + thread_id, + ) { + Ok(prior_messages) if !prior_messages.is_empty() => { + let pairs: Vec<(String, String)> = prior_messages + .into_iter() + .map(|m| (m.sender, m.content)) + .collect(); + if let Err(err) = agent.seed_resume_from_messages(pairs, message) { + log::warn!( + "[web-channel] failed to seed agent resume from conversation log \ + thread={} err={}", + thread_id, + err + ); + } + } + Ok(_) => { + log::debug!( + "[web-channel] no prior messages to seed for thread={} — first turn", + thread_id + ); + } + Err(err) => { log::warn!( - "[web-channel] failed to seed agent resume from conversation log \ - thread={} err={}", + "[web-channel] failed to read conversation log for resume thread={} err={}", thread_id, err ); } } - Ok(_) => { - log::debug!( - "[web-channel] no prior messages to seed for thread={} — first turn", - thread_id - ); - } - Err(err) => { - log::warn!( - "[web-channel] failed to read conversation log for resume thread={} err={}", - thread_id, - err - ); - } } }