fix(slack): drop synthetic thread_ts to avoid invalid_thread_ts

packages/core/kernel/request-run.ts

@@ -6,10 +6,12 @@ import {
   createActiveRequest,
   failActiveRequest,
   getPendingQuestion,
+  loadSession,
   saveSession,
   setPendingQuestion,
   updateActiveRequest,
   type ActiveRequest,
+  type PendingQuestion,
   type PersistedSession,
   type TrackedTodo,
   type TrackedTool,
@@ -27,6 +29,7 @@ import { maybeGenerateSessionTitle } from "@/core/runtime/session-title";
 import type { AgentAdapter, IMAdapter, StatusStreamChunk } from "@/core/types";
 import type { RuntimeRequestContext } from "@/core/kernel/request-context";
 import { formatSingleQuestionPrompt } from "@/core/runtime/helpers";
+import { isSyntheticOwner } from "@/ims/shared/synthetic-owner";
 import {
   buildSessionMessageState,
   createStatusStreamDiffer,
@@ -56,6 +59,34 @@ function isStatusStreamingEnabled(
   return getSlackStatusModeForChannel(channelId) !== "legacy";
 }
 
+function mirrorPendingQuestionToRealThread(params: {
+  channelId: string;
+  syntheticThreadId: string;
+  realThreadId: string | undefined;
+  pendingQuestion: PendingQuestion;
+}): void {
+  const { channelId, syntheticThreadId, realThreadId, pendingQuestion } = params;
+  if (!realThreadId || realThreadId === syntheticThreadId || !isSyntheticOwner(syntheticThreadId)) {
+    return;
+  }
+
+  const syntheticSession = loadSession(channelId, syntheticThreadId);
+  if (!syntheticSession) return;
+
+  const existingRealSession = loadSession(channelId, realThreadId);
+  if (existingRealSession) {
+    existingRealSession.pendingQuestion = pendingQuestion;
+    saveSession(existingRealSession);
+    return;
+  }
+
+  saveSession({
+    ...syntheticSession,
+    threadId: realThreadId,
+    pendingQuestion,
+  });
+}
+
 /**
  * Guard against publishing the user's own prompt as the bot's final reply.
  *
@@ -568,11 +599,12 @@ async function startKernelEventStreamWatcher(params: {
       }
 
       void (async () => {
+        let questionMessageTs: string | undefined;
         try {
           const first = normalized[0]!;
           const prefix = normalized.length > 1 ? `(1/${normalized.length}) ` : "";
           if (typeof deps.im.sendQuestion === "function") {
-            await deps.im.sendQuestion(
+            questionMessageTs = await deps.im.sendQuestion(
               request.channelId,
               request.replyThreadId,
               first.question,
@@ -581,7 +613,7 @@ async function startKernelEventStreamWatcher(params: {
             );
           } else {
             const promptText = formatSingleQuestionPrompt(first, 0, normalized.length);
-            await deps.im.sendMessage(request.channelId, request.replyThreadId, promptText);
+            questionMessageTs = await deps.im.sendMessage(request.channelId, request.replyThreadId, promptText);
           }
         } catch (err) {
           log.warn("Failed to post ask_user question", {
@@ -592,14 +624,21 @@ async function startKernelEventStreamWatcher(params: {
           });
         }
 
-        setPendingQuestion(request.channelId, request.threadId, {
+        const pendingQuestion: PendingQuestion = {
           requestId,
           sessionId: properties.sessionID ?? request.sessionId,
           askedAt: Date.now(),
           questions: normalized,
-          messageTs: request.statusMessageTs,
+          messageTs: questionMessageTs ?? request.statusMessageTs,
           collectedAnswers: [],
           questionDetailId,
+        };
+        setPendingQuestion(request.channelId, request.threadId, pendingQuestion);
+        mirrorPendingQuestionToRealThread({
+          channelId: request.channelId,
+          syntheticThreadId: request.threadId,
+          realThreadId: questionMessageTs,
+          pendingQuestion,
         });
       })();
       return;

packages/ims/slack/api.test.ts

@@ -1,16 +1,39 @@
-import { beforeEach, describe, expect, it, mock } from "bun:test";
+import { afterEach, beforeEach, describe, expect, it, mock } from "bun:test";
 
+// Both test groups below import `./api`, which imports `./client`. We mock
+// the single `./client` module so we don't need a real Slack connection,
+// and expose both the raw `apiCall` surface (used by the streaming helpers)
+// and the typed `chat.postMessage` surface (used by `postSlackQuestion`).
+//
+// IMPORTANT: `mock.module(...)` is process-wide in Bun, so the stub stays in
+// place for every later test file that loads `./client`. To avoid breaking
+// downstream tests that pull other exports (e.g. `sendChannelMessage` used by
+// `packages/core/tasks/scheduler.ts` and `packages/core/test/web-routes.test.ts`),
+// we start from the real module's exports and only override `getApp` /
+// `getSlackBotToken`. This way later imports keep finding every real export.
 const apiCalls: Array<{ method: string; args: Record<string, unknown> }> = [];
+const postMessageCalls: Array<Record<string, unknown>> = [];
+
+const realClient = await import("./client");
 
 mock.module("./client", () => ({
+  ...realClient,
   getApp: () => ({
     client: {
       apiCall: async (method: string, args: Record<string, unknown>) => {
         apiCalls.push({ method, args });
         return method === "chat.startStream" ? { ts: "111.222" } : {};
       },
+      chat: {
+        postMessage: async (args: Record<string, unknown>) => {
+          postMessageCalls.push(args);
+          return { ok: true, ts: "1700000000.000100" };
+        },
+      },
     },
   }),
+  // Override so the test never accidentally picks up a stale workspace token
+  // during the test run.
   getSlackBotToken: () => "xoxb-test",
 }));
 
@@ -67,3 +90,60 @@ describe("Slack streaming API helpers", () => {
     });
   });
 });
+
+describe("postSlackQuestion thread_ts handling", () => {
+  afterEach(() => {
+    postMessageCalls.length = 0;
+  });
+
+  it("omits thread_ts when the thread id is a synthetic cron-job placeholder", async () => {
+    const { postSlackQuestion } = await import("./api");
+    await postSlackQuestion({
+      channelId: "C0ATGCJ0YK0",
+      threadId: "cron-job:a86fbdc5-01df-4caf-9e0c-c0c199f00379:1780441200000",
+      question: "Ready to deploy?",
+      options: ["Yes", "No"],
+      token: "xoxb-test",
+    });
+
+    expect(postMessageCalls.length).toBeGreaterThan(0);
+    for (const call of postMessageCalls) {
+      // Slack rejects synthetic placeholders with `invalid_thread_ts`; the
+      // adapter must drop `thread_ts` and fall back to a top-level post.
+      expect(call).not.toHaveProperty("thread_ts");
+      expect(call.channel).toBe("C0ATGCJ0YK0");
+    }
+  });
+
+  it("omits thread_ts for synthetic task placeholders too", async () => {
+    const { postSlackQuestion } = await import("./api");
+    await postSlackQuestion({
+      channelId: "C0ATGCJ0YK0",
+      threadId: "task:abc-123",
+      question: "Continue?",
+      token: "xoxb-test",
+    });
+
+    expect(postMessageCalls.length).toBeGreaterThan(0);
+    for (const call of postMessageCalls) {
+      expect(call).not.toHaveProperty("thread_ts");
+    }
+  });
+
+  it("passes thread_ts through for real Slack timestamps", async () => {
+    const { postSlackQuestion } = await import("./api");
+    const realTs = "1717000000.000200";
+    await postSlackQuestion({
+      channelId: "C0ATGCJ0YK0",
+      threadId: realTs,
+      question: "Ready to deploy?",
+      options: ["Yes", "No"],
+      token: "xoxb-test",
+    });
+
+    expect(postMessageCalls.length).toBeGreaterThan(0);
+    for (const call of postMessageCalls) {
+      expect(call.thread_ts).toBe(realTs);
+    }
+  });
+});

packages/ims/slack/api.ts

@@ -2,6 +2,7 @@ import { basename } from "path";
 import { getApp, getSlackBotToken } from "./client";
 import { hasSimpleOptions } from "@/core/runtime/helpers";
 import type { StatusStreamChunk } from "@/core/types";
+import { isSyntheticOwner } from "@/ims/shared/synthetic-owner";
 
 // ---------------------------------------------------------------------------
 // Slack IM helper module.
@@ -90,11 +91,18 @@ export async function postSlackQuestion(args: {
   const displayPrefix = prefix ?? "";
   const questionText = `${displayPrefix}${question}`;
 
+  // Synthetic placeholder thread ids (`task:` / `cron-job:` / `cron:`) are
+  // not valid Slack `thread_ts` values; if the runtime hands us one during
+  // a task/cron run, fall back to posting at the top of the channel rather
+  // than letting Slack reject the call with `invalid_thread_ts`.
+  const threadIsSynthetic = isSyntheticOwner(threadId);
+  const threadField = threadIsSynthetic ? {} : { thread_ts: threadId };
+
   const postPlainText = async (): Promise<string | undefined> => {
     const optionText = options.length > 0 ? `\nOptions: ${options.join(" / ")}` : "";
     const result = await client.chat.postMessage({
       channel: channelId,
-      thread_ts: threadId,
+      ...threadField,
       text: `${questionText}${optionText}`,
       token,
     });
@@ -112,7 +120,7 @@ export async function postSlackQuestion(args: {
     try {
       const result = await client.chat.postMessage({
         channel: channelId,
-        thread_ts: threadId,
+        ...threadField,
         text: questionText,
         blocks: [
           {

packages/ims/slack/client.ts

@@ -309,7 +309,29 @@ export async function sendMessage(
   const formattedText = markdownToSlack(text);
   const chunks = splitForSlack(formattedText);
   const workspace = slackAuthRegistry.getChannelWorkspaceName(rawChannelId) || "unknown";
-  const botToken = getSlackBotTokenForProcessor(processorId) ?? getSlackBotToken(channelId, threadId);
+
+  // A "synthetic" thread id (`task:{id}` / `cron-job:{id}:{run}` / `cron:{id}`)
+  // is an internal placeholder used by Ode's task/cron schedulers before a
+  // real Slack `thread_ts` exists. Slack rejects these with `invalid_thread_ts`
+  // because they are not valid message timestamps. When the agent runtime
+  // emits intermediate output during a task/cron run, fall back to a
+  // top-level channel post (no `thread_ts`) so the message still lands in
+  // the channel instead of being lost + captured as a Sentry error.
+  const threadIsSynthetic = isSyntheticOwner(threadId);
+
+  // Token resolution. For real threads, prefer the registry-bound token for
+  // this (channel, thread) — that's the token the inbound router observed
+  // delivering the parent message and is the safest choice for replies. For
+  // synthetic placeholders the call has degenerated to a top-level channel
+  // post (see below); the registry has no entry for a fake `thread_ts`, so
+  // resolve via the channel's workspace first (mirrors `sendChannelMessage`)
+  // before falling back to `getSlackBotToken` to avoid the multi-workspace
+  // edge case where `getSlackBotToken` may return the first registered token
+  // instead of the one for `rawChannelId`.
+  const botToken = getSlackBotTokenForProcessor(processorId)
+    ?? (threadIsSynthetic
+      ? (getWorkspaceBotTokenForChannel(channelId) ?? getSlackBotToken(channelId))
+      : getSlackBotToken(channelId, threadId));
 
   if (!botToken) {
     log.warn("No Slack bot token available for channel", { channelId });
@@ -320,6 +342,7 @@ export async function sendMessage(
       workspace,
       channel: channelId,
       thread: threadId,
+      threadIsSynthetic,
       botTokenLast6: tokenLast6(botToken),
       text,
       chunks: chunks.length,
@@ -329,6 +352,7 @@ export async function sendMessage(
       workspace,
       channel: channelId,
       thread: threadId,
+      threadIsSynthetic,
       botTokenLast6: tokenLast6(botToken),
       text,
       chunks: chunks.length,
@@ -346,7 +370,7 @@ export async function sendMessage(
     try {
       const result = await slackApp.client.chat.postMessage({
         channel: rawChannelId,
-        thread_ts: threadId,
+        ...(threadIsSynthetic ? {} : { thread_ts: threadId }),
         text: chunk,
         token: botToken,
       });
@@ -358,7 +382,15 @@ export async function sendMessage(
       });
       lastTs = result.ts;
       if (botToken && result.ts) {
-        slackAuthRegistry.setThreadBotToken(rawChannelId, threadId, botToken);
+        // Don't bind a real bot token to a synthetic placeholder thread id —
+        // the real platform-assigned thread is whatever `result.ts` is for
+        // the first message of the run. `setMessageBotToken` covers the
+        // outgoing message itself; the cron/task scheduler seeds the
+        // session for the real thread via `seedCronChannelThreadSession` /
+        // `seedChannelThreadSession`.
+        if (!threadIsSynthetic) {
+          slackAuthRegistry.setThreadBotToken(rawChannelId, threadId, botToken);
+        }
         slackAuthRegistry.setMessageBotToken(rawChannelId, result.ts, botToken);
       }
     } catch (err) {
@@ -627,22 +659,55 @@ function createSlackAdapter(processorId?: string): IMAdapter {
       options: string[] | undefined,
       prefix?: string
     ) => {
-      const token = getSlackBotTokenForProcessor(processorId) ?? getSlackBotToken(channelId, threadId);
+      // Mirror the token-resolution logic in `sendMessage`: when the thread
+      // id is a synthetic placeholder (`task:` / `cron-job:` / `cron:`), the
+      // posted question degenerates to a top-level channel post inside
+      // `postSlackQuestion`, so the registry has no entry for that fake
+      // `thread_ts`. Resolve via the channel's workspace first to avoid the
+      // multi-workspace edge case where `getSlackBotToken(channelId, fakeTs)`
+      // can fall back to the first registered token instead of the one bound
+      // to `channelId`.
+      const threadIsSynthetic = isSyntheticOwner(threadId);
+      const token = getSlackBotTokenForProcessor(processorId)
+        ?? (threadIsSynthetic
+          ? (getWorkspaceBotTokenForChannel(channelId) ?? getSlackBotToken(channelId))
+          : getSlackBotToken(channelId, threadId));
       if (!token) {
         // No token -> fall through to plain-text sendMessage so the question
         // still gets delivered through whatever channel/path the caller has.
         const optionText = options && options.length > 0 ? `\nOptions: ${options.join(" / ")}` : "";
         return sendMessage(channelId, threadId, `${prefix ?? ""}${question}${optionText}`, processorId);
       }
       const { postSlackQuestion } = await import("./api");
-      return postSlackQuestion({
+      const questionMessageTs = await postSlackQuestion({
         channelId,
         threadId,
         question,
         options,
         prefix,
         token,
       });
+      // The synthetic branch above turns the question into a top-level channel
+      // post. The handler that processes the user's button click / threaded
+      // reply later looks up the bot token via BOTH
+      //   getMessageBotToken(channel, selectionMessageTs)   → this is the ts
+      //     of the user's selection reply (not the question), created inside
+      //     `commands.ts` via `chat.postMessage`, so no registry entry exists.
+      //   getThreadBotToken(channel, threadId)              → for a top-level
+      //     question, Slack derives `threadId` from
+      //     `body.message?.thread_ts || body.message?.ts`, which resolves to
+      //     the question message's own ts.
+      // Register the resolved `token` under both the message-token map (for
+      // parity with sendMessage's outgoing binding) AND the thread-token map
+      // keyed by the question ts, so the button-selection path can recover
+      // the correct workspace token in multi-workspace installs where the
+      // channel has not yet been bound by the message router (Codex P2 on
+      // PR #212).
+      if (questionMessageTs) {
+        slackAuthRegistry.setMessageBotToken(channelId, questionMessageTs, token);
+        slackAuthRegistry.setThreadBotToken(channelId, questionMessageTs, token);
+      }
+      return questionMessageTs;
     },
     updateMessage: (channelId: string, messageTs: string, text: string) =>
       updateMessage(channelId, messageTs, text, processorId),