ouyangbo/EvoScientist-WebUI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3 from EvoScientist/fix-hitl-approval-race

Browse Source 
Fix HITL approval recovery
This commit is contained in:
Xi Zhang
2026-06-15 23:42:08 +01:00
committed by GitHub
parent 62f32cb492 0f75a5f351
commit 6f7a5738f9
3 changed files with 131 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/components/ChatInterface.tsx
+57 -20
View File
@@ -102,6 +102,23 @@ interface UploadedWorkspaceFile {
size: number;
}
function parseToolArgs(rawArgs: unknown): Record<string, unknown> {
if (rawArgs && typeof rawArgs === "object") {
return rawArgs as Record<string, unknown>;
}
if (typeof rawArgs !== "string") {
return {};
}
try {
const parsed = JSON.parse(rawArgs);
return parsed && typeof parsed === "object"
? (parsed as Record<string, unknown>)
: {};
} 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<string, unknown>)
: {},
args: parseToolArgs(rawArgs),
};
});
}
@@ -641,6 +655,18 @@ export const ChatInterface = React.memo<ChatInterfaceProps>(
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<ChatInterfaceProps>(
seenAsyncUpdates.add(key);
return true;
});
const completedToolCallIds = new Set<string>();
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<string, number>();
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<ChatInterfaceProps>(
)
.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<ChatInterfaceProps>(
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<ChatInterfaceProps>(
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<Set<string>>(() => new Set());
useEffect(() => {
src/app/components/ChatMessage.tsx
+1
View File
@@ -129,6 +129,7 @@ export const ChatMessage = React.memo<ChatMessageProps>(
});
const cursor = new Map<string, number>();
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;
src/app/hooks/useChat.ts
+73 -19
View File
@@ -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<Message[] | null>(
null
);
const [fetchedThreadId, setFetchedThreadId] = useState<string | null>(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<typeof setTimeout> | 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,