fix(e2e): require causal telegram rtt canary

vincentkoc · vincentkoc · commit cd9c643dc6ef · 2026-06-07T08:25:11.000+02:00
diff --git a/scripts/e2e/npm-telegram-rtt-config.mjs b/scripts/e2e/npm-telegram-rtt-config.mjs
@@ -91,6 +91,7 @@ config.channels.telegram = {
     id: "TELEGRAM_BOT_TOKEN",
   },
   streaming: { mode: "off" },
+  replyToMode: "first",
   dmPolicy: "allowlist",
   allowFrom: [driverId],
   defaultTo: driverId,
diff --git a/scripts/e2e/npm-telegram-rtt-driver.mjs b/scripts/e2e/npm-telegram-rtt-driver.mjs
@@ -210,8 +210,8 @@ async function waitForSutReply(params) {
         continue;
       }
       const replyMatches = message.reply_to_message?.message_id === params.requestMessageId;
-      const anySutReplyMatches = params.allowAnySutReply;
-      if (replyMatches || anySutReplyMatches || params.matchText) {
+      const textMatches = params.matchText ? text.includes(params.matchText) : false;
+      if (replyMatches || textMatches) {
         return message;
       }
     }
@@ -371,7 +371,7 @@ async function main() {
 
   const scenarios = [];
   const canary = await runScenario({
-    allowAnySutReply: true,
+    allowAnySutReply: false,
     id: "telegram-canary",
     input: `/status@${sutMe.username}`,
     sutId: sutMe.id,
diff --git a/test/scripts/npm-telegram-rtt-driver.test.ts b/test/scripts/npm-telegram-rtt-driver.test.ts
@@ -1,6 +1,7 @@
 // Npm Telegram Rtt Driver tests cover npm telegram rtt driver script behavior.
 import { spawn, spawnSync, type ChildProcessWithoutNullStreams } from "node:child_process";
 import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
+import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import { setTimeout as delay } from "node:timers/promises";
@@ -103,6 +104,208 @@ type DriverCaseResult = {
   elapsedMs?: number;
 };
 
+type AsyncDriverResult = {
+  status: number | null;
+  signal: NodeJS.Signals | null;
+  stdout: string;
+  stderr: string;
+  timedOut: boolean;
+};
+
+type TelegramUpdate = {
+  update_id: number;
+  message: {
+    message_id: number;
+    date: number;
+    chat: { id: number };
+    from: { id: number; username: string };
+    reply_to_message?: { message_id: number };
+    text: string;
+  };
+};
+
+function runDriverAsync(env: Record<string, string>, timeoutMs = 5000): Promise<AsyncDriverResult> {
+  return new Promise((resolve) => {
+    const child = spawn(process.execPath, [DRIVER_SCRIPT], {
+      env: {
+        ...process.env,
+        OPENCLAW_QA_TELEGRAM_DRIVER_BOT_TOKEN: "driver-token",
+        OPENCLAW_QA_TELEGRAM_GROUP_ID: "-100123",
+        OPENCLAW_QA_TELEGRAM_SUT_BOT_TOKEN: "sut-token",
+        ...env,
+      },
+      stdio: "pipe",
+    });
+    let stdout = "";
+    let stderr = "";
+    let settled = false;
+    const timeout = setTimeout(() => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      child.kill("SIGKILL");
+      resolve({ status: null, signal: "SIGKILL", stdout, stderr, timedOut: true });
+    }, timeoutMs);
+    timeout.unref?.();
+    child.stdout.on("data", (chunk) => {
+      stdout += String(chunk);
+    });
+    child.stderr.on("data", (chunk) => {
+      stderr += String(chunk);
+    });
+    child.on("close", (status, signal) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      clearTimeout(timeout);
+      resolve({ status, signal, stdout, stderr, timedOut: false });
+    });
+  });
+}
+
+function readRequestJson(req: IncomingMessage): Promise<Record<string, unknown>> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    req.on("data", (chunk) => {
+      chunks.push(Buffer.from(chunk));
+    });
+    req.on("error", reject);
+    req.on("end", () => {
+      const raw = Buffer.concat(chunks).toString("utf8").trim();
+      if (!raw) {
+        resolve({});
+        return;
+      }
+      try {
+        resolve(JSON.parse(raw) as Record<string, unknown>);
+      } catch (error) {
+        reject(error);
+      }
+    });
+  });
+}
+
+function writeTelegramJson(res: ServerResponse, payload: unknown): void {
+  const body = `${JSON.stringify(payload)}\n`;
+  res.writeHead(200, {
+    "content-length": Buffer.byteLength(body),
+    "content-type": "application/json",
+  });
+  res.end(body);
+}
+
+function closeServer(server: Server): Promise<void> {
+  return new Promise((resolve, reject) => {
+    server.close((error) => {
+      if (error) {
+        reject(error);
+        return;
+      }
+      resolve();
+    });
+  });
+}
+
+async function startTelegramApiServer(options: {
+  canaryReplyToRequest: boolean;
+}): Promise<{ baseUrl: string; close: () => Promise<void> }> {
+  const pendingUpdates: TelegramUpdate[] = [];
+  let nextMessageId = 100;
+  let nextUpdateId = 1;
+  const chatId = -100123;
+  const nowUnixSeconds = () => Math.floor(Date.now() / 1000);
+  const pushSutUpdate = (params: { replyToMessageId?: number; text: string }) => {
+    pendingUpdates.push({
+      update_id: nextUpdateId++,
+      message: {
+        message_id: nextMessageId++,
+        date: nowUnixSeconds(),
+        chat: { id: chatId },
+        from: { id: 222, username: "sut_bot" },
+        ...(params.replyToMessageId === undefined
+          ? {}
+          : { reply_to_message: { message_id: params.replyToMessageId } }),
+        text: params.text,
+      },
+    });
+  };
+
+  const server = createServer((req, res) => {
+    void (async () => {
+      const match = req.url?.match(/^\/bot([^/]+)\/([^/?]+)/u);
+      if (!match) {
+        res.writeHead(404).end();
+        return;
+      }
+      const [, token, method] = match;
+      const body = await readRequestJson(req);
+
+      if (method === "getMe") {
+        writeTelegramJson(res, {
+          ok: true,
+          result:
+            token === "sut-token"
+              ? { id: 222, username: "sut_bot" }
+              : { id: 111, username: "driver_bot" },
+        });
+        return;
+      }
+
+      if (method === "sendMessage") {
+        const messageId = nextMessageId++;
+        const text = String(body.text ?? "");
+        if (token === "driver-token" && text.startsWith("/status@")) {
+          pushSutUpdate({
+            replyToMessageId: options.canaryReplyToRequest ? messageId : undefined,
+            text: "status ok",
+          });
+        } else if (token === "driver-token" && text.includes("Reply with exactly ")) {
+          const marker = text.match(/Reply with exactly ([^.]+)\./u)?.[1] ?? "OPENCLAW_E2E_OK_1";
+          pushSutUpdate({ replyToMessageId: messageId, text: marker });
+        }
+        writeTelegramJson(res, {
+          ok: true,
+          result: {
+            message_id: messageId,
+            date: nowUnixSeconds(),
+            chat: { id: chatId },
+            text,
+          },
+        });
+        return;
+      }
+
+      if (method === "getUpdates") {
+        const offset = typeof body.offset === "number" ? body.offset : 0;
+        const updates = pendingUpdates.filter((update) => update.update_id >= offset);
+        if (updates.length === 0) {
+          await delay(25);
+        }
+        writeTelegramJson(res, { ok: true, result: updates });
+        return;
+      }
+
+      writeTelegramJson(res, { ok: false, description: `unexpected method ${method}` });
+    })().catch((error) => {
+      res.writeHead(500, { "content-type": "text/plain" }).end(String(error));
+    });
+  });
+
+  await new Promise<void>((resolve) => {
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("fake Telegram server did not bind a TCP port");
+  }
+  return {
+    baseUrl: `http://127.0.0.1:${address.port}`,
+    close: () => closeServer(server),
+  };
+}
+
 async function runStalledTelegramBodyCase(): Promise<DriverCaseResult> {
   const root = mkdtempSync(path.join(tmpdir(), "openclaw-telegram-rtt-driver-"));
   const portPath = path.join(root, "port.txt");
@@ -239,4 +442,77 @@ describe("npm Telegram RTT driver", () => {
     expect(result.stderr).toContain("Telegram Bot API getMe response body exceeded 16 bytes");
     expect(result.stderr).not.toContain("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx");
   });
+
+  it("rejects unrelated SUT messages during the Telegram canary", async () => {
+    const root = mkdtempSync(path.join(tmpdir(), "openclaw-telegram-rtt-causal-"));
+    const server = await startTelegramApiServer({ canaryReplyToRequest: false });
+
+    try {
+      const outputDir = path.join(root, "out");
+      const result = await runDriverAsync({
+        OPENCLAW_NPM_TELEGRAM_OUTPUT_DIR: outputDir,
+        OPENCLAW_NPM_TELEGRAM_WARM_SAMPLES: "1",
+        OPENCLAW_QA_TELEGRAM_API_BASE_URL: server.baseUrl,
+        OPENCLAW_QA_TELEGRAM_CANARY_TIMEOUT_MS: "250",
+        OPENCLAW_QA_TELEGRAM_SCENARIO_TIMEOUT_MS: "250",
+      });
+      const summary = JSON.parse(
+        readFileSync(path.join(outputDir, "telegram-qa-summary.json"), "utf8"),
+      ) as {
+        scenarios: Array<{ id: string; status: string; details?: string }>;
+        status: string;
+      };
+      const canary = summary.scenarios.find((scenario) => scenario.id === "telegram-canary");
+
+      expect(result.timedOut).toBe(false);
+      expect(result.status).not.toBe(0);
+      expect(summary.status).toBe("fail");
+      expect(canary).toMatchObject({
+        id: "telegram-canary",
+        status: "fail",
+      });
+      expect(canary?.details).toContain("timed out");
+    } finally {
+      await server.close();
+      rmSync(root, { force: true, recursive: true });
+    }
+  });
+
+  it("accepts reply-threaded SUT messages during the Telegram canary", async () => {
+    const root = mkdtempSync(path.join(tmpdir(), "openclaw-telegram-rtt-causal-"));
+    const server = await startTelegramApiServer({ canaryReplyToRequest: true });
+
+    try {
+      const outputDir = path.join(root, "out");
+      const result = await runDriverAsync({
+        OPENCLAW_NPM_TELEGRAM_OUTPUT_DIR: outputDir,
+        OPENCLAW_NPM_TELEGRAM_WARM_SAMPLES: "1",
+        OPENCLAW_QA_TELEGRAM_API_BASE_URL: server.baseUrl,
+        OPENCLAW_QA_TELEGRAM_CANARY_TIMEOUT_MS: "1000",
+        OPENCLAW_QA_TELEGRAM_SCENARIO_TIMEOUT_MS: "1000",
+      });
+      const summary = JSON.parse(
+        readFileSync(path.join(outputDir, "telegram-qa-summary.json"), "utf8"),
+      ) as {
+        scenarios: Array<{ id: string; status: string }>;
+        status: string;
+      };
+
+      expect(result).toMatchObject({
+        signal: null,
+        status: 0,
+        timedOut: false,
+      });
+      expect(summary.status).toBe("pass");
+      expect(summary.scenarios).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({ id: "telegram-canary", status: "pass" }),
+          expect.objectContaining({ id: "telegram-mentioned-message-reply", status: "pass" }),
+        ]),
+      );
+    } finally {
+      await server.close();
+      rmSync(root, { force: true, recursive: true });
+    }
+  });
 });
diff --git a/test/scripts/rtt-harness.test.ts b/test/scripts/rtt-harness.test.ts
@@ -466,6 +466,7 @@ describe("RTT harness", () => {
     ]);
 
     const config = JSON.parse(await fs.readFile(configPath, "utf8"));
+    expect(config.channels.telegram.replyToMode).toBe("first");
     expect(config.channels.telegram.streaming).toEqual({ mode: "off" });
     expect(config.messages.groupChat.visibleReplies).toBe("automatic");
   });

Original file line number	Diff line number	Diff line change
`@@ -210,8 +210,8 @@ async function waitForSutReply(params) {`
`210`	`210`	`continue;`
`211`	`211`	`}`
`212`	`212`	`const replyMatches = message.reply_to_message?.message_id === params.requestMessageId;`
`213`		`- const anySutReplyMatches = params.allowAnySutReply;`
`214`		`- if (replyMatches \|\| anySutReplyMatches \|\| params.matchText) {`
	`213`	`+ const textMatches = params.matchText ? text.includes(params.matchText) : false;`
	`214`	`+ if (replyMatches \|\| textMatches) {`
`215`	`215`	`return message;`
`216`	`216`	`}`
`217`	`217`	`}`
`@@ -371,7 +371,7 @@ async function main() {`
`371`	`371`
`372`	`372`	`const scenarios = [];`
`373`	`373`	`const canary = await runScenario({`
`374`		`- allowAnySutReply: true,`
	`374`	`+ allowAnySutReply: false,`
`375`	`375`	`id: "telegram-canary",`
`376`	`376`	input: `/status@${sutMe.username}`,
`377`	`377`	`sutId: sutMe.id,`