fix: preserve Telegram topic routing for exec completions (#64580)

jalehman · web-flow · commit 29142a9d4764 · 2026-04-11T15:47:53.000-07:00
* clawdbot-a2c: pin exec completion delivery context Regeneration-Prompt: | Fix a Telegram forum topic misroute where delayed exec completion or similar async completion text could be delivered into the wrong topic after the session's stored route drifted. Keep the patch surgical. Preserve immutable origin deliveryContext when background exec completion events are queued, thread that context from the exec tool's ambient channel/session defaults into the process session, and ensure the queued system event carries it instead of relying on later heartbeat fallback to mutable session lastTo/lastThreadId data. Add one focused unit assertion that notifyOnExit events keep the original Telegram topic delivery context and one heartbeat regression that proves work started in topic 47 still delivers back to topic 47 even if the session store later points at topic 2175. * fix: note Telegram exec topic routing Regeneration-Prompt: | Prepare PR #64580 after review-pr with no blocking findings. The only required prep change was the workflow-mandated changelog entry under CHANGELOG.md -> Unreleased -> Fixes. Preserve the review conclusion that the code change is already acceptable, do not widen scope beyond the changelog, and include the PR number plus thanks attribution in the changelog line for the Telegram exec forum-topic completion routing fix.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -178,6 +178,7 @@ Docs: https://docs.openclaw.ai
 - Models/vLLM: ignore empty `tool_calls` arrays from reasoning-model OpenAI-compatible replies, reset false `toolUse` stop reasons when no actual tool calls were parsed, and stop sending `tool_choice` unless tools are present so vLLM reasoning responses no longer hang indefinitely. (#61197, #61534) Thanks @balajisiva.
 - Heartbeat/scheduling: spread interval heartbeats across stable per-agent phases derived from gateway identity, so provider traffic is distributed more uniformly across the configured interval instead of clustering around startup-relative times. (#64560) Thanks @odysseus0.
 - Config/media: accept `tools.media.asyncCompletion.directSend` in strict config validation so gateways no longer reject the generated-schema-backed async media completion setting at startup. (#63618) Thanks @qiziAI.
+- Telegram/exec: preserve delayed exec completion routing for forum topics by pinning background exec completions to the topic where the run started even if the session route later drifts. (#64580) thanks @jalehman.
 
 ## 2026.4.9
 
diff --git a/src/agents/bash-process-registry.ts b/src/agents/bash-process-registry.ts
@@ -1,4 +1,5 @@
 import type { ChildProcessWithoutNullStreams } from "node:child_process";
+import type { DeliveryContext } from "../utils/delivery-context.js";
 import { createSessionSlug as createSessionSlugId } from "./session-slug.js";
 
 const DEFAULT_JOB_TTL_MS = 30 * 60 * 1000; // 30 minutes
@@ -30,6 +31,7 @@ export interface ProcessSession {
   command: string;
   scopeKey?: string;
   sessionKey?: string;
+  notifyDeliveryContext?: DeliveryContext;
   notifyOnExit?: boolean;
   notifyOnExitEmptySuccess?: boolean;
   exitNotified?: boolean;
diff --git a/src/agents/bash-tools.exec-runtime.test.ts b/src/agents/bash-tools.exec-runtime.test.ts
@@ -305,11 +305,21 @@ describe("emitExecSystemEvent", () => {
     emitExecSystemEvent("Exec finished", {
       sessionKey: "agent:ops:main",
       contextKey: "exec:run-1",
+      deliveryContext: {
+        channel: "telegram",
+        to: "telegram:-100123:topic:47",
+        threadId: 47,
+      },
     });
 
     expect(enqueueSystemEventMock).toHaveBeenCalledWith("Exec finished", {
       sessionKey: "agent:ops:main",
       contextKey: "exec:run-1",
+      deliveryContext: {
+        channel: "telegram",
+        to: "telegram:-100123:topic:47",
+        threadId: 47,
+      },
     });
     expect(requestHeartbeatNowMock).toHaveBeenCalledWith({
       reason: "exec-event",
diff --git a/src/agents/bash-tools.exec-runtime.ts b/src/agents/bash-tools.exec-runtime.ts
@@ -27,6 +27,7 @@ import { logWarn } from "../logger.js";
 import type { ManagedRun } from "../process/supervisor/index.js";
 import { getProcessSupervisor } from "../process/supervisor/index.js";
 import type { RunExit, TerminationReason } from "../process/supervisor/types.js";
+import { normalizeDeliveryContext, type DeliveryContext } from "../utils/delivery-context.js";
 import {
   addSession,
   appendOutput,
@@ -343,7 +344,11 @@ function maybeNotifyOnExit(session: ProcessSession, status: "completed" | "faile
   const summary = output
     ? `Exec ${status} (${session.id.slice(0, 8)}, ${exitLabel}) :: ${output}`
     : `Exec ${status} (${session.id.slice(0, 8)}, ${exitLabel})`;
-  enqueueSystemEvent(summary, { sessionKey, trusted: false });
+  enqueueSystemEvent(summary, {
+    sessionKey,
+    deliveryContext: session.notifyDeliveryContext,
+    trusted: false,
+  });
   requestHeartbeatNow(scopedHeartbeatWakeOptions(sessionKey, { reason: "exec-event" }));
 }
 
@@ -409,13 +414,17 @@ export function resolveApprovalRunningNoticeMs(value?: number) {
 
 export function emitExecSystemEvent(
   text: string,
-  opts: { sessionKey?: string; contextKey?: string },
+  opts: { sessionKey?: string; contextKey?: string; deliveryContext?: DeliveryContext },
 ) {
   const sessionKey = opts.sessionKey?.trim();
   if (!sessionKey) {
     return;
   }
-  enqueueSystemEvent(text, { sessionKey, contextKey: opts.contextKey });
+  enqueueSystemEvent(text, {
+    sessionKey,
+    contextKey: opts.contextKey,
+    deliveryContext: opts.deliveryContext,
+  });
   requestHeartbeatNow(scopedHeartbeatWakeOptions(sessionKey, { reason: "exec-event" }));
 }
 
@@ -547,6 +556,7 @@ export async function runExecProcess(opts: {
   notifyOnExitEmptySuccess?: boolean;
   scopeKey?: string;
   sessionKey?: string;
+  notifyDeliveryContext?: DeliveryContext;
   timeoutSec: number | null;
   onUpdate?: (partialResult: AgentToolResult<ExecToolDetails>) => void;
 }): Promise<ExecProcessHandle> {
@@ -564,6 +574,7 @@ export async function runExecProcess(opts: {
     command: opts.command,
     scopeKey: opts.scopeKey,
     sessionKey: opts.sessionKey,
+    notifyDeliveryContext: normalizeDeliveryContext(opts.notifyDeliveryContext),
     notifyOnExit: opts.notifyOnExit,
     notifyOnExitEmptySuccess: opts.notifyOnExitEmptySuccess === true,
     exitNotified: false,
diff --git a/src/agents/bash-tools.exec.ts b/src/agents/bash-tools.exec.ts
@@ -16,6 +16,7 @@ import {
   normalizeOptionalLowercaseString,
   normalizeOptionalString,
 } from "../shared/string-coerce.js";
+import { normalizeDeliveryContext } from "../utils/delivery-context.js";
 import { splitShellArgs } from "../utils/shell-argv.js";
 import { markBackgrounded } from "./bash-process-registry.js";
 import { describeExecTool } from "./bash-tools.descriptions.js";
@@ -1322,6 +1323,12 @@ export function createExecTool(
   const notifyOnExit = defaults?.notifyOnExit !== false;
   const notifyOnExitEmptySuccess = defaults?.notifyOnExitEmptySuccess === true;
   const notifySessionKey = normalizeOptionalString(defaults?.sessionKey);
+  const notifyDeliveryContext = normalizeDeliveryContext({
+    channel: defaults?.messageProvider,
+    to: defaults?.currentChannelId,
+    accountId: defaults?.accountId,
+    threadId: defaults?.currentThreadTs,
+  });
   const approvalRunningNoticeMs = resolveApprovalRunningNoticeMs(defaults?.approvalRunningNoticeMs);
   // Derive agentId only when sessionKey is an agent session key.
   const parsedAgentSession = parseAgentSessionKey(defaults?.sessionKey);
@@ -1668,6 +1675,7 @@ export function createExecTool(
         notifyOnExitEmptySuccess,
         scopeKey: defaults?.scopeKey,
         sessionKey: notifySessionKey,
+        notifyDeliveryContext,
         timeoutSec: effectiveTimeout,
         onUpdate,
       });
diff --git a/src/agents/bash-tools.test.ts b/src/agents/bash-tools.test.ts
@@ -206,18 +206,18 @@ const requireRunningSessionId = (result: { details: unknown }) => {
   return requireSessionId(result.details as { sessionId?: string });
 };
 
-function hasNotifyEventForPrefix(prefix: string): boolean {
-  return peekSystemEvents(DEFAULT_NOTIFY_SESSION_KEY).some((event) => event.includes(prefix));
+function hasNotifyEventForPrefix(prefix: string, sessionKey = DEFAULT_NOTIFY_SESSION_KEY): boolean {
+  return peekSystemEvents(sessionKey).some((event) => event.includes(prefix));
 }
 
-async function waitForNotifyEvent(sessionId: string) {
+async function waitForNotifyEvent(sessionId: string, sessionKey = DEFAULT_NOTIFY_SESSION_KEY) {
   const prefix = sessionId.slice(0, 8);
   let finished = getFinishedSession(sessionId);
-  let hasEvent = hasNotifyEventForPrefix(prefix);
+  let hasEvent = hasNotifyEventForPrefix(prefix, sessionKey);
   await expect
     .poll(() => {
       finished = getFinishedSession(sessionId);
-      hasEvent = hasNotifyEventForPrefix(prefix);
+      hasEvent = hasNotifyEventForPrefix(prefix, sessionKey);
       return Boolean(finished && hasEvent);
     }, NOTIFY_POLL_OPTIONS)
     .toBe(true);
@@ -579,6 +579,32 @@ describe("exec notifyOnExit", () => {
     expect(formatted).toContain("System (untrusted):");
   });
 
+  it("preserves the origin delivery context on background exec completion events", async () => {
+    const sessionKey = "agent:main:telegram:group:-1003774691294:topic:47";
+    const tool = createNotifyOnExitExecTool({
+      sessionKey,
+      messageProvider: "telegram",
+      currentChannelId: "telegram:-1003774691294:topic:47",
+      currentThreadTs: "47",
+    });
+
+    const sessionId = await startBackgroundCommand(tool, echoAfterDelay("notify"));
+
+    await waitForNotifyEvent(sessionId, sessionKey);
+    const queuedEvent = peekSystemEventEntries(sessionKey).find((event) =>
+      event.text.includes(sessionId.slice(0, 8)),
+    );
+
+    expect(queuedEvent).toMatchObject({
+      trusted: false,
+      deliveryContext: {
+        channel: "telegram",
+        to: "telegram:-1003774691294:topic:47",
+        threadId: "47",
+      },
+    });
+  });
+
   it("scopes notifyOnExit heartbeat wake to the exec session key", async () => {
     const tool = createNotifyOnExitExecTool();
     const wakeHandler = vi.fn().mockResolvedValue({ status: "skipped", reason: "disabled" });
diff --git a/src/infra/heartbeat-runner.ghost-reminder.test.ts b/src/infra/heartbeat-runner.ghost-reminder.test.ts
@@ -358,4 +358,71 @@ describe("Ghost reminder bug (issue #13317)", () => {
       expect(options?.messageThreadId).toBeUndefined();
     });
   });
+
+  it("keeps exec-event delivery pinned to the original Telegram topic when session route drifts", async () => {
+    await withTempHeartbeatSandbox(async ({ tmpDir, storePath }) => {
+      const cfg: OpenClawConfig = {
+        agents: {
+          defaults: {
+            workspace: tmpDir,
+            heartbeat: {
+              every: "5m",
+              target: "last",
+            },
+          },
+        },
+        channels: { telegram: { allowFrom: ["*"] } },
+        session: { store: storePath },
+      };
+      const sessionKey = "agent:main:telegram:group:-1003774691294:topic:47";
+      await fs.writeFile(
+        storePath,
+        JSON.stringify({
+          [sessionKey]: {
+            sessionId: "sid",
+            updatedAt: Date.now(),
+            lastChannel: "telegram",
+            lastTo: "telegram:-1003774691294:topic:2175",
+            lastThreadId: 2175,
+          },
+        }),
+      );
+
+      const sendTelegram = vi.fn().mockResolvedValue({
+        messageId: "m1",
+        chatId: "-1003774691294",
+      });
+      const getReplySpy = vi.fn().mockResolvedValue({
+        text: "The review-worker spawn finished successfully.",
+      });
+      enqueueSystemEvent("Exec completed (review-run, code 0)", {
+        sessionKey,
+        trusted: false,
+        deliveryContext: {
+          channel: "telegram",
+          to: "telegram:-1003774691294:topic:47",
+          threadId: 47,
+        },
+      });
+
+      const result = await runHeartbeatOnce({
+        cfg,
+        agentId: "main",
+        sessionKey,
+        reason: "exec-event",
+        deps: {
+          getReplyFromConfig: getReplySpy,
+          telegram: sendTelegram,
+        },
+      });
+
+      expect(result.status).toBe("ran");
+      expect(sendTelegram).toHaveBeenCalledTimes(1);
+      expect(sendTelegram).toHaveBeenCalledWith(
+        "telegram:-1003774691294:topic:47",
+        "The review-worker spawn finished successfully.",
+        expect.objectContaining({ messageThreadId: 47 }),
+      );
+    });
+  });
 });
