diff --git a/src/messaging/process-message.ts b/src/messaging/process-message.ts index 08ab205..8aaf9a5 100644 --- a/src/messaging/process-message.ts +++ b/src/messaging/process-message.ts @@ -24,7 +24,6 @@ import { isDebugMode } from "./debug-mode.js"; import { sendWeixinErrorNotice } from "./error-notice.js"; import { applyWeixinMessageSendingHook, emitWeixinMessageSent } from "./outbound-hooks.js"; import { - setContextToken, weixinMessageToMsgContext, getContextTokenFromMsgContext, isMediaItem, @@ -38,6 +37,16 @@ import { handleSlashCommand } from "./slash-commands.js"; const MEDIA_OUTBOUND_TEMP_DIR = path.join(resolvePreferredOpenClawTmpDir(), "weixin/media/outbound-temp"); +type DispatchReplyOptions = NonNullable< + Parameters[0]["replyOptions"] +> & { + queuedFollowupLifecycle?: { + onEnqueued?: () => void; + onComplete?: () => void; + }; + onTurnAdopted?: () => void | Promise; +}; + /** Dependencies for processOneMessage, injected by the monitor loop. */ export type ProcessMessageDeps = { accountId: string; @@ -49,6 +58,7 @@ export type ProcessMessageDeps = { typingTicket?: string; log: (msg: string) => void; errLog: (m: string) => void; + onReplyAdmitted?: () => void; }; /** Extract text body from item_list (for slash command detection). */ @@ -166,7 +176,7 @@ export async function processOneMessage( const ctx = weixinMessageToMsgContext(full, deps.accountId, mediaOpts); // --- Framework command authorization --- - const rawBody = ctx.Body?.trim() ?? ""; + const rawBody = textBody.trim() || (ctx.Body?.trim() ?? ""); ctx.CommandBody = rawBody; const senderId = full.from_user_id ?? ""; @@ -245,7 +255,10 @@ export async function processOneMessage( agentId: route.agentId, }); const finalized = deps.channelRuntime.reply.finalizeInboundContext( - ctx as Parameters[0], + { + ...ctx, + BodyForAgent: ctx.Body, + } as Parameters[0], ); logger.info( @@ -270,9 +283,6 @@ export async function processOneMessage( ); const contextToken = getContextTokenFromMsgContext(ctx); - if (contextToken) { - setContextToken(deps.accountId, full.from_user_id ?? "", contextToken); - } const runId = randomUUID(); const replyProgressSender = resolveReplyProgressMessagesEnabled(deps.config) ? new WeixinReplyProgressSender({ @@ -446,6 +456,23 @@ export async function processOneMessage( }, }); + let queuedFollowup = false; + const dispatchReplyOptions: DispatchReplyOptions = { + ...replyOptions, + ...(replyProgressSender?.replyOptions ?? {}), + // Newer hosts use this marker for active-run admission; older hosts ignore it. + queuedFollowupLifecycle: { + onEnqueued: () => { + queuedFollowup = true; + deps.onReplyAdmitted?.(); + }, + onComplete: () => void replyProgressSender?.finalize(), + }, + onAgentRunStart: () => deps.onReplyAdmitted?.(), + onTurnAdopted: deps.onReplyAdmitted, + disableBlockStreaming: true, + }; + logger.debug(`dispatchReplyFromConfig: starting agentId=${route.agentId ?? "(none)"}`); try { await deps.channelRuntime.reply.withReplyDispatcher({ @@ -455,11 +482,7 @@ export async function processOneMessage( ctx: finalized, cfg: deps.config, dispatcher, - replyOptions: { - ...replyOptions, - ...(replyProgressSender?.replyOptions ?? {}), - disableBlockStreaming: true, - }, + replyOptions: dispatchReplyOptions, }), }); logger.debug(`dispatchReplyFromConfig: done agentId=${route.agentId ?? "(none)"}`); @@ -470,7 +493,7 @@ export async function processOneMessage( throw err; } finally { markDispatchIdle(); - await replyProgressSender?.finalize(); + if (!queuedFollowup) await replyProgressSender?.finalize(); logger.info( `debug-check: accountId=${deps.accountId} debug=${String(debug)} hasContextToken=${Boolean(contextToken)}`, diff --git a/src/monitor/monitor.test.ts b/src/monitor/monitor.test.ts new file mode 100644 index 0000000..cf60044 --- /dev/null +++ b/src/monitor/monitor.test.ts @@ -0,0 +1,172 @@ +import { describe, expect, it, vi } from "vitest"; + +import { MessageItemType, type GetUpdatesResp, type WeixinMessage } from "../api/types.js"; +import type { ProcessMessageDeps } from "../messaging/process-message.js"; + +const getUpdatesMock = vi.fn<(opts: { abortSignal?: AbortSignal }) => Promise>(); +const getForUserMock = vi.fn< + (userId: string, contextToken?: string) => Promise<{ typingTicket: string }> +>(); +const processOneMessageMock = + vi.fn<(message: WeixinMessage, deps: ProcessMessageDeps) => Promise>(); +const saveGetUpdatesBufMock = vi.fn<(filePath: string, value: string) => void>(); +const setContextTokenMock = vi.fn<(accountId: string, userId: string, token: string) => void>(); + +vi.mock("../api/api.js", () => ({ + getUpdates: (opts: { abortSignal?: AbortSignal }) => getUpdatesMock(opts), + classifyFetchError: (err: unknown) => ({ + type: "mock", + description: String(err), + code: undefined, + }), +})); + +vi.mock("../api/config-cache.js", () => ({ + WeixinConfigManager: class { + async getForUser(userId: string, contextToken?: string): Promise<{ typingTicket: string }> { + return getForUserMock(userId, contextToken); + } + }, +})); + +vi.mock("../messaging/process-message.js", () => ({ + processOneMessage: (message: WeixinMessage, deps: ProcessMessageDeps) => + processOneMessageMock(message, deps), +})); + +vi.mock("../messaging/inbound.js", () => ({ + setContextToken: (accountId: string, userId: string, token: string) => + setContextTokenMock(accountId, userId, token), +})); + +vi.mock("../storage/sync-buf.js", () => ({ + getSyncBufFilePath: () => "sync-buf", + loadGetUpdatesBuf: () => undefined, + saveGetUpdatesBuf: (filePath: string, value: string) => + saveGetUpdatesBufMock(filePath, value), +})); + +vi.mock("../util/logger.js", () => ({ + logger: { + withAccount: () => ({ + info: vi.fn(), + debug: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }), + }, +})); + +describe("monitorWeixinProvider", () => { + it("orders ordinary admission while approvals bypass an active ordinary turn", async () => { + vi.resetModules(); + const { monitorWeixinProvider } = await import("./monitor.js"); + const abortController = new AbortController(); + const firstPreprocessing = createDeferred(); + const firstRun = createDeferred(); + const approvalStarted = createDeferred(); + const secondStarted = createDeferred(); + const started: string[] = []; + const responses: GetUpdatesResp[] = [ + { + ret: 0, + msgs: [makeMessage("first", { message_id: 101, context_token: "token-1" })], + get_updates_buf: "cursor-1", + }, + { + ret: 0, + msgs: [ + makeMessage("second", { message_id: 102, context_token: "token-2" }), + makeMessage("third", { message_id: 104, context_token: "token-4" }), + makeMessage("/approve plugin:test approve", { + message_id: 103, + context_token: "token-3", + }), + ], + get_updates_buf: "cursor-2", + }, + ]; + + getUpdatesMock.mockImplementation(async ({ abortSignal }) => { + const next = responses.shift(); + if (next) return next; + return await new Promise((_, reject) => { + if (abortSignal?.aborted) { + reject(new Error("aborted")); + return; + } + abortSignal?.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); + }); + }); + getForUserMock.mockResolvedValue({ typingTicket: "ticket" }); + processOneMessageMock.mockImplementation(async (message, deps) => { + const text = getText(message); + started.push(text); + if (text === "first") { + await firstPreprocessing.promise; + deps.onReplyAdmitted?.(); + await firstRun.promise; + return; + } + if (text.startsWith("/approve plugin:")) { + approvalStarted.resolve(); + return; + } + if (text === "second") { + secondStarted.resolve(); + abortController.abort(); + } + }); + + const monitor = monitorWeixinProvider({ + baseUrl: "https://example.test", + cdnBaseUrl: "https://cdn.example.test", + accountId: "acc-monitor", + config: {} as never, + channelRuntime: {} as never, + abortSignal: abortController.signal, + runtime: { log: vi.fn(), error: vi.fn() }, + }); + + try { + await approvalStarted.promise; + expect(started).toEqual(["first", "/approve plugin:test approve"]); + firstPreprocessing.resolve(); + await secondStarted.promise; + await monitor; + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(started).toEqual(["first", "/approve plugin:test approve", "second"]); + expect(saveGetUpdatesBufMock).toHaveBeenLastCalledWith("sync-buf", "cursor-2"); + expect(setContextTokenMock.mock.calls).toEqual([ + ["acc-monitor", "user-a", "token-1"], + ["acc-monitor", "user-a", "token-2"], + ["acc-monitor", "user-a", "token-4"], + ["acc-monitor", "user-a", "token-3"], + ]); + } finally { + firstPreprocessing.resolve(); + firstRun.resolve(); + await monitor; + } + }); +}); + +function makeMessage(text: string, overrides: Partial = {}): WeixinMessage { + return { + from_user_id: "user-a", + item_list: [{ type: MessageItemType.TEXT, text_item: { text } }], + ...overrides, + }; +} + +function getText(message: WeixinMessage): string { + return message.item_list?.[0]?.text_item?.text ?? ""; +} + +function createDeferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((innerResolve) => { + resolve = innerResolve; + }); + return { promise, resolve }; +} diff --git a/src/monitor/monitor.ts b/src/monitor/monitor.ts index 58e1497..eabded9 100644 --- a/src/monitor/monitor.ts +++ b/src/monitor/monitor.ts @@ -2,8 +2,10 @@ import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contrac import type { PluginRuntime } from "openclaw/plugin-sdk/core"; import { getUpdates, classifyFetchError } from "../api/api.js"; +import { MessageItemType, type WeixinMessage } from "../api/types.js"; import { WeixinConfigManager } from "../api/config-cache.js"; import { STALE_TOKEN_ERRCODE, pauseSession, getRemainingPauseMs } from "../api/session-guard.js"; +import { setContextToken } from "../messaging/inbound.js"; import { processOneMessage } from "../messaging/process-message.js"; import { getSyncBufFilePath, loadGetUpdatesBuf, saveGetUpdatesBuf } from "../storage/sync-buf.js"; import { logger } from "../util/logger.js"; @@ -14,6 +16,7 @@ const DEFAULT_LONG_POLL_TIMEOUT_MS = 35_000; const MAX_CONSECUTIVE_FAILURES = 3; const BACKOFF_DELAY_MS = 30_000; const RETRY_DELAY_MS = 2_000; +const PLUGIN_APPROVAL_RE = /^\/approve\s+plugin:/i; export type MonitorWeixinOpts = { baseUrl: string; @@ -36,7 +39,7 @@ export type MonitorWeixinOpts = { }; /** - * Long-poll loop: getUpdates -> normalize -> recordInboundSession -> dispatchReplyFromConfig. + * Long-poll loop: getUpdates -> dispatchReplyFromConfig. * Runs until abort. */ export async function monitorWeixinProvider(opts: MonitorWeixinOpts): Promise { @@ -66,7 +69,6 @@ export async function monitorWeixinProvider(opts: MonitorWeixinOpts): Promise void, + ): Promise => { + if (abortSignal?.aborted) return; + aLog.info( + `inbound message: from=${full.from_user_id} types=${full.item_list?.map((i) => i.type).join(",") ?? "none"}`, + ); + + const now = Date.now(); + setStatus?.({ accountId, lastEventAt: now, lastInboundAt: now }); + + // allowFrom filtering is delegated to processOneMessage via the framework + // authorization pipeline (resolveSenderCommandAuthorizationWithRuntime). + + const fromUserId = full.from_user_id ?? ""; + const cachedConfig = await configManager.getForUser(fromUserId, full.context_token); + if (abortSignal?.aborted) return; + + await processOneMessage(full, { + accountId, + config, + channelRuntime, + baseUrl, + cdnBaseUrl, + token, + typingTicket: cachedConfig.typingTicket, + log: opts.runtime?.log ?? (() => {}), + errLog, + onReplyAdmitted, + }); + }; + // Serialize preprocessing until core accepts the turn; approvals use an independent lane. + let ordinaryLane = Promise.resolve(); + let approvalLane = Promise.resolve(); + const scheduleInboundMessage = (full: WeixinMessage): void => { + const isApproval = isPluginApprovalMessage(full); + const previous = isApproval ? approvalLane : ordinaryLane; + const next = previous.then( + () => + new Promise((releaseLane) => { + let released = false; + const releaseOnce = () => { + if (released) return; + released = true; + releaseLane(); + }; + void processInboundMessage(full, releaseOnce) + .catch((err) => { + errLog(`weixin inbound message failed: ${String(err)}`); + aLog.error( + `Inbound message failed: ${String(err)}, stack=${(err as Error).stack ?? "none"}`, + ); + }) + .finally(releaseOnce); + }), + ); + if (isApproval) { + approvalLane = next; + } else { + ordinaryLane = next; + } + }; let nextTimeoutMs = longPollTimeoutMs ?? DEFAULT_LONG_POLL_TIMEOUT_MS; let consecutiveFailures = 0; @@ -154,32 +219,11 @@ export async function monitorWeixinProvider(opts: MonitorWeixinOpts): Promise i.type).join(",") ?? "none"}`, - ); - - const now = Date.now(); - setStatus?.({ accountId, lastEventAt: now, lastInboundAt: now }); - - // allowFrom filtering is delegated to processOneMessage via the framework - // authorization pipeline (resolveSenderCommandAuthorizationWithRuntime). - - const fromUserId = full.from_user_id ?? ""; - const cachedConfig = await configManager.getForUser(fromUserId, full.context_token); - - await processOneMessage(full, { - accountId, - config, - channelRuntime, - baseUrl, - cdnBaseUrl, - token, - typingTicket: cachedConfig.typingTicket, - log: opts.runtime?.log ?? (() => {}), - errLog, - }); + for (const full of resp.msgs ?? []) { + if (full.context_token) { + setContextToken(accountId, full.from_user_id ?? "", full.context_token); + } + scheduleInboundMessage(full); } } catch (err) { if (abortSignal?.aborted) { @@ -209,6 +253,11 @@ export async function monitorWeixinProvider(opts: MonitorWeixinOpts): Promise item.type === MessageItemType.TEXT)?.text_item?.text; + return PLUGIN_APPROVAL_RE.test(String(text ?? "").trim()); +} + function sleep(ms: number, signal?: AbortSignal): Promise { return new Promise((resolve, reject) => { const t = setTimeout(resolve, ms);