mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-04 16:38:49 +00:00
1465 lines
50 KiB
TypeScript
1465 lines
50 KiB
TypeScript
import { useCallback, useEffect, useRef, useState } from "react";
|
|
|
|
import { useClient } from "@/providers/ClientProvider";
|
|
import { toMediaAttachment } from "@/lib/media";
|
|
import {
|
|
mergeToolProgressEvents,
|
|
mergeUniqueToolTraceLines,
|
|
normalizeToolProgressEvents,
|
|
toolTraceLinesFromEvents,
|
|
} from "@/lib/tool-traces";
|
|
import { hasPendingAgentActivity } from "@/lib/activity-timeline";
|
|
import type { StreamError } from "@/lib/nanobot-client";
|
|
import { formatQuotedUserMessage } from "@/lib/user-message-quote";
|
|
import type {
|
|
InboundEvent,
|
|
OutboundCliAppMention,
|
|
OutboundMcpPresetMention,
|
|
OutboundMedia,
|
|
GoalStateWsPayload,
|
|
MessageDeliveryStatus,
|
|
ToolProgressEvent,
|
|
UIMediaAttachment,
|
|
UIFileEdit,
|
|
UIMessage,
|
|
UITurnPhase,
|
|
WorkspaceScopePayload,
|
|
} from "@/lib/types";
|
|
|
|
interface StreamBuffer {
|
|
/** ID of the assistant message currently receiving deltas (cleared when its segment closes). */
|
|
messageId: string;
|
|
}
|
|
|
|
interface ActiveAssistantCursor {
|
|
id: string;
|
|
index: number;
|
|
}
|
|
|
|
type PendingStreamEvent =
|
|
| { kind: "delta"; text: string; turn: UIMessageTurnFields }
|
|
| { 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``.
|
|
*
|
|
* Lookup rule: reasoning can only extend the current reasoning placeholder.
|
|
* Once ordinary answer text has appeared, the next reasoning chunk starts a
|
|
* fresh Thought block so streamed output stays in arrival order:
|
|
* Thought -> answer -> Thought -> answer.
|
|
*/
|
|
function attachReasoningChunk(
|
|
prev: UIMessage[],
|
|
chunk: string,
|
|
segments?: {
|
|
ensure: () => string;
|
|
},
|
|
turn: UIMessageTurnFields = {},
|
|
): UIMessage[] {
|
|
for (let i = prev.length - 1; i >= 0; i -= 1) {
|
|
const candidate = prev[i];
|
|
// A user turn is a hard boundary: reasoning after it belongs to the new
|
|
// assistant turn, never to an earlier assistant reply.
|
|
if (candidate.role === "user") break;
|
|
// A trace row (e.g. Used tools) is also a phase boundary. Reasoning after
|
|
// tools belongs to the next assistant iteration, not the assistant turn
|
|
// that produced those tool calls.
|
|
if (candidate.kind === "trace") break;
|
|
if (candidate.role !== "assistant") continue;
|
|
if (!matchesTurn(candidate, turn)) break;
|
|
const activitySegmentId = candidate.activitySegmentId ?? segments?.ensure();
|
|
const hasAnswer = candidate.content.length > 0;
|
|
if (hasAnswer) break;
|
|
if (
|
|
candidate.reasoningStreaming
|
|
|| candidate.reasoning !== undefined
|
|
|| candidate.isStreaming
|
|
) {
|
|
const merged: UIMessage = {
|
|
...candidate,
|
|
reasoning: (candidate.reasoning ?? "") + chunk,
|
|
reasoningStreaming: true,
|
|
...(activitySegmentId ? { activitySegmentId } : {}),
|
|
...turn,
|
|
};
|
|
return [...prev.slice(0, i), merged, ...prev.slice(i + 1)];
|
|
}
|
|
break;
|
|
}
|
|
const activitySegmentId = segments?.ensure();
|
|
return [
|
|
...prev,
|
|
{
|
|
id: crypto.randomUUID(),
|
|
role: "assistant",
|
|
content: "",
|
|
isStreaming: true,
|
|
reasoning: chunk,
|
|
reasoningStreaming: true,
|
|
...(activitySegmentId ? { activitySegmentId } : {}),
|
|
...turn,
|
|
createdAt: Date.now(),
|
|
},
|
|
];
|
|
}
|
|
|
|
/**
|
|
* 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">,
|
|
): UIMessage[] {
|
|
const last = prev[prev.length - 1];
|
|
if (!last || !isReasoningOnlyPlaceholder(last) || !matchesTurn(last, message)) {
|
|
return [
|
|
...prev,
|
|
{
|
|
id: crypto.randomUUID(),
|
|
role: "assistant",
|
|
createdAt: Date.now(),
|
|
...message,
|
|
},
|
|
];
|
|
}
|
|
return [
|
|
...prev.slice(0, -1),
|
|
{
|
|
...last,
|
|
...message,
|
|
isStreaming: false,
|
|
reasoningStreaming: false,
|
|
},
|
|
];
|
|
}
|
|
|
|
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
|
|
* separately (e.g. via ``fetchWebuiThread``) since the server only replays
|
|
* live events.
|
|
*/
|
|
/** Payload passed to ``send`` when the user attaches one or more files.
|
|
*
|
|
* ``media`` is handed to the wire client verbatim; ``preview`` powers the
|
|
* optimistic user bubble. Keeping the two separate lets the bubble re-use the
|
|
* local data URL even after the server persists the file under a different
|
|
* name. */
|
|
export interface SendAttachment {
|
|
media: OutboundMedia;
|
|
preview: UIMediaAttachment;
|
|
}
|
|
|
|
export interface SendOptions {
|
|
cliApps?: OutboundCliAppMention[];
|
|
mcpPresets?: OutboundMcpPresetMention[];
|
|
quotedContext?: string;
|
|
workspaceScope?: WorkspaceScopePayload | null;
|
|
sideChannel?: boolean;
|
|
finalizeActiveTurn?: boolean;
|
|
/** Append guidance to the running turn without detaching its active answer segment. */
|
|
continueActiveTurn?: boolean;
|
|
}
|
|
|
|
export interface SubmittedTurn {
|
|
turnId: string;
|
|
userMessageId: string;
|
|
sideChannel: boolean;
|
|
}
|
|
|
|
function eventExtendsModelActivity(ev: InboundEvent): boolean {
|
|
if (
|
|
ev.event === "delta"
|
|
|| ev.event === "reasoning_delta"
|
|
|| ev.event === "file_edit"
|
|
) return true;
|
|
return ev.event === "message"
|
|
&& (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;
|
|
}
|
|
|
|
function transitionTurnDelivery(
|
|
messages: UIMessage[],
|
|
turnId: string,
|
|
status: MessageDeliveryStatus,
|
|
): UIMessage[] {
|
|
let changed = false;
|
|
const next = messages.map((message) => {
|
|
if (
|
|
message.role !== "user"
|
|
|| message.turnId !== turnId
|
|
|| message.deliveryStatus === status
|
|
|| (status === "accepted" && message.deliveryStatus !== "sending")
|
|
) {
|
|
return message;
|
|
}
|
|
changed = true;
|
|
return { ...message, deliveryStatus: status };
|
|
});
|
|
return changed ? next : messages;
|
|
}
|
|
|
|
export function useNanobotStream(
|
|
chatId: string | null,
|
|
initialMessages: UIMessage[] = [],
|
|
hasPendingToolCalls = false,
|
|
onTurnEnd?: () => void,
|
|
): {
|
|
messages: UIMessage[];
|
|
/** Whether ``messages`` belongs to the current ``chatId`` after a session switch. */
|
|
messagesReady: boolean;
|
|
isStreaming: boolean;
|
|
/** Unix epoch seconds when the current user turn started (WebSocket ``goal_status``). */
|
|
runStartedAt: number | null;
|
|
/** Latest sustained goal for this ``chatId`` (``goal_state`` WS events). */
|
|
goalState: GoalStateWsPayload | undefined;
|
|
send: (
|
|
content: string,
|
|
images?: SendAttachment[],
|
|
options?: SendOptions,
|
|
) => SubmittedTurn | null;
|
|
transcribeAudio: (dataUrl: string, options?: { durationMs?: number }) => Promise<string>;
|
|
stop: () => void;
|
|
/** Mark an accepted canonical snapshot as the definitive end of the active turn. */
|
|
reconcileTurnComplete: () => void;
|
|
setMessages: React.Dispatch<React.SetStateAction<UIMessage[]>>;
|
|
/** Latest transport-level fault raised since the last ``dismissStreamError``.
|
|
* ``null`` when there is nothing to show. */
|
|
streamError: StreamError | null;
|
|
/** Clear the current ``streamError`` (e.g. after the user dismisses the
|
|
* notification or starts a fresh action). */
|
|
dismissStreamError: () => void;
|
|
} {
|
|
const { client } = useClient();
|
|
const initialRunStartedAt = chatId ? client.getRunStartedAt(chatId) : null;
|
|
const [messages, setMessages] = useState<UIMessage[]>(initialMessages);
|
|
const [messageOwnerChatId, setMessageOwnerChatId] = useState(chatId);
|
|
/** If history ends in unfinished agent activity, keep the loading spinner alive. */
|
|
const initialStreaming = hasPendingAgentActivity(initialMessages);
|
|
const [isStreaming, setIsStreaming] = useState(
|
|
initialStreaming || hasPendingToolCalls || initialRunStartedAt !== null,
|
|
);
|
|
/** Unix epoch seconds when the current user turn started; cleared on ``idle``. */
|
|
const [runStartedAt, setRunStartedAt] = useState<number | null>(initialRunStartedAt);
|
|
const [goalState, setGoalState] = useState<GoalStateWsPayload | undefined>(undefined);
|
|
const [streamError, setStreamError] = useState<StreamError | null>(null);
|
|
const buffer = useRef<StreamBuffer | null>(null);
|
|
const activeAssistantRef = useRef<ActiveAssistantCursor | null>(null);
|
|
const closedAssistantStreamIdsRef = useRef<Set<string>>(new Set());
|
|
const activitySegmentRef = useRef<string | null>(null);
|
|
const fileEditSegmentRef = useRef<string | null>(null);
|
|
const activitySegmentCounterRef = useRef(0);
|
|
const pendingStreamEventsRef = useRef<PendingStreamEvent[]>([]);
|
|
const streamFrameRef = useRef<number | null>(null);
|
|
const streamTimerRef = useRef<number | null>(null);
|
|
const suppressStreamUntilTurnEndRef = useRef(false);
|
|
const sideChannelTurnIdsRef = useRef<Set<string>>(new Set());
|
|
/** Timer that defers ``isStreaming = false`` after ``stream_end``.
|
|
*
|
|
* When the model finishes a text segment and calls a tool, the server
|
|
* sends ``stream_end`` but the agent is still "thinking" while the tool
|
|
* executes. By deferring the flag reset by a short window (1 s) we keep
|
|
* the loading spinner alive across tool-call boundaries without needing
|
|
* backend changes. */
|
|
const streamEndTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
|
|
|
const dismissStreamError = useCallback(() => setStreamError(null), []);
|
|
|
|
const clearPendingStreamWork = useCallback(() => {
|
|
if (streamFrameRef.current !== null) {
|
|
window.cancelAnimationFrame(streamFrameRef.current);
|
|
streamFrameRef.current = null;
|
|
}
|
|
if (streamTimerRef.current !== null) {
|
|
window.clearTimeout(streamTimerRef.current);
|
|
streamTimerRef.current = null;
|
|
}
|
|
pendingStreamEventsRef.current = [];
|
|
}, []);
|
|
|
|
const cancelStreamEndTimer = useCallback(() => {
|
|
if (streamEndTimerRef.current === null) return;
|
|
clearTimeout(streamEndTimerRef.current);
|
|
streamEndTimerRef.current = null;
|
|
}, []);
|
|
|
|
const isSideChannelEvent = useCallback((ev: InboundEvent) => {
|
|
const turnId = eventTurnId(ev);
|
|
return turnId !== undefined && sideChannelTurnIdsRef.current.has(turnId);
|
|
}, []);
|
|
|
|
const scheduleStreamEndTimer = useCallback((turn: UIMessageTurnFields = {}) => {
|
|
cancelStreamEndTimer();
|
|
streamEndTimerRef.current = setTimeout(() => {
|
|
streamEndTimerRef.current = null;
|
|
setIsStreaming(false);
|
|
setMessages((prev) => finalizeStreamedTurn(prev, turn));
|
|
}, STREAM_END_IDLE_DELAY_MS);
|
|
}, [cancelStreamEndTimer]);
|
|
|
|
const createActivitySegmentId = useCallback((activate = true) => {
|
|
activitySegmentCounterRef.current += 1;
|
|
const id = `activity-${activitySegmentCounterRef.current}`;
|
|
if (activate) activitySegmentRef.current = id;
|
|
return id;
|
|
}, []);
|
|
|
|
const freshActivitySegmentId = useCallback(
|
|
() => createActivitySegmentId(true),
|
|
[createActivitySegmentId],
|
|
);
|
|
|
|
const detachedActivitySegmentId = useCallback(
|
|
() => createActivitySegmentId(false),
|
|
[createActivitySegmentId],
|
|
);
|
|
|
|
const ensureActivitySegmentId = useCallback(() => {
|
|
if (activitySegmentRef.current) return activitySegmentRef.current;
|
|
return freshActivitySegmentId();
|
|
}, [freshActivitySegmentId]);
|
|
|
|
const clearActivitySegment = useCallback(() => {
|
|
activitySegmentRef.current = null;
|
|
fileEditSegmentRef.current = null;
|
|
}, []);
|
|
|
|
const closeActiveAssistantStream = useCallback(() => {
|
|
const closedStreamId = buffer.current?.messageId ?? activeAssistantRef.current?.id;
|
|
if (closedStreamId) closedAssistantStreamIdsRef.current.add(closedStreamId);
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
return !!closedStreamId;
|
|
}, []);
|
|
|
|
const applyStreamError = useCallback((err: StreamError) => {
|
|
// One multiplexed client serves every thread. A correlated send fault
|
|
// belongs only to its target chat. An uncorrelated transport close can
|
|
// still be shown in the mounted thread, but cannot roll back any turn.
|
|
if (!chatId || (err.chatId && err.chatId !== chatId)) return;
|
|
setStreamError(err);
|
|
if (!err.turnId) return;
|
|
|
|
const rejectedTurnId = err.turnId;
|
|
pendingStreamEventsRef.current = pendingStreamEventsRef.current.filter(
|
|
(event) => event.turn.turnId !== rejectedTurnId,
|
|
);
|
|
sideChannelTurnIdsRef.current.delete(rejectedTurnId);
|
|
cancelStreamEndTimer();
|
|
setMessages((prev) => {
|
|
const rejectedRows = prev.filter((message) => message.turnId === rejectedTurnId);
|
|
if (rejectedRows.length === 0) return prev;
|
|
const rejectedIds = new Set(rejectedRows.map((message) => message.id));
|
|
const rejectedSegments = new Set(
|
|
rejectedRows
|
|
.map((message) => message.activitySegmentId)
|
|
.filter((segmentId): segmentId is string => typeof segmentId === "string"),
|
|
);
|
|
if (
|
|
activeAssistantRef.current
|
|
&& rejectedIds.has(activeAssistantRef.current.id)
|
|
) {
|
|
activeAssistantRef.current = null;
|
|
}
|
|
if (buffer.current && rejectedIds.has(buffer.current.messageId)) {
|
|
buffer.current = null;
|
|
}
|
|
for (const id of rejectedIds) closedAssistantStreamIdsRef.current.delete(id);
|
|
if (
|
|
activitySegmentRef.current
|
|
&& rejectedSegments.has(activitySegmentRef.current)
|
|
) {
|
|
activitySegmentRef.current = null;
|
|
}
|
|
if (
|
|
fileEditSegmentRef.current
|
|
&& rejectedSegments.has(fileEditSegmentRef.current)
|
|
) {
|
|
fileEditSegmentRef.current = null;
|
|
}
|
|
return prev.flatMap((message) => {
|
|
if (message.turnId !== rejectedTurnId) return [message];
|
|
if (message.role !== "user") return [];
|
|
return [{
|
|
...message,
|
|
deliveryStatus: "failed",
|
|
deliveryErrorKind: err.kind,
|
|
}];
|
|
});
|
|
});
|
|
|
|
const remainingStartedAt = client.getRunStartedAt(chatId);
|
|
const hasRemainingRun = (
|
|
remainingStartedAt !== null
|
|
|| client.hasUnsettledRun(chatId)
|
|
);
|
|
setRunStartedAt(remainingStartedAt);
|
|
setIsStreaming(hasRemainingRun);
|
|
if (!hasRemainingRun) suppressStreamUntilTurnEndRef.current = false;
|
|
}, [cancelStreamEndTimer, chatId, client]);
|
|
|
|
useEffect(() => client.onError(applyStreamError), [applyStreamError, client]);
|
|
|
|
const resolveActiveAssistantIndex = useCallback((
|
|
prev: UIMessage[],
|
|
turn: UIMessageTurnFields = {},
|
|
): number | null => {
|
|
const cursor = activeAssistantRef.current;
|
|
if (!cursor) return null;
|
|
const indexed = prev[cursor.index];
|
|
if (
|
|
indexed?.id === cursor.id
|
|
&& indexed.role === "assistant"
|
|
&& indexed.kind !== "trace"
|
|
&& indexed.isStreaming
|
|
&& matchesTurn(indexed, turn)
|
|
) {
|
|
return cursor.index;
|
|
}
|
|
const idx = prev.findIndex((m) => m.id === cursor.id);
|
|
if (idx === -1) {
|
|
activeAssistantRef.current = null;
|
|
return null;
|
|
}
|
|
const found = prev[idx];
|
|
if (
|
|
found.role !== "assistant"
|
|
|| found.kind === "trace"
|
|
|| !found.isStreaming
|
|
|| !matchesTurn(found, turn)
|
|
) {
|
|
activeAssistantRef.current = null;
|
|
return null;
|
|
}
|
|
activeAssistantRef.current = { id: cursor.id, index: idx };
|
|
return idx;
|
|
}, []);
|
|
|
|
const appendAnswerChunk = useCallback(
|
|
(prev: UIMessage[], chunk: string, turn: UIMessageTurnFields = {}): UIMessage[] => {
|
|
let next = prev;
|
|
let targetIndex = resolveActiveAssistantIndex(next, turn);
|
|
|
|
if (targetIndex === null) {
|
|
targetIndex = findActiveAssistantPlaceholderIndex(next, turn);
|
|
}
|
|
if (targetIndex === null) {
|
|
targetIndex = findStreamingAssistantIndex(next, closedAssistantStreamIdsRef.current, turn);
|
|
}
|
|
if (targetIndex === null) {
|
|
const id = crypto.randomUUID();
|
|
next = [
|
|
...next,
|
|
{
|
|
id,
|
|
role: "assistant",
|
|
content: "",
|
|
isStreaming: true,
|
|
createdAt: Date.now(),
|
|
},
|
|
];
|
|
targetIndex = next.length - 1;
|
|
}
|
|
|
|
const target = next[targetIndex];
|
|
const merged: UIMessage = {
|
|
...target,
|
|
content: target.content + chunk,
|
|
isStreaming: true,
|
|
...turn,
|
|
};
|
|
closedAssistantStreamIdsRef.current.delete(merged.id);
|
|
activeAssistantRef.current = { id: merged.id, index: targetIndex };
|
|
buffer.current = { messageId: merged.id };
|
|
return replaceMessageAt(next, targetIndex, merged);
|
|
},
|
|
[resolveActiveAssistantIndex],
|
|
);
|
|
|
|
const applyPendingStreamEvents = useCallback(
|
|
(prev: UIMessage[], events: PendingStreamEvent[]): UIMessage[] => {
|
|
let next = prev;
|
|
for (const event of events) {
|
|
if (event.kind === "delta") {
|
|
next = appendAnswerChunk(next, event.text, event.turn);
|
|
} else {
|
|
if (closeActiveAssistantStream()) clearActivitySegment();
|
|
next = attachReasoningChunk(
|
|
next,
|
|
event.text,
|
|
{ ensure: ensureActivitySegmentId },
|
|
event.turn,
|
|
);
|
|
}
|
|
}
|
|
return next;
|
|
},
|
|
[appendAnswerChunk, clearActivitySegment, closeActiveAssistantStream, ensureActivitySegmentId],
|
|
);
|
|
|
|
const flushPendingStreamEvents = useCallback((options?: {
|
|
closeAnswerSegment?: boolean;
|
|
finalAnswerText?: string;
|
|
turn?: UIMessageTurnFields;
|
|
}) => {
|
|
if (streamFrameRef.current !== null) {
|
|
window.cancelAnimationFrame(streamFrameRef.current);
|
|
streamFrameRef.current = null;
|
|
}
|
|
if (streamTimerRef.current !== null) {
|
|
window.clearTimeout(streamTimerRef.current);
|
|
streamTimerRef.current = null;
|
|
}
|
|
const events = pendingStreamEventsRef.current;
|
|
const finalAnswerText = options?.finalAnswerText;
|
|
const turn = options?.turn ?? {};
|
|
if (events.length === 0 && finalAnswerText === undefined) {
|
|
if (options?.closeAnswerSegment) closeActiveAssistantStream();
|
|
return;
|
|
}
|
|
pendingStreamEventsRef.current = [];
|
|
setMessages((prev) => {
|
|
let next = events.length > 0 ? applyPendingStreamEvents(prev, events) : prev;
|
|
if (finalAnswerText !== undefined) {
|
|
const targetIndex =
|
|
resolveActiveAssistantIndex(next, turn)
|
|
?? findStreamingAssistantIndex(next, closedAssistantStreamIdsRef.current, turn);
|
|
if (targetIndex !== null) {
|
|
const target = next[targetIndex];
|
|
const merged = {
|
|
...target,
|
|
content: finalAnswerText,
|
|
isStreaming: true,
|
|
...turn,
|
|
};
|
|
next = replaceMessageAt(next, targetIndex, merged);
|
|
if (!options?.closeAnswerSegment) {
|
|
closedAssistantStreamIdsRef.current.delete(merged.id);
|
|
activeAssistantRef.current = { id: merged.id, index: targetIndex };
|
|
buffer.current = { messageId: merged.id };
|
|
}
|
|
} else {
|
|
const id = crypto.randomUUID();
|
|
next = [
|
|
...next,
|
|
{
|
|
id,
|
|
role: "assistant",
|
|
content: finalAnswerText,
|
|
isStreaming: true,
|
|
...turn,
|
|
createdAt: Date.now(),
|
|
},
|
|
];
|
|
if (options?.closeAnswerSegment) {
|
|
closedAssistantStreamIdsRef.current.add(id);
|
|
} else {
|
|
activeAssistantRef.current = { id, index: next.length - 1 };
|
|
buffer.current = { messageId: id };
|
|
}
|
|
}
|
|
}
|
|
if (options?.closeAnswerSegment) closeActiveAssistantStream();
|
|
return next;
|
|
});
|
|
}, [applyPendingStreamEvents, closeActiveAssistantStream, resolveActiveAssistantIndex]);
|
|
|
|
const schedulePendingStreamFlush = useCallback(() => {
|
|
if (streamFrameRef.current !== null || streamTimerRef.current !== null) return;
|
|
if (document.visibilityState === "hidden") {
|
|
streamTimerRef.current = window.setTimeout(() => {
|
|
streamTimerRef.current = null;
|
|
const events = pendingStreamEventsRef.current;
|
|
if (events.length === 0) return;
|
|
pendingStreamEventsRef.current = [];
|
|
setMessages((prev) => applyPendingStreamEvents(prev, events));
|
|
}, BACKGROUND_STREAM_FLUSH_INTERVAL_MS);
|
|
return;
|
|
}
|
|
streamFrameRef.current = window.requestAnimationFrame(() => {
|
|
streamFrameRef.current = null;
|
|
const events = pendingStreamEventsRef.current;
|
|
if (events.length === 0) return;
|
|
pendingStreamEventsRef.current = [];
|
|
setMessages((prev) => applyPendingStreamEvents(prev, events));
|
|
});
|
|
}, [applyPendingStreamEvents]);
|
|
|
|
useEffect(() => {
|
|
const flushOnReturn = () => {
|
|
if (document.visibilityState !== "visible") return;
|
|
if (pendingStreamEventsRef.current.length === 0) return;
|
|
flushPendingStreamEvents();
|
|
};
|
|
document.addEventListener("visibilitychange", flushOnReturn);
|
|
return () => document.removeEventListener("visibilitychange", flushOnReturn);
|
|
}, [flushPendingStreamEvents]);
|
|
|
|
useEffect(() => {
|
|
return client.onStatus((status) => {
|
|
if (status !== "reconnecting" && status !== "closed") return;
|
|
// A transport drop does not prove the backend turn completed. Keep the
|
|
// semantic running state intact so queued guidance is not flushed early.
|
|
cancelStreamEndTimer();
|
|
});
|
|
}, [cancelStreamEndTimer, client]);
|
|
|
|
// Reset local state when switching chats. Do not reset on every
|
|
// ``initialMessages`` update: a brand-new chat can receive an empty/404
|
|
// history response after the optimistic first message has already rendered.
|
|
useEffect(() => {
|
|
const restoredRunStartedAt = chatId ? client.getRunStartedAt(chatId) : null;
|
|
setMessages(initialMessages);
|
|
setMessageOwnerChatId(chatId);
|
|
setIsStreaming(
|
|
hasPendingAgentActivity(initialMessages)
|
|
|| hasPendingToolCalls
|
|
|| restoredRunStartedAt !== null,
|
|
);
|
|
setStreamError(null);
|
|
setRunStartedAt(restoredRunStartedAt);
|
|
setGoalState(chatId ? client.getGoalState(chatId) : undefined);
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
closedAssistantStreamIdsRef.current.clear();
|
|
clearActivitySegment();
|
|
clearPendingStreamWork();
|
|
sideChannelTurnIdsRef.current.clear();
|
|
suppressStreamUntilTurnEndRef.current = false;
|
|
cancelStreamEndTimer();
|
|
// eslint-disable-next-line react-hooks/exhaustive-deps
|
|
}, [chatId, client, cancelStreamEndTimer, clearActivitySegment, clearPendingStreamWork]);
|
|
|
|
useEffect(() => {
|
|
if (hasPendingToolCalls) setIsStreaming(true);
|
|
}, [hasPendingToolCalls]);
|
|
|
|
useEffect(() => {
|
|
if (!chatId) return;
|
|
|
|
const handle = (ev: InboundEvent) => {
|
|
if (ev.event === "error") {
|
|
if (ev.detail === "message_too_big") {
|
|
applyStreamError({
|
|
kind: "message_too_big",
|
|
chatId,
|
|
turnId: ev.turn_id,
|
|
});
|
|
} else if (ev.detail === "workspace_scope_rejected") {
|
|
applyStreamError({
|
|
kind: "workspace_scope_rejected",
|
|
reason: ev.reason,
|
|
chatId,
|
|
turnId: ev.turn_id,
|
|
});
|
|
} else if (ev.turn_id) {
|
|
applyStreamError({
|
|
kind: "turn_rejected",
|
|
detail: ev.detail,
|
|
reason: ev.reason,
|
|
chatId,
|
|
turnId: ev.turn_id,
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
const turnId = eventTurnId(ev);
|
|
if (turnId) {
|
|
setMessages((prev) => transitionTurnDelivery(prev, turnId, "accepted"));
|
|
}
|
|
if (ev.event === "message_accepted") return;
|
|
const sideChannelEvent = isSideChannelEvent(ev);
|
|
if (
|
|
streamEndTimerRef.current !== null
|
|
&& !sideChannelEvent
|
|
&& eventExtendsModelActivity(ev)
|
|
) cancelStreamEndTimer();
|
|
|
|
if (ev.event === "delta") {
|
|
if (suppressStreamUntilTurnEndRef.current) return;
|
|
const chunk = typeof ev.text === "string" ? ev.text : "";
|
|
if (!chunk) return;
|
|
clearActivitySegment();
|
|
setIsStreaming(true);
|
|
pendingStreamEventsRef.current.push({
|
|
kind: "delta",
|
|
text: chunk,
|
|
turn: turnFieldsFromEvent(ev, "answer"),
|
|
});
|
|
schedulePendingStreamFlush();
|
|
return;
|
|
}
|
|
|
|
if (ev.event === "reasoning_delta") {
|
|
if (suppressStreamUntilTurnEndRef.current) return;
|
|
const chunk = ev.text;
|
|
if (!chunk) return;
|
|
if (fileEditSegmentRef.current) clearActivitySegment();
|
|
setIsStreaming(true);
|
|
pendingStreamEventsRef.current.push({
|
|
kind: "reasoning",
|
|
text: chunk,
|
|
turn: turnFieldsFromEvent(ev, "reasoning"),
|
|
});
|
|
schedulePendingStreamFlush();
|
|
return;
|
|
}
|
|
|
|
if (ev.event === "stream_end") {
|
|
const turn = turnFieldsFromEvent(ev, "answer");
|
|
const mergeNext = ev.resuming === true && ev.merge_next === true;
|
|
flushPendingStreamEvents({
|
|
closeAnswerSegment: !mergeNext,
|
|
...(typeof ev.text === "string" ? { finalAnswerText: ev.text } : {}),
|
|
turn,
|
|
});
|
|
if (suppressStreamUntilTurnEndRef.current) return;
|
|
if (ev.resuming) {
|
|
cancelStreamEndTimer();
|
|
setIsStreaming(true);
|
|
if (!mergeNext) {
|
|
setMessages((prev) => finalizeStreamedTurn(prev, turn));
|
|
}
|
|
return;
|
|
}
|
|
scheduleStreamEndTimer(turn);
|
|
return;
|
|
}
|
|
|
|
const shouldCloseAnswerBeforeEvent =
|
|
ev.event === "file_edit"
|
|
|| (
|
|
ev.event === "message"
|
|
&& (ev.kind === "tool_hint" || ev.kind === "progress")
|
|
);
|
|
flushPendingStreamEvents({ closeAnswerSegment: shouldCloseAnswerBeforeEvent });
|
|
|
|
if (ev.event === "reasoning_end") {
|
|
if (suppressStreamUntilTurnEndRef.current) return;
|
|
setMessages((prev) => closeReasoningStream(prev));
|
|
return;
|
|
}
|
|
|
|
if (ev.event === "goal_state") {
|
|
setGoalState(ev.goal_state);
|
|
return;
|
|
}
|
|
|
|
if (ev.event === "goal_status") {
|
|
if (ev.status === "running" && typeof ev.started_at === "number") {
|
|
setRunStartedAt(ev.started_at);
|
|
setIsStreaming(true);
|
|
} else {
|
|
setRunStartedAt(null);
|
|
setIsStreaming(false);
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (ev.event === "turn_end") {
|
|
if ("goal_state" in ev && ev.goal_state != null && typeof ev.goal_state === "object") {
|
|
setGoalState(ev.goal_state);
|
|
}
|
|
setRunStartedAt(null);
|
|
// Definitive signal that the turn is fully complete. Cancel any
|
|
// pending debounce timer and stop the loading indicator immediately.
|
|
cancelStreamEndTimer();
|
|
setIsStreaming(false);
|
|
const completedAt = Date.now();
|
|
setMessages((prev) => {
|
|
let finalized = prev.map((m) => (m.isStreaming ? { ...m, isStreaming: false } : m));
|
|
finalized = pruneReasoningOnlyPlaceholders(finalized);
|
|
const latencyMs =
|
|
typeof ev.latency_ms === "number" && ev.latency_ms >= 0
|
|
? Math.round(ev.latency_ms)
|
|
: undefined;
|
|
finalized = stampLastAssistantCompletion(
|
|
finalized,
|
|
{
|
|
...(latencyMs !== undefined ? { latencyMs } : {}),
|
|
completedAt,
|
|
},
|
|
ev.turn_id,
|
|
);
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
clearActivitySegment();
|
|
closedAssistantStreamIdsRef.current.clear();
|
|
return finalized;
|
|
});
|
|
suppressStreamUntilTurnEndRef.current = false;
|
|
onTurnEnd?.();
|
|
return;
|
|
}
|
|
|
|
if (ev.event === "message") {
|
|
if (
|
|
suppressStreamUntilTurnEndRef.current &&
|
|
(ev.kind === "tool_hint" || ev.kind === "progress" || ev.kind === "reasoning")
|
|
) {
|
|
return;
|
|
}
|
|
// Back-compat: a legacy ``kind: "reasoning"`` message (no streaming
|
|
// partner) is treated as one complete delta + immediate end so the
|
|
// bubble renders identically to the streaming path.
|
|
if (ev.kind === "reasoning") {
|
|
const line = ev.text;
|
|
if (!line) return;
|
|
if (fileEditSegmentRef.current) clearActivitySegment();
|
|
setMessages((prev) => closeReasoningStream(attachReasoningChunk(
|
|
prev,
|
|
line,
|
|
{ ensure: ensureActivitySegmentId },
|
|
turnFieldsFromEvent(ev, "reasoning"),
|
|
)));
|
|
return;
|
|
}
|
|
// Intermediate agent breadcrumbs (tool-call hints, raw progress).
|
|
// Attach them to the last trace row if it was the last emitted item
|
|
// so a sequence of calls collapses into one compact trace group.
|
|
if (ev.kind === "tool_hint" || ev.kind === "progress") {
|
|
const structuredEvents = normalizeToolProgressEvents(ev.tool_events);
|
|
const turn = turnFieldsFromEvent(ev, "activity");
|
|
setMessages((prev) => {
|
|
const segmentId = ensureActivitySegmentId();
|
|
const base = prev;
|
|
const visibleStructuredEvents = filterCoveredFileEditToolEvents(base, structuredEvents);
|
|
const structuredLines = toolTraceLinesFromEvents(visibleStructuredEvents);
|
|
const lines = structuredLines.length > 0
|
|
? structuredLines
|
|
: structuredEvents.length > 0
|
|
? []
|
|
: ev.text
|
|
? [ev.text]
|
|
: [];
|
|
if (lines.length === 0) return base;
|
|
const last = base[base.length - 1];
|
|
if (
|
|
last
|
|
&& last.kind === "trace"
|
|
&& !last.isStreaming
|
|
&& (!last.activitySegmentId || last.activitySegmentId === segmentId)
|
|
) {
|
|
const previousTraces = last.traces?.length
|
|
? last.traces
|
|
: last.content
|
|
? [last.content]
|
|
: [];
|
|
const mergedLines = visibleStructuredEvents.length > 0
|
|
? mergeUniqueToolTraceLines(previousTraces, structuredLines)
|
|
: null;
|
|
const merged: UIMessage = {
|
|
...last,
|
|
traces: mergedLines ? mergedLines.traces : [...previousTraces, ...lines],
|
|
content: mergedLines
|
|
? mergedLines.traces[mergedLines.traces.length - 1]
|
|
: lines[lines.length - 1],
|
|
toolEvents: visibleStructuredEvents.length
|
|
? mergeToolProgressEvents(last.toolEvents, visibleStructuredEvents)
|
|
: last.toolEvents,
|
|
activitySegmentId: last.activitySegmentId ?? segmentId,
|
|
...turn,
|
|
};
|
|
return [...base.slice(0, -1), merged];
|
|
}
|
|
return [
|
|
...base,
|
|
{
|
|
id: crypto.randomUUID(),
|
|
role: "tool",
|
|
kind: "trace",
|
|
content: lines[lines.length - 1],
|
|
traces: lines,
|
|
...(visibleStructuredEvents.length ? { toolEvents: visibleStructuredEvents } : {}),
|
|
activitySegmentId: segmentId,
|
|
...turn,
|
|
createdAt: Date.now(),
|
|
},
|
|
];
|
|
});
|
|
return;
|
|
}
|
|
|
|
const media = ev.media_urls?.length
|
|
? ev.media_urls.map((m) => toMediaAttachment(m))
|
|
: ev.media?.map((url) => toMediaAttachment({ url }));
|
|
const hasMedia = !!media && media.length > 0;
|
|
if (sideChannelEvent) {
|
|
setMessages((prev) => absorbCompleteAssistantMessage(prev, {
|
|
content: ev.text,
|
|
...(hasMedia ? { media } : {}),
|
|
...(ev.source ? { source: ev.source } : {}),
|
|
...turnFieldsFromEvent(ev, "answer"),
|
|
}));
|
|
if (typeof ev.turn_id === "string") sideChannelTurnIdsRef.current.delete(ev.turn_id);
|
|
return;
|
|
}
|
|
|
|
// A complete (non-streamed) assistant message. If a stream was in
|
|
// flight, drop the placeholder so we don't render the text twice.
|
|
// Streaming state is closed by ``stream_end`` when present, or by
|
|
// ``turn_end`` for non-streamed and tool-heavy turns.
|
|
clearActivitySegment();
|
|
setMessages((prev) => {
|
|
const activeId = buffer.current?.messageId;
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
const filtered = activeId ? prev.filter((m) => m.id !== activeId) : prev;
|
|
const content = ev.text;
|
|
const lat =
|
|
typeof ev.latency_ms === "number" && ev.latency_ms >= 0
|
|
? Math.round(ev.latency_ms)
|
|
: undefined;
|
|
return absorbCompleteAssistantMessage(filtered, {
|
|
content,
|
|
...(hasMedia ? { media } : {}),
|
|
...(lat !== undefined ? { latencyMs: lat } : {}),
|
|
...(ev.source ? { source: ev.source } : {}),
|
|
...turnFieldsFromEvent(ev, "answer"),
|
|
});
|
|
});
|
|
if (hasMedia) {
|
|
suppressStreamUntilTurnEndRef.current = true;
|
|
}
|
|
return;
|
|
}
|
|
if (ev.event === "file_edit") {
|
|
const edits = Array.isArray(ev.edits) ? ev.edits : [];
|
|
if (edits.length === 0) return;
|
|
const normalized = mergeFileEdits(undefined, edits);
|
|
if (normalized.length === 0) return;
|
|
const turn = turnFieldsFromEvent(ev, "activity");
|
|
const opensFileEditPhase = normalized.some(
|
|
(edit) => edit.status === "editing" || edit.phase === "start",
|
|
);
|
|
let eventSegmentId = fileEditSegmentRef.current;
|
|
if (!eventSegmentId && opensFileEditPhase) {
|
|
eventSegmentId = detachedActivitySegmentId();
|
|
fileEditSegmentRef.current = eventSegmentId;
|
|
}
|
|
setMessages((prev) => {
|
|
let segmentId = eventSegmentId;
|
|
const base = stripCoveredFileEditToolHintsFromMessages(prev, normalized, turn);
|
|
const targetIndex = findFileEditTraceIndex(base, segmentId, normalized);
|
|
if (targetIndex !== null) {
|
|
const target = base[targetIndex];
|
|
segmentId = target.activitySegmentId ?? segmentId ?? detachedActivitySegmentId();
|
|
if (opensFileEditPhase) fileEditSegmentRef.current = segmentId;
|
|
const merged: UIMessage = {
|
|
...target,
|
|
fileEdits: mergeFileEdits(target.fileEdits, normalized),
|
|
activitySegmentId: segmentId,
|
|
...turn,
|
|
};
|
|
return replaceMessageAt(base, targetIndex, merged);
|
|
}
|
|
segmentId = segmentId ?? detachedActivitySegmentId();
|
|
if (opensFileEditPhase) fileEditSegmentRef.current = segmentId;
|
|
return [
|
|
...base,
|
|
{
|
|
id: crypto.randomUUID(),
|
|
role: "tool",
|
|
kind: "trace",
|
|
content: "",
|
|
traces: [],
|
|
fileEdits: normalized,
|
|
activitySegmentId: segmentId,
|
|
...turn,
|
|
createdAt: Date.now(),
|
|
},
|
|
];
|
|
});
|
|
return;
|
|
}
|
|
// ``attached`` frames aren't actionable here.
|
|
};
|
|
|
|
const unsub = client.onChat(chatId, handle);
|
|
return () => {
|
|
unsub();
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
closedAssistantStreamIdsRef.current.clear();
|
|
clearActivitySegment();
|
|
clearPendingStreamWork();
|
|
cancelStreamEndTimer();
|
|
};
|
|
}, [
|
|
applyStreamError,
|
|
cancelStreamEndTimer,
|
|
chatId,
|
|
client,
|
|
clearActivitySegment,
|
|
clearPendingStreamWork,
|
|
detachedActivitySegmentId,
|
|
ensureActivitySegmentId,
|
|
flushPendingStreamEvents,
|
|
isSideChannelEvent,
|
|
onTurnEnd,
|
|
schedulePendingStreamFlush,
|
|
scheduleStreamEndTimer,
|
|
]);
|
|
|
|
const send = useCallback(
|
|
(content: string, images?: SendAttachment[], options?: SendOptions) => {
|
|
if (!chatId) return null;
|
|
const hasAttachments = !!images && images.length > 0;
|
|
// Text is optional when files are attached — the agent will still see
|
|
// them via ``media`` paths.
|
|
if (!hasAttachments && !content.trim()) return null;
|
|
|
|
const sideChannel = options?.sideChannel === true;
|
|
const finalizeActiveTurn = options?.finalizeActiveTurn === true;
|
|
const continueActiveTurn = options?.continueActiveTurn === true;
|
|
const outboundContent = options?.quotedContext
|
|
? formatQuotedUserMessage(content, options.quotedContext)
|
|
: content;
|
|
flushPendingStreamEvents();
|
|
if (finalizeActiveTurn) {
|
|
cancelStreamEndTimer();
|
|
setIsStreaming(false);
|
|
}
|
|
const turnId = crypto.randomUUID();
|
|
const userMessageId = crypto.randomUUID();
|
|
if (sideChannel) sideChannelTurnIdsRef.current.add(turnId);
|
|
const previews = hasAttachments ? images!.map((i) => i.preview) : undefined;
|
|
setMessages((prev) => {
|
|
if ((!sideChannel && !continueActiveTurn) || finalizeActiveTurn) {
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
closedAssistantStreamIdsRef.current.clear();
|
|
clearActivitySegment();
|
|
suppressStreamUntilTurnEndRef.current = false;
|
|
} else if (continueActiveTurn) {
|
|
// Guidance belongs to the active backend turn. Preserve the answer
|
|
// cursor so its resuming stream_end can finalize the text already
|
|
// shown before the new user row, while starting fresh activity after it.
|
|
clearActivitySegment();
|
|
}
|
|
const base = finalizeActiveTurn ? finalizeStreamedTurn(prev) : prev;
|
|
return [
|
|
...(sideChannel || continueActiveTurn ? base : pruneReasoningOnlyPlaceholders(base)),
|
|
{
|
|
id: userMessageId,
|
|
role: "user",
|
|
content: outboundContent,
|
|
turnId,
|
|
turnPhase: "user",
|
|
turnSeq: 0,
|
|
deliveryStatus: "sending",
|
|
createdAt: Date.now(),
|
|
...(previews ? { media: previews } : {}),
|
|
...(options?.cliApps?.length ? { cliApps: options.cliApps } : {}),
|
|
...(options?.mcpPresets?.length ? { mcpPresets: options.mcpPresets } : {}),
|
|
},
|
|
];
|
|
});
|
|
if (!sideChannel) setIsStreaming(true);
|
|
const wireMedia = hasAttachments ? images!.map((i) => i.media) : undefined;
|
|
const clientOptions = {
|
|
...options,
|
|
turnId,
|
|
...((sideChannel || continueActiveTurn) ? { startsNewRun: false } : {}),
|
|
};
|
|
delete clientOptions.quotedContext;
|
|
delete clientOptions.sideChannel;
|
|
delete clientOptions.finalizeActiveTurn;
|
|
delete clientOptions.continueActiveTurn;
|
|
client.sendMessage(chatId, outboundContent, wireMedia, clientOptions);
|
|
return { turnId, userMessageId, sideChannel };
|
|
},
|
|
[cancelStreamEndTimer, chatId, clearActivitySegment, client, flushPendingStreamEvents],
|
|
);
|
|
|
|
const stop = useCallback(() => {
|
|
if (!chatId) return;
|
|
flushPendingStreamEvents();
|
|
setIsStreaming(false);
|
|
setMessages((prev) => {
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
closedAssistantStreamIdsRef.current.clear();
|
|
clearActivitySegment();
|
|
return prev.map((m) => (m.isStreaming ? { ...m, isStreaming: false } : m));
|
|
});
|
|
suppressStreamUntilTurnEndRef.current = false;
|
|
client.sendMessage(chatId, "/stop");
|
|
}, [chatId, clearActivitySegment, client, flushPendingStreamEvents]);
|
|
|
|
const reconcileTurnComplete = useCallback(() => {
|
|
cancelStreamEndTimer();
|
|
clearPendingStreamWork();
|
|
buffer.current = null;
|
|
activeAssistantRef.current = null;
|
|
closedAssistantStreamIdsRef.current.clear();
|
|
clearActivitySegment();
|
|
suppressStreamUntilTurnEndRef.current = false;
|
|
setRunStartedAt(null);
|
|
setIsStreaming(false);
|
|
}, [cancelStreamEndTimer, clearActivitySegment, clearPendingStreamWork]);
|
|
|
|
const transcribeAudio = useCallback(
|
|
(dataUrl: string, options?: { durationMs?: number }) =>
|
|
client.transcribeAudio(dataUrl, options),
|
|
[client],
|
|
);
|
|
|
|
return {
|
|
messages,
|
|
messagesReady: messageOwnerChatId === chatId,
|
|
isStreaming,
|
|
runStartedAt,
|
|
goalState,
|
|
send,
|
|
transcribeAudio,
|
|
stop,
|
|
reconcileTurnComplete,
|
|
setMessages,
|
|
streamError,
|
|
dismissStreamError,
|
|
};
|
|
}
|