mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-10 14:28:38 +03:00
Code Issues Packages Projects Releases Wiki Activity

refactor(webui): extract event projection helpers

Browse Source
This commit is contained in:
chengyongru
2026-08-10 16:24:07 +08:00
committed by chengyongru
parent 5d733b1c7c
commit 05d73803e7
5 changed files with 1126 additions and 351 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/utils/test_webui_event_projection_equivalence.py
+65
View File
@@ -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"]
webui/src/hooks/useNanobotStream.ts
+27 -349
View File
@@ -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<UIMessage, "turnId" | "turnPhase" | "turnSeq">;
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<string>,
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<UIMessage, "latencyMs" | "completedAt">,
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<UIMessage, "id" | "role" | "createdAt">,
@@ -275,193 +148,6 @@ function absorbCompleteAssistantMessage(
];
}
function fileEditKey(edit: Pick<UIFileEdit, "call_id" | "tool" | "path">): 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<UIFileEdit, "call_id" | "tool" | "path">): 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<string>();
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).
webui/src/lib/thread-event-projection.ts
+357
View File
@@ -0,0 +1,357 @@
import { toolTraceLinesFromEvents } from "@/lib/tool-traces";
import type {
ToolProgressEvent,
UIFileEdit,
UIMessage,
UITurnPhase,
} from "@/lib/types";
export type UIMessageTurnFields = Pick<UIMessage, "turnId" | "turnPhase" | "turnSeq">;
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<string>,
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<UIMessage, "latencyMs" | "completedAt">,
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<UIFileEdit, "call_id" | "tool" | "path">): 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<UIFileEdit, "call_id" | "tool" | "path">,
): 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<string>();
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,
);
}
webui/src/tests/fixtures/live-replay-event-projection.json
Vendored
+600
View File
@@ -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
}
]
}
]
}
webui/src/tests/useNanobotStream.test.tsx
+77 -2
View File
@@ -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<Record<string, unknown>>;
}
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<keyof UIMessage>;
function normalizeProjection(messages: UIMessage[]): Array<Record<string, unknown>> {
const segmentAliases = new Map<string, string>();
return messages.map((message) => {
const row: Record<string, unknown> = {};
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<string, Set<(ev: InboundEvent) => 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);
});
});