fix(telegram): 接入官方 IM 定时正文投递与持久回执

apps/desktop/src/main/hook-control/__tests__/transport-manager.test.ts

@@ -9,11 +9,13 @@ import { createServer, type IncomingMessage, type ServerResponse } from 'node:ht
 import { once } from 'node:events';
 import type { Duplex } from 'node:stream';
 
-import { WebSocketServer, type WebSocket as ServerSocket } from 'ws';
+import { WebSocket, WebSocketServer, type WebSocket as ServerSocket } from 'ws';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 
 import {
   HOOK_FEATURE_GROUP_RELAY,
+  HOOK_FEATURE_MESSAGE_OPS,
+  makeMessageOpResult,
   HOOK_FEATURE_GROUP_RELAY_RECIPIENT,
   HOOK_FEATURE_LIFECYCLE_ANNOUNCEMENT,
   HOOK_FEATURE_MULTI_TEAM,
@@ -4209,3 +4211,59 @@ describe('多 workspace 绑定(multi-team)', () => {
     await expect(promise).resolves.toEqual({ bound: true, prefs: [] });
   });
 });
+
+
+describe('official Telegram scheduled message transport', () => {
+  it('uses the negotiated Telegram connection, matches real op receipts and gates on disconnect', async () => {
+    const { wss, url } = await startServer();
+    const key = 'telegram:dm:bot-1:telegram-user-1:g1';
+    const manager = makeManager(memoryStore({ url, enabled: false, telegramEnabled: true }), {
+      listTelegramDeliveryKeys: () => [key], toolTimeoutMs: 200,
+    });
+    cleanups.push(() => manager.dispose());
+    const connection = once(wss, 'connection') as Promise<[ServerSocket]>;
+    manager.sync();
+    const [socket] = await connection;
+    const server = collectFrames(socket);
+    await server.waitFor('hello');
+    socket.send(serializeHookMessage(makeWelcome({ serverName: 'telegram-server', features: TELEGRAM_FEATURES })));
+    socket.send(serializeHookMessage(makeProviderBindState(TELEGRAM_CONFIRMED)));
+    await expect.poll(() => manager.telegramDeliveryStatus().connected).toBe(true);
+    expect(manager.telegramDeliveryStatus().supported).toBe(false);
+    socket.send(serializeHookMessage(makeWelcome({ serverName: 'telegram-server', features: [...TELEGRAM_FEATURES, HOOK_FEATURE_MESSAGE_OPS] })));
+    socket.send(serializeHookMessage(makeProviderBindState(TELEGRAM_CONFIRMED)));
+    await new Promise(resolve => setTimeout(resolve, 30));
+    expect(manager.telegramDeliveryStatus().supported).toBe(false); // msg-op-v1 is reaction-only on old servers
+    expect(await manager.sendTelegramDelivery({ opId: 'old-server', scope: { externalKey: key }, action: { kind: 'send', text: 'test' } })).toMatchObject({ deliveryState: 'not_sent' });
+    expect(server.frames.filter(f => f.type === 'msg.op')).toHaveLength(0);
+    socket.send(serializeHookMessage(makeWelcome({ serverName: 'telegram-server', features: [...TELEGRAM_FEATURES, HOOK_FEATURE_MESSAGE_OPS, 'telegram-dm-send-v1', 'telegram-send-epoch:test-epoch'] })));
+    socket.send(serializeHookMessage(makeProviderBindState(TELEGRAM_CONFIRMED)));
+    await expect.poll(() => manager.telegramDeliveryStatus().supported).toBe(true);
+    expect(manager.telegramDeliveryStatus().target).toMatchObject({ externalKey: key, botName: 'cindy_example_bot' });
+    const payload = { opId: 'telegram-delivery:test', scope: { externalKey: key }, action: { kind: 'send' as const, text: '<b>test</b>', tier: 'html' as const, delivery: { bindingId: TELEGRAM_CONFIRMED.bindingId!, epoch: 'test-epoch', expiresAt: Date.now() + 60000 } } };
+    // The manager's connected snapshot can outlive the socket's OPEN state.
+    const closedSocket = vi.spyOn(WebSocket.prototype, 'readyState', 'get').mockReturnValue(WebSocket.CLOSING);
+    const refused = manager.sendTelegramDelivery(payload);
+    closedSocket.mockRestore();
+    expect(await refused).toMatchObject({ opId: payload.opId, deliveryState: 'not_sent' });
+    expect(server.frames.filter(f => f.type === 'msg.op')).toHaveLength(0);
+
+    // send() can throw after attempting a write: that is not proof of non-delivery.
+    const failedWrite = vi.spyOn(WebSocket.prototype, 'send').mockImplementation(() => { throw new Error('uncertain write'); });
+    const uncertain = manager.sendTelegramDelivery({ ...payload, opId: 'telegram-delivery:write-error' });
+    failedWrite.mockRestore();
+    expect(await uncertain).toBeNull();
+    // A definite pre-write refusal must release the pending op for a same-key retry.
+    const sent = manager.sendTelegramDelivery(payload);
+    const frame = await server.waitFor('msg.op');
+    expect(frame.type === 'msg.op' ? frame.payload : null).toEqual(payload);
+    socket.send(serializeHookMessage(makeMessageOpResult({ opId: 'unrelated', ok: true, messageId: 'wrong' })));
+    socket.send(serializeHookMessage(makeMessageOpResult({ opId: payload.opId, ok: true, messageId: '42' })));
+    expect(await sent).toEqual({ opId: payload.opId, ok: true, messageId: '42' });
+    const interrupted = manager.sendTelegramDelivery({ ...payload, opId: 'telegram-delivery:interrupted' });
+    await manager.deactivateAccount();
+    expect(await interrupted).toBeNull();
+    expect(manager.telegramDeliveryStatus().connected).toBe(false);
+    expect(await manager.sendTelegramDelivery(payload)).toMatchObject({ deliveryState: 'not_sent' });
+  });
+});

apps/desktop/src/main/hook-control/bindings.ts

@@ -23,6 +23,7 @@ import fs from 'node:fs';
 import path from 'node:path';
 
 export interface HookBindingStore {
+  listKeys?(connectionId: string): string[];
   findBySession?(sessionId: string): Array<{ connectionId: string; externalKey: string }>;
   get(connectionId: string, externalKey: string): string | null;
   /**
@@ -90,6 +91,13 @@ export function createHookBindingStore(deps: {
   }
 
   return {
+    listKeys(connectionId) {
+      const ns: unknown = readAll()[connectionId];
+      if (!ns || typeof ns !== 'object' || Array.isArray(ns)) return [];
+      return Object.entries(ns).filter(([, row]) =>
+        row && typeof row === 'object' && typeof row.sessionId === 'string'
+      ).map(([key]) => key);
+    },
     findBySession(sessionId) {
       const found: Array<{ connectionId: string; externalKey: string }> = [];
       for (const [connectionId, rows] of Object.entries(readAll())) {

apps/desktop/src/main/hook-control/ipc.ts

@@ -94,6 +94,7 @@ import {
 } from './manager.js';
 import { createHookTransport } from './transport.js';
 import { registerSlackToolBridge, unregisterSlackToolBridge } from './slackToolBridge.js';
+import { createTelegramDeliveryBridge, purgeTelegramDeliveryJournals, registerTelegramDeliveryBridge, type TelegramDeliveryBridge } from './telegramDelivery.js';
 import { createHookBindingStore } from './bindings.js';
 import { createHookRequestLedger } from './requestLedger.js';
 import {
@@ -125,6 +126,8 @@ let codexMcpRefreshPending = false;
 let codexMcpRefreshRunning = false;
 let codexMcpRefreshRetryTimer: NodeJS.Timeout | null = null;
 let latestSlackToolProviderEnabled = false;
+/** Owner-scoped journal directory of the active Telegram delivery bridge (sent text lives in `.sent`). */
+let telegramDeliveryDirectory: string | null = null;
 
 const CODEX_MCP_REFRESH_RETRY_MS = 2_000;
 
@@ -435,6 +438,10 @@ function ensureInstances(): { store: SlackHookStore; manager: HookControlManager
     });
   }
   if (!manager) {
+    let telegramDelivery: TelegramDeliveryBridge | null = null;
+    const deliveryBindings = createHookBindingStore({
+      filePath: ownerScopedUserDataPath('hook-bindings.json'), log,
+    });
     const dispatcher = createHookDispatcher({
       // 两个 provider 复用 dispatcher，但连接身份和服务地址彼此隔离。
       getConnection: (connectionId) => {
@@ -566,6 +573,8 @@ function ensureInstances(): { store: SlackHookStore; manager: HookControlManager
     manager = createHookControlManager({
       store,
       isAvailable: hookControlAvailable,
+      onTelegramDeliveryResult: (result) => telegramDelivery?.onResult(result),
+      listTelegramDeliveryKeys: (connectionId) => deliveryBindings.listKeys?.(connectionId) ?? [],
       createTransport: createHookTransport,
       getTelegramUrl: () => getClientEndpoint('telegramHookWsUrl'),
       getXUrl: () => getClientEndpoint('xHookWsUrl'),
@@ -659,6 +668,13 @@ function ensureInstances(): { store: SlackHookStore; manager: HookControlManager
     // Slack 网关工具桥: lizi_slack provider 经叶子注册表取用(不直接 import
     // 本模块, 避免 mcp-providers <-> ipc 的静态引用闭环)
     const m = manager;
+    telegramDeliveryDirectory = ownerScopedUserDataPath('telegram-delivery-receipts');
+    telegramDelivery = createTelegramDeliveryBridge({
+      directory: telegramDeliveryDirectory,
+      status: () => m.telegramDeliveryStatus(),
+      send: (payload) => m.sendTelegramDelivery(payload),
+    });
+    registerTelegramDeliveryBridge(telegramDelivery);
     registerSlackToolBridge({
       availability: () =>
         hookControlAvailable()
@@ -1215,6 +1231,16 @@ export async function stopHookControlAccount(): Promise<void> {
 export function resetHookControlOwnerBoundary(options?: { clearPersisted?: boolean }): void {
   mirrorWorkspacePrefs.invalidateOwnerBoundary();
   unregisterSlackToolBridge();
+  registerTelegramDeliveryBridge(null);
+  // Account deletion removes the owner's delivery journals (`.sent` keeps the actual Telegram
+  // text and entities). The bridge is unregistered first and journal writes are synchronous, so
+  // nothing recreates the directory afterwards; a late in-flight result write fails on the
+  // missing directory instead. Ordinary logout / quit keep the journals for idempotent replay.
+  const deliveryDirectory = telegramDeliveryDirectory;
+  telegramDeliveryDirectory = null;
+  void purgeTelegramDeliveryJournals(deliveryDirectory, options?.clearPersisted).catch((error: unknown) => {
+    log.warn(`telegram delivery journal purge failed: ${String(error)}`);
+  });
   resetGroupContextCursorsSafely(options);
   resetTelegramSpeakerRegistrationCache();
   manager?.dispose();

apps/desktop/src/main/hook-control/manager.ts

@@ -14,12 +14,15 @@
  * 测试用内存 harness 直接驱动(规则 14); Electron 绑定在 ipc.ts 组装。
  */
 
+import { selectTelegramDeliveryTarget, type TelegramDeliveryStatus } from './telegramDelivery.js';
 import { randomUUID } from 'node:crypto';
 
 import {
   HOOK_FEATURE_GROUP_RELAY,
   DEFAULT_TELEGRAM_BEHAVIOR,
   HOOK_FEATURE_MESSAGE_OPS,
+  HOOK_FEATURE_TELEGRAM_DM_SEND,
+  HOOK_TELEGRAM_SEND_EPOCH_PREFIX,
   HOOK_FEATURE_GROUP_RELAY_RECIPIENT,
   HOOK_FEATURE_LIFECYCLE_ANNOUNCEMENT,
   HOOK_FEATURE_MULTI_TEAM,
@@ -53,6 +56,9 @@ import {
   makeQueryResponse,
   makeTaskAck,
   makeToolRequest,
+  makeMessageOp,
+  type MessageOpPayload,
+  type MessageOpResultPayload,
   type BindUpdatePayload,
   type GroupMessagePayload,
   type HelloInput,
@@ -139,6 +145,8 @@ export function telegramGroupMessageOwner(
 }
 
 export interface HookControlManagerDeps {
+  onTelegramDeliveryResult?: (result: MessageOpResultPayload) => void;
+  listTelegramDeliveryKeys?: (connectionId: string) => string[];
   store: SlackHookStore;
   /** Runtime capability gate; false stops the transport without changing user prefs/bindings. */
   isAvailable?: () => boolean;
@@ -218,6 +226,8 @@ export interface HookControlManagerDeps {
 }
 
 export interface HookControlManager {
+  telegramDeliveryStatus(): TelegramDeliveryStatus;
+  sendTelegramDelivery(payload: MessageOpPayload): Promise<MessageOpResultPayload | null>;
   /** 按当前配置 + 登录态同步连接(启动 / 开关 / 配置变更 / 登录态变化后调用)。 */
   sync(): void;
   /**
@@ -651,6 +661,7 @@ export function createHookControlManager(deps: HookControlManagerDeps): HookCont
       // 只给 Telegram 声明: msg.op 目前只有 Telegram 的执行器, X 的渲染路径
       // 不接入(#1855 的红线之一)。
       HOOK_FEATURE_MESSAGE_OPS,
+      HOOK_FEATURE_TELEGRAM_DM_SEND,
       // 进度消息、成功终稿、交互卡由客户端渲染并经 msg.op 发布, 命令菜单以客户端注册表
       // 为准(dispatcher 的 telegramTurnCarrier / telegramCardOps / provider.commands.set)。
       // 每项都要服务端同时宣告才启用; 否则那一项继续由服务端渲染。
@@ -903,6 +914,35 @@ export function createHookControlManager(deps: HookControlManagerDeps): HookCont
     return store.get().lifecycleAnnouncementOverride ?? DEFAULT_SLACK_LIFECYCLE_ANNOUNCEMENT;
   }
 
+  const pendingDeliveries = new Map<string, {
+    resolve: (result: MessageOpResultPayload | null) => void;
+    timer: ReturnType<typeof setTimeout>;
+  }>();
+  function drainDeliveries(): void {
+    for (const pending of pendingDeliveries.values()) {
+      clearTimeout(pending.timer);
+      pending.resolve(null);
+    }
+    pendingDeliveries.clear();
+  }
+  function telegramDeliveryStatus(): TelegramDeliveryStatus {
+    const lane = telegramLane;
+    const connected = accountActive && !disposed && isAvailable() && store.get().telegramEnabled &&
+      lane.status === 'connected' && lane.serverWelcomeReceived && !lane.awaitingStateSnapshot &&
+      laneCapabilityReady(lane);
+    const epochs = lane.serverFeatures.filter(f => f.startsWith(HOOK_TELEGRAM_SEND_EPOCH_PREFIX));
+    const sendEpoch = epochs.length === 1 ? epochs[0]!.slice(HOOK_TELEGRAM_SEND_EPOCH_PREFIX.length) : undefined;
+    const supported = connected && lane.serverFeatures.includes(HOOK_FEATURE_MESSAGE_OPS) &&
+      lane.serverFeatures.includes(HOOK_FEATURE_TELEGRAM_DM_SEND) && !!sendEpoch;
+    const target = connected && lane.binding?.state === 'confirmed'
+      ? selectTelegramDeliveryTarget(lane.binding, deps.listTelegramDeliveryKeys?.(dispatchId('telegram')) ?? [])
+      : null;
+    return { connected, supported, target, sendEpoch,
+      ...(!connected ? { code: 'HOOK_NOT_CONNECTED' } : !supported ? { code: 'SERVER_TOO_OLD' }
+        : !target ? { code: 'OWNER_DM_TARGET_UNAVAILABLE' } : {}),
+    };
+  }
+
   function dispatchId(provider: HookProvider): string {
     const account = getAccountFingerprint?.() ?? 'no-account';
     return `${id}:${account}:${provider}`;
@@ -2179,6 +2219,16 @@ export function createHookControlManager(deps: HookControlManagerDeps): HookCont
       return;
     }
     if (msg.type === 'msg.op.result') {
+      if (expectedProvider === 'telegram') {
+        deps.onTelegramDeliveryResult?.(msg.payload);
+        const pending = pendingDeliveries.get(msg.payload.opId);
+        if (pending) {
+          pendingDeliveries.delete(msg.payload.opId);
+          clearTimeout(pending.timer);
+          pending.resolve(msg.payload);
+          return;
+        }
+      }
       // 表情回执: 纯装饰动作的结果, 失败只记一行 —— 不重试也不影响任务本身。
       dispatcher?.onMessageOpResult(msg.payload);
       return;
@@ -2630,6 +2680,7 @@ export function createHookControlManager(deps: HookControlManagerDeps): HookCont
   }
 
   function stopLane(lane: NeutralProviderLane): void {
+    if (lane.config.provider === 'telegram') drainDeliveries();
     clearLaneBindWatchdog(lane);
     lane.openRequestId = null;
     lane.bindRequest = null;
@@ -2796,6 +2847,7 @@ export function createHookControlManager(deps: HookControlManagerDeps): HookCont
         if (created === null || lane.transport !== created) return;
         lane.status = s;
         lane.lastError = err;
+        if (provider === 'telegram' && s !== 'connected') drainDeliveries();
         if (s !== 'connected' || !laneCapabilityReady(lane)) {
           dispatcher?.onDisconnected(dispatchId(provider));
           drainLanePendingPrefs(lane);
@@ -2948,6 +3000,36 @@ export function createHookControlManager(deps: HookControlManagerDeps): HookCont
       notifyStatus(toView());
       return true;
     },
+    telegramDeliveryStatus,
+    sendTelegramDelivery(payload) {
+      if (pendingDeliveries.has(payload.opId)) return Promise.resolve(null);
+      const status = telegramDeliveryStatus();
+      if (!status.supported || !status.target ||
+          status.target.externalKey !== payload.scope.externalKey || payload.action.kind !== 'send' ||
+          payload.action.delivery?.epoch !== status.sendEpoch ||
+          payload.action.delivery?.bindingId !== status.target.bindingId) {
+        return Promise.resolve({ opId: payload.opId, ok: false, deliveryState: 'not_sent',
+          error: 'Telegram target or capability changed before transport submission' });
+      }
+      return new Promise((resolve) => {
+        const timer = setTimeout(() => {
+          pendingDeliveries.delete(payload.opId);
+          resolve(null);
+        }, toolTimeoutMs ?? DEFAULT_TOOL_TIMEOUT_MS);
+        pendingDeliveries.set(payload.opId, { resolve, timer });
+        try {
+          if (telegramLane.transport?.send(makeMessageOp(payload), { throwOnWriteFailure: true })) return;
+          clearTimeout(timer);
+          pendingDeliveries.delete(payload.opId);
+          resolve({ opId: payload.opId, ok: false, deliveryState: 'not_sent',
+            error: 'Telegram transport closed before submission' });
+        } catch { // A write was attempted: preserve unknown delivery rather than permit a retry.
+          clearTimeout(timer);
+          pendingDeliveries.delete(payload.opId);
+          resolve(null);
+        }
+      });
+    },
     async setSlackCommunications(teamId, enabled) {
       if (!serverFeatures.includes('slack-communications')) {
         return { ok: false, error: { code: 'SERVER_TOO_OLD', message: 'Slack server does not support device communications' } };