feat: telegram group topics support (message_thread_id) #509

lib/dispatch/index.ts

@@ -132,7 +132,11 @@ export async function dispatchTask(
   // Compute session key deterministically (avoids waiting for gateway)
   // Slot name provides both collision prevention and human-readable identity
   const botName = slotName(project.name, role, level, slotIndex);
-  const sessionKey = `agent:${agentId ?? "unknown"}:subagent:${project.name}-${role}-${level}-${botName.toLowerCase()}`;
+  // Use project.slug (always lowercase) to build session key.
+  // project.name may have mixed case (e.g. "UpMoltWork"), which caused heartbeat
+  // mismatches when the gateway stores session keys in lowercase format.
+  const projectKey = (project.slug ?? project.name).toLowerCase();
+  const sessionKey = `agent:${agentId ?? "unknown"}:subagent:${projectKey}-${role}-${level}-${botName.toLowerCase()}`;
 
   // Clear stale session key if it doesn't match the current deterministic key
   // (handles migration from old numeric format like ...-0 to name-based ...-Cordelia)
@@ -271,6 +275,7 @@ export async function dispatchTask(
       channel: notifyTarget?.channel ?? "telegram",
       runtime,
       accountId: notifyTarget?.accountId,
+      messageThreadId: notifyTarget?.messageThreadId,
       runCommand: rc,
     },
   ).catch((err) => {
@@ -294,6 +299,7 @@ export async function dispatchTask(
     dispatchTimeoutMs: timeouts.dispatchMs,
     extraSystemPrompt: roleInstructions.trim() || undefined,
     runCommand: rc,
+    notifyTarget,
   });
 
   // Step 5: Update worker state

lib/dispatch/notify.ts

@@ -85,6 +85,14 @@ export type NotifyEvent =
       issueUrl: string;
       issueTitle: string;
       prUrl?: string;
+    }
+  | {
+      type: "issueComplete";
+      project: string;
+      issueId: number;
+      issueUrl: string;
+      issueTitle: string;
+      prUrl?: string;
     };
 
 /**
@@ -245,6 +253,15 @@ function buildMessage(event: NotifyEvent): string {
       msg += `\n→ Moving to To Improve for developer attention`;
       return msg;
     }
+
+    case "issueComplete": {
+      let msg = `🏁 Issue completed: #${event.issueId} — ${event.issueTitle}`;
+      msg += `\n📦 Project: ${event.project}`;
+      if (event.prUrl) msg += `\n🔗 ${prLink(event.prUrl)}`;
+      msg += `\n📋 [Issue #${event.issueId}](${event.issueUrl})`;
+      msg += `\n✅ Issue closed — work delivered.`;
+      return msg;
+    }
   }
 }
 
@@ -262,13 +279,16 @@ async function sendMessage(
   runtime?: PluginRuntime,
   accountId?: string,
   runCommand?: RunCommand,
+  messageThreadId?: number,
 ): Promise<boolean> {
   try {
     // Use runtime API when available (avoids CLI subprocess timeouts)
     if (runtime) {
       if (channel === "telegram") {
-        // Cast to any to bypass TypeScript type limitation; disableWebPagePreview is valid in Telegram API
-        await runtime.channel.telegram.sendMessageTelegram(target, message, { silent: true, disableWebPagePreview: true, accountId } as any);
+        // Cast to any to bypass TypeScript type limitation; disableWebPagePreview and messageThreadId are valid in Telegram API
+        const telegramOpts: Record<string, unknown> = { silent: true, disableWebPagePreview: true, accountId };
+        if (messageThreadId != null) telegramOpts.messageThreadId = messageThreadId;
+        await runtime.channel.telegram.sendMessageTelegram(target, message, telegramOpts as any);
         return true;
       }
       if (channel === "whatsapp") {
@@ -341,6 +361,8 @@ export async function notify(
     accountId?: string;
     /** Injected runCommand for dependency injection. */
     runCommand?: RunCommand;
+    /** Optional Telegram forum topic ID for per-topic routing */
+    messageThreadId?: number;
   },
 ): Promise<boolean> {
   if (opts.config?.[event.type] === false) return true;
@@ -364,7 +386,7 @@ export async function notify(
     message,
   });
 
-  return sendMessage(target, message, channel, opts.workspaceDir, opts.runtime, opts.accountId, opts.runCommand);
+  return sendMessage(target, message, channel, opts.workspaceDir, opts.runtime, opts.accountId, opts.runCommand, opts.messageThreadId);
 }
 
 /**

lib/dispatch/session.ts

@@ -82,12 +82,56 @@ export function ensureSessionFireAndForget(sessionKey: string, model: string, wo
   });
 }
 
+/** Same shape as `resolveNotifyChannel()` — used to pass Telegram chat/topic into the gateway agent run. */
+export type NotifyRoutingTarget = {
+  channelId: string;
+  channel: string;
+  accountId?: string;
+  messageThreadId?: number;
+};
+
+function applyNotifyRoutingToGatewayParams(
+  params: Record<string, unknown>,
+  target: NotifyRoutingTarget | undefined,
+): void {
+  if (!target?.channelId) {
+    return;
+  }
+  params.to = target.channelId;
+  params.channel = target.channel;
+  if (target.accountId) {
+    params.accountId = target.accountId;
+  }
+  if (target.messageThreadId != null && Number.isFinite(Number(target.messageThreadId))) {
+    params.threadId = String(Math.trunc(Number(target.messageThreadId)));
+  }
+}
+
 export function sendToAgent(
   sessionKey: string, taskMessage: string,
-  opts: { agentId?: string; projectName: string; issueId: number; role: string; level?: string; slotIndex?: number; fromLabel?: string; orchestratorSessionKey?: string; workspaceDir: string; dispatchTimeoutMs?: number; extraSystemPrompt?: string; runCommand: RunCommand },
+  opts: {
+    agentId?: string;
+    projectName: string;
+    issueId: number;
+    role: string;
+    level?: string;
+    slotIndex?: number;
+    fromLabel?: string;
+    orchestratorSessionKey?: string;
+    workspaceDir: string;
+    dispatchTimeoutMs?: number;
+    extraSystemPrompt?: string;
+    runCommand: RunCommand;
+    /**
+     * When set (e.g. from `resolveNotifyChannel`), forwarded to the gateway `agent` call as
+     * `to`, `channel`, `accountId`, and `threadId` so plugin tools get `messageThreadId` injection
+     * (Telegram forum topics) on the worker run.
+     */
+    notifyTarget?: NotifyRoutingTarget;
+  },
 ): void {
   const rc = opts.runCommand;
-  const gatewayParams = JSON.stringify({
+  const gatewayParamsRecord: Record<string, unknown> = {
     idempotencyKey: `devclaw-${opts.projectName}-${opts.issueId}-${opts.role}-${opts.level ?? "unknown"}-${opts.slotIndex ?? 0}-${opts.fromLabel ?? "unknown"}-${sessionKey}`,
     agentId: opts.agentId ?? "devclaw",
     sessionKey,
@@ -96,7 +140,9 @@ export function sendToAgent(
     lane: "subagent",
     ...(opts.orchestratorSessionKey ? { spawnedBy: opts.orchestratorSessionKey } : {}),
     ...(opts.extraSystemPrompt ? { extraSystemPrompt: opts.extraSystemPrompt } : {}),
-  });
+  };
+  applyNotifyRoutingToGatewayParams(gatewayParamsRecord, opts.notifyTarget);
+  const gatewayParams = JSON.stringify(gatewayParamsRecord);
   // Fire-and-forget: long-running agent turn, don't await
   rc(
     ["openclaw", "gateway", "call", "agent", "--params", gatewayParams, "--expect-final", "--json"],

lib/json-result.ts

@@ -0,0 +1,14 @@
+/**
+ * OpenClaw tool success shape — mirrors `jsonResult` from `openclaw/plugin-sdk`.
+ * Implemented locally so DevClaw tools work when the host resolves `openclaw/plugin-sdk`
+ * in a way that breaks named imports (e.g. interop / bundling).
+ */
+export function jsonResult(payload: unknown): {
+  content: Array<{ type: "text"; text: string }>;
+  details: unknown;
+} {
+  return {
+    content: [{ type: "text", text: JSON.stringify(payload, null, 2) }],
+    details: payload,
+  };
+}

lib/projects/io.ts

@@ -102,36 +102,105 @@ export async function writeProjects(
 }
 
 /**
- * Resolve a project by slug or channelId (for backward compatibility).
- * Returns the slug of the found project.
+ * Build a stable scope key for a channel binding.
+ * Used for topic-aware project resolution.
+ */
+export function resolveProjectChannelScope(opts: {
+  channel: string;
+  channelId: string;
+  accountId?: string;
+  messageThreadId?: number | string | null;
+}): string {
+  const account = opts.accountId ?? "default";
+  const topic = opts.messageThreadId ?? "root";
+  return `${opts.channel}:${account}:${opts.channelId}:topic:${topic}`;
+}
+
+/**
+ * Resolve a project by slug or channel scope (for backward compatibility).
+ * When given a bare string, treats it as slug or channelId (legacy behavior).
+ * When given a scope object, performs topic-aware resolution.
  */
 export function resolveProjectSlug(
   data: ProjectsData,
-  slugOrChannelId: string,
+  slugOrChannelIdOrScope: string | {
+    channelId: string;
+    channel?: string;
+    accountId?: string;
+    messageThreadId?: number | string | null;
+  },
 ): string | undefined {
-  // Direct lookup by slug
-  if (data.projects[slugOrChannelId]) {
-    return slugOrChannelId;
+  // String input: legacy mode (slug or channelId)
+  if (typeof slugOrChannelIdOrScope === "string") {
+    const slugOrChannelId = slugOrChannelIdOrScope;
+    // Direct lookup by slug
+    if (data.projects[slugOrChannelId]) {
+      return slugOrChannelId;
+    }
+
+    // Reverse lookup by channelId in channels
+    for (const [slug, project] of Object.entries(data.projects)) {
+      if (project.channels.some((ch) => ch.channelId === slugOrChannelId)) {
+        return slug;
+      }
+    }
+
+    return undefined;
   }
 
-  // Reverse lookup by channelId in channels
+  // Scoped input: topic-aware resolution
+  const { channelId, channel, accountId, messageThreadId } = slugOrChannelIdOrScope;
+  const requestedChannel = channel ?? "telegram";
+  const requestedKey = resolveProjectChannelScope({
+    channel: requestedChannel,
+    channelId,
+    accountId,
+    messageThreadId,
+  });
+
+  let fallbackSlug: string | undefined;
+
   for (const [slug, project] of Object.entries(data.projects)) {
-    if (project.channels.some(ch => ch.channelId === slugOrChannelId)) {
-      return slug;
+    for (const ch of project.channels) {
+      const scopeKey = resolveProjectChannelScope({
+        channel: ch.channel,
+        channelId: ch.channelId,
+        accountId: ch.accountId,
+        messageThreadId: ch.messageThreadId,
+      });
+
+      // Exact topic match wins immediately
+      if (scopeKey === requestedKey) {
+        return slug;
+      }
+
+      // Record chat-level fallback when messageThreadId is undefined/root
+      const isRootScope =
+        ch.channel === requestedChannel &&
+        ch.channelId === channelId &&
+        (ch.messageThreadId == null || ch.messageThreadId === ("root" as any));
+      if (isRootScope && !fallbackSlug) {
+        fallbackSlug = slug;
+      }
     }
   }
 
-  return undefined;
+  return fallbackSlug;
 }
 
 /**
- * Get a project by slug or channelId (dual-mode resolution).
+ * Get a project by slug or channel scope (dual-mode resolution).
  */
 export function getProject(
   data: ProjectsData,
-  slugOrChannelId: string,
+  slugOrChannelIdOrScope: string | {
+    channelId: string;
+    channel?: string;
+    accountId?: string;
+    messageThreadId?: number | string | null;
+  },
 ): Project | undefined {
-  const slug = resolveProjectSlug(data, slugOrChannelId);
+  const slug = resolveProjectSlug(data, slugOrChannelIdOrScope);
   return slug ? data.projects[slug] : undefined;
 }
 

lib/projects/migrations.ts

@@ -12,7 +12,7 @@
  * - projects.json format: flat slots → per-level format
  */
 
-import type { RoleWorkerState, SlotState, Project } from "./types.js";
+import type { RoleWorkerState, SlotState, Project, Channel } from "./types.js";
 
 // ---------------------------------------------------------------------------
 // Role aliases — old role IDs → canonical IDs
@@ -195,6 +195,7 @@ function parseWorkerState(worker: Record<string, unknown>, role: string): RoleWo
  * 3. Old level names in worker state
  * 4. Old slot-based format → per-level format
  * 5. Missing channel field defaults to "telegram"
+ * 6. Telegram: legacy topicId → messageThreadId; topicId removed from stored shape
  */
 /**
  * Returns true if any migration was applied (caller should persist).
@@ -230,6 +231,19 @@ export function migrateProject(project: Project): boolean {
     project.workers = {};
   }
 
+  // Telegram channels: legacy topicId → messageThreadId; drop topicId (canonical field only)
+  if (project.channels) {
+    for (const ch of project.channels) {
+      const rawCh = ch as unknown as Record<string, unknown> & Channel;
+      if (rawCh.channel !== "telegram" || rawCh.topicId == null) continue;
+      if (rawCh.messageThreadId == null) {
+        rawCh.messageThreadId = Number(rawCh.topicId);
+      }
+      delete rawCh.topicId;
+      changed = true;
+    }
+  }
+
   // Migrate legacy `groupId` field to `channelId` in channel objects.
   // Before the rename, channels were stored with { groupId: "..." } on disk.
   if (project.channels) {