@@ -51,6 +51,7 @@ vi.mock("openclaw/plugin-sdk/runtime-env", () => ({
5151}));
52525353let TelegramPollingSession: typeof import("./polling-session.js").TelegramPollingSession;
54+let claimTelegramSpooledUpdate: typeof import("./telegram-ingress-spool.js").claimTelegramSpooledUpdate;
5455let listTelegramSpooledUpdates: typeof import("./telegram-ingress-spool.js").listTelegramSpooledUpdates;
5556let writeTelegramSpooledUpdate: typeof import("./telegram-ingress-spool.js").writeTelegramSpooledUpdate;
5657@@ -310,10 +311,111 @@ async function waitForApiMiddleware(
310311throw new Error("Telegram API middleware was not installed");
311312}
312313314+type TestTelegramUpdate = {
315+update_id: number;
316+message: {
317+text: string;
318+chat: { id: number; type: "supergroup" };
319+message_thread_id?: number;
320+is_topic_message?: boolean;
321+};
322+};
323+324+function topicUpdate(updateId: number, threadId: number, text: string): TestTelegramUpdate {
325+return {
326+update_id: updateId,
327+message: {
328+ text,
329+message_thread_id: threadId,
330+is_topic_message: true,
331+chat: { id: -100, type: "supergroup" },
332+},
333+};
334+}
335+336+async function writeSpooledTestUpdates(
337+spoolDir: string,
338+updates: readonly TestTelegramUpdate[],
339+): Promise<void> {
340+for (const update of updates) {
341+await writeTelegramSpooledUpdate({ spoolDir, update });
342+}
343+}
344+345+async function pendingUpdateIds(spoolDir: string, limit: number | "all" = 100): Promise<number[]> {
346+return (await listTelegramSpooledUpdates({ spoolDir, limit })).map((update) => update.updateId);
347+}
348+349+async function withTempSpool<T>(fn: (spoolDir: string) => Promise<T>): Promise<T> {
350+const spoolDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
351+try {
352+return await fn(spoolDir);
353+} finally {
354+await fs.rm(spoolDir, { recursive: true, force: true });
355+}
356+}
357+358+function createIdleIngressWorker() {
359+let stopWorker: (() => void) | undefined;
360+const workerDone = new Promise<void>((resolve) => {
361+stopWorker = resolve;
362+});
363+const createWorker = vi.fn(() => ({
364+onMessage: vi.fn(() => () => undefined),
365+stop: vi.fn(async () => {
366+stopWorker?.();
367+}),
368+task: vi.fn(async () => {
369+await workerDone;
370+}),
371+}));
372+return {
373+ createWorker,
374+stop: () => stopWorker?.(),
375+};
376+}
377+378+function startIsolatedIngressSession(params: {
379+abort: AbortController;
380+spoolDir: string;
381+handleUpdate: (update: { update_id?: number }) => Promise<void>;
382+drainIntervalMs?: number;
383+log?: (message: string) => void;
384+stop?: () => Promise<void>;
385+}) {
386+const worker = createIdleIngressWorker();
387+const bot = {
388+api: {
389+deleteWebhook: vi.fn(async () => true),
390+config: { use: vi.fn() },
391+},
392+init: vi.fn(async () => undefined),
393+handleUpdate: vi.fn(params.handleUpdate),
394+stop: vi.fn(params.stop ?? (async () => undefined)),
395+};
396+createTelegramBotMock.mockReturnValueOnce(bot);
397+const session = createPollingSession({
398+abortSignal: params.abort.signal,
399+log: params.log,
400+isolatedIngress: {
401+enabled: true,
402+spoolDir: params.spoolDir,
403+createWorker: worker.createWorker,
404+drainIntervalMs: params.drainIntervalMs ?? 10,
405+},
406+});
407+return {
408+ bot,
409+createWorker: worker.createWorker,
410+runPromise: session.runUntilAbort(),
411+stopWorker: worker.stop,
412+};
413+}
414+313415describe("TelegramPollingSession", () => {
314416beforeAll(async () => {
315417({ TelegramPollingSession } = await import("./polling-session.js"));
316-({ listTelegramSpooledUpdates, writeTelegramSpooledUpdate } =
418+({ claimTelegramSpooledUpdate, listTelegramSpooledUpdates, writeTelegramSpooledUpdate } =
317419await import("./telegram-ingress-spool.js"));
318420});
319421@@ -538,6 +640,128 @@ describe("TelegramPollingSession", () => {
538640await runPromise;
539641});
540642643+it("keeps failed lanes blocked for the rest of the drain pass", async () => {
644+await withTempSpool(async (tempDir) => {
645+const abort = new AbortController();
646+const log = vi.fn();
647+const events: string[] = [];
648+await writeSpooledTestUpdates(tempDir, [
649+topicUpdate(42, 10, "first topic 10 turn"),
650+topicUpdate(43, 11, "topic 11 turn"),
651+topicUpdate(44, 10, "second topic 10 turn"),
652+]);
653+654+const { runPromise, stopWorker } = startIsolatedIngressSession({
655+ abort,
656+spoolDir: tempDir,
657+ log,
658+drainIntervalMs: 500,
659+handleUpdate: async (update) => {
660+if (update.update_id === 42) {
661+events.push("topic10:first");
662+throw new Error("handler boom");
663+}
664+if (update.update_id === 43) {
665+events.push("topic11");
666+return;
667+}
668+if (update.update_id === 44) {
669+events.push("topic10:second");
670+}
671+},
672+});
673+674+await vi.waitFor(() => expect(events).toEqual(["topic10:first", "topic11"]));
675+expect(await pendingUpdateIds(tempDir, "all")).toEqual([42, 44]);
676+expectLogIncludes(log, "spooled update 42 failed; keeping for retry");
677+abort.abort();
678+stopWorker();
679+await runPromise;
680+});
681+});
682+683+it("recovers restart processing claims before draining later same-lane updates", async () => {
684+await withTempSpool(async (tempDir) => {
685+const abort = new AbortController();
686+const events: string[] = [];
687+await writeSpooledTestUpdates(tempDir, [
688+topicUpdate(42, 10, "interrupted topic 10 turn"),
689+topicUpdate(43, 10, "later topic 10 turn"),
690+topicUpdate(44, 11, "topic 11 turn"),
691+]);
692+const interrupted = (await listTelegramSpooledUpdates({ spoolDir: tempDir })).find(
693+(update) => update.updateId === 42,
694+);
695+if (!interrupted) {
696+throw new Error("Expected interrupted update");
697+}
698+await claimTelegramSpooledUpdate(interrupted);
699+700+const { runPromise, stopWorker } = startIsolatedIngressSession({
701+ abort,
702+spoolDir: tempDir,
703+handleUpdate: async (update) => {
704+events.push(`handled:${update.update_id}`);
705+if (update.update_id === 44) {
706+abort.abort();
707+}
708+},
709+});
710+711+await runPromise;
712+expect(events).toEqual(["handled:42", "handled:44"]);
713+expect(await pendingUpdateIds(tempDir)).toEqual([43]);
714+expect((await fs.readdir(tempDir)).toSorted()).toEqual(["0000000000000043.json"]);
715+stopWorker();
716+});
717+});
718+719+it("scans past active-lane backlogs to start unrelated lanes", async () => {
720+await withTempSpool(async (tempDir) => {
721+const abort = new AbortController();
722+const events: string[] = [];
723+let releaseTopicTenTurn: (() => void) | undefined;
724+const topicTenTurnDone = new Promise<void>((resolve) => {
725+releaseTopicTenTurn = resolve;
726+});
727+await writeSpooledTestUpdates(tempDir, [topicUpdate(0, 10, "active topic 10 turn")]);
728+for (let updateId = 1; updateId <= 100; updateId += 1) {
729+await writeTelegramSpooledUpdate({
730+spoolDir: tempDir,
731+update: topicUpdate(updateId, 10, `blocked topic 10 turn ${updateId}`),
732+});
733+}
734+await writeTelegramSpooledUpdate({
735+spoolDir: tempDir,
736+update: topicUpdate(101, 11, "topic 11 turn"),
737+});
738+739+const { runPromise, stopWorker } = startIsolatedIngressSession({
740+ abort,
741+spoolDir: tempDir,
742+handleUpdate: async (update) => {
743+if (update.update_id === 0) {
744+events.push("topic10:start");
745+await topicTenTurnDone;
746+events.push("topic10:end");
747+return;
748+}
749+if (update.update_id === 101) {
750+events.push("handled:101");
751+abort.abort();
752+}
753+},
754+});
755+756+await vi.waitFor(() => expect(events).toEqual(["topic10:start", "handled:101"]));
757+releaseTopicTenTurn?.();
758+await runPromise;
759+expect(events).toEqual(["topic10:start", "handled:101", "topic10:end"]);
760+releaseTopicTenTurn?.();
761+stopWorker();
762+});
763+});
764+541765it("lets isolated ingress drain interleave different Telegram topic lanes", async () => {
542766const abort = new AbortController();
543767const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
@@ -618,7 +842,7 @@ describe("TelegramPollingSession", () => {
618842await vi.waitFor(() => expect(events).toEqual(["topic10:start", "topic11"]));
619843expect(
620844(await listTelegramSpooledUpdates({ spoolDir: tempDir })).map((update) => update.updateId),
621-).toEqual([42, 44]);
845+).toEqual([44]);
622846623847releaseTopicTenTurn?.();
624848await vi.waitFor(() =>
@@ -837,7 +1061,7 @@ describe("TelegramPollingSession", () => {
8371061await vi.waitFor(() => expect(events).toEqual(["regular:start", "status", "stop"]));
8381062expect(
8391063(await listTelegramSpooledUpdates({ spoolDir: tempDir })).map((update) => update.updateId),
840-).toEqual([42]);
1064+).toEqual([]);
84110658421066releaseRegularTurn?.();
8431067await vi.waitFor(async () =>
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。