From 05d73803e7f68786b611e27d6906ad536fb3888f Mon Sep 17 00:00:00 2001 From: chengyongru Date: Mon, 10 Aug 2026 15:43:54 +0800 Subject: [PATCH] refactor(webui): extract event projection helpers --- ...test_webui_event_projection_equivalence.py | 65 ++ webui/src/hooks/useNanobotStream.ts | 376 +---------- webui/src/lib/thread-event-projection.ts | 357 +++++++++++ .../live-replay-event-projection.json | 600 ++++++++++++++++++ webui/src/tests/useNanobotStream.test.tsx | 79 ++- 5 files changed, 1126 insertions(+), 351 deletions(-) create mode 100644 tests/utils/test_webui_event_projection_equivalence.py create mode 100644 webui/src/lib/thread-event-projection.ts create mode 100644 webui/src/tests/fixtures/live-replay-event-projection.json diff --git a/tests/utils/test_webui_event_projection_equivalence.py b/tests/utils/test_webui_event_projection_equivalence.py new file mode 100644 index 000000000..e20384b98 --- /dev/null +++ b/tests/utils/test_webui_event_projection_equivalence.py @@ -0,0 +1,65 @@ +"""Shared characterization cases for live and persisted WebUI projection.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from nanobot.webui.transcript import replay_transcript_to_ui_messages + +_FIXTURE_PATH = ( + Path(__file__).parents[2] + / "webui" + / "src" + / "tests" + / "fixtures" + / "live-replay-event-projection.json" +) +_SEMANTIC_MESSAGE_FIELDS = ( + "role", + "content", + "kind", + "traces", + "toolEvents", + "fileEdits", + "images", + "media", + "cliApps", + "mcpPresets", + "sessionMentions", + "reasoning", + "latencyMs", + "source", + "turnId", + "turnPhase", + "turnSeq", +) + + +def _normalize_projection(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + segment_aliases: dict[str, str] = {} + normalized: list[dict[str, Any]] = [] + for message in messages: + row = { + field: message[field] + for field in _SEMANTIC_MESSAGE_FIELDS + if field in message and message[field] is not None + } + segment_id = message.get("activitySegmentId") + if isinstance(segment_id, str) and segment_id: + row["activitySegmentId"] = segment_aliases.setdefault( + segment_id, + f"segment-{len(segment_aliases) + 1}", + ) + normalized.append(row) + return normalized + + +def test_replay_matches_shared_live_projection_before_canonical_revision_migration() -> None: + """Lock the known-equivalent subset without defining the future snapshot protocol.""" + fixture = json.loads(_FIXTURE_PATH.read_text(encoding="utf-8")) + + for case in fixture["cases"]: + actual = replay_transcript_to_ui_messages(case["transcript"]) + assert _normalize_projection(actual) == case["expected"], case["name"] diff --git a/webui/src/hooks/useNanobotStream.ts b/webui/src/hooks/useNanobotStream.ts index a7e1adeac..6ecea4194 100644 --- a/webui/src/hooks/useNanobotStream.ts +++ b/webui/src/hooks/useNanobotStream.ts @@ -10,6 +10,23 @@ import { } from "@/lib/tool-traces"; import { hasPendingAgentActivity } from "@/lib/activity-timeline"; import type { StreamError } from "@/lib/nanobot-client"; +import { + closeReasoningStream, + filterCoveredFileEditToolEvents, + finalizeStreamedTurn, + findActiveAssistantPlaceholderIndex, + findFileEditTraceIndex, + findStreamingAssistantIndex, + isReasoningOnlyPlaceholder, + matchesTurn, + mergeFileEdits, + pruneReasoningOnlyPlaceholders, + replaceMessageAt, + stampLastAssistantCompletion, + stripCoveredFileEditToolHintsFromMessages, + turnFieldsFromEvent, +} from "@/lib/thread-event-projection"; +import type { UIMessageTurnFields } from "@/lib/thread-event-projection"; import { formatQuotedUserMessage } from "@/lib/user-message-quote"; import type { InboundEvent, @@ -19,11 +36,8 @@ import type { SessionMention, GoalStateWsPayload, MessageDeliveryStatus, - ToolProgressEvent, UIMediaAttachment, - UIFileEdit, UIMessage, - UITurnPhase, WorkspaceScopePayload, } from "@/lib/types"; @@ -41,54 +55,9 @@ type PendingStreamEvent = | { kind: "delta"; text: string; turn: UIMessageTurnFields; source?: UIMessage["source"] } | { kind: "reasoning"; text: string; turn: UIMessageTurnFields }; -type UIMessageTurnFields = Pick; - -const FILE_EDIT_TOOL_NAMES = new Set(["write_file", "edit_file", "apply_patch"]); const STREAM_END_IDLE_DELAY_MS = 1000; const BACKGROUND_STREAM_FLUSH_INTERVAL_MS = 1_000; -function turnFieldsFromEvent( - ev: { turn_id?: string; turn_phase?: UITurnPhase; turn_seq?: number }, - fallbackPhase?: UITurnPhase, -): UIMessageTurnFields { - const fields: UIMessageTurnFields = {}; - if (typeof ev.turn_id === "string" && ev.turn_id.length > 0) { - fields.turnId = ev.turn_id; - } - const phase = ev.turn_phase ?? fallbackPhase; - if (phase) fields.turnPhase = phase; - if (typeof ev.turn_seq === "number" && Number.isFinite(ev.turn_seq)) { - fields.turnSeq = ev.turn_seq; - } - return fields; -} - -function matchesTurn(message: UIMessage, turn: UIMessageTurnFields): boolean { - return !turn.turnId || !message.turnId || message.turnId === turn.turnId; -} - -/** Find a still-open streamed assistant turn. Closed stream segments stay visible - * as streaming until ``turn_end`` for visual continuity, but they must not - * receive later delta segments. */ -function findStreamingAssistantIndex( - prev: UIMessage[], - closedStreamIds: ReadonlySet, - turn: UIMessageTurnFields = {}, -): number | null { - for (let i = prev.length - 1; i >= 0; i -= 1) { - const m = prev[i]; - if (m.kind === "trace") continue; - if ( - m.role === "assistant" - && m.isStreaming - && !closedStreamIds.has(m.id) - && matchesTurn(m, turn) - ) return i; - if (m.role === "user") break; - } - return null; -} - /** * Append a reasoning chunk to the last open reasoning stream in ``prev``. * @@ -152,102 +121,6 @@ function attachReasoningChunk( ]; } -/** - * Find the most recent assistant placeholder that an incoming answer - * delta should adopt instead of spawning a parallel row. We look for an - * empty-content assistant turn that is still marked ``isStreaming`` — - * typically created earlier by ``reasoning_delta``. Anything else means - * the model already produced an answer in a previous turn, so the new - * delta belongs in a fresh row. - */ -function findActiveAssistantPlaceholderIndex( - prev: UIMessage[], - turn: UIMessageTurnFields = {}, -): number | null { - const last = prev[prev.length - 1]; - if (!last) return null; - if (last.role !== "assistant" || last.kind === "trace") return null; - if (last.content.length > 0) return null; - if (!last.isStreaming) return null; - if (!matchesTurn(last, turn)) return null; - return prev.length - 1; -} - -function replaceMessageAt(prev: UIMessage[], index: number, message: UIMessage): UIMessage[] { - const next = prev.slice(); - next[index] = message; - return next; -} - -/** - * Close the active reasoning stream segment, if any. Idempotent: a - * ``reasoning_end`` with no preceding deltas is a harmless no-op. - */ -function closeReasoningStream(prev: UIMessage[]): UIMessage[] { - for (let i = prev.length - 1; i >= 0; i -= 1) { - const candidate = prev[i]; - if (!candidate.reasoningStreaming) continue; - const latencyMs = - candidate.latencyMs === undefined - && Number.isFinite(candidate.createdAt) - && candidate.createdAt > 1_000_000_000_000 - ? Math.max(0, Math.round(Date.now() - candidate.createdAt)) - : candidate.latencyMs; - const merged: UIMessage = { - ...candidate, - reasoningStreaming: false, - ...(latencyMs !== undefined ? { latencyMs } : {}), - }; - return [...prev.slice(0, i), merged, ...prev.slice(i + 1)]; - } - return prev; -} - -function isReasoningOnlyPlaceholder(message: UIMessage): boolean { - return ( - message.role === "assistant" - && message.kind !== "trace" - && message.content.trim().length === 0 - && !!message.reasoning - && !message.reasoningStreaming - && !message.media?.length - ); -} - -function isToolTrace(message: UIMessage | undefined): boolean { - return message?.kind === "trace"; -} - -function pruneReasoningOnlyPlaceholders(prev: UIMessage[]): UIMessage[] { - return prev.filter((message, index) => { - if (!isReasoningOnlyPlaceholder(message)) return true; - // A reasoning-only assistant row immediately followed by tool traces is - // the live equivalent of a persisted assistant tool-call message with - // empty content, reasoning_content, and tool_calls. Keep it so live render - // and history replay stay isomorphic. - return isToolTrace(prev[index + 1]); - }); -} - -function stampLastAssistantCompletion( - prev: UIMessage[], - completion: Pick, - turnId?: string, -): UIMessage[] { - for (let i = prev.length - 1; i >= 0; i -= 1) { - const m = prev[i]; - if ( - m.role === "assistant" - && m.kind !== "trace" - && (!turnId || !m.turnId || m.turnId === turnId) - ) { - const merged: UIMessage = { ...m, ...completion, isStreaming: false }; - return [...prev.slice(0, i), merged, ...prev.slice(i + 1)]; - } - } - return prev; -} - function absorbCompleteAssistantMessage( prev: UIMessage[], message: Omit, @@ -275,193 +148,6 @@ function absorbCompleteAssistantMessage( ]; } -function fileEditKey(edit: Pick): string { - if (edit.call_id && edit.path) return `${edit.call_id}|${edit.tool}|${edit.path}`; - if (edit.call_id) return `${edit.call_id}|${edit.tool}`; - return `${edit.tool}|${edit.path}`; -} - -function fileEditToolEventKey(edit: Pick): string { - if (edit.call_id) return `${edit.call_id}|${edit.tool}`; - return fileEditKey(edit); -} - -function toolEventFileEditKey(event: ToolProgressEvent): string | null { - const fn = (event as { function?: { name?: unknown } }).function; - const name = typeof event.name === "string" - ? event.name - : typeof fn?.name === "string" - ? fn.name - : ""; - const callId = typeof event.call_id === "string" ? event.call_id : ""; - if (!name || !callId || !FILE_EDIT_TOOL_NAMES.has(name)) return null; - return `${callId}|${name}`; -} - -function hasFileEditForToolEvent(messages: UIMessage[], event: ToolProgressEvent): boolean { - const key = toolEventFileEditKey(event); - if (!key) return false; - return messages.some((message) => - message.fileEdits?.some((edit) => fileEditToolEventKey(edit) === key), - ); -} - -function filterCoveredFileEditToolEvents( - messages: UIMessage[], - events: ToolProgressEvent[], -): ToolProgressEvent[] { - if (events.length === 0) return events; - return events.filter((event) => !hasFileEditForToolEvent(messages, event)); -} - -function stripCoveredFileEditToolHints(message: UIMessage, edits: UIFileEdit[]): UIMessage { - const incomingKeys = new Set(edits.map(fileEditToolEventKey)); - const events = message.toolEvents ?? []; - if (!events.length || incomingKeys.size === 0) return message; - - const removedTraceLines = new Set(); - const keptEvents: ToolProgressEvent[] = []; - let changed = false; - for (const event of events) { - const key = toolEventFileEditKey(event); - if (key && incomingKeys.has(key)) { - changed = true; - for (const line of toolTraceLinesFromEvents([event])) { - removedTraceLines.add(line); - } - continue; - } - keptEvents.push(event); - } - if (!changed) return message; - - const previousTraces = message.traces?.length - ? message.traces - : message.content - ? [message.content] - : []; - const nextTraces = previousTraces.filter((line) => !removedTraceLines.has(line)); - return { - ...message, - traces: nextTraces, - content: nextTraces[nextTraces.length - 1] ?? "", - toolEvents: keptEvents.length ? keptEvents : undefined, - }; -} - -function traceMessageIsEmpty(message: UIMessage): boolean { - const traces = message.traces; - const hasTrace = traces?.length - ? traces.some((line) => line.trim().length > 0) - : (message.content ?? "").trim().length > 0; - return ( - message.kind === "trace" - && !hasTrace - && !message.toolEvents?.length - && !message.fileEdits?.length - && !message.media?.length - ); -} - -function stripCoveredFileEditToolHintsFromMessages( - messages: UIMessage[], - edits: UIFileEdit[], - turn: UIMessageTurnFields, -): UIMessage[] { - if (edits.length === 0) return messages; - let next = messages; - for (let i = next.length - 1; i >= 0; i -= 1) { - const candidate = next[i]; - if (candidate.role === "user") break; - if (candidate.kind !== "trace") continue; - if (!matchesTurn(candidate, turn)) continue; - const cleaned = stripCoveredFileEditToolHints(candidate, edits); - if (cleaned === candidate) continue; - if (next === messages) next = [...messages]; - if (traceMessageIsEmpty(cleaned)) { - next.splice(i, 1); - } else { - next[i] = cleaned; - } - } - return next; -} - -function normalizeFileEdit(edit: UIFileEdit): UIFileEdit | null { - if (!edit || !edit.tool || (!edit.path && !edit.pending)) return null; - const inferredStatus = - edit.phase === "error" - ? "error" - : edit.phase === "end" - ? "done" - : "editing"; - const normalized: UIFileEdit = { - ...edit, - call_id: edit.call_id || `${edit.tool}:${edit.path}`, - added: Number.isFinite(edit.added) ? Math.max(0, Math.round(edit.added)) : 0, - deleted: Number.isFinite(edit.deleted) ? Math.max(0, Math.round(edit.deleted)) : 0, - status: edit.status === "error" || edit.status === "done" || edit.status === "editing" - ? edit.status - : inferredStatus, - }; - if (edit.pending && !edit.path) normalized.pending = true; - return normalized; -} - -function mergeFileEdits(existing: UIFileEdit[] | undefined, incoming: UIFileEdit[]): UIFileEdit[] { - const next = [...(existing ?? [])]; - const indexByKey = new Map(next.map((edit, index) => [fileEditKey(edit), index])); - for (const raw of incoming) { - const edit = normalizeFileEdit(raw); - if (!edit) continue; - const key = fileEditKey(edit); - let existingIndex = indexByKey.get(key); - if (existingIndex === undefined && edit.path) { - const eventKey = fileEditToolEventKey(edit); - const pendingIndex = next.findIndex((existing) => - !existing.path && existing.pending && fileEditToolEventKey(existing) === eventKey, - ); - if (pendingIndex >= 0) existingIndex = pendingIndex; - } - if (existingIndex === undefined) { - indexByKey.set(key, next.length); - next.push(edit); - continue; - } - const merged = { ...next[existingIndex], ...edit }; - if (edit.path && !edit.pending) delete merged.pending; - next[existingIndex] = merged; - indexByKey.set(key, existingIndex); - } - return next; -} - -function findFileEditTraceIndex( - prev: UIMessage[], - segmentId: string | null, - incoming: UIFileEdit[], -): number | null { - const incomingKeys = new Set(incoming.map(fileEditKey)); - const incomingToolEventKeys = new Set(incoming.map(fileEditToolEventKey)); - for (let i = prev.length - 1; i >= 0; i -= 1) { - const candidate = prev[i]; - if (candidate.role === "user") break; - if (candidate.kind !== "trace") continue; - if (segmentId && candidate.activitySegmentId === segmentId) return i; - for (const existing of candidate.fileEdits ?? []) { - if ( - incomingKeys.has(fileEditKey(existing)) - || ( - !existing.path - && existing.pending - && incomingToolEventKeys.has(fileEditToolEventKey(existing)) - ) - ) return i; - } - } - return null; -} - /** * Subscribe to a chat by ID. Returns the in-memory message list for the chat, * a streaming flag, and a ``send`` function. Initial history must be seeded @@ -507,17 +193,6 @@ function eventExtendsModelActivity(ev: InboundEvent): boolean { && (ev.kind === "tool_hint" || ev.kind === "progress" || ev.kind === "reasoning"); } -function finalizeStreamedTurn( - prev: UIMessage[], - turn: UIMessageTurnFields = {}, -): UIMessage[] { - return prev.map((m) => - m.isStreaming && matchesTurn(m, turn) - ? { ...m, isStreaming: false, reasoningStreaming: false } - : m, - ); -} - function eventTurnId(ev: InboundEvent): string | undefined { return "turn_id" in ev && typeof ev.turn_id === "string" ? ev.turn_id : undefined; } @@ -1104,7 +779,7 @@ export function useNanobotStream( if (ev.event === "reasoning_end") { if (suppressStreamUntilTurnEndRef.current) return; - setMessages((prev) => closeReasoningStream(prev)); + setMessages((prev) => closeReasoningStream(prev, Date.now())); return; } @@ -1174,12 +849,15 @@ export function useNanobotStream( const line = ev.text; if (!line) return; if (fileEditSegmentRef.current) clearActivitySegment(); - setMessages((prev) => closeReasoningStream(attachReasoningChunk( - prev, - line, - { ensure: ensureActivitySegmentId }, - turnFieldsFromEvent(ev, "reasoning"), - ))); + setMessages((prev) => closeReasoningStream( + attachReasoningChunk( + prev, + line, + { ensure: ensureActivitySegmentId }, + turnFieldsFromEvent(ev, "reasoning"), + ), + Date.now(), + )); return; } // Intermediate agent breadcrumbs (tool-call hints, raw progress). diff --git a/webui/src/lib/thread-event-projection.ts b/webui/src/lib/thread-event-projection.ts new file mode 100644 index 000000000..d577c70b9 --- /dev/null +++ b/webui/src/lib/thread-event-projection.ts @@ -0,0 +1,357 @@ +import { toolTraceLinesFromEvents } from "@/lib/tool-traces"; +import type { + ToolProgressEvent, + UIFileEdit, + UIMessage, + UITurnPhase, +} from "@/lib/types"; + +export type UIMessageTurnFields = Pick; + +const FILE_EDIT_TOOL_NAMES = new Set(["write_file", "edit_file", "apply_patch"]); + +/** + * PR3 projection seam: replay can share these folds once GatewayContext exposes + * an ordered canonical-event sequence and a monotonic per-thread revision. + * Snapshot acceptance and revision comparison stay outside this projection; + * until then, history continues to consume server-projected UIMessage snapshots. + */ + +export function turnFieldsFromEvent( + ev: { turn_id?: string; turn_phase?: UITurnPhase; turn_seq?: number }, + fallbackPhase?: UITurnPhase, +): UIMessageTurnFields { + const fields: UIMessageTurnFields = {}; + if (typeof ev.turn_id === "string" && ev.turn_id.length > 0) { + fields.turnId = ev.turn_id; + } + const phase = ev.turn_phase ?? fallbackPhase; + if (phase) fields.turnPhase = phase; + if (typeof ev.turn_seq === "number" && Number.isFinite(ev.turn_seq)) { + fields.turnSeq = ev.turn_seq; + } + return fields; +} + +export function matchesTurn(message: UIMessage, turn: UIMessageTurnFields): boolean { + return !turn.turnId || !message.turnId || message.turnId === turn.turnId; +} + +/** Find a still-open streamed assistant turn. Closed stream segments stay visible + * as streaming until ``turn_end`` for visual continuity, but they must not + * receive later delta segments. */ +export function findStreamingAssistantIndex( + prev: UIMessage[], + closedStreamIds: ReadonlySet, + turn: UIMessageTurnFields = {}, +): number | null { + for (let i = prev.length - 1; i >= 0; i -= 1) { + const m = prev[i]; + if (m.kind === "trace") continue; + if ( + m.role === "assistant" + && m.isStreaming + && !closedStreamIds.has(m.id) + && matchesTurn(m, turn) + ) return i; + if (m.role === "user") break; + } + return null; +} + +/** + * Find the most recent assistant placeholder that an incoming answer + * delta should adopt instead of spawning a parallel row. + */ +export function findActiveAssistantPlaceholderIndex( + prev: UIMessage[], + turn: UIMessageTurnFields = {}, +): number | null { + const last = prev[prev.length - 1]; + if (!last) return null; + if (last.role !== "assistant" || last.kind === "trace") return null; + if (last.content.length > 0) return null; + if (!last.isStreaming) return null; + if (!matchesTurn(last, turn)) return null; + return prev.length - 1; +} + +export function replaceMessageAt( + prev: UIMessage[], + index: number, + message: UIMessage, +): UIMessage[] { + const next = prev.slice(); + next[index] = message; + return next; +} + +/** Close the active reasoning stream segment. ``now`` is supplied by the caller + * so the projection remains deterministic for replay and fixture tests. */ +export function closeReasoningStream(prev: UIMessage[], now: number): UIMessage[] { + for (let i = prev.length - 1; i >= 0; i -= 1) { + const candidate = prev[i]; + if (!candidate.reasoningStreaming) continue; + const latencyMs = + candidate.latencyMs === undefined + && Number.isFinite(candidate.createdAt) + && candidate.createdAt > 1_000_000_000_000 + ? Math.max(0, Math.round(now - candidate.createdAt)) + : candidate.latencyMs; + const merged: UIMessage = { + ...candidate, + reasoningStreaming: false, + ...(latencyMs !== undefined ? { latencyMs } : {}), + }; + return [...prev.slice(0, i), merged, ...prev.slice(i + 1)]; + } + return prev; +} + +export function isReasoningOnlyPlaceholder(message: UIMessage): boolean { + return ( + message.role === "assistant" + && message.kind !== "trace" + && message.content.trim().length === 0 + && !!message.reasoning + && !message.reasoningStreaming + && !message.media?.length + ); +} + +function isToolTrace(message: UIMessage | undefined): boolean { + return message?.kind === "trace"; +} + +export function pruneReasoningOnlyPlaceholders(prev: UIMessage[]): UIMessage[] { + return prev.filter((message, index) => { + if (!isReasoningOnlyPlaceholder(message)) return true; + // A reasoning-only assistant row immediately followed by tool traces is + // the live equivalent of a persisted assistant tool-call message with + // empty content, reasoning_content, and tool_calls. Keep it so live render + // and history replay stay isomorphic. + return isToolTrace(prev[index + 1]); + }); +} + +export function stampLastAssistantCompletion( + prev: UIMessage[], + completion: Pick, + turnId?: string, +): UIMessage[] { + for (let i = prev.length - 1; i >= 0; i -= 1) { + const m = prev[i]; + if ( + m.role === "assistant" + && m.kind !== "trace" + && (!turnId || !m.turnId || m.turnId === turnId) + ) { + const merged: UIMessage = { ...m, ...completion, isStreaming: false }; + return [...prev.slice(0, i), merged, ...prev.slice(i + 1)]; + } + } + return prev; +} + +function fileEditKey(edit: Pick): string { + if (edit.call_id && edit.path) return `${edit.call_id}|${edit.tool}|${edit.path}`; + if (edit.call_id) return `${edit.call_id}|${edit.tool}`; + return `${edit.tool}|${edit.path}`; +} + +function fileEditToolEventKey( + edit: Pick, +): string { + if (edit.call_id) return `${edit.call_id}|${edit.tool}`; + return fileEditKey(edit); +} + +function toolEventFileEditKey(event: ToolProgressEvent): string | null { + const fn = (event as { function?: { name?: unknown } }).function; + const name = typeof event.name === "string" + ? event.name + : typeof fn?.name === "string" + ? fn.name + : ""; + const callId = typeof event.call_id === "string" ? event.call_id : ""; + if (!name || !callId || !FILE_EDIT_TOOL_NAMES.has(name)) return null; + return `${callId}|${name}`; +} + +function hasFileEditForToolEvent(messages: UIMessage[], event: ToolProgressEvent): boolean { + const key = toolEventFileEditKey(event); + if (!key) return false; + return messages.some((message) => + message.fileEdits?.some((edit) => fileEditToolEventKey(edit) === key), + ); +} + +export function filterCoveredFileEditToolEvents( + messages: UIMessage[], + events: ToolProgressEvent[], +): ToolProgressEvent[] { + if (events.length === 0) return events; + return events.filter((event) => !hasFileEditForToolEvent(messages, event)); +} + +function stripCoveredFileEditToolHints(message: UIMessage, edits: UIFileEdit[]): UIMessage { + const incomingKeys = new Set(edits.map(fileEditToolEventKey)); + const events = message.toolEvents ?? []; + if (!events.length || incomingKeys.size === 0) return message; + + const removedTraceLines = new Set(); + const keptEvents: ToolProgressEvent[] = []; + let changed = false; + for (const event of events) { + const key = toolEventFileEditKey(event); + if (key && incomingKeys.has(key)) { + changed = true; + for (const line of toolTraceLinesFromEvents([event])) { + removedTraceLines.add(line); + } + continue; + } + keptEvents.push(event); + } + if (!changed) return message; + + const previousTraces = message.traces?.length + ? message.traces + : message.content + ? [message.content] + : []; + const nextTraces = previousTraces.filter((line) => !removedTraceLines.has(line)); + return { + ...message, + traces: nextTraces, + content: nextTraces[nextTraces.length - 1] ?? "", + toolEvents: keptEvents.length ? keptEvents : undefined, + }; +} + +function traceMessageIsEmpty(message: UIMessage): boolean { + const traces = message.traces; + const hasTrace = traces?.length + ? traces.some((line) => line.trim().length > 0) + : (message.content ?? "").trim().length > 0; + return ( + message.kind === "trace" + && !hasTrace + && !message.toolEvents?.length + && !message.fileEdits?.length + && !message.media?.length + ); +} + +export function stripCoveredFileEditToolHintsFromMessages( + messages: UIMessage[], + edits: UIFileEdit[], + turn: UIMessageTurnFields, +): UIMessage[] { + if (edits.length === 0) return messages; + let next = messages; + for (let i = next.length - 1; i >= 0; i -= 1) { + const candidate = next[i]; + if (candidate.role === "user") break; + if (candidate.kind !== "trace") continue; + if (!matchesTurn(candidate, turn)) continue; + const cleaned = stripCoveredFileEditToolHints(candidate, edits); + if (cleaned === candidate) continue; + if (next === messages) next = [...messages]; + if (traceMessageIsEmpty(cleaned)) { + next.splice(i, 1); + } else { + next[i] = cleaned; + } + } + return next; +} + +function normalizeFileEdit(edit: UIFileEdit): UIFileEdit | null { + if (!edit || !edit.tool || (!edit.path && !edit.pending)) return null; + const inferredStatus = + edit.phase === "error" + ? "error" + : edit.phase === "end" + ? "done" + : "editing"; + const normalized: UIFileEdit = { + ...edit, + call_id: edit.call_id || `${edit.tool}:${edit.path}`, + added: Number.isFinite(edit.added) ? Math.max(0, Math.round(edit.added)) : 0, + deleted: Number.isFinite(edit.deleted) ? Math.max(0, Math.round(edit.deleted)) : 0, + status: edit.status === "error" || edit.status === "done" || edit.status === "editing" + ? edit.status + : inferredStatus, + }; + if (edit.pending && !edit.path) normalized.pending = true; + return normalized; +} + +export function mergeFileEdits( + existing: UIFileEdit[] | undefined, + incoming: UIFileEdit[], +): UIFileEdit[] { + const next = [...(existing ?? [])]; + const indexByKey = new Map(next.map((edit, index) => [fileEditKey(edit), index])); + for (const raw of incoming) { + const edit = normalizeFileEdit(raw); + if (!edit) continue; + const key = fileEditKey(edit); + let existingIndex = indexByKey.get(key); + if (existingIndex === undefined && edit.path) { + const eventKey = fileEditToolEventKey(edit); + const pendingIndex = next.findIndex((existing) => + !existing.path && existing.pending && fileEditToolEventKey(existing) === eventKey, + ); + if (pendingIndex >= 0) existingIndex = pendingIndex; + } + if (existingIndex === undefined) { + indexByKey.set(key, next.length); + next.push(edit); + continue; + } + const merged = { ...next[existingIndex], ...edit }; + if (edit.path && !edit.pending) delete merged.pending; + next[existingIndex] = merged; + indexByKey.set(key, existingIndex); + } + return next; +} + +export function findFileEditTraceIndex( + prev: UIMessage[], + segmentId: string | null, + incoming: UIFileEdit[], +): number | null { + const incomingKeys = new Set(incoming.map(fileEditKey)); + const incomingToolEventKeys = new Set(incoming.map(fileEditToolEventKey)); + for (let i = prev.length - 1; i >= 0; i -= 1) { + const candidate = prev[i]; + if (candidate.role === "user") break; + if (candidate.kind !== "trace") continue; + if (segmentId && candidate.activitySegmentId === segmentId) return i; + for (const existing of candidate.fileEdits ?? []) { + if ( + incomingKeys.has(fileEditKey(existing)) + || ( + !existing.path + && existing.pending + && incomingToolEventKeys.has(fileEditToolEventKey(existing)) + ) + ) return i; + } + } + return null; +} + +export function finalizeStreamedTurn( + prev: UIMessage[], + turn: UIMessageTurnFields = {}, +): UIMessage[] { + return prev.map((m) => + m.isStreaming && matchesTurn(m, turn) + ? { ...m, isStreaming: false, reasoningStreaming: false } + : m, + ); +} diff --git a/webui/src/tests/fixtures/live-replay-event-projection.json b/webui/src/tests/fixtures/live-replay-event-projection.json new file mode 100644 index 000000000..bc2f593c9 --- /dev/null +++ b/webui/src/tests/fixtures/live-replay-event-projection.json @@ -0,0 +1,600 @@ +{ + "cases": [ + { + "name": "reasoning_then_streamed_answer", + "chat_id": "fixture-reasoning", + "initial_messages": [ + { + "id": "fixture-user-reasoning", + "role": "user", + "content": "Explain event projection.", + "turnId": "turn-reasoning", + "turnPhase": "user", + "turnSeq": 1, + "createdAt": 1700000000000 + } + ], + "live_events": [ + { + "event": "reasoning_delta", + "chat_id": "fixture-reasoning", + "text": "Compare ", + "turn_id": "turn-reasoning", + "turn_phase": "reasoning", + "turn_seq": 2 + }, + { + "event": "reasoning_delta", + "chat_id": "fixture-reasoning", + "text": "state.", + "turn_id": "turn-reasoning", + "turn_phase": "reasoning", + "turn_seq": 3 + }, + { + "event": "reasoning_end", + "chat_id": "fixture-reasoning", + "turn_id": "turn-reasoning", + "turn_phase": "reasoning", + "turn_seq": 4 + }, + { + "event": "delta", + "chat_id": "fixture-reasoning", + "text": "Use one ", + "turn_id": "turn-reasoning", + "turn_phase": "answer", + "turn_seq": 5 + }, + { + "event": "delta", + "chat_id": "fixture-reasoning", + "text": "fold.", + "turn_id": "turn-reasoning", + "turn_phase": "answer", + "turn_seq": 6 + }, + { + "event": "stream_end", + "chat_id": "fixture-reasoning", + "turn_id": "turn-reasoning", + "turn_phase": "answer", + "turn_seq": 7 + }, + { + "event": "turn_end", + "chat_id": "fixture-reasoning", + "latency_ms": 42, + "turn_id": "turn-reasoning", + "turn_phase": "complete", + "turn_seq": 8 + } + ], + "transcript": [ + { + "event": "user", + "chat_id": "fixture-reasoning", + "text": "Explain event projection.", + "turn_id": "turn-reasoning", + "turn_phase": "user", + "turn_seq": 1, + "created_at_ms": 1700000000000 + }, + { + "event": "reasoning_delta", + "chat_id": "fixture-reasoning", + "text": "Compare ", + "turn_id": "turn-reasoning", + "turn_phase": "reasoning", + "turn_seq": 2 + }, + { + "event": "reasoning_delta", + "chat_id": "fixture-reasoning", + "text": "state.", + "turn_id": "turn-reasoning", + "turn_phase": "reasoning", + "turn_seq": 3 + }, + { + "event": "reasoning_end", + "chat_id": "fixture-reasoning", + "turn_id": "turn-reasoning", + "turn_phase": "reasoning", + "turn_seq": 4 + }, + { + "event": "delta", + "chat_id": "fixture-reasoning", + "text": "Use one ", + "turn_id": "turn-reasoning", + "turn_phase": "answer", + "turn_seq": 5 + }, + { + "event": "delta", + "chat_id": "fixture-reasoning", + "text": "fold.", + "turn_id": "turn-reasoning", + "turn_phase": "answer", + "turn_seq": 6 + }, + { + "event": "stream_end", + "chat_id": "fixture-reasoning", + "turn_id": "turn-reasoning", + "turn_phase": "answer", + "turn_seq": 7 + }, + { + "event": "turn_end", + "chat_id": "fixture-reasoning", + "latency_ms": 42, + "turn_id": "turn-reasoning", + "turn_phase": "complete", + "turn_seq": 8 + } + ], + "expected": [ + { + "role": "user", + "content": "Explain event projection.", + "turnId": "turn-reasoning", + "turnPhase": "user", + "turnSeq": 1 + }, + { + "role": "assistant", + "content": "Use one fold.", + "reasoning": "Compare state.", + "activitySegmentId": "segment-1", + "latencyMs": 42, + "turnId": "turn-reasoning", + "turnPhase": "answer", + "turnSeq": 6 + } + ] + }, + { + "name": "length_recovery_merges_answer_segments", + "chat_id": "fixture-length", + "initial_messages": [ + { + "id": "fixture-user-length", + "role": "user", + "content": "Continue after the limit.", + "turnId": "turn-length", + "turnPhase": "user", + "turnSeq": 1, + "createdAt": 1700000001000 + } + ], + "live_events": [ + { + "event": "delta", + "chat_id": "fixture-length", + "text": "first ", + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 2 + }, + { + "event": "stream_end", + "chat_id": "fixture-length", + "text": "first ", + "resuming": true, + "merge_next": true, + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 3 + }, + { + "event": "delta", + "chat_id": "fixture-length", + "text": "second", + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 4 + }, + { + "event": "stream_end", + "chat_id": "fixture-length", + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 5 + }, + { + "event": "turn_end", + "chat_id": "fixture-length", + "latency_ms": 17, + "turn_id": "turn-length", + "turn_phase": "complete", + "turn_seq": 6 + } + ], + "transcript": [ + { + "event": "user", + "chat_id": "fixture-length", + "text": "Continue after the limit.", + "turn_id": "turn-length", + "turn_phase": "user", + "turn_seq": 1, + "created_at_ms": 1700000001000 + }, + { + "event": "delta", + "chat_id": "fixture-length", + "text": "first ", + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 2 + }, + { + "event": "stream_end", + "chat_id": "fixture-length", + "text": "first ", + "resuming": true, + "merge_next": true, + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 3 + }, + { + "event": "delta", + "chat_id": "fixture-length", + "text": "second", + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 4 + }, + { + "event": "stream_end", + "chat_id": "fixture-length", + "turn_id": "turn-length", + "turn_phase": "answer", + "turn_seq": 5 + }, + { + "event": "turn_end", + "chat_id": "fixture-length", + "latency_ms": 17, + "turn_id": "turn-length", + "turn_phase": "complete", + "turn_seq": 6 + } + ], + "expected": [ + { + "role": "user", + "content": "Continue after the limit.", + "turnId": "turn-length", + "turnPhase": "user", + "turnSeq": 1 + }, + { + "role": "assistant", + "content": "first second", + "latencyMs": 17, + "turnId": "turn-length", + "turnPhase": "answer", + "turnSeq": 4 + } + ] + }, + { + "name": "tool_activity_then_complete_answer", + "chat_id": "fixture-activity", + "initial_messages": [ + { + "id": "fixture-user-activity", + "role": "user", + "content": "Inspect the projection code.", + "turnId": "turn-activity", + "turnPhase": "user", + "turnSeq": 1, + "createdAt": 1700000002000 + } + ], + "live_events": [ + { + "event": "message", + "chat_id": "fixture-activity", + "kind": "tool_hint", + "text": "search projection helpers", + "turn_id": "turn-activity", + "turn_phase": "activity", + "turn_seq": 2 + }, + { + "event": "reasoning_delta", + "chat_id": "fixture-activity", + "text": "Review results.", + "turn_id": "turn-activity", + "turn_phase": "reasoning", + "turn_seq": 4 + }, + { + "event": "reasoning_end", + "chat_id": "fixture-activity", + "turn_id": "turn-activity", + "turn_phase": "reasoning", + "turn_seq": 5 + }, + { + "event": "message", + "chat_id": "fixture-activity", + "text": "Projection matches.", + "turn_id": "turn-activity", + "turn_phase": "answer", + "turn_seq": 6 + }, + { + "event": "turn_end", + "chat_id": "fixture-activity", + "latency_ms": 8, + "turn_id": "turn-activity", + "turn_phase": "complete", + "turn_seq": 7 + } + ], + "transcript": [ + { + "event": "user", + "chat_id": "fixture-activity", + "text": "Inspect the projection code.", + "turn_id": "turn-activity", + "turn_phase": "user", + "turn_seq": 1, + "created_at_ms": 1700000002000 + }, + { + "event": "message", + "chat_id": "fixture-activity", + "kind": "tool_hint", + "text": "search projection helpers", + "turn_id": "turn-activity", + "turn_phase": "activity", + "turn_seq": 2 + }, + { + "event": "reasoning_delta", + "chat_id": "fixture-activity", + "text": "Review results.", + "turn_id": "turn-activity", + "turn_phase": "reasoning", + "turn_seq": 4 + }, + { + "event": "reasoning_end", + "chat_id": "fixture-activity", + "turn_id": "turn-activity", + "turn_phase": "reasoning", + "turn_seq": 5 + }, + { + "event": "message", + "chat_id": "fixture-activity", + "text": "Projection matches.", + "turn_id": "turn-activity", + "turn_phase": "answer", + "turn_seq": 6 + }, + { + "event": "turn_end", + "chat_id": "fixture-activity", + "latency_ms": 8, + "turn_id": "turn-activity", + "turn_phase": "complete", + "turn_seq": 7 + } + ], + "expected": [ + { + "role": "user", + "content": "Inspect the projection code.", + "turnId": "turn-activity", + "turnPhase": "user", + "turnSeq": 1 + }, + { + "role": "tool", + "content": "search projection helpers", + "kind": "trace", + "traces": [ + "search projection helpers" + ], + "activitySegmentId": "segment-1", + "turnId": "turn-activity", + "turnPhase": "activity", + "turnSeq": 2 + }, + { + "role": "assistant", + "content": "Projection matches.", + "reasoning": "Review results.", + "activitySegmentId": "segment-1", + "latencyMs": 8, + "turnId": "turn-activity", + "turnPhase": "answer", + "turnSeq": 6 + } + ] + }, + { + "name": "file_edit_lifecycle_merges_by_call_and_path", + "chat_id": "fixture-file-edit", + "initial_messages": [ + { + "id": "fixture-user-file-edit", + "role": "user", + "content": "Update app.py.", + "turnId": "turn-file-edit", + "turnPhase": "user", + "turnSeq": 1, + "createdAt": 1700000003000 + } + ], + "live_events": [ + { + "event": "file_edit", + "chat_id": "fixture-file-edit", + "edits": [ + { + "version": 1, + "call_id": "call-edit", + "tool": "edit_file", + "path": "app.py", + "phase": "start", + "added": 0, + "deleted": 0, + "status": "editing" + } + ], + "turn_id": "turn-file-edit", + "turn_phase": "activity", + "turn_seq": 2 + }, + { + "event": "file_edit", + "chat_id": "fixture-file-edit", + "edits": [ + { + "version": 1, + "call_id": "call-edit", + "tool": "edit_file", + "path": "app.py", + "phase": "end", + "added": 3, + "deleted": 1, + "status": "done" + } + ], + "turn_id": "turn-file-edit", + "turn_phase": "activity", + "turn_seq": 3 + }, + { + "event": "message", + "chat_id": "fixture-file-edit", + "text": "Updated app.py.", + "turn_id": "turn-file-edit", + "turn_phase": "answer", + "turn_seq": 4 + }, + { + "event": "turn_end", + "chat_id": "fixture-file-edit", + "latency_ms": 9, + "turn_id": "turn-file-edit", + "turn_phase": "complete", + "turn_seq": 5 + } + ], + "transcript": [ + { + "event": "user", + "chat_id": "fixture-file-edit", + "text": "Update app.py.", + "turn_id": "turn-file-edit", + "turn_phase": "user", + "turn_seq": 1, + "created_at_ms": 1700000003000 + }, + { + "event": "file_edit", + "chat_id": "fixture-file-edit", + "edits": [ + { + "version": 1, + "call_id": "call-edit", + "tool": "edit_file", + "path": "app.py", + "phase": "start", + "added": 0, + "deleted": 0, + "status": "editing" + } + ], + "turn_id": "turn-file-edit", + "turn_phase": "activity", + "turn_seq": 2 + }, + { + "event": "file_edit", + "chat_id": "fixture-file-edit", + "edits": [ + { + "version": 1, + "call_id": "call-edit", + "tool": "edit_file", + "path": "app.py", + "phase": "end", + "added": 3, + "deleted": 1, + "status": "done" + } + ], + "turn_id": "turn-file-edit", + "turn_phase": "activity", + "turn_seq": 3 + }, + { + "event": "message", + "chat_id": "fixture-file-edit", + "text": "Updated app.py.", + "turn_id": "turn-file-edit", + "turn_phase": "answer", + "turn_seq": 4 + }, + { + "event": "turn_end", + "chat_id": "fixture-file-edit", + "latency_ms": 9, + "turn_id": "turn-file-edit", + "turn_phase": "complete", + "turn_seq": 5 + } + ], + "expected": [ + { + "role": "user", + "content": "Update app.py.", + "turnId": "turn-file-edit", + "turnPhase": "user", + "turnSeq": 1 + }, + { + "role": "tool", + "content": "", + "kind": "trace", + "traces": [], + "fileEdits": [ + { + "version": 1, + "call_id": "call-edit", + "tool": "edit_file", + "path": "app.py", + "phase": "end", + "added": 3, + "deleted": 1, + "status": "done" + } + ], + "activitySegmentId": "segment-1", + "turnId": "turn-file-edit", + "turnPhase": "activity", + "turnSeq": 3 + }, + { + "role": "assistant", + "content": "Updated app.py.", + "latencyMs": 9, + "turnId": "turn-file-edit", + "turnPhase": "answer", + "turnSeq": 4 + } + ] + } + ] +} diff --git a/webui/src/tests/useNanobotStream.test.tsx b/webui/src/tests/useNanobotStream.test.tsx index 80b821c3c..a88aa59fa 100644 --- a/webui/src/tests/useNanobotStream.test.tsx +++ b/webui/src/tests/useNanobotStream.test.tsx @@ -4,10 +4,67 @@ import { describe, expect, it, vi } from "vitest"; import { useNanobotStream } from "@/hooks/useNanobotStream"; import type { StreamError } from "@/lib/nanobot-client"; -import type { ConnectionStatus, InboundEvent, GoalStateWsPayload } from "@/lib/types"; +import type { + ConnectionStatus, + GoalStateWsPayload, + InboundEvent, + UIMessage, +} from "@/lib/types"; import { ClientProvider } from "@/providers/ClientProvider"; +import projectionFixture from "./fixtures/live-replay-event-projection.json"; -const EMPTY_MESSAGES: import("@/lib/types").UIMessage[] = []; +const EMPTY_MESSAGES: UIMessage[] = []; + +interface ProjectionFixtureCase { + name: string; + chat_id: string; + initial_messages: UIMessage[]; + live_events: InboundEvent[]; + expected: Array>; +} + +const PROJECTION_FIXTURE_CASES = ( + projectionFixture as unknown as { cases: ProjectionFixtureCase[] } +).cases; +const SEMANTIC_MESSAGE_FIELDS = [ + "role", + "content", + "kind", + "traces", + "toolEvents", + "fileEdits", + "images", + "media", + "cliApps", + "mcpPresets", + "sessionMentions", + "reasoning", + "latencyMs", + "source", + "turnId", + "turnPhase", + "turnSeq", +] as const satisfies ReadonlyArray; + +function normalizeProjection(messages: UIMessage[]): Array> { + const segmentAliases = new Map(); + return messages.map((message) => { + const row: Record = {}; + for (const field of SEMANTIC_MESSAGE_FIELDS) { + const value = message[field]; + if (value !== undefined && value !== null) row[field] = value; + } + if (message.activitySegmentId) { + let alias = segmentAliases.get(message.activitySegmentId); + if (!alias) { + alias = `segment-${segmentAliases.size + 1}`; + segmentAliases.set(message.activitySegmentId, alias); + } + row.activitySegmentId = alias; + } + return row; + }); +} function fakeClient() { const handlers = new Map void>>(); @@ -2841,3 +2898,21 @@ describe("useNanobotStream", () => { }); }); + +describe("live/replay projection before canonical-event revision migration", () => { + it.each(PROJECTION_FIXTURE_CASES)("matches the shared $name fixture", (fixtureCase) => { + const fake = fakeClient(); + const { result } = renderHook( + () => useNanobotStream(fixtureCase.chat_id, fixtureCase.initial_messages), + { wrapper: wrap(fake.client) }, + ); + + for (const event of fixtureCase.live_events) { + act(() => { + fake.emit(fixtureCase.chat_id, event); + }); + } + + expect(normalizeProjection(result.current.messages)).toEqual(fixtureCase.expected); + }); +});