diff --git a/extensions/qqbot/src/engine/gateway/typing-keepalive.test.ts b/extensions/qqbot/src/engine/gateway/typing-keepalive.test.ts index 65adae4fe029..a8e73ca4d7eb 100644 --- a/extensions/qqbot/src/engine/gateway/typing-keepalive.test.ts +++ b/extensions/qqbot/src/engine/gateway/typing-keepalive.test.ts @@ -54,4 +54,65 @@ describe("TypingKeepAlive", () => { await vi.advanceTimersByTimeAsync(10_000); expect(sendInputNotify).toHaveBeenCalledTimes(TYPING_RENEWAL_LIMIT); }); + + it("counts token-refresh retry attempts against the renewal budget", async () => { + vi.useFakeTimers(); + const clearCache = vi.fn(); + const sendInputNotify = vi + .fn(async () => undefined) + .mockRejectedValueOnce(new Error("11244 token expired")); + const keepAlive = new TypingKeepAlive( + async () => "token-1", + clearCache, + sendInputNotify, + "openid-1", + "msg-1", + ); + + keepAlive.start(); + + // First tick: the failed attempt and its token-refresh retry both spend budget. + await vi.advanceTimersByTimeAsync(5_000); + expect(clearCache).toHaveBeenCalledTimes(1); + expect(sendInputNotify).toHaveBeenCalledTimes(2); + + // Only one renewal remains before the reserved final-reply slot. + await vi.advanceTimersByTimeAsync(20_000); + expect(sendInputNotify).toHaveBeenCalledTimes(TYPING_RENEWAL_LIMIT); + }); + + it("suppresses overlapping renewals while a send is still in flight", async () => { + vi.useFakeTimers(); + let release: (() => void) | undefined; + const sendInputNotify = vi.fn( + () => + new Promise((resolve) => { + release = resolve; + }), + ); + const keepAlive = new TypingKeepAlive( + async () => "token-1", + vi.fn(), + sendInputNotify, + "openid-1", + "msg-1", + ); + + keepAlive.start(); + + await vi.advanceTimersByTimeAsync(5_000); + expect(sendInputNotify).toHaveBeenCalledTimes(1); + + // A stalled RPC must not double-send or burn extra reply budget. + await vi.advanceTimersByTimeAsync(10_000); + expect(sendInputNotify).toHaveBeenCalledTimes(1); + + release?.(); + // Let the stalled tick settle so the next interval tick is not suppressed. + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(5_000); + expect(sendInputNotify).toHaveBeenCalledTimes(2); + + keepAlive.stop(); + }); }); diff --git a/extensions/qqbot/src/engine/gateway/typing-keepalive.ts b/extensions/qqbot/src/engine/gateway/typing-keepalive.ts index c205777b3afd..b01129cdf35a 100644 --- a/extensions/qqbot/src/engine/gateway/typing-keepalive.ts +++ b/extensions/qqbot/src/engine/gateway/typing-keepalive.ts @@ -1,10 +1,11 @@ /** * Periodically refresh C2C typing state while a response is in progress. * - * All I/O operations are injected via constructor parameters so this - * module has zero external dependencies and can run in both plugin versions. + * Interval scheduling comes from the core typing keepalive loop; this module + * owns the QQ passive-reply budget accounting and token-refresh retry. */ +import { createTypingKeepaliveLoop } from "openclaw/plugin-sdk/channel-outbound"; import { formatErrorMessage } from "../utils/format.js"; /** Function that sends a typing indicator to one user. */ @@ -25,9 +26,14 @@ export const TYPING_RENEWAL_LIMIT = QQ_C2C_PASSIVE_REPLY_LIMIT - INITIAL_TYPING_NOTIFY_COUNT - FINAL_REPLY_RESERVE_COUNT; export class TypingKeepAlive { - private timer: ReturnType | null = null; - private stopped = false; private renewalsRemaining = TYPING_RENEWAL_LIMIT; + private stopped = false; + // Core loop owns the interval and in-flight tick suppression; budget + // accounting in sendAttempt() decides when it must stop for good. + private readonly loop = createTypingKeepaliveLoop({ + intervalMs: TYPING_INTERVAL_MS, + onTick: () => this.send(), + }); constructor( private readonly getToken: () => Promise, @@ -35,36 +41,24 @@ export class TypingKeepAlive { private readonly sendInputNotify: SendInputNotifyFn, private readonly openid: string, private readonly msgId: string | undefined, - private readonly log?: { - info: (msg: string) => void; - error: (msg: string) => void; - debug?: (msg: string) => void; - }, + private readonly log?: { debug?: (msg: string) => void }, ) {} /** Start periodic keep-alive sends. */ start(): void { - if (this.stopped) { - return; + // stop() is a permanent latch: a stopped keepalive must never spend more budget. + if (!this.stopped) { + this.loop.start(); } - this.timer = setInterval(() => { - if (this.stopped) { - this.stop(); - return; - } - this.send().catch(() => {}); - }, TYPING_INTERVAL_MS); } /** Stop periodic keep-alive sends. */ stop(): void { this.stopped = true; - if (this.timer) { - clearInterval(this.timer); - this.timer = null; - } + this.loop.stop(); } + // Never rejects: the core loop does not catch onTick errors. private async send(): Promise { try { const token = await this.getToken(); @@ -88,6 +82,8 @@ export class TypingKeepAlive { return; } + // Count attempts, not successes (token-refresh retries included): a failed + // call may still consume QQ's msg_id budget, so keep FINAL_REPLY_RESERVE_COUNT safe. this.renewalsRemaining--; try { await this.sendInputNotify(token, this.openid, this.msgId, TYPING_INPUT_SECOND); diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index 1f09661f8c15..51cbe682c138 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -147,8 +147,8 @@ const defaultPublicDeprecatedExportsByEntrypointBudget = Object.freeze({ "channel-mention-gating": 7, "channel-lifecycle": 23, "channel-ingress": 8, - "channel-message": 230, - "channel-message-runtime": 227, + "channel-message": 231, + "channel-message-runtime": 228, "channel-pairing-paths": 1, "channel-policy": 8, "channel-route": 5, @@ -195,17 +195,17 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { ), publicExports: readPluginSdkSurfaceBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", - 10523, + 10526, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", - 5243, + 5246, env, ), publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_DEPRECATED_EXPORTS", - 3272, + 3274, env, ), publicWildcardReexports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/plugin-sdk/channel-outbound.ts b/src/plugin-sdk/channel-outbound.ts index 3e782ab0b869..201e2655629c 100644 --- a/src/plugin-sdk/channel-outbound.ts +++ b/src/plugin-sdk/channel-outbound.ts @@ -31,6 +31,9 @@ export { createChannelReplyPipeline as createChannelMessageReplyPipeline, resolveChannelSourceReplyDeliveryMode as resolveChannelMessageSourceReplyDeliveryMode, } from "../channels/message/index.js"; +// Bare interval/stop orchestration for channels that own their typing renewal +// policy (e.g. per-message reply budgets) instead of the createTypingCallbacks lifecycle. +export { createTypingKeepaliveLoop } from "../channels/typing-lifecycle.js"; export { createFinalizableDraftLifecycle,