fix(chat): surface ask cards for bot-delegated tasks

apps/api/src/app.ts

@@ -183,7 +183,11 @@ export async function createApp(
   const pipedream =
     pipedreamOverride ??
     (isPipedreamEnabled(pipedreamConfig) ? new PipedreamConnector(pipedreamConfig) : undefined);
-  const messagingPlatforms = messagingPlatformsFromEnv(env);
+  // This process registers the inbound sink (messaging.onInbound below),
+  // so it's the one that must hold Telegram's live getUpdates connection —
+  // see messagingPlatformsFromEnv's docstring for why a second poller
+  // elsewhere (e.g. the worker) would actively break this.
+  const messagingPlatforms = messagingPlatformsFromEnv(env, { pollInboundMessages: true });
   const messaging =
     messagingOverride ??
     (isMessagingSurfaceEnabled(messagingPlatforms, {
@@ -396,6 +400,10 @@ export async function createApp(
     );
   });
   mountWebhookHttpRoutes(app, { prisma, secrets, events, jobs });
+  // Shared with stop so a shutdown during retry delays does not restart polling.
+  let messagingStopped = false;
+  let clearMessagingRetryDelay: (() => void) | undefined;
+  let messagingInitTask: Promise<void> | undefined;
   // Messaging webhooks only exist when the surface is enabled.
   if (messaging) {
     const inbound = createMessagingInboundHandler({
@@ -429,6 +437,44 @@ export async function createApp(
       else await applyMessagingOutboundStatus(prisma, event);
     });
     mountMessagingWebhookRoutes(app, { messaging });
+    // Start polling-mode adapters (e.g. Telegram with no public webhook URL
+    // registered) immediately rather than waiting for the first webhook
+    // POST or outbound send to lazily trigger it. This is the process that
+    // owns the inbound sink registered just above, so it must be the one
+    // holding the live connection — a second poller elsewhere (e.g. the
+    // worker) would only fight this one for Telegram's single getUpdates
+    // slot without ever seeing the messages itself.
+    // Bounded retries cover transient Telegram startup failures; polling-only
+    // bots otherwise stay dark until an unrelated outbound send re-inits.
+    messagingInitTask = (async () => {
+      const delayMs = [0, 2_000, 10_000];
+      for (let attempt = 0; attempt < delayMs.length; attempt += 1) {
+        if (messagingStopped) return;
+        if (delayMs[attempt]! > 0) {
+          await new Promise<void>((resolve) => {
+            const timer = setTimeout(resolve, delayMs[attempt]);
+            clearMessagingRetryDelay = () => {
+              clearTimeout(timer);
+              clearMessagingRetryDelay = undefined;
+              resolve();
+            };
+          });
+          clearMessagingRetryDelay = undefined;
+        }
+        if (messagingStopped) return;
+        try {
+          await messaging.initialize?.();
+          return;
+        } catch (error) {
+          console.error(
+            attempt === delayMs.length - 1
+              ? "messaging surface initialize failed"
+              : "messaging surface initialize failed; retrying",
+            error,
+          );
+        }
+      }
+    })();
   }
 
   app.get("/health", (c) =>
@@ -459,6 +505,10 @@ export async function createApp(
     executor,
     stop: async () => {
       oauthLogins.abortAll();
+      messagingStopped = true;
+      clearMessagingRetryDelay?.();
+      await messagingInitTask?.catch(() => undefined);
+      await messaging?.shutdown?.();
       await email?.drain?.();
       await reconciler?.stop();
       await jobs.close();

apps/api/src/router.ts

@@ -138,7 +138,12 @@ import {
   UpdaterProxyError,
 } from "./server-update.js";
 import { assertTeachingSendAllowed, createTaughtSkillsService } from "./taught-skills.js";
-import { isPeerRun, loadAllMessages, loadMessagePage } from "./thread-message-pages.js";
+import {
+  isPeerRun,
+  loadAllMessages,
+  loadMessagePage,
+  shouldForwardPeerThreadEvent,
+} from "./thread-message-pages.js";
 import {
   reactToThreadMessage,
   resolveThreadTarget,
@@ -1051,25 +1056,7 @@ export function createRouter(deps: RouterDeps) {
           context.signal,
         )) {
           if (await isPeerRun(deps.prisma, event.runId, peerRunCache)) {
-            // Keep terminal peer-run events so clients can clear working state.
-            // Keep compact peer receipts for mobile; drop peer activity/replies.
-            const isTerminal =
-              event.type === "run.completed" ||
-              event.type === "run.failed" ||
-              event.type === "run.cancelled";
-            const blocks = event.payload.blocks;
-            const isReceipt =
-              (event.type === "thread.message.created" ||
-                event.type === "thread.message.updated") &&
-              Array.isArray(blocks) &&
-              blocks.some(
-                (block) =>
-                  !!block &&
-                  typeof block === "object" &&
-                  "kind" in block &&
-                  (block.kind === "bot_message_received" || block.kind === "bot_message_sent"),
-              );
-            if (!isTerminal && !isReceipt) continue;
+            if (!shouldForwardPeerThreadEvent(event)) continue;
           }
           yield event;
         }

apps/api/src/thread-message-pages.test.ts

@@ -1,6 +1,11 @@
 import type { PrismaClient } from "@rakazo/db";
 import { describe, expect, it, vi } from "vitest";
-import { isPeerRun, loadAllMessages, loadMessagePage } from "./thread-message-pages.js";
+import {
+  isPeerRun,
+  loadAllMessages,
+  loadMessagePage,
+  shouldForwardPeerThreadEvent,
+} from "./thread-message-pages.js";
 
 describe("thread message pages", () => {
   it("caches peer-run classification for live events", async () => {
@@ -13,6 +18,31 @@ describe("thread message pages", () => {
     expect(findUnique).toHaveBeenCalledTimes(1);
   });
 
+  it("forwards peer waiting, ask, and text events on an open thread", () => {
+    expect(shouldForwardPeerThreadEvent({ type: "run.waiting_input", payload: {} })).toBe(true);
+    expect(shouldForwardPeerThreadEvent({ type: "computer.takeover.requested", payload: {} })).toBe(
+      true,
+    );
+    expect(
+      shouldForwardPeerThreadEvent({
+        type: "thread.message.created",
+        payload: { blocks: [{ kind: "ask", text: "Pick one" }] },
+      }),
+    ).toBe(true);
+    expect(
+      shouldForwardPeerThreadEvent({
+        type: "thread.message.created",
+        payload: { blocks: [{ kind: "text", text: "peer body" }] },
+      }),
+    ).toBe(true);
+    expect(
+      shouldForwardPeerThreadEvent({
+        type: "thread.progress",
+        payload: {},
+      }),
+    ).toBe(false);
+  });
+
   it("keeps peer receipt rows when filtering peer-run output from pages", async () => {
     const findMany = vi.fn(async () => [
       {
@@ -66,17 +96,18 @@ describe("thread message pages", () => {
     expect(page.messages.map((message) => message.id)).toEqual([
       "message-user",
       "message-received",
+      "message-reply",
     ]);
   });
 
-  it("filters peer-run output when its receipt is outside the loaded page", async () => {
+  it("filters peer-run activity when its receipt is outside the loaded page", async () => {
     const findMany = vi.fn(async () => [
       {
         id: "message-peer",
         threadId: "thread-1",
         seq: 2,
         role: "bot",
-        blocks: [{ kind: "text", text: "Echoed peer reply" }],
+        blocks: [{ kind: "steps", steps: [{ label: "Echoed peer reply", count: 1 }] }],
         botId: "bot-1",
         replyToMessageId: null,
         runId: "run-peer",
@@ -104,7 +135,7 @@ describe("thread message pages", () => {
     expect(page.messages.map((message) => message.id)).toEqual(["message-user"]);
   });
 
-  it("omits peer around-page targets from the normal transcript", async () => {
+  it("keeps peer text around-page targets and omits peer activity", async () => {
     const findMany = vi.fn(async () => [
       {
         id: "message-user",
@@ -152,11 +183,14 @@ describe("thread message pages", () => {
       seq: 6,
     });
 
-    expect(page.messages.map((message) => message.id)).toEqual(["message-user"]);
+    expect(page.messages.map((message) => message.id)).toEqual([
+      "message-user",
+      "message-peer-target",
+    ]);
     expect(runFindMany).toHaveBeenCalled();
   });
 
-  it("keeps peer receipt around-page targets in the normal transcript page", async () => {
+  it("keeps peer receipt and text around-page targets in the normal transcript page", async () => {
     const findMany = vi.fn(async () => [
       {
         id: "message-user",
@@ -213,6 +247,7 @@ describe("thread message pages", () => {
     expect(page.messages.map((message) => message.id)).toEqual([
       "message-user",
       "message-peer-receipt",
+      "message-peer-text",
     ]);
   });
 
@@ -246,21 +281,28 @@ describe("thread message pages", () => {
     expect(page.messages.map((message) => message.id)).toEqual(["message-peer"]);
   });
 
-  it("scans past a page containing only peer-run output", async () => {
-    const row = (seq: number, runId: string) => ({
+  it("scans past a page containing only hidden peer-run activity", async () => {
+    const row = (seq: number, runId: string, kind: "text" | "steps" = "text") => ({
       id: `message-${seq}`,
       threadId: "thread-1",
       seq,
       role: "bot",
-      blocks: [{ kind: "text", text: String(seq) }],
+      blocks:
+        kind === "steps"
+          ? [{ kind: "steps", steps: [{ label: String(seq), count: 1 }] }]
+          : [{ kind: "text", text: String(seq) }],
       botId: "bot-1",
       replyToMessageId: null,
       runId,
       createdAt: new Date("2026-08-16T00:00:00.000Z"),
     });
     const findMany = vi
       .fn()
-      .mockResolvedValueOnce([row(4, "run-peer"), row(3, "run-peer"), row(2, "run-peer")])
+      .mockResolvedValueOnce([
+        row(4, "run-peer", "steps"),
+        row(3, "run-peer", "steps"),
+        row(2, "run-peer", "steps"),
+      ])
       .mockResolvedValueOnce([row(1, "run-user")]);
     const prisma = {
       message: { findMany },

apps/api/src/thread-message-pages.ts

@@ -106,10 +106,14 @@ async function withoutPeerRunMessages<T extends { runId: string | null; blocks:
   const peerRunIds = new Set(peerRuns.map((run) => run.id));
   return rows.filter((row) => {
     if (!row.runId || !peerRunIds.has(row.runId)) return true;
-    // Keep compact sent/received receipts; clients render them as chips.
+    // Keep peer receipts (chips), ask cards, and the bot's own text reply.
     const blocks = row.blocks as MessageBlock[];
     return blocks.some(
-      (block) => block.kind === "bot_message_sent" || block.kind === "bot_message_received",
+      (block) =>
+        block.kind === "bot_message_sent" ||
+        block.kind === "bot_message_received" ||
+        block.kind === "ask" ||
+        block.kind === "text",
     );
   });
 }
@@ -130,6 +134,39 @@ export async function isPeerRun(
   return peerRun;
 }
 
+/** Peer-run SSE events that must still reach an open thread (terminals, waits, receipts, asks, text). */
+export function shouldForwardPeerThreadEvent(event: {
+  type: string;
+  payload: { blocks?: unknown };
+}): boolean {
+  if (
+    event.type === "run.completed" ||
+    event.type === "run.failed" ||
+    event.type === "run.cancelled" ||
+    event.type === "run.waiting_input" ||
+    event.type === "computer.takeover.requested"
+  ) {
+    return true;
+  }
+  if (event.type !== "thread.message.created" && event.type !== "thread.message.updated") {
+    return false;
+  }
+  const blocks = event.payload.blocks;
+  return (
+    Array.isArray(blocks) &&
+    blocks.some(
+      (block) =>
+        !!block &&
+        typeof block === "object" &&
+        "kind" in block &&
+        (block.kind === "bot_message_received" ||
+          block.kind === "bot_message_sent" ||
+          block.kind === "ask" ||
+          block.kind === "text"),
+    )
+  );
+}
+
 function toThreadMessage(row: {
   id: string;
   threadId: string;