fix: harden agent and WeChat run completion
This commit is contained in:
john
+130
-86
@@ -69,6 +69,7 @@ const DEFAULT_WECHAT_JSAPI_TICKET_URL = 'https://api.weixin.qq.com/cgi-bin/ticke
|
||||
const DEFAULT_ASR_TARGET = process.env.H5_ASR_TARGET ?? 'https://asr.tkmind.cn';
|
||||
const WECHAT_RECENT_MEDIA_TTL_MS = 15 * 60 * 1000;
|
||||
const WECHAT_RECENT_IMAGE_MAX_COUNT = 10;
|
||||
const DEFAULT_WECHAT_AGENT_REPLY_TIMEOUT_MS = 15 * 60 * 1000;
|
||||
export { loadWechatMpConfig };
|
||||
const PUBLIC_HTML_LINK_PATTERN =
|
||||
/https?:\/\/[^\s<>"')\]]+\/MindSpace\/([0-9a-f-]{36}|[a-z0-9._-]+)\/public\/([^\s<>"')\]]+\.html)/gi;
|
||||
@@ -229,110 +230,137 @@ function pushMessage(messages, incoming) {
|
||||
return [...messages, incoming];
|
||||
}
|
||||
|
||||
async function executeSessionReply(
|
||||
export async function executeSessionReply(
|
||||
apiFetch,
|
||||
sessionId,
|
||||
requestId,
|
||||
prompt,
|
||||
metadata = {},
|
||||
{ submitReply = null, prepareUserMessage = null } = {},
|
||||
{ submitReply = null, prepareUserMessage = null, timeoutMs = 0 } = {},
|
||||
) {
|
||||
const eventsResponse = await apiFetch(`/sessions/${sessionId}/events`, {
|
||||
method: 'GET',
|
||||
headers: { Accept: 'text/event-stream' },
|
||||
});
|
||||
if (!eventsResponse.ok || !eventsResponse.body) {
|
||||
const text = await eventsResponse.text().catch(() => '');
|
||||
throw new Error(text || '无法建立公众号消息事件流');
|
||||
}
|
||||
const normalizedTimeoutMs = Math.max(0, Number(timeoutMs) || 0);
|
||||
const abortController = new AbortController();
|
||||
let timedOut = false;
|
||||
let reader = null;
|
||||
const timeout = normalizedTimeoutMs > 0
|
||||
? setTimeout(() => {
|
||||
timedOut = true;
|
||||
abortController.abort();
|
||||
void reader?.cancel?.().catch?.(() => {});
|
||||
}, normalizedTimeoutMs)
|
||||
: null;
|
||||
|
||||
let userMessage = createUserMessage(prompt, metadata);
|
||||
if (prepareUserMessage) {
|
||||
userMessage = (await prepareUserMessage(userMessage)) ?? userMessage;
|
||||
}
|
||||
if (submitReply) {
|
||||
await submitReply({ sessionId, requestId, userMessage });
|
||||
} else {
|
||||
const replyResponse = await apiFetch(`/sessions/${sessionId}/reply`, {
|
||||
method: 'POST',
|
||||
body: JSON.stringify({
|
||||
request_id: requestId,
|
||||
user_message: userMessage,
|
||||
}),
|
||||
try {
|
||||
const eventsResponse = await apiFetch(`/sessions/${sessionId}/events`, {
|
||||
method: 'GET',
|
||||
headers: { Accept: 'text/event-stream' },
|
||||
signal: abortController.signal,
|
||||
});
|
||||
if (!replyResponse.ok) {
|
||||
const text = await replyResponse.text().catch(() => '');
|
||||
throw new Error(text || 'Agent reply 请求失败');
|
||||
if (!eventsResponse.ok || !eventsResponse.body) {
|
||||
const text = await eventsResponse.text().catch(() => '');
|
||||
throw new Error(text || '无法建立公众号消息事件流');
|
||||
}
|
||||
replyResponse.body?.cancel().catch?.(() => {});
|
||||
}
|
||||
|
||||
const reader = eventsResponse.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
let messages = [];
|
||||
let requestMessages = [];
|
||||
let hasScopedAssistantUpdate = false;
|
||||
|
||||
while (true) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const frames = buffer.split('\n\n');
|
||||
buffer = frames.pop() ?? '';
|
||||
for (const frame of frames) {
|
||||
let data = '';
|
||||
for (const line of frame.split('\n')) {
|
||||
if (line.startsWith('data:')) data += line.slice(5).trim();
|
||||
let userMessage = createUserMessage(prompt, metadata);
|
||||
if (prepareUserMessage) {
|
||||
userMessage = (await prepareUserMessage(userMessage)) ?? userMessage;
|
||||
}
|
||||
if (submitReply) {
|
||||
await submitReply({ sessionId, requestId, userMessage, signal: abortController.signal });
|
||||
} else {
|
||||
const replyResponse = await apiFetch(`/sessions/${sessionId}/reply`, {
|
||||
method: 'POST',
|
||||
body: JSON.stringify({
|
||||
request_id: requestId,
|
||||
user_message: userMessage,
|
||||
}),
|
||||
signal: abortController.signal,
|
||||
});
|
||||
if (!replyResponse.ok) {
|
||||
const text = await replyResponse.text().catch(() => '');
|
||||
throw new Error(text || 'Agent reply 请求失败');
|
||||
}
|
||||
if (!data) continue;
|
||||
let event;
|
||||
try {
|
||||
event = JSON.parse(data);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
const routingId = event.chat_request_id ?? event.request_id;
|
||||
if (routingId && routingId !== requestId) continue;
|
||||
replyResponse.body?.cancel().catch?.(() => {});
|
||||
}
|
||||
|
||||
if (event.type === 'Message' && event.message?.metadata?.userVisible) {
|
||||
const hasActionRequired = event.message.content?.some((item) => item.type === 'actionRequired');
|
||||
if (hasActionRequired) {
|
||||
throw new Error('当前回复需要人工确认，公众号通道暂不支持');
|
||||
reader = eventsResponse.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
let messages = [];
|
||||
let requestMessages = [];
|
||||
let hasScopedAssistantUpdate = false;
|
||||
|
||||
while (true) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const frames = buffer.split('\n\n');
|
||||
buffer = frames.pop() ?? '';
|
||||
for (const frame of frames) {
|
||||
let data = '';
|
||||
for (const line of frame.split('\n')) {
|
||||
if (line.startsWith('data:')) data += line.slice(5).trim();
|
||||
}
|
||||
if (event.message.role === 'assistant') hasScopedAssistantUpdate = true;
|
||||
messages = pushMessage(messages, event.message);
|
||||
requestMessages = pushMessage(requestMessages, event.message);
|
||||
} else if (event.type === 'UpdateConversation') {
|
||||
// Ignore unscoped snapshots until this request has yielded an assistant update.
|
||||
// Otherwise a stale session snapshot can overwrite the current reply with a
|
||||
// previous page/link from the same WeChat-dedicated session.
|
||||
if (hasScopedAssistantUpdate) {
|
||||
messages = (event.conversation ?? []).filter((item) => item.metadata?.userVisible);
|
||||
if (!data) continue;
|
||||
let event;
|
||||
try {
|
||||
event = JSON.parse(data);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
} else if (event.type === 'Error') {
|
||||
throw new Error(event.error || '任务执行失败');
|
||||
} else if (event.type === 'Finish') {
|
||||
const assistant = [...messages].reverse().find((item) => item.role === 'assistant');
|
||||
if (!hasScopedAssistantUpdate || !assistant) {
|
||||
throw new Error('本轮未收到可发送的新回复，请稍后重试');
|
||||
const routingId = event.chat_request_id ?? event.request_id;
|
||||
if (routingId && routingId !== requestId) continue;
|
||||
|
||||
if (event.type === 'Message' && event.message?.metadata?.userVisible) {
|
||||
const hasActionRequired = event.message.content?.some((item) => item.type === 'actionRequired');
|
||||
if (hasActionRequired) {
|
||||
throw new Error('当前回复需要人工确认，公众号通道暂不支持');
|
||||
}
|
||||
if (event.message.role === 'assistant') hasScopedAssistantUpdate = true;
|
||||
messages = pushMessage(messages, event.message);
|
||||
requestMessages = pushMessage(requestMessages, event.message);
|
||||
} else if (event.type === 'UpdateConversation') {
|
||||
// Ignore unscoped snapshots until this request has yielded an assistant update.
|
||||
// Otherwise a stale session snapshot can overwrite the current reply with a
|
||||
// previous page/link from the same WeChat-dedicated session.
|
||||
if (hasScopedAssistantUpdate) {
|
||||
messages = (event.conversation ?? []).filter((item) => item.metadata?.userVisible);
|
||||
}
|
||||
} else if (event.type === 'Error') {
|
||||
throw new Error(event.error || '任务执行失败');
|
||||
} else if (event.type === 'Finish') {
|
||||
const assistant = [...messages].reverse().find((item) => item.role === 'assistant');
|
||||
if (!hasScopedAssistantUpdate || !assistant) {
|
||||
throw new Error('本轮未收到可发送的新回复，请稍后重试');
|
||||
}
|
||||
const reply = {
|
||||
text: messageVisibleText(assistant),
|
||||
tokenState: event.token_state ?? null,
|
||||
messages,
|
||||
// Keep request-scoped stream messages separate from a later full
|
||||
// UpdateConversation snapshot. Artifact delivery must never inspect
|
||||
// historical tool calls from the whole dedicated session.
|
||||
requestMessages,
|
||||
};
|
||||
assertWechatAgentReplyIsSendable(reply);
|
||||
return reply;
|
||||
}
|
||||
const reply = {
|
||||
text: messageVisibleText(assistant),
|
||||
tokenState: event.token_state ?? null,
|
||||
messages,
|
||||
// Keep request-scoped stream messages separate from a later full
|
||||
// UpdateConversation snapshot. Artifact delivery must never inspect
|
||||
// historical tool calls from the whole dedicated session.
|
||||
requestMessages,
|
||||
};
|
||||
assertWechatAgentReplyIsSendable(reply);
|
||||
return reply;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error('公众号消息事件流提前结束');
|
||||
throw new Error('公众号消息事件流提前结束');
|
||||
} catch (err) {
|
||||
if (timedOut) {
|
||||
const timeoutError = new Error(`公众号消息处理超时（${normalizedTimeoutMs}ms），请稍后重试`);
|
||||
timeoutError.code = 'WECHAT_AGENT_REPLY_TIMEOUT';
|
||||
timeoutError.retryable = true;
|
||||
throw timeoutError;
|
||||
}
|
||||
throw err;
|
||||
} finally {
|
||||
if (timeout) clearTimeout(timeout);
|
||||
await reader?.cancel?.().catch?.(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
const WECHAT_CUSTOMER_TEXT_MAX_BYTES = 2048;
|
||||
@@ -1580,6 +1608,13 @@ export function createWechatMpService({
|
||||
mediaAnalysisGrayUsers: Array.isArray(config.mediaAnalysisGrayUsers)
|
||||
? config.mediaAnalysisGrayUsers
|
||||
: [],
|
||||
reliabilityGrayUsers: Array.isArray(config.reliabilityGrayUsers)
|
||||
? config.reliabilityGrayUsers
|
||||
: [],
|
||||
agentReplyTimeoutMs: Math.max(
|
||||
0,
|
||||
Number(config.agentReplyTimeoutMs ?? DEFAULT_WECHAT_AGENT_REPLY_TIMEOUT_MS),
|
||||
),
|
||||
requireFreshPageThumbnail,
|
||||
repairFreshPageThumbnail,
|
||||
asrTarget: config.asrTarget || DEFAULT_ASR_TARGET,
|
||||
@@ -2321,6 +2356,8 @@ export function createWechatMpService({
|
||||
const runIntentMessage = async ({ inbound, intent, user }) => {
|
||||
const wechatIntent = classifyWechatIntent(intent);
|
||||
const mediaAnalysisEnabled = isWechatMediaGrayUser(user, config.mediaAnalysisGrayUsers);
|
||||
const reliabilityEnabled = isWechatMediaGrayUser(user, config.reliabilityGrayUsers);
|
||||
const agentReplyTimeoutMs = reliabilityEnabled ? config.agentReplyTimeoutMs : 0;
|
||||
const resetCandidate =
|
||||
intent.msgType === 'text' || intent.msgType === 'voice' ? intent.agentText : '';
|
||||
const imagePolicy = resolveWechatImageGenerationPolicy({
|
||||
@@ -2396,6 +2433,7 @@ export function createWechatMpService({
|
||||
options: { requireHistoricalImageIsolation: true },
|
||||
})
|
||||
: null,
|
||||
timeoutMs: agentReplyTimeoutMs,
|
||||
},
|
||||
);
|
||||
const generatedImages = collectWechatGeneratedImages(replyRequestMessages(reply));
|
||||
@@ -2549,6 +2587,11 @@ export function createWechatMpService({
|
||||
return { sessionId };
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
if (err?.code === 'WECHAT_AGENT_REPLY_TIMEOUT') {
|
||||
await userAuth.clearWechatAgentRoute(config.appId, inbound.fromUserName).catch((clearErr) => {
|
||||
logger.warn?.('WeChat MP timed-out session route clear failed:', clearErr);
|
||||
});
|
||||
}
|
||||
// A Page Data request must fail closed. Retrying a poisoned completion in
|
||||
// another session while the finish guard is also active can turn one
|
||||
// request into repairs against historical pages. Drop only this user's
|
||||
@@ -2611,6 +2654,7 @@ export function createWechatMpService({
|
||||
options: { requireHistoricalImageIsolation: true },
|
||||
})
|
||||
: null,
|
||||
timeoutMs: agentReplyTimeoutMs,
|
||||
},
|
||||
);
|
||||
const generatedImages = collectWechatGeneratedImages(replyRequestMessages(reply));
|
||||
|
||||
Reference in New Issue
Block a user