fix: unify chat process and reply presentation

This commit is contained in:
2026-10-29 17:54:34 +08:00
parent 7621bf0f86
commit 9474788db6
46 changed files with 3160 additions and 500 deletions
@@ -2,6 +2,21 @@ import { describe, expect, it } from "vitest";
import { applySessionStreamEvent, toAgentChatMessages, toAgentUiMessages } from "./agent-session-message-state";
describe("task context in persisted answers", () => {
it("restores submitted answer cards only from snapshots bound to their user message", () => {
const reply = { version: "agent-interaction-request@1", interaction_id: "question", message_id: "source", turn_id: "turn", call_id: "call",
status: "answered", revision: 1, draft_revision: 0, draft: [], created_at: "2026-10-28T10:00:00Z",
questions: [{ question_id: "q", title: "核查时段", body: "请选择", mode: "single", required: true, allow_custom: false, options: [{ id: "a", label: "前一天" }] }],
answers: [{ question_id: "q", selected_option_ids: ["a"], selected_options: [{ id: "a", label: "前一天" }], custom_text: "" }],
delivery: { native_message_id: "reply", revision: 2, status: "delivered" } };
const parts = [{ type: "text", text: "核查时段: 前一天" }, { type: "data-interaction_reply", data: reply }];
const messages = toAgentChatMessages(toAgentUiMessages([
{ id: "reply", role: "user", parts }, { id: "other", role: "user", parts },
{ id: "reply", role: "assistant", parts },
{ id: "pending", role: "user", parts: [{ type: "data-interaction_reply", data: { ...reply, status: "pending", delivery: { ...reply.delivery, native_message_id: "pending" } } }] }
]), {}, {});
expect(messages[0].interactionReply).toEqual(reply);
expect(messages.slice(1).every((message) => message.interactionReply === undefined)).toBe(true);
});
it("inserts consumed user input once before the active assistant in live and resumed streams", () => {
const messages = toAgentUiMessages([
{ id: "original", role: "user", parts: [{ type: "text", text: "原问题" }] },
@@ -60,6 +75,20 @@ describe("task context in persisted answers", () => {
describe("rich answer replay", () => {
const block = { version: "answer-ui@1", block_id: "reading", fallback_text: "演练读数", spec: { root: "metric", elements: { metric: { type: "Metric", props: { label: "压力", value: "0.3", unit: "MPa", source: "演练数据" }, children: [] } } } };
it("restores anchored final content without process UI, duplicate blocks or internal reference markers", () => {
const messages = toAgentUiMessages([{ id: "answer", role: "assistant", parts: [
{ type: "data-answer_ui", data: { ...block, block_id: "query", phase: "process", fallback_text: "查询过程资料" } },
{ type: "data-answer_ui", data: { ...block, phase: "final" } },
{ type: "data-answer_ui", data: { ...block, phase: "final", fallback_text: "更新后的最终读数" } },
{ type: "data-final_answer", data: { content: "最终说明\n\n[[ui:reading]]\n\n后续解读\n\n[[ui:query]]", ui_block_ids: ["reading", "query"] } },
] }]);
const restored = toAgentChatMessages(messages, {}, {})[0];
expect(restored.answerParts?.filter((part) => part.type === "answer-ui")).toHaveLength(2);
expect(restored.content).toContain("更新后的最终读数");
expect(restored.content).not.toContain("查询过程资料");
expect(restored.content).not.toContain("[[ui:");
expect(restored.content.indexOf("更新后的最终读数")).toBeLessThan(restored.content.indexOf("后续解读"));
});
it("keeps text, component and summary in their original order across history restoration", () => {
const messages = toAgentUiMessages([{ id: "answer", role: "assistant", parts: [
{ type: "text", text: "下面是演练数据。" },
@@ -2,6 +2,8 @@ import type { UIMessage } from "ai";
import type { Dispatch, SetStateAction } from "react";
import {
parseAnswerUiBlock,
readInteractionReply,
getFinalAnswerTranscript,
type AnswerUiPart,
applyPermissionResponse,
agentTaskContextReferenceSchema,
@@ -31,6 +33,8 @@ type AgentUiDataParts = {
todo_update: Record<string, unknown>;
permission_request: Record<string, unknown>;
permission_response: Record<string, unknown>;
interaction_request: Record<string, unknown>;
interaction_reply: Record<string, unknown>;
question_request: Record<string, unknown>;
question_response: Record<string, unknown>;
stream_token: Record<string, unknown>;
@@ -137,7 +141,10 @@ export function toAgentChatMessages(
if (!isAgentDataPart(part)) {
continue;
}
if (part.type === "data-task_context") {
if (part.type === "data-interaction_reply" && message.role === "user") {
const reply = readInteractionReply(part.data, message.id);
if (reply) next = { ...next, interactionReply: reply };
} else if (part.type === "data-task_context") {
const parsed = agentTaskContextReferenceSchema.safeParse(part.data);
if (parsed.success) next = { ...next, taskContext: parsed.data };
} else if (part.type === "data-final_answer") {
@@ -184,8 +191,7 @@ export function toAgentChatMessages(
}
if (next.finalAnswer) {
const selected = new Set(next.finalAnswer.uiBlockIds);
next.content = [next.finalAnswer.content, ...(next.answerParts ?? []).flatMap((part) => part.type === "answer-ui" && selected.has(part.block.block_id) ? [part.block.fallback_text] : [])].join("\n\n");
next.content = getFinalAnswerTranscript(next);
}
return [applyQuestionOverrides(applyPermissionOverrides(next, permissionOverrides), questionOverrides)];
});
@@ -526,9 +532,15 @@ function collectAnswerParts(message: AgentUiMessage): AnswerUiPart[] | undefined
if (part.type === "text" && part.text) result.push({ type: "text", text: part.text });
if (part.type === "data-answer_ui") {
const data = typeof part.data === "object" && part.data !== null ? part.data : {};
const { version, block_id, fallback_text, spec } = data;
const block = parseAnswerUiBlock({ version, block_id, fallback_text, spec });
if (block) result.push({ type: "answer-ui", block });
const { version, block_id, phase, fallback_text, spec } = data;
const block = parseAnswerUiBlock({ version, block_id, ...(phase === undefined ? {} : { phase }), fallback_text, spec });
if (block) {
const existing = result.findIndex((item) => item.type === "answer-ui" && item.block.block_id === block.block_id);
if (existing >= 0) {
const previous = result[existing];
if (previous.type === "answer-ui" && previous.block.phase === block.phase) result[existing] = { type: "answer-ui", block };
} else result.push({ type: "answer-ui", block });
}
else result.push({ type: "text", text: typeof fallback_text === "string" ? fallback_text.slice(0, 4000) : "这段结构化内容无法显示。" });
}
@@ -1,6 +1,6 @@
import { Chat } from "@ai-sdk/react";
import type { ChatTransport } from "ai";
import { parseFrontendActionRequest, type AgentApiClient, type AgentRunStatus, type AgentLoadedChatSession,
import { AgentSessionStreamError, AgentStreamDisconnectedError, parseFrontendActionRequest, type AgentApiClient, type AgentRunStatus, type AgentLoadedChatSession,
type AgentMessagePagination, type AgentSessionStreamEvent, type AgentSessionPurpose, type AgentTaskDiscussionContext, type AnswerUiSelection } from "@/features/agent";
import { showMapNotice } from "@/features/map/core";
import { applySessionStreamEvent, getDataString, toAgentChatMessages, toAgentUiMessages,
@@ -8,6 +8,15 @@ import { applySessionStreamEvent, getDataString, toAgentChatMessages, toAgentUiM
import type { WorkbenchSessionResources } from "./workbench-session-resources";
export type AgentReadingPosition = { messageId?: string; offset: number; atBottom: boolean };
export function isPendingInteractionError(error: unknown) {
const message = error instanceof Error ? error.message : error;
if (typeof message !== "string") return false;
try {
const value: unknown = JSON.parse(message);
return typeof value === "object" && value !== null && "code" in value && value.code === "INTERACTION_PENDING";
} catch { return false; }
}
export type AgentSessionRuntime = {
key: string;
scope: string;
@@ -29,6 +38,9 @@ export type AgentSessionRuntime = {
updatedAt: number;
resumed: boolean;
controller?: AbortController;
recoveryTimer?: ReturnType<typeof setTimeout>;
transportInterrupted?: boolean;
connectionError?: string;
error?: string;
runStatus?: AgentRunStatus;
handlers?: {
@@ -36,6 +48,7 @@ export type AgentSessionRuntime = {
onEvent: (event: AgentSessionStreamEvent) => void;
onError: (error: Error) => void;
onFinish: (status: AgentRunStatus) => void;
onReconnect?: () => void;
};
};
@@ -84,6 +97,7 @@ export function createAgentRuntime(resources: WorkbenchSessionResources, scope:
const sessionId = getDataString(data.data, "session_id");
if (sessionId && runtime.sessionId && sessionId !== runtime.sessionId) return;
if (sessionId) runtime.sessionId = sessionId;
if (data.type === "data-interaction_request") runtime.runStatus = "waiting_input";
if (data.type === "data-session_title") runtime.title = getDataString(data.data, "title") ?? runtime.title;
if (data.type === "data-credential_refresh_required") {
void refreshRuntimeCredential(runtime, data.data);
@@ -92,19 +106,35 @@ export function createAgentRuntime(resources: WorkbenchSessionResources, scope:
} else if (active()) runtime.handlers?.onData(data);
notify();
},
onError: (error) => { if (resources.disposed) return; runtime.error = error.message; runtime.runStatus = "error"; if (active()) runtime.handlers?.onError(error); notify(); },
onError: (error) => { if (resources.disposed) return; runtime.transportInterrupted = error instanceof AgentStreamDisconnectedError || error instanceof TypeError;
runtime.error = error.message; runtime.runStatus = isPendingInteractionError(error) ? "waiting_input" : "error";
if (active()) {
if (runtime.transportInterrupted && runtime.sessionId && runtime.handlers?.onReconnect) runtime.handlers.onReconnect();
else runtime.handlers?.onError(error);
}
notify(); },
onFinish: ({ isAbort, isError }) => {
if (resources.disposed) return;
runtime.loadedAt = Date.now();
runtime.updatedAt = Date.now();
runtime.runStatus = isAbort ? "aborted" : isError ? "error" : "completed";
if (active()) runtime.handlers?.onFinish(runtime.runStatus);
else notifyBackgroundCompletion(runtime);
runtime.runStatus = isAbort ? "aborted" : isError
? isPendingInteractionError(runtime.error) ? "waiting_input" : "error"
: runtime.runStatus === "waiting_input" ? "waiting_input" : "completed";
const recovering = isError && !isAbort && runtime.transportInterrupted && runtime.sessionId;
if (active()) { if (!recovering) runtime.handlers?.onFinish(runtime.runStatus); }
else if (!recovering) notifyBackgroundCompletion(runtime);
if (recovering) {
runtime.recoveryTimer = setTimeout(() => {
runtime.recoveryTimer = undefined;
if (!resources.disposed && resources.agents.get(runtime.key) === runtime) resumeAgentRuntime(resources, runtime);
}, 0);
}
runtime.transportInterrupted = false;
notify();
},
});
resources.agents.set(runtime.key, runtime);
resources.disposers.add(() => { runtime.handlers = undefined; runtime.controller?.abort(); void runtime.chat.stop(); });
resources.disposers.add(() => { runtime.handlers = undefined; clearTimeout(runtime.recoveryTimer); runtime.controller?.abort(); void runtime.chat.stop(); });
return runtime;
}
@@ -132,15 +162,13 @@ async function rejectBackgroundAction(resources: WorkbenchSessionResources, runt
}
export function resumeAgentRuntime(resources: WorkbenchSessionResources, runtime: AgentSessionRuntime) {
if (!runtime.sessionId || runtime.controller || runtime.chat.status === "submitted" || runtime.chat.status === "streaming") return;
if (resources.disposed || !runtime.sessionId || runtime.controller || runtime.chat.status === "submitted" || runtime.chat.status === "streaming") return;
const controller = new AbortController();
runtime.controller = controller;
runtime.resumed = true;
runtime.runStatus = "running";
runtime.error = undefined;
void runtime.client.streamSession(runtime.sessionId, { signal: controller.signal, messageLimit: 40,
onEvent: (event) => {
if (resources.disposed) return;
const onEvent = (event: AgentSessionStreamEvent) => {
if (resources.disposed || controller.signal.aborted || runtime.controller !== controller) return;
const sessionId = event.type === "state" ? event.sessionId : event.sessionId ?? getDataString(event.data, "session_id");
if (sessionId && sessionId !== runtime.sessionId) return;
if (event.type === "state") {
@@ -155,27 +183,47 @@ export function resumeAgentRuntime(resources: WorkbenchSessionResources, runtime
if (!runtime.messagePage) runtime.messagePage = event.messagePage;
runtime.resumed = event.isStreaming;
runtime.runStatus = event.runStatus ?? (event.isStreaming ? "running" : "completed");
if (runtime.runStatus !== "error") runtime.error = undefined;
terminal = !event.isStreaming;
runtime.connectionError = undefined;
} else {
if (event.type === "credential_refresh_required") void refreshRuntimeCredential(runtime, event.data);
else if (resources.activeAgent !== runtime && event.type === "frontend_action") void rejectBackgroundAction(resources, runtime, event.data);
runtime.chat.messages = applySessionStreamEvent(runtime.chat.messages, event);
if (event.type === "session_title") runtime.title = getDataString(event.data, "title") ?? runtime.title;
if (event.type === "done" || event.type === "error") {
terminal = true;
runtime.resumed = false;
runtime.runStatus = event.type === "error" ? "error" : "completed";
runtime.runStatus = event.type === "error" ? "error" : event.data.reason === "waiting_input" ? "waiting_input" : "completed";
runtime.updatedAt = Date.now();
}
}
if (resources.activeAgent === runtime && event.type !== "credential_refresh_required") runtime.handlers?.onEvent(event);
resources.notify();
},
}).catch((error: unknown) => {
if (!controller.signal.aborted) {
runtime.error = error instanceof Error ? error.message : "会话连接中断";
if (resources.activeAgent === runtime) runtime.handlers?.onError(new Error(runtime.error));
};
let terminal = false;
void (async () => {
let failures = 0;
while (!resources.disposed && !controller.signal.aborted) {
terminal = false;
try {
await runtime.client.streamSession(runtime.sessionId!, { signal: controller.signal, messageLimit: 40, onEvent });
if (terminal || controller.signal.aborted || resources.disposed) break;
throw new AgentStreamDisconnectedError("会话连接已断开,正在自动恢复");
} catch (error) {
if (controller.signal.aborted || resources.disposed || terminal) break;
runtime.connectionError = error instanceof Error ? error.message : "会话连接中断";
if (error instanceof AgentSessionStreamError && [400, 401, 403, 404, 410].includes(error.status)) {
runtime.runStatus = "error";
if (resources.activeAgent === runtime) runtime.handlers?.onError(error);
break;
}
if (resources.activeAgent === runtime) runtime.handlers?.onReconnect?.();
resources.notify();
await waitForReconnect(Math.min(1_000 * 2 ** Math.min(failures++, 4), 15_000), controller.signal);
}
}
}).finally(() => {
if (runtime.resumed && !controller.signal.aborted && !runtime.error) runtime.error = "会话连接已断开,返回会话后重新连接";
})().finally(() => {
if (runtime.controller === controller) { runtime.controller = undefined; runtime.resumed = false; }
runtime.loadedAt = Date.now();
if (!resources.disposed && !controller.signal.aborted && resources.activeAgent !== runtime) notifyBackgroundCompletion(runtime);
@@ -183,6 +231,23 @@ export function resumeAgentRuntime(resources: WorkbenchSessionResources, runtime
});
}
function waitForReconnect(delay: number, signal: AbortSignal): Promise<void> {
return new Promise((resolve) => {
if (signal.aborted) { resolve(); return; }
const finish = () => {
clearTimeout(timer);
signal.removeEventListener("abort", finish);
window.removeEventListener("online", finish);
window.removeEventListener("focus", finish);
resolve();
};
const timer = navigator.onLine ? setTimeout(finish, delay) : undefined;
signal.addEventListener("abort", finish, { once: true });
window.addEventListener("online", finish, { once: true });
window.addEventListener("focus", finish, { once: true });
});
}
function notifyBackgroundCompletion(runtime: AgentSessionRuntime) {
showMapNotice({ id: `agent-background-${runtime.key}`, tone: runtime.error || runtime.runStatus === "error" ? "error" : "info",
title: runtime.error || runtime.runStatus === "error" ? "后台对话处理失败" : runtime.runStatus === "aborted" ? "后台对话已停止" : "后台对话已完成",
@@ -10,11 +10,14 @@ import { showMapNotice } from "@/features/map/core";
import { findOwnedWorkbenchProject } from "../api/project-client";
import type { InvestigationDiscussion } from "../api/investigation-report";
import {
useInteractionRequests,
useAgentConnection,
FrontendActionExecutor,
agentTaskDiscussionContextSchema,
type AgentTaskDiscussionContext,
agentBootstrapKey,
createAgentApiClient,
AgentResponseFinishedError,
useAgentMessageInbox,
fetchAgentBootstrap,
resolveAgentModelSelection,
@@ -59,6 +62,7 @@ import { useWorkbenchSessionResources } from "./workbench-session-resources";
import { useAgentTopics } from "./use-agent-topics";
import { useAgentHistory } from "./use-agent-history";
import { agentRuntimes, createAgentRuntime, mergeHistoryMessages, resumeAgentRuntime, runtimeNeedsInput, runtimeRunning,
isPendingInteractionError,
type AgentReadingPosition, type AgentSessionRuntime } from "./agent-session-runtime";
const AGENT_PANEL_COLLAPSE_MS = 180;
@@ -98,6 +102,7 @@ export function useWorkbenchAgent({
clientRef.current = scopeClient;
const projectScopeKey = `project:${getWorkbenchProject(getAccessToken)?.project_id ?? "default"}`;
const scopeKey = projectScopeKey;
const { connectionStatus, recheck: checkAgentConnection, recoveryVersion } = useAgentConnection(scopeClient, scopeKey);
const history = useAgentHistory(scopeClient, scopeKey);
const topics = useAgentTopics(scopeClient, scopeKey);
const runtimeSummaries: AgentChatSessionSummary[] = agentRuntimes(resources).filter((entry) => entry.scope === scopeKey && entry.sessionId && entry.purpose !== "task_execution")
@@ -142,8 +147,6 @@ export function useWorkbenchAgent({
const [registry, setRegistry] = useState<UIRegistry | null>(null);
const [sessionTitle, setSessionTitle] = useState("新对话");
const [statusLabel, setStatusLabel] = useState("Agent 后端连接中");
const [connectionStatus, setConnectionStatus] =
useState<AgentConnectionStatus>("connecting");
const [modelOptions, setModelOptions] = useState<AgentModelOption[]>([]);
const [selectedModel, setSelectedModel] = useState("");
const [inputDelivery, setInputDelivery] = useState<AgentInputDelivery>("steer");
@@ -162,6 +165,44 @@ export function useWorkbenchAgent({
const [questionOverrides, setQuestionOverrides] = useState<Record<string, QuestionOverride>>({});
const agentMessageInbox = useAgentMessageInbox(clientRef.current, sessionPurpose === "ordinary_chat" ? sessionId : null);
const refreshAgentMessageInbox = agentMessageInbox.refresh;
const continueInteraction = useCallback(async (owningSessionId: string, interactionId: string) => {
const target = agentRuntimes(resources).find((entry) => entry.scope === scopeKey && entry.sessionId === owningSessionId);
if (!target || target.client !== scopeClient || resources.disposed) throw new Error("原会话已关闭,请重新打开后继续。");
const record = await target.client.getInteractionRequest(owningSessionId, interactionId);
if (target.client !== scopeClient || resources.disposed || !agentRuntimes(resources).includes(target)) return;
if (!record.delivery || ["delivered", "cancelled"].includes(record.delivery.status)) return;
const text = record.status === "skipped" ? "已跳过本组提问。" : record.questions.map((question) => {
const answer = record.answers?.find((item) => item.question_id === question.question_id);
return `${question.title}:${[...(answer?.selected_options.map((option) => option.label) ?? []), answer?.custom_text].filter(Boolean).join(";")}`;
}).join("\n");
const alreadyPresent = target.chat.messages.some((message) => message.id === record.delivery!.native_message_id);
const replyMessage: AgentUiMessage = { id: record.delivery.native_message_id, role: "user", parts: [
{ type: "text", text }, { type: "data-interaction_reply", data: record },
] };
if (alreadyPresent) target.chat.messages = target.chat.messages.map((message) => message.id === replyMessage.id ? replyMessage : message);
target.runStatus = "running";
try {
await target.chat.sendMessage(alreadyPresent ? undefined : replyMessage, { body: {
session_id: owningSessionId, interaction_id: interactionId, model: selectedModel || undefined,
capabilities: frontendActionEnabledRef.current ? ["host-question@1", "frontend-action@1"] : ["host-question@1"],
approval_mode: approvalMode, task_context: target.taskDiscussion?.context,
} });
} catch (error) { target.runStatus = "error"; throw error; }
finally {
if (!resources.disposed && target.client === scopeClient && !runtimeRunning(target)) {
try {
const loaded = await target.client.loadSession(owningSessionId, { messageLimit: 40 });
if (loaded && !resources.disposed && target.client === scopeClient) {
target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages));
resources.notify();
}
} catch { /* A failed reconciliation leaves the accepted answer intact. */ }
}
}
}, [resources, scopeKey, scopeClient, selectedModel, approvalMode]);
const interactionRequests = useInteractionRequests(scopeClient, scopeKey, sessionId, continueInteraction);
const refreshInteractionRequests = interactionRequests.refresh;
useEffect(() => {
sessionIdRef.current = sessionId;
@@ -232,7 +273,6 @@ export function useWorkbenchAgent({
const handleDataPart = useCallback(
(part: AgentDataPart) => {
setConnectionStatus("online");
const eventSessionId = getDataString(part.data, "session_id");
if (eventSessionId && sessionIdRef.current && eventSessionId !== sessionIdRef.current) {
return;
@@ -247,6 +287,7 @@ export function useWorkbenchAgent({
);
return;
}
if (part.type === "data-interaction_request") { void refreshInteractionRequests(); return; }
if (part.type === "data-agent_message") { refreshAgentMessageInbox(); return; }
if (part.type === "data-user_input" && part.data.status === "delivered") {
chatRef.current.setMessages((current) => applySessionStreamEvent(current, { type: "user_input", data: part.data }));
@@ -295,7 +336,7 @@ export function useWorkbenchAgent({
});
});
},
[refreshAgentMessageInbox, applySessionId, claimEnvelope, handleCredentialRefresh, handleFrontendAction, onUiEnvelope]
[refreshInteractionRequests, refreshAgentMessageInbox, applySessionId, claimEnvelope, handleCredentialRefresh, handleFrontendAction, onUiEnvelope]
);
const transport = useMemo(() => createWorkbenchAgentTransport(scopeClient, getAccessToken,
@@ -355,9 +396,14 @@ export function useWorkbenchAgent({
const {
data: backendConnection,
error: backendConnectionError
error: backendConnectionError,
mutate: refreshBootstrap
} = useSWR([...agentBootstrapKey, getAccessToken, scopeKey], () => fetchAgentBootstrap(scopeClient), {
shouldRetryOnError: false
shouldRetryOnError: true,
errorRetryInterval: 5_000,
errorRetryCount: 3,
revalidateOnFocus: true,
revalidateOnReconnect: true
});
const clearCollapseTimer = useCallback(() => {
if (collapseTimerRef.current !== null) {
@@ -374,6 +420,7 @@ export function useWorkbenchAgent({
}, []);
const stopSessionStreamSubscription = useCallback(() => {
clearTimeout(runtimeRef.current?.recoveryTimer);
runtimeRef.current?.controller?.abort();
if (runtimeRef.current) { runtimeRef.current.controller = undefined; runtimeRef.current.resumed = false; }
}, []);
@@ -402,8 +449,7 @@ export function useWorkbenchAgent({
setRegistry(backendConnection.registry);
setModelOptions(backendConnection.models);
setSelectedModel((current) => resolveAgentModelSelection(backendConnection, current));
if (chatRef.current.status === "ready") {
setConnectionStatus("online");
if (chatRef.current.status === "ready" && runtimeRef.current && !runtimeRunning(runtimeRef.current) && !runtimeRef.current.error) {
setStatusLabel("准备就绪");
}
}, [backendConnection]);
@@ -413,8 +459,7 @@ export function useWorkbenchAgent({
return;
}
setConnectionStatus("offline");
setStatusLabel("Agent 后端不可用");
if (runtimeRef.current && !runtimeRunning(runtimeRef.current) && !runtimeRef.current.error) setStatusLabel("Agent 初始化信息暂不可用");
}, [backendConnectionError]);
const collapsePanel = useCallback(() => {
@@ -469,6 +514,12 @@ export function useWorkbenchAgent({
() => clientRef.current.listResponseInputs(sessionId!),
{ refreshInterval: streaming ? 1000 : 0, shouldRetryOnError: false }
);
useEffect(() => {
if (!recoveryVersion) return;
void refreshBootstrap();
void refreshInteractionRequests();
void refreshPendingInputs();
}, [recoveryVersion, refreshBootstrap, refreshInteractionRequests, refreshPendingInputs]);
const updateInputDelivery = useCallback(async (id: string, delivery: AgentInputDelivery) => {
if (!sessionIdRef.current) return;
try {
@@ -504,10 +555,38 @@ export function useWorkbenchAgent({
if (sessionIdRef.current === activeSessionId) void refreshPendingInputs().catch(() => {});
}
}, [refreshPendingInputs, resources, setPromptDraft]);
const checkedLegacyQuestions = useRef(new Set<string>());
const checkingLegacyQuestions = useRef(new Set<string>());
const legacyQuestionScope = useRef(`${scopeKey}:${sessionId}`);
if (legacyQuestionScope.current !== `${scopeKey}:${sessionId}`) {
legacyQuestionScope.current = `${scopeKey}:${sessionId}`;
checkedLegacyQuestions.current.clear();
checkingLegacyQuestions.current.clear();
}
const messages = useMemo(
() => toAgentChatMessages(chat.messages, permissionOverrides, questionOverrides),
[chat.messages, permissionOverrides, questionOverrides]
() => toAgentChatMessages(chat.messages, permissionOverrides, questionOverrides).map((message) => ({ ...message,
questions: message.questions?.map((question) => question.status === "pending" && !checkedLegacyQuestions.current.has(`${scopeKey}:${question.sessionId}:${question.requestId}`)
? { ...question, status: "checking" as const } : question),
})),
[chat.messages, permissionOverrides, questionOverrides, scopeKey]
);
useEffect(() => {
const target = runtime;
for (const message of messages) for (const question of message.questions ?? []) {
const identity = `${scopeKey}:${question.sessionId}:${question.requestId}`;
if (question.status !== "checking" || checkingLegacyQuestions.current.has(identity)) continue;
checkingLegacyQuestions.current.add(identity);
void target.client.getQuestionStatus(question.sessionId, question.requestId).then(({ status }) => {
checkedLegacyQuestions.current.add(identity);
checkingLegacyQuestions.current.delete(identity);
target.questionOverrides = { ...target.questionOverrides, [question.requestId]: { status } };
if (resources.activeAgent === target) setQuestionOverrides(target.questionOverrides);
}).catch(() => {
// Uncertainty keeps historical controls disabled until the session is reopened.
checkingLegacyQuestions.current.delete(identity);
});
}
}, [messages, resources, runtime, scopeKey]);
const personaState = useMemo(
() => deriveAgentPersonaState(messages, {
chatStatus: chat.status,
@@ -521,13 +600,15 @@ export function useWorkbenchAgent({
const handleSessionStreamEvent = useCallback(
(event: AgentSessionStreamEvent, expectedSessionId: string) => {
if (event.type === "interaction_request") { void refreshInteractionRequests(); return; }
if (event.type === "state") {
void refreshInteractionRequests();
if (sessionIdRef.current !== expectedSessionId && sessionIdRef.current !== event.sessionId) {
return;
}
const nextMessages = runtimeRef.current?.chat.messages ?? toAgentUiMessages(event.messages);
setStreamRenderState(createCompletedStreamRenderState(nextMessages));
setStatusLabel(event.isStreaming ? "Agent 会话流已恢复" : "Agent 历史会话已打开");
setStatusLabel(event.runStatus === "waiting_input" ? "等待补充信息" : event.isStreaming ? "Agent 会话流已恢复" : "Agent 历史会话已打开");
return;
}
@@ -576,8 +657,9 @@ export function useWorkbenchAgent({
setStatusLabel(title);
}
} else if (event.type === "done") {
void refreshInteractionRequests();
markLastAssistantStreamDone(chatRef.current.messages, setStreamRenderState);
setStatusLabel("Agent 分析完成");
setStatusLabel(event.data.reason === "waiting_input" ? "等待补充信息" : "Agent 分析完成");
runtimeRef.current!.loadedAt = Date.now();
} else if (event.type === "error") {
markLastAssistantStreamDone(chatRef.current.messages, setStreamRenderState);
@@ -587,7 +669,7 @@ export function useWorkbenchAgent({
// Message changes are applied once by the owning runtime, including background streams.
},
[refreshAgentMessageInbox, claimEnvelope, handleCredentialRefresh, handleFrontendAction, onUiEnvelope]
[refreshInteractionRequests, refreshAgentMessageInbox, claimEnvelope, handleCredentialRefresh, handleFrontendAction, onUiEnvelope]
);
useLayoutEffect(() => {
@@ -595,23 +677,35 @@ export function useWorkbenchAgent({
onData: handleDataPart,
onEvent: (event) => handleSessionStreamEvent(event, runtime.sessionId ?? ""),
onError: (error) => {
void refreshInteractionRequests();
markLastAssistantStreamDone(runtime.chat.messages, setStreamRenderState);
setConnectionStatus("offline"); setStatusLabel(error.message);
if (isPendingInteractionError(error)) {
setStatusLabel("等待补充信息");
showMapNotice({ id: "agent-stream-status", tone: "warning", title: "有待回答的问题", message: "请在对话中的提问卡片中选择或填写后提交,也可以取消提问。" });
return;
}
checkAgentConnection();
setStatusLabel(error.message);
showMapNotice({ id: "agent-stream-status", tone: "error", title: "Agent 请求失败", message: error.message });
},
onFinish: (status) => {
void refreshInteractionRequests();
markLastAssistantStreamDone(runtime.chat.messages, setStreamRenderState);
setConnectionStatus(status === "error" ? "offline" : "online");
setStatusLabel(status === "error" ? runtime.error ?? "Agent 处理失败" : status === "aborted" ? "Agent 已停止" : "Agent 分析完成");
if (status === "error") checkAgentConnection();
setStatusLabel(status === "error" ? runtime.error ?? "Agent 处理失败" : status === "aborted" ? "Agent 已停止" : status === "waiting_input" ? "等待补充信息" : "Agent 分析完成");
},
onReconnect: () => {
checkAgentConnection();
setStatusLabel("连接已中断,正在恢复会话");
},
};
return () => { runtime.handlers = undefined; };
}, [handleDataPart, handleSessionStreamEvent, runtime]);
}, [checkAgentConnection, handleDataPart, handleSessionStreamEvent, refreshInteractionRequests, runtime]);
const loadHistorySession = useCallback(async (nextSessionId: string) => {
const generation = ++loadingGenerationRef.current;
let target = agentRuntimes(resources).find((entry) => entry.scope === scopeKey && entry.sessionId === nextSessionId);
try {
if (!target || target.error || (!runtimeRunning(target) && Date.now() - target.loadedAt >= 60_000)) {
if (!target || target.error || target.connectionError || (!runtimeRunning(target) && Date.now() - target.loadedAt >= 60_000)) {
const loaded = await resources.cache.load(JSON.stringify(["session", scopeKey, nextSessionId]),
() => scopeClient.loadSession(nextSessionId, { messageLimit: 40, signal: AbortSignal.timeout(15_000) }), 60_000, { force: Boolean(target?.error) });
if (!loaded) throw new Error("这个历史会话不存在或已被删除。");
@@ -732,7 +826,6 @@ export function useWorkbenchAgent({
activateRuntime(next);
setTaskDiscussion(null);
setStatusLabel(nextStatusLabel);
setConnectionStatus("online");
}, [activateRuntime, resources, scopeClient, scopeKey, transport]);
const selectTopic = (topicId: string | null) => {
@@ -798,7 +891,9 @@ export function useWorkbenchAgent({
return;
}
if (streaming) {
if (streaming || interactionRequests.awaitingInput) {
const owningRuntime = runtimeRef.current;
const owningSessionId = sessionIdRef.current;
try {
if (!sessionIdRef.current) throw new Error("Agent 正在建立会话,请稍后追加消息。");
const receipt = await clientRef.current.submitResponseInput(sessionIdRef.current, normalizedPrompt, inputDelivery, `msg_${crypto.randomUUID().replaceAll("-", "")}`);
@@ -808,15 +903,23 @@ export function useWorkbenchAgent({
? [...current.filter((input) => input.id !== receipt.id), receipt]
: current.filter((input) => input.id !== receipt.id), { revalidate: false })
.then(() => refreshPendingInputs()).catch(() => {});
return;
} catch (error) {
showMapNotice({ tone: "error", title: "追加消息失败", message: error instanceof Error ? error.message : "请稍后重试。" });
throw error;
if (error instanceof AgentResponseFinishedError) {
await refreshInteractionRequests();
if (runtimeRef.current !== owningRuntime || sessionIdRef.current !== owningSessionId) {
throw new Error("原会话已切换,请回到原会话后重新发送。");
}
} else {
showMapNotice({ tone: "error", title: "追加消息失败", message: error instanceof Error ? error.message : "请稍后重试。" });
throw error;
}
}
return;
}
setStatusLabel("Agent 正在分析");
runtimeRef.current!.error = undefined;
runtimeRef.current!.runStatus = "running";
await chat.sendMessage(
{ text: normalizedPrompt },
@@ -826,13 +929,13 @@ export function useWorkbenchAgent({
topic_id: !sessionIdRef.current ? runtimeRef.current?.topicId ?? undefined : undefined,
model: selectedModel || undefined,
approval_mode: approvalMode,
capabilities: frontendActionEnabledRef.current ? ["frontend-action@1"] : [],
capabilities: frontendActionEnabledRef.current ? ["frontend-action@1", "host-question@1"] : ["host-question@1"],
task_context: taskDiscussion?.sessionId === sessionIdRef.current ? taskDiscussion.context : undefined
}
}
);
},
[approvalMode, chat, inputDelivery, refreshPendingInputs, selectedModel, streaming, taskDiscussion]
[approvalMode, chat, inputDelivery, interactionRequests.awaitingInput, refreshInteractionRequests, refreshPendingInputs, selectedModel, streaming, taskDiscussion]
);
const triggerEvent = useCallback(async (trigger: AgentEventTriggerContext, prompt: string) => {
@@ -846,17 +949,19 @@ export function useWorkbenchAgent({
if (loaded.runStatus === "running" || loaded.isStreaming) resumeAgentRuntime(resources, target);
}
activateRuntime(target);
setConnectionStatus("online");
if (runtimeRunning(target) || target.chat.messages.length) return { sessionId: id, reused: true };
setStatusLabel("Agent 正在调查水质投诉事件");
void target.chat.sendMessage({ text: prompt }, { body: {
session_id: id, model: selectedModel || undefined, approval_mode: approvalMode, trigger_context: trigger,
capabilities: frontendActionEnabledRef.current ? ["frontend-action@1"] : [],
capabilities: frontendActionEnabledRef.current ? ["frontend-action@1", "host-question@1"] : ["host-question@1"],
} }).catch(() => {});
return { sessionId: id, reused: reserved.event_duplicate === true };
}, [activateRuntime, approvalMode, resources, scopeClient, scopeKey, selectedModel, transport]);
const stopPrompt = useCallback(() => {
for (const entry of interactionRequests.entries) {
if (entry.record.status === "pending") void interactionRequests.act(entry.record.interaction_id, "cancel");
}
chat.stop();
stopSessionStreamSubscription();
const activeSessionId = sessionIdRef.current;
@@ -864,7 +969,7 @@ export function useWorkbenchAgent({
void clientRef.current.abort(activeSessionId).then(() => refreshPendingInputs()).catch(() => undefined);
}
markLastAssistantStreamDone(chatRef.current.messages, setStreamRenderState);
}, [chat, refreshPendingInputs, stopSessionStreamSubscription]);
}, [chat, interactionRequests, refreshPendingInputs, stopSessionStreamSubscription]);
const replyPermission = useCallback(async (permission: AgentPermissionRequest, reply: AgentPermissionReply) => {
const target = resources.activeAgent;
@@ -933,6 +1038,7 @@ export function useWorkbenchAgent({
inputDelivery,
setInputDelivery,
pendingInputs,
interactionRequests,
updateInputDelivery,
withdrawInput,
agentMessageInbox,
@@ -946,6 +1052,7 @@ export function useWorkbenchAgent({
openMobilePanel,
closeMobilePanel,
connectionStatus,
connectionRecoveryVersion: recoveryVersion,
panelCollapsing,
panelOpen,
personaState,
@@ -1063,9 +1170,9 @@ function createWorkbenchAgentTransport(client: AgentApiClient, access?: Workbenc
const model = resolveAgentModelSelection(catalog, readBodyString(body, "model") ?? "");
if (!model) throw new Error("Agent 后端当前没有可用模型,请检查模型配置。");
return { body: createResponseRequest(messages, {
conversation: readBodyString(body, "session_id"), model, topicId: readBodyString(body, "topic_id"),
conversation: readBodyString(body, "session_id"), interactionId: readBodyString(body, "interaction_id"), model, topicId: readBodyString(body, "topic_id"),
approvalMode: readBodyString(body, "approval_mode") ?? "request",
capabilities: Array.isArray(body?.capabilities) ? body.capabilities.filter((value): value is string => value === "frontend-action@1") : [], triggerContext: body?.trigger_context, taskContext: body?.task_context,
capabilities: Array.isArray(body?.capabilities) ? body.capabilities.filter((value): value is string => value === "frontend-action@1" || value === "host-question@1") : [], triggerContext: body?.trigger_context, taskContext: body?.task_context,
}) };
},
});