diff --git a/src/app/components/ChatInterface.tsx b/src/app/components/ChatInterface.tsx index 09849c9..ab52759 100644 --- a/src/app/components/ChatInterface.tsx +++ b/src/app/components/ChatInterface.tsx @@ -102,6 +102,23 @@ interface UploadedWorkspaceFile { size: number; } +function parseToolArgs(rawArgs: unknown): Record { + if (rawArgs && typeof rawArgs === "object") { + return rawArgs as Record; + } + if (typeof rawArgs !== "string") { + return {}; + } + try { + const parsed = JSON.parse(rawArgs); + return parsed && typeof parsed === "object" + ? (parsed as Record) + : {}; + } catch { + return {}; + } +} + function getMessageToolCalls(message: Message): Array<{ id?: string; name: string; @@ -147,10 +164,7 @@ function getMessageToolCalls(message: Message): Array<{ return { id: toolCall.id, name: toolCall.function?.name || toolCall.name || toolCall.type || "", - args: - rawArgs && typeof rawArgs === "object" - ? (rawArgs as Record) - : {}, + args: parseToolArgs(rawArgs), }; }); } @@ -641,6 +655,18 @@ export const ChatInterface = React.memo( resumeInterrupt({ status: "cancelled" }); }, [resumeInterrupt]); + // Ordered list of pending tool-approval requests from the interrupt. We hand + // ChatMessage the ORDER (not a name-keyed map) so two calls to the same tool + // in one turn (e.g. two `execute`) each bind to their OWN request/args instead + // of both collapsing onto the last one. `Array.isArray` guards a malformed + // payload — a non-array `action_requests` here would otherwise throw and blank + // the whole page. + const actionRequests: ActionRequest[] = useMemo(() => { + const raw = + interrupt?.value && (interrupt.value as any)["action_requests"]; + return Array.isArray(raw) ? (raw as ActionRequest[]) : []; + }, [interrupt]); + // TODO: can we make this part of the hook? const processedMessages = useMemo(() => { /* @@ -679,6 +705,19 @@ export const ChatInterface = React.memo( seenAsyncUpdates.add(key); return true; }); + const completedToolCallIds = new Set(); + for (const message of visibleMessages) { + if (message.type !== "tool") continue; + const toolCallId = message.tool_call_id; + if (toolCallId) completedToolCallIds.add(toolCallId); + } + const pendingActionCounts = new Map(); + for (const ar of actionRequests) { + pendingActionCounts.set( + ar.name, + (pendingActionCounts.get(ar.name) ?? 0) + 1 + ); + } visibleMessages.forEach((message: Message) => { if (message.type === "ai") { const toolCallsWithStatus = getMessageToolCalls(message) @@ -695,16 +734,25 @@ export const ChatInterface = React.memo( ) .map((toolCall, toolCallIndex) => { const name = toolCall.name || "unknown"; + const id = + toolCall.id || + `${message.id ?? "ai-message"}-tool-${toolCallIndex}-${name}`; + const pendingCount = pendingActionCounts.get(name) ?? 0; + const hasPendingAction = + pendingCount > 0 && !completedToolCallIds.has(id); + if (hasPendingAction) { + pendingActionCounts.set(name, pendingCount - 1); + } return { - id: - toolCall.id || - `${message.id ?? "ai-message"}-tool-${toolCallIndex}-${name}`, + id, name, args: toolCall.args, // The selector call only survives the filter above while the run is // actively selecting (!interrupt), so this resolves to a spinner for // it without a special case. - status: interrupt ? "interrupted" : ("pending" as const), + status: hasPendingAction + ? "interrupted" + : ("pending" as const), } as ToolCall; }); messageMap.set(message.id!, { @@ -746,7 +794,7 @@ export const ChatInterface = React.memo( showAvatar: data.message.type !== prevMessage?.type, }; }); - }, [messages, interrupt, isLoading, stream]); + }, [messages, actionRequests, interrupt, isLoading, stream]); // Where to anchor the "Conversation compacted" block. The event's // cutoffIndex points into the raw `messages` array (messages[0:cutoff] were @@ -776,17 +824,6 @@ export const ChatInterface = React.memo( const hasTasks = todos.length > 0; const hasFiles = Object.keys(files).length > 0; - // Ordered list of pending tool-approval requests from the interrupt. We hand - // ChatMessage the ORDER (not a name-keyed map) so two calls to the same tool - // in one turn (e.g. two `execute`) each bind to their OWN request/args instead - // of both collapsing onto the last one. `Array.isArray` guards a malformed - // payload — a non-array `action_requests` here would otherwise throw and blank - // the whole page. - const actionRequests: ActionRequest[] = useMemo(() => { - const raw = - interrupt?.value && (interrupt.value as any)["action_requests"]; - return Array.isArray(raw) ? (raw as ActionRequest[]) : []; - }, [interrupt]); const [submittedActionRequestKeys, setSubmittedActionRequestKeys] = useState>(() => new Set()); useEffect(() => { diff --git a/src/app/components/ChatMessage.tsx b/src/app/components/ChatMessage.tsx index b562073..8ff6003 100644 --- a/src/app/components/ChatMessage.tsx +++ b/src/app/components/ChatMessage.tsx @@ -129,6 +129,7 @@ export const ChatMessage = React.memo( }); const cursor = new Map(); for (const tc of toolCalls) { + if (tc.status !== "interrupted") continue; const list = queues.get(tc.name); if (!list) continue; const i = cursor.get(tc.name) ?? 0; diff --git a/src/app/hooks/useChat.ts b/src/app/hooks/useChat.ts index 4903314..59f01ee 100644 --- a/src/app/hooks/useChat.ts +++ b/src/app/hooks/useChat.ts @@ -1,6 +1,6 @@ "use client"; -import { useCallback, useEffect, useState } from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; import { useStream } from "@langchain/langgraph-sdk/react"; import { type Message, type Assistant } from "@langchain/langgraph-sdk"; import { v4 as uuidv4 } from "uuid"; @@ -101,6 +101,30 @@ function interruptValueKey(i: unknown): string | null { } } +function hasActionableInterrupt(i: unknown): boolean { + if (!i || typeof i !== "object") return false; + const value = (i as { value?: unknown }).value; + if (!value || typeof value !== "object") return false; + const v = value as { type?: unknown; action_requests?: unknown }; + return ( + v.type === "ask_user" || + (Array.isArray(v.action_requests) && v.action_requests.length > 0) + ); +} + +function latestTaskInterrupt( + tasks: Array<{ interrupts?: unknown[] }> | undefined +): unknown { + if (!Array.isArray(tasks)) return undefined; + for (let i = tasks.length - 1; i >= 0; i--) { + const interrupts = tasks[i]?.interrupts; + if (Array.isArray(interrupts) && interrupts.length > 0) { + return interrupts[interrupts.length - 1]; + } + } + return undefined; +} + export function useChat({ activeAssistant, onHistoryRevalidate, @@ -177,10 +201,19 @@ export function useChat({ const [fetchedMessages, setFetchedMessages] = useState( null ); + const [fetchedThreadId, setFetchedThreadId] = useState(null); + const recoveryRunRef = useRef(0); useEffect(() => { if (!threadId) { setFetchedInterrupt(undefined); setFetchedMessages(null); + setFetchedThreadId(null); + setResolvedInterruptKey(null); + return; + } + if (stream.isLoading) { + recoveryRunRef.current += 1; + setFetchedInterrupt(undefined); setResolvedInterruptKey(null); return; } @@ -190,6 +223,7 @@ export function useChat({ // plus its approval interrupt. Either way we backfill from thread state // (the same data a thread-switch re-fetch would pull in). const baseline = stream.messages.length; + const recoveryRunId = ++recoveryRunRef.current; let cancelled = false; let tries = 0; let timer: ReturnType | undefined; @@ -202,36 +236,42 @@ export function useChat({ next?: unknown[]; values?: { messages?: Message[] }; }; - if (cancelled) return; + if (cancelled || recoveryRunRef.current !== recoveryRunId) return; const msgs = state.values?.messages; - const interrupts = state.tasks?.at(-1)?.interrupts; - const pending: unknown = - Array.isArray(interrupts) && interrupts.length > 0 - ? interrupts[interrupts.length - 1] - : undefined; + const pending = latestTaskInterrupt(state.tasks); const stillPending = Array.isArray(state.next) && state.next.length > 0; - // Backfill whenever the server is ahead of what the stream delivered. - if (Array.isArray(msgs) && msgs.length > baseline) { - setFetchedMessages(msgs); - } const safePending = normalizePendingInterrupt(pending); - if (safePending) { - // Tool-approval interrupt reached — surface it (+ messages) and stop. - // There IS a live pending interrupt, so drop any stale suppression. + if (safePending && hasActionableInterrupt(safePending)) { + // Tool-approval interrupt reached — surface it and its matching message + // snapshot together. Mixing live messages with fetched interrupts is the + // race that hides approval cards for repeated execute calls. setFetchedInterrupt( safePending as unknown as typeof stream.interrupt ); setResolvedInterruptKey(null); - if (Array.isArray(msgs)) setFetchedMessages(msgs); + if (Array.isArray(msgs)) { + setFetchedThreadId(threadId); + setFetchedMessages(msgs); + } return; } + // Backfill only after the live stream is idle. During active streaming the + // live message list owns rendering; this recovery loop is for dropped tail + // state after the stream has settled. + if (Array.isArray(msgs) && msgs.length > baseline) { + setFetchedThreadId(threadId); + setFetchedMessages(msgs); + } if (!stillPending) { // The server has no pending task/interrupt anymore. Record the stale // live interrupt's identity so the getter suppresses ONLY that one // (composer unlocks after approving) — a new interrupt still shows. setFetchedInterrupt(undefined); setResolvedInterruptKey(interruptValueKey(stream.interrupt)); - if (Array.isArray(msgs)) setFetchedMessages(msgs); + if (Array.isArray(msgs)) { + setFetchedThreadId(threadId); + setFetchedMessages(msgs); + } return; } // Keep polling only while the run is still working server-side; a @@ -240,7 +280,13 @@ export function useChat({ timer = setTimeout(attempt, 1000); } } catch { - if (!cancelled && tries < MAX_TRIES) timer = setTimeout(attempt, 1000); + if ( + !cancelled && + recoveryRunRef.current === recoveryRunId && + tries < MAX_TRIES + ) { + timer = setTimeout(attempt, 1000); + } } }; void attempt(); @@ -262,7 +308,9 @@ export function useChat({ (resolvedInterruptKey === null || interruptValueKey(liveInterrupt) !== resolvedInterruptKey) ? liveInterrupt - : fetchedInterrupt ?? undefined; + : fetchedThreadId === threadId + ? fetchedInterrupt ?? undefined + : undefined; // Prefer the backfilled snapshot when it is "ahead" of the live stream — i.e. // the stream ended early and dropped the tail. "Ahead" means either MORE // messages, or (once settled) the SAME number of messages but MORE total text: @@ -272,7 +320,9 @@ export function useChat({ // refresh). The equal-count/more-text rule is gated on `!isLoading` so a // mid-stream poll snapshot never flickers over the actively updating stream. const messages = (() => { - if (!fetchedMessages) return stream.messages; + if (!fetchedMessages || fetchedThreadId !== threadId) + return stream.messages; + if (fetchedInterrupt) return fetchedMessages; if (fetchedMessages.length > stream.messages.length) return fetchedMessages; if ( !stream.isLoading && @@ -292,7 +342,9 @@ export function useChat({ // just-added optimistic user message — making it flicker/vanish. setFetchedInterrupt(undefined); setFetchedMessages(null); + setFetchedThreadId(null); setResolvedInterruptKey(null); + recoveryRunRef.current += 1; const newMessage: Message = { id: uuidv4(), type: "human", content }; stream.submit( { messages: [newMessage] }, @@ -330,7 +382,9 @@ export function useChat({ // approval card or shadow the resumed run's messages. setFetchedInterrupt(undefined); setFetchedMessages(null); + setFetchedThreadId(null); setResolvedInterruptKey(null); + recoveryRunRef.current += 1; stream.submit(null, { command: { resume: value }, streamSubgraphs: true,