diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 678d010bd..1f05a6356 100644 --- a/apps/api/src/app.ts +++ b/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 | 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((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(); diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index 09e10d2c0..c50bb2868 100644 --- a/apps/api/src/router.ts +++ b/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; } diff --git a/apps/api/src/thread-message-pages.test.ts b/apps/api/src/thread-message-pages.test.ts index 2f9af85b1..b735bfeed 100644 --- a/apps/api/src/thread-message-pages.test.ts +++ b/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,13 +281,16 @@ 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, @@ -260,7 +298,11 @@ describe("thread message pages", () => { }); 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 }, diff --git a/apps/api/src/thread-message-pages.ts b/apps/api/src/thread-message-pages.ts index c1e6a63de..c40d3bd48 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -106,10 +106,14 @@ async function withoutPeerRunMessages 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; diff --git a/apps/api/src/thread-target.test.ts b/apps/api/src/thread-target.test.ts index a696f5779..85dd736a7 100644 --- a/apps/api/src/thread-target.test.ts +++ b/apps/api/src/thread-target.test.ts @@ -5,6 +5,7 @@ import { describe, expect, it, vi } from "vitest"; import { cancelSupersededQueuedRuns, reactToThreadMessage, + sendThreadMessage, stopThreadRuns, type ThreadTarget, threadHead, @@ -178,7 +179,7 @@ describe("threadSnapshot", () => { findFirst: vi.fn().mockResolvedValue({ seq: 4 }), findMany: findManyEvents, }, - run: { findFirst: vi.fn().mockResolvedValue(run) }, + run: { findFirst: botRunFindFirst([run]) }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), @@ -230,11 +231,7 @@ describe("threadSnapshot", () => { createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const findManyEvents = vi.fn(); - const findFirstRun = vi - .fn() - .mockResolvedValueOnce(run) - // The failure is itself the newest terminal run, so it stays visible. - .mockResolvedValueOnce({ id: run.id }); + const findFirstRun = botRunFindFirst([run]); const tx = { $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), message: { findMany: vi.fn().mockResolvedValue([]) }, @@ -268,6 +265,15 @@ describe("threadSnapshot", () => { }), }), ); + expect(findFirstRun).toHaveBeenCalledWith( + expect.objectContaining({ + where: { + botId: "bot-1", + threadId: "thread-1", + status: { in: ["waiting_input", "waiting_takeover"] }, + }, + }), + ); expect(snapshot.run).toEqual( expect.objectContaining({ id: "run-failed", @@ -278,6 +284,64 @@ describe("threadSnapshot", () => { expect(findManyEvents).not.toHaveBeenCalled(); }); + it("prefers a waiting peer ask over a concurrent user run", async () => { + const waitingPeer = { + id: "run-peer-waiting", + botId: "bot-1", + threadId: "thread-1", + taskId: "task-peer", + status: "waiting_input", + trigger: "bot_message", + modelProvider: null, + modelId: null, + error: null, + startedAt: new Date("2026-08-23T00:00:02.000Z"), + completedAt: null, + createdAt: new Date("2026-08-23T00:00:02.000Z"), + }; + const olderUser = { + id: "run-user", + botId: "bot-1", + threadId: "thread-1", + taskId: "task-user", + status: "running", + trigger: "user", + modelProvider: null, + modelId: null, + error: null, + startedAt: new Date("2026-08-23T00:00:01.000Z"), + completedAt: null, + createdAt: new Date("2026-08-23T00:00:01.000Z"), + }; + const snapshot = await threadSnapshot( + { + prisma: { + $transaction: vi.fn(async (callback: (client: unknown) => unknown) => + callback({ + $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + message: { findMany: vi.fn().mockResolvedValue([]) }, + event: { + findFirst: vi.fn().mockResolvedValue(null), + findMany: vi.fn().mockResolvedValue([]), + }, + run: { findFirst: botRunFindFirst([waitingPeer, olderUser]) }, + }), + ), + } as unknown as PrismaClient, + }, + { + kind: "bot", + botId: "bot-1", + threadId: "thread-1", + bot: { computer: null }, + } as ThreadTarget, + ); + + expect(snapshot.run).toEqual( + expect.objectContaining({ id: "run-peer-waiting", status: "waiting_input" }), + ); + }); + it("drops a failed run once a newer run has finished", async () => { const failed = { id: "run-failed", @@ -293,11 +357,21 @@ describe("threadSnapshot", () => { completedAt: new Date("2026-08-23T00:00:01.000Z"), createdAt: new Date("2026-08-23T00:00:00.000Z"), }; - const findFirstRun = vi - .fn() - .mockResolvedValueOnce(failed) - // The supersession probe finds a newer completed run. - .mockResolvedValueOnce({ id: "run-completed" }); + const completed = { + id: "run-completed", + botId: "bot-1", + threadId: "thread-1", + taskId: "task-2", + status: "completed", + trigger: "user", + modelProvider: null, + modelId: null, + error: null, + startedAt: null, + completedAt: new Date("2026-08-23T00:00:03.000Z"), + createdAt: new Date("2026-08-23T00:00:02.000Z"), + }; + const findFirstRun = botRunFindFirst([failed, completed]); const tx = { $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), message: { findMany: vi.fn().mockResolvedValue([]) }, @@ -319,8 +393,7 @@ describe("threadSnapshot", () => { const snapshot = await threadSnapshot({ prisma }, target); - expect(findFirstRun).toHaveBeenNthCalledWith( - 2, + expect(findFirstRun).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ trigger: { not: "bot_message" }, @@ -334,7 +407,7 @@ describe("threadSnapshot", () => { it("does not return a cancelled or completed run", async () => { const findManyEvents = vi.fn(); - const findFirstRun = vi.fn().mockResolvedValue(null); + const findFirstRun = botRunFindFirst([]); const tx = { $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), message: { findMany: vi.fn().mockResolvedValue([]) }, @@ -401,8 +474,11 @@ describe("threadSnapshot", () => { expect.objectContaining({ where: { threadId: "thread-1", - trigger: { not: "bot_message" }, status: { in: ["queued", "leased", "running", "waiting_input", "waiting_takeover"] }, + OR: [ + { trigger: { not: "bot_message" } }, + { status: { in: ["waiting_input", "waiting_takeover"] } }, + ], }, }), ); @@ -452,7 +528,10 @@ describe("threadSnapshot", () => { expect(findManyRuns).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ - trigger: { not: "bot_message" }, + OR: [ + { trigger: { not: "bot_message" } }, + { status: { in: ["waiting_input", "waiting_takeover"] } }, + ], status: { in: ["queued", "leased", "running", "waiting_input", "waiting_takeover"] }, }), }), @@ -467,6 +546,29 @@ describe("threadSnapshot", () => { ); }); + it("includes waiting peer bot_message runs in group activeRuns", async () => { + const peerWaiting = { + id: "run-peer-waiting", + botId: "bot-a", + threadId: "thread-1", + taskId: "task-peer", + status: "waiting_input", + trigger: "bot_message", + modelProvider: null, + modelId: null, + error: null, + startedAt: new Date("2026-08-23T00:00:05.000Z"), + completedAt: null, + createdAt: new Date("2026-08-23T00:00:05.000Z"), + }; + const findManyRuns = groupRunFindMany({ active: [peerWaiting] }); + const snapshot = await threadSnapshot({ prisma: groupPrisma(findManyRuns) }, groupTarget()); + + expect(snapshot.activeRuns).toEqual([ + expect.objectContaining({ id: "run-peer-waiting", status: "waiting_input" }), + ]); + }); + it("does not revive an older group failure after a newer run completed", async () => { const failed = { id: "run-old-failed", @@ -701,22 +803,72 @@ function isTerminalRunQuery(where: { status?: { in?: string[] } } | undefined) { return Array.isArray(statuses) && statuses.includes("failed") && statuses.includes("completed"); } -function excludesPeerRuns(where: { trigger?: { not?: string } } | undefined) { - return where?.trigger?.not === "bot_message"; +function matchesPeerActiveFilter( + row: { trigger?: string; status?: string }, + where: + | { + trigger?: { not?: string }; + OR?: Array<{ trigger?: { not?: string }; status?: { in?: string[] } }>; + } + | undefined, +) { + if (where?.trigger?.not === "bot_message") return row.trigger !== "bot_message"; + if (!where?.OR) return true; + return where.OR.some((clause) => { + if (clause.trigger?.not === "bot_message") return row.trigger !== "bot_message"; + if (clause.status?.in) return clause.status.in.includes(row.status ?? ""); + return false; + }); +} + +function botRunFindFirst( + rows: Array<{ + id: string; + status: string; + trigger?: string; + createdAt?: Date; + }>, +) { + return vi.fn().mockImplementation( + async (args: { + where?: { + status?: { in?: string[] }; + trigger?: { not?: string }; + }; + select?: { id?: boolean }; + }) => { + const statuses = args.where?.status?.in; + const matched = rows + .filter((row) => !statuses || statuses.includes(row.status)) + .filter((row) => + args.where?.trigger?.not === "bot_message" ? row.trigger !== "bot_message" : true, + ) + .sort((a, b) => { + const byCreated = (b.createdAt?.getTime() ?? 0) - (a.createdAt?.getTime() ?? 0); + return byCreated !== 0 ? byCreated : b.id.localeCompare(a.id); + }); + const row = matched[0] ?? null; + if (!row) return null; + return args.select?.id ? { id: row.id } : row; + }, + ); } function groupRunFindMany(input: { active?: unknown[]; terminals?: unknown[] }) { - return vi - .fn() - .mockImplementation( - async (args: { where?: { status?: { in?: string[] }; trigger?: { not?: string } } }) => { - const rows = isTerminalRunQuery(args.where) - ? (input.terminals ?? []) - : (input.active ?? []); - if (!excludesPeerRuns(args.where)) return rows; - return rows.filter((row) => (row as { trigger?: string }).trigger !== "bot_message"); - }, - ); + return vi.fn().mockImplementation( + async (args: { + where?: { + status?: { in?: string[] }; + trigger?: { not?: string }; + OR?: Array<{ trigger?: { not?: string }; status?: { in?: string[] } }>; + }; + }) => { + const rows = isTerminalRunQuery(args.where) ? (input.terminals ?? []) : (input.active ?? []); + return rows.filter((row) => + matchesPeerActiveFilter(row as { trigger?: string; status?: string }, args.where), + ); + }, + ); } function groupPrisma(findManyRuns: ReturnType) { @@ -744,6 +896,70 @@ function groupTarget() { } as unknown as ThreadTarget; } +describe("sendThreadMessage", () => { + it("rejects a new bot message while a run is waiting on input", async () => { + const tx = { + thread: { + update: vi.fn().mockResolvedValue({ nextMessageSeq: 2 }), + }, + message: { + create: vi.fn().mockResolvedValue({ + id: "msg-1", + threadId: "thread-1", + seq: 1, + role: "user", + blocks: [{ kind: "text", text: "hi" }], + botId: null, + replyToMessageId: null, + runId: null, + thumbsUp: false, + createdAt: new Date(), + }), + update: vi.fn(), + }, + run: { + findMany: vi + .fn() + .mockResolvedValue([{ id: "run-waiting", taskId: "task-1", status: "waiting_input" }]), + }, + steeringMessage: { create: vi.fn() }, + event: { create: vi.fn() }, + task: { create: vi.fn() }, + }; + const prisma = { + message: { findUnique: vi.fn().mockResolvedValue(null) }, + $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), + } as unknown as PrismaClient; + const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; + const target = { + kind: "bot", + botId: "bot-1", + threadId: "thread-1", + bot: { computer: null }, + } as ThreadTarget; + + await expect( + sendThreadMessage( + { + prisma, + events: { notify: vi.fn() } as never, + jobs: { enqueue: vi.fn() } as never, + }, + actor, + target, + { + text: "hi", + clientNonce: "nonce-1", + }, + ), + ).rejects.toMatchObject({ + code: "CONFLICT", + message: "Answer the pending ask first.", + }); + expect(tx.steeringMessage.create).not.toHaveBeenCalled(); + }); +}); + describe("stopThreadRuns", () => { it("releases every active group member screen immediately", async () => { const releaseScreen = vi.fn().mockResolvedValue(undefined); diff --git a/apps/api/src/thread-target.ts b/apps/api/src/thread-target.ts index 8974404a8..576df0fea 100644 --- a/apps/api/src/thread-target.ts +++ b/apps/api/src/thread-target.ts @@ -1,3 +1,4 @@ +import { ORPCError } from "@orpc/server"; import { type JobPublisher, runContinueJob } from "@rakazo/adapter-kit"; import { cancelComputerRunWork, screenLeaseIdForRun, toComputerRef } from "@rakazo/adapters"; import { @@ -55,6 +56,8 @@ export type ThreadTarget = const THREAD_MESSAGE_PAGE_SIZE = 100; const RUNS_NEEDING_CONTINUE = new Set(["queued"]); +const STEERABLE_RUN_STATUSES = new Set(["queued", "leased", "running"]); + type MentionTargetInput = string | { kind: "bot" | "group" | "routine" | "connector"; id: string }; function splitMentionTargets(mentions: MentionTargetInput[] | undefined) { @@ -319,17 +322,27 @@ export async function threadSnapshot( }), deps.prisma.$transaction(async (tx) => { await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR SHARE`; - const [messagePage, last, run] = await Promise.all([ + const [messagePage, last, waitingRun, busyOrFailed] = await Promise.all([ loadMessagePage(tx, target.threadId, undefined, THREAD_MESSAGE_PAGE_SIZE), tx.event.findFirst({ where: { threadId: target.threadId }, orderBy: { seq: "desc" }, select: { seq: true }, }), + // Waiting asks win over a concurrent busy run (including peer bot_message). + tx.run.findFirst({ + where: { + botId: target.botId, + threadId: target.threadId, + status: { in: ["waiting_input", "waiting_takeover"] }, + }, + orderBy: [{ createdAt: "desc" }, { id: "desc" }], + }), tx.run.findFirst({ where: { botId: target.botId, threadId: target.threadId, + // Hide peer bot_message busy/failed noise; waiting is handled above. trigger: { not: "bot_message" }, status: { in: [...ACTIVE_RUN_STATUSES, "failed"] }, }, @@ -338,6 +351,7 @@ export async function threadSnapshot( orderBy: [{ createdAt: "desc" }, { id: "desc" }], }), ]); + const run = waitingRun ?? busyOrFailed; // A failed run is only the thread's word while it is still the newest // terminal run; otherwise a stale failure would resurface in the // composer error strip on every load, forever. Instead of comparing @@ -350,7 +364,7 @@ export async function threadSnapshot( where: { botId: target.botId, threadId: target.threadId, - // Match the selection query — peer bot_message runs must not bury a user-visible failure. + // Peer bot_message failures must not bury a user-visible failure. trigger: { not: "bot_message" }, status: { in: ["failed", "completed", "cancelled"] }, }, @@ -396,8 +410,12 @@ export async function threadSnapshot( tx.run.findMany({ where: { threadId: target.threadId, - trigger: { not: "bot_message" }, status: { in: [...ACTIVE_RUN_STATUSES] }, + // Include waiting peer runs so their ask cards stay answerable. + OR: [ + { trigger: { not: "bot_message" } }, + { status: { in: ["waiting_input", "waiting_takeover"] } }, + ], }, orderBy: { createdAt: "desc" }, }), @@ -572,7 +590,7 @@ export async function sendThreadMessage( replyToMessageId: input.replyToMessageId, clientNonce: input.clientNonce, }); - const active = await tx.run.findFirst({ + const activeRuns = await tx.run.findMany({ where: { threadId: target.threadId, botId: target.botId, @@ -580,6 +598,12 @@ export async function sendThreadMessage( }, select: { id: true, taskId: true, status: true }, }); + if (activeRuns.some((run) => !STEERABLE_RUN_STATUSES.has(run.status))) { + throw new ORPCError("CONFLICT", { + message: "Answer the pending ask first.", + }); + } + const active = activeRuns[0]; if (active) { await tx.steeringMessage.create({ data: { @@ -687,7 +711,15 @@ export async function sendThreadMessage( }, select: { id: true, taskId: true, botId: true, status: true }, }); - const activeByBotId = new Map(activeRuns.map((run) => [run.botId, run])); + const activeByBotId = new Map(); + for (const run of activeRuns) { + if (!STEERABLE_RUN_STATUSES.has(run.status)) { + throw new ORPCError("CONFLICT", { + message: "Answer the pending ask first.", + }); + } + if (!activeByBotId.has(run.botId)) activeByBotId.set(run.botId, run); + } const runs: Array<{ id: string; taskId: string; botId: string; status: string }> = []; for (const botId of targetBotIds) { const active = activeByBotId.get(botId); diff --git a/apps/mobile/app/thread.tsx b/apps/mobile/app/thread.tsx index 81505e4ee..4fbabf3ff 100644 --- a/apps/mobile/app/thread.tsx +++ b/apps/mobile/app/thread.tsx @@ -766,6 +766,7 @@ function Thread() { event.type === "thread.subagent" || event.type === "thread.cleared" || event.type === "run.waiting_input" || + event.type === "computer.takeover.requested" || isRunTerminalEvent(event) ) { if (event.type === "thread.cleared") { diff --git a/apps/mobile/lib/api.test.ts b/apps/mobile/lib/api.test.ts index bfea5511f..48a3eb6e0 100644 --- a/apps/mobile/lib/api.test.ts +++ b/apps/mobile/lib/api.test.ts @@ -1062,6 +1062,56 @@ describe("mobile thread event reduction", () => { expect(repeated?.run).toBe(waiting?.run); }); + it("applies computer takeover requests as waiting_takeover", () => { + const progress = mobileMessage("progress:run-1", [{ kind: "progress", text: "working…" }], 1); + const initial: MobileSnapshot = { + ...snapshot([progress]), + run: { id: "run-1", status: "running" }, + activeRuns: [{ id: "run-1", status: "running" }], + }; + const waiting = applyMobileThreadEvent(initial, { + type: "computer.takeover.requested", + runId: "run-1", + seq: 10, + }); + + expect(waiting?.run?.status).toBe("waiting_takeover"); + expect(waiting?.activeRuns?.[0]?.status).toBe("waiting_takeover"); + expect(waiting?.messages.some((message) => message.id.startsWith("progress:"))).toBe(false); + expect(waiting?.cursor).toBe(10); + }); + + it("inserts a peer takeover run that was absent from the open snapshot", () => { + const progress = mobileMessage("progress:run-peer", [{ kind: "progress", text: "working…" }], 1); + const initial: MobileSnapshot = { + ...snapshot([progress]), + run: { id: "run-user", status: "running" }, + activeRuns: [{ id: "run-user", status: "running" }], + computer: { + state: "running", + controlHolder: "bot", + screenAvailable: true, + mode: "team", + busyBotName: "Peer", + }, + }; + + const waiting = applyMobileThreadEvent(initial, { + type: "computer.takeover.requested", + runId: "run-peer", + botId: "bot-peer", + seq: 12, + }); + + expect(waiting?.run).toEqual({ id: "run-peer", botId: "bot-peer", status: "waiting_takeover" }); + expect(waiting?.activeRuns).toEqual([ + { id: "run-user", status: "running" }, + { id: "run-peer", botId: "bot-peer", status: "waiting_takeover" }, + ]); + expect(waiting?.messages.some((message) => message.id.startsWith("progress:"))).toBe(false); + expect(waiting?.computer?.busyBotName).toBeNull(); + }); + it("advances the cursor for durable message events", () => { const next = applyMobileThreadEvent(snapshot(), { type: "thread.message.created", diff --git a/apps/mobile/lib/api.ts b/apps/mobile/lib/api.ts index 0acc3ffbb..daf38b32f 100644 --- a/apps/mobile/lib/api.ts +++ b/apps/mobile/lib/api.ts @@ -652,27 +652,60 @@ export function applyMobileThreadEvent( activeRuns: [], }; } - if (event.type === "run.waiting_input") { + if (event.type === "run.waiting_input" || event.type === "computer.takeover.requested") { + const status = event.type === "run.waiting_input" ? "waiting_input" : "waiting_takeover"; const progressId = progressMessageId(event); + // Waiting pauses drop live progress server-side; clear a leftover bubble so + // the waiting footer is not hidden behind a stale "Working…" row. const messages = prev.messages.filter((message) => message.id !== progressId); const progressCleared = messages.length !== prev.messages.length; - const runChanged = Boolean( - prev.run && prev.run.id === event.runId && prev.run.status !== "waiting_input", - ); - const activeRunChanged = prev.activeRuns?.some( - (candidate) => candidate.id === event.runId && candidate.status !== "waiting_input", + const runId = event.runId; + const knownInRun = Boolean(runId && prev.run?.id === runId); + const knownInActive = Boolean(runId && prev.activeRuns?.some((candidate) => candidate.id === runId)); + // Peer bot_message runs are omitted from snapshots while busy; the first wait + // event is how an open thread learns they need ask/takeover UI. + const needsInsert = Boolean(runId) && !knownInRun && !knownInActive; + const runChanged = Boolean(knownInRun && prev.run && prev.run.status !== status); + const activeRunChanged = Boolean( + knownInActive && + prev.activeRuns?.some((candidate) => candidate.id === runId && candidate.status !== status), ); + const computer = + event.type === "computer.takeover.requested" && prev.computer?.busyBotName + ? { ...prev.computer, busyBotName: null } + : prev.computer; + const computerChanged = computer !== prev.computer; const cursor = event.seq ?? prev.cursor; - if (!runChanged && !activeRunChanged && !progressCleared) { + if (!runChanged && !activeRunChanged && !progressCleared && !needsInsert && !computerChanged) { return cursor === prev.cursor ? prev : { ...prev, cursor }; } - const run = runChanged && prev.run ? { ...prev.run, status: "waiting_input" } : prev.run; + if (needsInsert && runId) { + const waitingRun = { + id: runId, + status, + ...(event.botId ? { botId: event.botId } : {}), + }; + const baseActive = prev.activeRuns ?? (prev.run ? [prev.run] : []); + const activeRuns = [...baseActive.filter((candidate) => candidate.id !== runId), waitingRun]; + const promoteWaiting = + !prev.run || + (prev.run.status !== "waiting_input" && prev.run.status !== "waiting_takeover"); + return { + ...prev, + cursor, + messages, + computer, + run: promoteWaiting ? waitingRun : prev.run, + activeRuns, + }; + } + const run = runChanged && prev.run ? { ...prev.run, status } : prev.run; const activeRuns = activeRunChanged ? prev.activeRuns?.map((candidate) => - candidate.id === event.runId ? { ...candidate, status: "waiting_input" } : candidate, + candidate.id === runId ? { ...candidate, status } : candidate, ) : prev.activeRuns; - return { ...prev, cursor, run, activeRuns, messages }; + return { ...prev, cursor, run, activeRuns, messages, computer }; } if (isRunTerminalEvent(event)) { const activeRuns = prev.activeRuns?.filter((candidate) => candidate.id !== event.runId); diff --git a/apps/web/src/lib/thread-events.test.ts b/apps/web/src/lib/thread-events.test.ts index 74d78466a..298225134 100644 --- a/apps/web/src/lib/thread-events.test.ts +++ b/apps/web/src/lib/thread-events.test.ts @@ -350,6 +350,36 @@ describe("thread event reduction", () => { expect(waiting?.activeRuns?.[0]?.status).toBe("waiting_takeover"); }); + it("inserts a peer takeover run that was absent from the open snapshot", () => { + const userRun = threadRun("run-user"); + const initial: ThreadSnapshot = { + ...snapshot([]), + run: userRun, + activeRuns: [userRun], + }; + + const waiting = reduceThreadSnapshot( + initial, + event({ + type: "computer.takeover.requested", + seq: 12, + runId: "run-peer", + botId: "bot-peer", + }), + ); + + expect(waiting?.run).toMatchObject({ + id: "run-peer", + botId: "bot-peer", + status: "waiting_takeover", + trigger: "bot_message", + }); + expect(waiting?.activeRuns?.map((run) => ({ id: run.id, status: run.status }))).toEqual([ + { id: "run-user", status: "running" }, + { id: "run-peer", status: "waiting_takeover" }, + ]); + }); + it("keeps event-sourced waiting_takeover when a stale refresh still shows the bot busy", () => { const run = threadRun("run-1"); const waitingLocal: ThreadSnapshot = { diff --git a/apps/web/src/lib/thread-events.ts b/apps/web/src/lib/thread-events.ts index 750ccc239..9d926671e 100644 --- a/apps/web/src/lib/thread-events.ts +++ b/apps/web/src/lib/thread-events.ts @@ -270,11 +270,18 @@ export function reduceThreadSnapshot( } if (event.type === "run.waiting_input" || event.type === "computer.takeover.requested") { const status = event.type === "run.waiting_input" ? "waiting_input" : "waiting_takeover"; - const runChanged = Boolean( - prev.run && prev.run.id === event.runId && prev.run.status !== status, + const runId = event.runId; + const knownInRun = Boolean(runId && prev.run?.id === runId); + const knownInActive = Boolean( + runId && prev.activeRuns?.some((candidate) => candidate.id === runId), ); - const activeRunChanged = prev.activeRuns?.some( - (candidate) => candidate.id === event.runId && candidate.status !== status, + // Peer bot_message runs are omitted from snapshots while busy; the first wait + // event is how an open thread learns they need ask/takeover UI. + const needsInsert = Boolean(runId) && !knownInRun && !knownInActive; + const runChanged = Boolean(knownInRun && prev.run && prev.run.status !== status); + const activeRunChanged = Boolean( + knownInActive && + prev.activeRuns?.some((candidate) => candidate.id === runId && candidate.status !== status), ); const members = updateMemberStatus(prev.members, event.botId, status); // Ask pauses delete progress events server-side; drop the live bubble so a missed @@ -287,11 +294,42 @@ export function reduceThreadSnapshot( if ( !runChanged && !activeRunChanged && + !needsInsert && members === prev.members && messages === prev.messages ) { return prev; } + if (needsInsert && runId) { + const waitingRun: Run = { + id: runId, + botId: event.botId, + threadId: event.threadId, + taskId: runId, + status, + trigger: "bot_message", + routineId: null, + modelProvider: null, + modelId: null, + error: null, + startedAt: event.createdAt, + completedAt: null, + createdAt: event.createdAt, + }; + const baseActive = prev.activeRuns ?? (prev.run ? [prev.run] : []); + const activeRuns = [...baseActive.filter((candidate) => candidate.id !== runId), waitingRun]; + const promoteWaiting = + !prev.run || + (prev.run.status !== "waiting_input" && prev.run.status !== "waiting_takeover"); + return { + ...prev, + cursor: event.seq, + members, + messages, + run: promoteWaiting ? waitingRun : prev.run, + activeRuns, + }; + } return { ...prev, cursor: event.seq, @@ -300,7 +338,7 @@ export function reduceThreadSnapshot( run: runChanged && prev.run ? { ...prev.run, status } : prev.run, activeRuns: activeRunChanged ? prev.activeRuns?.map((candidate) => - candidate.id === event.runId ? { ...candidate, status } : candidate, + candidate.id === runId ? { ...candidate, status } : candidate, ) : prev.activeRuns, }; diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index ecd2b2280..75baa0ebb 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -95,6 +95,10 @@ async function main() { const pipedream = isPipedreamEnabled(pipedreamConfig) ? new PipedreamConnector(pipedreamConfig) : undefined; + // pollInboundMessages stays false (the default) here: this process + // only ever sends outbound (messaging.deliver jobs). It must never poll + // Telegram — that would steal the single getUpdates slot away from the + // API process, which is the one with the inbound sink actually wired up. const messagingPlatforms = messagingPlatformsFromEnv(messagingEnvFromProcess(process.env)); const messaging = isMessagingSurfaceEnabled(messagingPlatforms, { deploymentModelKey, diff --git a/packages/adapter-kit/src/interfaces.ts b/packages/adapter-kit/src/interfaces.ts index bc58fbece..946353706 100644 --- a/packages/adapter-kit/src/interfaces.ts +++ b/packages/adapter-kit/src/interfaces.ts @@ -323,6 +323,20 @@ export interface MessagingSurface { * message delivery, and silently no-ops on platforms without support. */ sendTyping(threadId: string, context: AdapterContext): Promise; + /** + * Start the surface eagerly instead of waiting for the first inbound + * webhook or outbound send to touch it. Needed for adapters that pull + * updates themselves (e.g. Telegram in "auto"/"polling" mode, used when + * no public webhook URL is reachable): without an explicit call here, + * nothing kicks off that pull loop until something else happens to + * initialize the adapter first. + */ + initialize?(): Promise; + /** + * Release long-lived inbound resources (Telegram getUpdates polling) so a + * replacement process can claim them. No-op when the surface never started. + */ + shutdown?(): Promise; } /** diff --git a/packages/adapters/src/builtin-tools.ts b/packages/adapters/src/builtin-tools.ts index 667d88eb0..09ec35106 100644 --- a/packages/adapters/src/builtin-tools.ts +++ b/packages/adapters/src/builtin-tools.ts @@ -593,7 +593,7 @@ export const builtinAgentTools: ConnectorTool[] = [ { name: "message_bot", description: - "Send a useful update, question, or result to another of the user's bots. Delivery is async and does not end your turn. Continue independent work; do not poll or send ack-only messages. Later updates only if they add something new.", + "Send a useful update, question, or result to another of the user's bots. You must call this tool to actually deliver it — writing the message in your own reply text (e.g. \"[to Comms] ...\") does not send anything and the recipient never sees it. Delivery is async and does not end your turn. Continue independent work; do not poll or send ack-only messages. Later updates only if they add something new.", inputSchema: { type: "object", properties: { diff --git a/packages/adapters/src/chat-sdk-surface.test.ts b/packages/adapters/src/chat-sdk-surface.test.ts index ffb18087b..4da7ed86d 100644 --- a/packages/adapters/src/chat-sdk-surface.test.ts +++ b/packages/adapters/src/chat-sdk-surface.test.ts @@ -376,4 +376,25 @@ describe("ChatSdkMessagingSurface shape", () => { expect(surface.describe().capabilities).toEqual({ providers: ["mock"] }); expect(providerOfThreadId("mock:C1:9")).toBe("mock"); }); + + it("retries initialize after a failure and stops polling on shutdown", async () => { + let attempts = 0; + const stopPolling = vi.fn(async () => undefined); + const { surface, adapter } = createSurface({}, { + initialize: vi.fn(async () => { + attempts += 1; + if (attempts === 1) throw new Error("telegram unavailable"); + }), + stopPolling, + } as Partial); + + await expect(surface.initialize()).rejects.toThrow(/telegram unavailable/); + // Partial multi-adapter init can leave Telegram polling; clear it on failure. + expect(stopPolling).toHaveBeenCalledTimes(1); + await expect(surface.initialize()).resolves.toBeUndefined(); + expect(adapter.initialize).toHaveBeenCalledTimes(2); + + await surface.shutdown(); + expect(stopPolling).toHaveBeenCalledTimes(2); + }); }); diff --git a/packages/adapters/src/chat-sdk-surface.ts b/packages/adapters/src/chat-sdk-surface.ts index b4b54f6ed..31d648164 100644 --- a/packages/adapters/src/chat-sdk-surface.ts +++ b/packages/adapters/src/chat-sdk-surface.ts @@ -154,13 +154,60 @@ export class ChatSdkMessagingSurface implements MessagingSurface { await raceWithSignal(platform.adapter.startTyping(threadId), context.signal); } + /** + * Proactively start the underlying Chat SDK instance. Webhook handling + * and outbound sends already trigger this lazily (see ensureInitialized + * below), but a polling-mode adapter (Telegram in "auto" mode with no + * public webhook URL registered) needs its pull loop running before the + * first inbound message can ever arrive, so long-running hosts call this + * explicitly at startup instead of waiting on the first send. + */ + async initialize(): Promise { + await this.ensureInitialized(); + } + + /** + * Stop Telegram getUpdates before exit. Chat.shutdown() only calls + * adapter.disconnect(), which Telegram does not implement. + */ + async shutdown(): Promise { + const pending = this.initialized; + this.initialized = undefined; + if (pending) await pending.catch(() => undefined); + await this.stopPollingAdapters(); + const chat = this.chat as { shutdown?: () => Promise }; + await chat.shutdown?.().catch(() => undefined); + } + private ensureInitialized(): Promise { // Webhook handling initializes lazily inside the Chat SDK; proactive - // sends from job runners need the explicit call. - this.initialized ??= this.chat.initialize(); + // sends from job runners need the explicit call. Chat keeps a rejected + // initPromise until shutdown(), so clear that before allowing a retry. + if (!this.initialized) { + this.initialized = this.chat.initialize().catch(async (error) => { + try { + // Telegram may already be polling if Promise.all failed on another + // adapter after Telegram's initialize started getUpdates. + await this.stopPollingAdapters(); + await (this.chat as { shutdown?: () => Promise }).shutdown?.(); + } catch { + // Best-effort reset of Chat's cached init failure. + } finally { + this.initialized = undefined; + } + throw error; + }); + } return this.initialized; } + private async stopPollingAdapters(): Promise { + for (const platform of this.byProvider.values()) { + const adapter = platform.adapter as { stopPolling?: () => Promise }; + await adapter.stopPolling?.().catch(() => undefined); + } + } + private async dispatchWebhook(platform: MessagingPlatform, request: Request): Promise { const declared = Number(request.headers.get("content-length") ?? 0); if (declared > MESSAGING_WEBHOOK_MAX_BODY_BYTES) { diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index a22e231c3..46900cded 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -250,7 +250,7 @@ describe("deliverMessagingOutbound", () => { expect(deps.sendToThread).not.toHaveBeenCalled(); }); - it("ignores non-messaging runs and runs without a messaging identity", async () => { + it("ignores in-app runs and runs without a messaging identity", async () => { const notMessaging = createDeps({ run: { ...messagingRun, trigger: "user" } }); await deliverMessagingOutbound(notMessaging, { runId: "run-1" }, context); expect(notMessaging.sendToThread).not.toHaveBeenCalled(); @@ -260,6 +260,15 @@ describe("deliverMessagingOutbound", () => { expect(noIdentity.sendToThread).not.toHaveBeenCalled(); }); + it("mirrors delegated bot_message replies to the linked DM", async () => { + const deps = createDeps({ run: { ...messagingRun, trigger: "bot_message" } }); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + expect(deps.sendToThread).toHaveBeenCalledWith( + { threadId: "sendblue:dm-1", body: "Hello from your bot" }, + context, + ); + }); + it("holds sendblue DM sends at the consecutive-outbound cap", async () => { const deps = createDeps({ identity: { diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index 3edc85977..ab640db5f 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -71,14 +71,21 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise => - block.kind === "channel_message", - ); - if (channelBlock) { - await mirrorChannelRun(deps, run, channelBlock); + if (!run) return; + + if (run.trigger === "messaging") { + const sourceBlocks = (run.sourceMessage?.blocks ?? []) as MessageBlock[]; + const channelBlock = sourceBlocks.find( + (block): block is Extract => + block.kind === "channel_message", + ); + if (channelBlock) { + await mirrorChannelRun(deps, run, channelBlock); + return; + } + } else if (run.trigger !== "bot_message") { + // Mirror inbound messaging runs and delegated bot_message replies only. + // Do not push in-app user/routine runs out to the linked chat. return; } @@ -87,6 +94,9 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise { - it("keeps bot-to-bot exchanges out of the user transcript", () => { - expect(userVisibleMessages(peerExchange).map((item) => item.id)).toEqual(["user", "answer"]); + it("hides peer activity but keeps the bot's text reply to the user", () => { + expect(userVisibleMessages(peerExchange).map((item) => item.id)).toEqual([ + "user", + "reply", + "answer", + ]); }); it("keeps compact peer receipts when includePeerReceipts is set", () => { expect( userVisibleMessages(peerExchange, { includePeerReceipts: true }).map((item) => item.id), - ).toEqual(["user", "sent", "received", "answer"]); + ).toEqual(["user", "sent", "received", "reply", "answer"]); }); it("uses authoritative peer run ids when the receipt is outside the loaded page", () => { const messages = [ + message("activity", "run-peer", [ + { kind: "steps", steps: [{ label: "Echoed peer reply", count: 1 }] }, + ]), message("reply", "run-peer", [{ kind: "text", text: "Echoed peer reply" }]), message("answer", "run-user", [{ kind: "text", text: "Visible answer" }]), ]; expect( userVisibleMessages(messages, { knownPeerRunIds: ["run-peer"] }).map((item) => item.id), - ).toEqual(["answer"]); + ).toEqual(["reply", "answer"]); + }); + + it("keeps a peer-run ask card and text reply while hiding other peer activity", () => { + const messages = [ + message("ask", "run-peer", [ + { + kind: "ask", + text: "Pick one", + status: "pending", + actions: [{ id: "a", label: "A" }], + }, + ]), + message("activity", "run-peer", [{ kind: "steps", steps: [{ label: "Work", count: 1 }] }]), + message("reply", "run-peer", [{ kind: "text", text: "Peer body" }]), + message("answer", "run-user", [{ kind: "text", text: "Visible answer" }]), + ]; + + expect( + userVisibleMessages(messages, { knownPeerRunIds: ["run-peer"] }).map((item) => item.id), + ).toEqual(["ask", "reply", "answer"]); }); }); diff --git a/packages/core/src/message-visibility.ts b/packages/core/src/message-visibility.ts index 8fbd03374..ce2a93475 100644 --- a/packages/core/src/message-visibility.ts +++ b/packages/core/src/message-visibility.ts @@ -36,6 +36,8 @@ export function userVisibleMessages( return messages.filter((message) => { if (isPeerReceiptBlocks(message.blocks)) return includePeerReceipts; - return !message.runId || !peerRunIds.has(message.runId); + if (!message.runId || !peerRunIds.has(message.runId)) return true; + // Keep peer-run ask cards and the bot's own text reply to the user. + return message.blocks.some((block) => block.kind === "ask" || block.kind === "text"); }); }