From d3263671ead2e855d7b58cab1f555e215d4cfc3a Mon Sep 17 00:00:00 2001 From: Joseph Harper <8pgg4z5yvz@privaterelay.appleid.com> Date: Thu, 3 Sep 2026 02:01:10 +0000 Subject: [PATCH 01/22] fix(chat): surface ask cards for bot-delegated tasks Runs started by another bot's message_bot call (trigger: "bot_message") were unconditionally hidden from a bot's own chat thread across three call sites, so a background peer run never bubbles up as busy/failed noise. But the same blanket filter also hid a genuine ask card whenever that run paused waiting on the human (waiting_input/waiting_takeover), since delegated work is the normal way bots get tasked in this app. The bot's sidebar status correctly showed waiting_input (that query has no trigger filter), while its own chat showed no run and no card at all, leaving it stuck forever with no visible way to unblock it. Carve out waiting_input/waiting_takeover in the run-selection queries (thread-target.ts), keep the message carrying an ask block through the peer-run message filter (thread-message-pages.ts), and do the same in the shared client/server message-visibility filter used by the web (and mobile) transcript. Verified live against a running instance: a stuck Comms bot's pending multiple-choice ask card now renders with working answer buttons. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019QKrK5ECpiQWEXaL4pikb5 --- apps/api/src/thread-message-pages.ts | 11 ++++++-- apps/api/src/thread-target.ts | 34 +++++++++++++++++++++++-- packages/core/src/message-visibility.ts | 7 ++++- 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/apps/api/src/thread-message-pages.ts b/apps/api/src/thread-message-pages.ts index c1e6a63de..6e3bcd264 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -106,10 +106,17 @@ 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 compact sent/received receipts; clients render them as chips. Also keep any + // ask card a peer-triggered run created — it needs the user's own answer (approval, + // secret, or choice) and has nowhere else to render. Hiding it here left bots that + // were delegated a task by another bot (the normal way work gets assigned) stuck in + // waiting_input forever with no visible way to unblock them. 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", ); }); } diff --git a/apps/api/src/thread-target.ts b/apps/api/src/thread-target.ts index 8974404a8..e80b1b92b 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) { @@ -330,8 +333,17 @@ export async function threadSnapshot( where: { botId: target.botId, threadId: target.threadId, - trigger: { not: "bot_message" }, status: { in: [...ACTIVE_RUN_STATUSES, "failed"] }, + // A run started by another bot's message_bot call is normally kept out of + // this thread's spinner/failure surface (see below) so background peer + // chatter doesn't bury what the human is looking at. But when that run is + // paused waiting on a human (waiting_input/waiting_takeover) it MUST still + // surface here — otherwise the ask card it created has nowhere to render + // and the bot sits stuck forever with no visible way to unblock it. + OR: [ + { trigger: { not: "bot_message" } }, + { status: { in: ["waiting_input", "waiting_takeover"] } }, + ], }, // The id tiebreak keeps ordering deterministic under equal // timestamps, matching the supersession probe below. @@ -396,8 +408,14 @@ export async function threadSnapshot( tx.run.findMany({ where: { threadId: target.threadId, - trigger: { not: "bot_message" }, status: { in: [...ACTIVE_RUN_STATUSES] }, + // See the matching comment in the bot-thread branch above: a bot_message-triggered + // run must still surface here once it's waiting on a human, or its ask card is + // invisible and the member is stuck with no way to unblock it. + OR: [ + { trigger: { not: "bot_message" } }, + { status: { in: ["waiting_input", "waiting_takeover"] } }, + ], }, orderBy: { createdAt: "desc" }, }), @@ -580,6 +598,12 @@ export async function sendThreadMessage( }, select: { id: true, taskId: true, status: true }, }); + if (active && !STEERABLE_RUN_STATUSES.has(active.status)) { + throw new ORPCError("CONFLICT", { + message: + "This bot has a pending approval waiting on you — resolve its ask card before sending a new message.", + }); + } if (active) { await tx.steeringMessage.create({ data: { @@ -691,6 +715,12 @@ export async function sendThreadMessage( const runs: Array<{ id: string; taskId: string; botId: string; status: string }> = []; for (const botId of targetBotIds) { const active = activeByBotId.get(botId); + if (active && !STEERABLE_RUN_STATUSES.has(active.status)) { + throw new ORPCError("CONFLICT", { + message: + "This bot has a pending approval waiting on you — resolve its ask card before sending a new message.", + }); + } if (active) { await tx.steeringMessage.create({ data: { messageId: message.id, botId, userId: actor.userId, runId: active.id }, diff --git a/packages/core/src/message-visibility.ts b/packages/core/src/message-visibility.ts index 8fbd03374..a72d12fc2 100644 --- a/packages/core/src/message-visibility.ts +++ b/packages/core/src/message-visibility.ts @@ -36,6 +36,11 @@ 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; + // A run started by another bot's message_bot call can still pause for a human + // answer (approval, secret, or multiple-choice ask) partway through. That ask + // card has nowhere else to render, so it must survive this filter even though + // the rest of that peer run's activity stays hidden from the transcript. + return message.blocks.some((block) => block.kind === "ask"); }); } From de4cb0d5e00b1bd2221dd316c63414776563ebb0 Mon Sep 17 00:00:00 2001 From: Joseph Harper <8pgg4z5yvz@privaterelay.appleid.com> Date: Thu, 3 Sep 2026 02:01:15 +0000 Subject: [PATCH 02/22] chore(deps): bump @composio/core to 0.18.0 Also keeps the local dev-script DNS fix (NODE_OPTIONS=--dns-result-order=ipv4first) that was already carried as an uncommitted change. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019QKrK5ECpiQWEXaL4pikb5 --- package.json | 2 +- packages/adapters/package.json | 2 +- pnpm-lock.yaml | 21 ++++++++++++--------- 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/package.json b/package.json index a22939cd0..ea6b58ef9 100644 --- a/package.json +++ b/package.json @@ -9,7 +9,7 @@ "node": "^22.22.2 || ^24.0.0 || >=26.0.0" }, "scripts": { - "dev": "cross-env COREPACK_ENABLE_DOWNLOAD_PROMPT=0 turbo dev --filter=@rakazo/api --filter=@rakazo/worker --filter=@rakazo/web --filter=@rakazo/sandbox-supervisor", + "dev": "cross-env COREPACK_ENABLE_DOWNLOAD_PROMPT=0 NODE_OPTIONS=--dns-result-order=ipv4first turbo dev --filter=@rakazo/api --filter=@rakazo/worker --filter=@rakazo/web --filter=@rakazo/sandbox-supervisor", "build": "turbo build", "check": "turbo check", "lint": "biome check .", diff --git a/packages/adapters/package.json b/packages/adapters/package.json index 833c4643e..dd69b919c 100644 --- a/packages/adapters/package.json +++ b/packages/adapters/package.json @@ -16,7 +16,7 @@ "@chat-adapter/state-memory": "4.39.0", "@chat-adapter/telegram": "4.39.0", "@chat-adapter/whatsapp": "4.39.0", - "@composio/core": "0.16.0", + "@composio/core": "0.18.0", "@daytona/sdk": "^0.204.1", "@e2b/desktop": "^2.3.1", "@earendil-works/pi-agent-core": "^0.84.3", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fb1508607..5d75fe8f4 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -501,8 +501,8 @@ importers: specifier: 4.39.0 version: 4.39.0(zod@4.4.3) '@composio/core': - specifier: 0.16.0 - version: 0.16.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3) + specifier: 0.18.0 + version: 0.18.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3) '@daytona/sdk': specifier: ^0.204.1 version: 0.204.1 @@ -1737,13 +1737,15 @@ packages: '@composio/client@0.1.0-alpha.76': resolution: {integrity: sha512-MXC5JGRVdiQ4EgLricy9o/mqBa1+1T7wHFZ6Q4ZJkrjzZqOMvxTgy21Zlb5J/1oGkB2bg9UDzpH8PkXCp/D4zA==} - '@composio/core@0.16.0': - resolution: {integrity: sha512-m4LEjOaH4LdhP3rz3NyNXZwh3AI915morLV1vrjscW2ACGqL+fTHACiuem0sz45tPoW5DkPYP8k/Z5D4Zk3KLA==} + '@composio/core@0.18.0': + resolution: {integrity: sha512-u8dfSDJV+BYiQYJBxbk6xRI8o8f7O6nb8xE0rJmvC+8zKNJdY/M119R8a30grLh07a6fcLeVFYCJlTRpCjwKyg==} + engines: {node: '>=22.22.3'} peerDependencies: zod: '>=3.25.76 <5' - '@composio/json-schema-to-zod@0.3.0': - resolution: {integrity: sha512-Lgz5QlclPsd9mo8FGn/12h3cwTpN6A9aYvucO1umjnuvyetb6nC//nHEu9TDU+BN5Sa0n+qmUUv4Tzd4wI4xeA==} + '@composio/json-schema-to-zod@0.3.1': + resolution: {integrity: sha512-ZmDksfKZrkJZj6XCgR+AmjsxqvFnfc4XDEQPfJJZMYxIqhE3MBZ2KimLuPT+YGahO4h9uzT2EYi9asMF4qR4+g==} + engines: {node: '>=22.22.3'} peerDependencies: zod: '>=3.25.76 <5' @@ -11024,16 +11026,17 @@ snapshots: '@composio/client@0.1.0-alpha.76': {} - '@composio/core@0.16.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3)': + '@composio/core@0.18.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3)': dependencies: '@composio/client': 0.1.0-alpha.76 - '@composio/json-schema-to-zod': 0.3.0(zod@4.4.3) + '@composio/json-schema-to-zod': 0.3.1(zod@4.4.3) '@types/json-schema': 7.0.15 is-fs-case-sensitive: 2.0.0 openai: 7.4.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3) picocolors: 1.1.1 pusher-js: 8.6.0 semver: 7.8.5 + undici: 7.29.0 zod: 4.4.3 zod-to-json-schema: 3.25.2(zod@4.4.3) transitivePeerDependencies: @@ -11042,7 +11045,7 @@ snapshots: - '@smithy/signature-v4' - ws - '@composio/json-schema-to-zod@0.3.0(zod@4.4.3)': + '@composio/json-schema-to-zod@0.3.1(zod@4.4.3)': dependencies: zod: 4.4.3 From 7e8c393ee33556b77899acd2ebadbfefa81d03de Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 3 Sep 2026 18:37:50 +0000 Subject: [PATCH 03/22] fix(chat): tighten ask-card visibility follow-up Revert the out-of-scope Composio and DNS changes. Prefer waiting asks in bot snapshots, keep short CONFLICT copy when a run is waiting, and cover peer ask visibility plus the send conflict in unit tests. Co-authored-by: Elie Steinbock --- apps/api/src/thread-message-pages.ts | 6 +- apps/api/src/thread-target.test.ts | 274 +++++++++++++++++-- apps/api/src/thread-target.ts | 54 ++-- package.json | 2 +- packages/adapters/package.json | 2 +- packages/core/src/message-visibility.test.ts | 20 ++ packages/core/src/message-visibility.ts | 5 +- pnpm-lock.yaml | 21 +- 8 files changed, 306 insertions(+), 78 deletions(-) diff --git a/apps/api/src/thread-message-pages.ts b/apps/api/src/thread-message-pages.ts index 6e3bcd264..88aad2440 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -106,11 +106,7 @@ 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. Also keep any - // ask card a peer-triggered run created — it needs the user's own answer (approval, - // secret, or choice) and has nowhere else to render. Hiding it here left bots that - // were delegated a task by another bot (the normal way work gets assigned) stuck in - // waiting_input forever with no visible way to unblock them. + // Keep peer receipts (chips) and ask cards that need a human answer. const blocks = row.blocks as MessageBlock[]; return blocks.some( (block) => diff --git a/apps/api/src/thread-target.test.ts b/apps/api/src/thread-target.test.ts index a696f5779..ff4978cdc 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 e80b1b92b..576df0fea 100644 --- a/apps/api/src/thread-target.ts +++ b/apps/api/src/thread-target.ts @@ -322,34 +322,36 @@ 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"] }, - // A run started by another bot's message_bot call is normally kept out of - // this thread's spinner/failure surface (see below) so background peer - // chatter doesn't bury what the human is looking at. But when that run is - // paused waiting on a human (waiting_input/waiting_takeover) it MUST still - // surface here — otherwise the ask card it created has nowhere to render - // and the bot sits stuck forever with no visible way to unblock it. - OR: [ - { trigger: { not: "bot_message" } }, - { status: { in: ["waiting_input", "waiting_takeover"] } }, - ], }, // The id tiebreak keeps ordering deterministic under equal // timestamps, matching the supersession probe below. 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 @@ -362,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"] }, }, @@ -409,9 +411,7 @@ export async function threadSnapshot( where: { threadId: target.threadId, status: { in: [...ACTIVE_RUN_STATUSES] }, - // See the matching comment in the bot-thread branch above: a bot_message-triggered - // run must still surface here once it's waiting on a human, or its ask card is - // invisible and the member is stuck with no way to unblock it. + // Include waiting peer runs so their ask cards stay answerable. OR: [ { trigger: { not: "bot_message" } }, { status: { in: ["waiting_input", "waiting_takeover"] } }, @@ -590,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, @@ -598,12 +598,12 @@ export async function sendThreadMessage( }, select: { id: true, taskId: true, status: true }, }); - if (active && !STEERABLE_RUN_STATUSES.has(active.status)) { + if (activeRuns.some((run) => !STEERABLE_RUN_STATUSES.has(run.status))) { throw new ORPCError("CONFLICT", { - message: - "This bot has a pending approval waiting on you — resolve its ask card before sending a new message.", + message: "Answer the pending ask first.", }); } + const active = activeRuns[0]; if (active) { await tx.steeringMessage.create({ data: { @@ -711,16 +711,18 @@ export async function sendThreadMessage( }, select: { id: true, taskId: true, botId: true, status: true }, }); - const activeByBotId = new Map(activeRuns.map((run) => [run.botId, run])); - const runs: Array<{ id: string; taskId: string; botId: string; status: string }> = []; - for (const botId of targetBotIds) { - const active = activeByBotId.get(botId); - if (active && !STEERABLE_RUN_STATUSES.has(active.status)) { + const activeByBotId = new Map(); + for (const run of activeRuns) { + if (!STEERABLE_RUN_STATUSES.has(run.status)) { throw new ORPCError("CONFLICT", { - message: - "This bot has a pending approval waiting on you — resolve its ask card before sending a new message.", + 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); if (active) { await tx.steeringMessage.create({ data: { messageId: message.id, botId, userId: actor.userId, runId: active.id }, diff --git a/package.json b/package.json index ea6b58ef9..a22939cd0 100644 --- a/package.json +++ b/package.json @@ -9,7 +9,7 @@ "node": "^22.22.2 || ^24.0.0 || >=26.0.0" }, "scripts": { - "dev": "cross-env COREPACK_ENABLE_DOWNLOAD_PROMPT=0 NODE_OPTIONS=--dns-result-order=ipv4first turbo dev --filter=@rakazo/api --filter=@rakazo/worker --filter=@rakazo/web --filter=@rakazo/sandbox-supervisor", + "dev": "cross-env COREPACK_ENABLE_DOWNLOAD_PROMPT=0 turbo dev --filter=@rakazo/api --filter=@rakazo/worker --filter=@rakazo/web --filter=@rakazo/sandbox-supervisor", "build": "turbo build", "check": "turbo check", "lint": "biome check .", diff --git a/packages/adapters/package.json b/packages/adapters/package.json index dd69b919c..833c4643e 100644 --- a/packages/adapters/package.json +++ b/packages/adapters/package.json @@ -16,7 +16,7 @@ "@chat-adapter/state-memory": "4.39.0", "@chat-adapter/telegram": "4.39.0", "@chat-adapter/whatsapp": "4.39.0", - "@composio/core": "0.18.0", + "@composio/core": "0.16.0", "@daytona/sdk": "^0.204.1", "@e2b/desktop": "^2.3.1", "@earendil-works/pi-agent-core": "^0.84.3", diff --git a/packages/core/src/message-visibility.test.ts b/packages/core/src/message-visibility.test.ts index 858b928bd..aae609587 100644 --- a/packages/core/src/message-visibility.test.ts +++ b/packages/core/src/message-visibility.test.ts @@ -53,4 +53,24 @@ describe("user-visible messages", () => { userVisibleMessages(messages, { knownPeerRunIds: ["run-peer"] }).map((item) => item.id), ).toEqual(["answer"]); }); + + it("keeps a peer-run ask card visible 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", "answer"]); + }); }); diff --git a/packages/core/src/message-visibility.ts b/packages/core/src/message-visibility.ts index a72d12fc2..365bdcf7f 100644 --- a/packages/core/src/message-visibility.ts +++ b/packages/core/src/message-visibility.ts @@ -37,10 +37,7 @@ export function userVisibleMessages( return messages.filter((message) => { if (isPeerReceiptBlocks(message.blocks)) return includePeerReceipts; if (!message.runId || !peerRunIds.has(message.runId)) return true; - // A run started by another bot's message_bot call can still pause for a human - // answer (approval, secret, or multiple-choice ask) partway through. That ask - // card has nowhere else to render, so it must survive this filter even though - // the rest of that peer run's activity stays hidden from the transcript. + // Keep peer-run ask cards so a waiting human answer stays visible. return message.blocks.some((block) => block.kind === "ask"); }); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5d75fe8f4..fb1508607 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -501,8 +501,8 @@ importers: specifier: 4.39.0 version: 4.39.0(zod@4.4.3) '@composio/core': - specifier: 0.18.0 - version: 0.18.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3) + specifier: 0.16.0 + version: 0.16.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3) '@daytona/sdk': specifier: ^0.204.1 version: 0.204.1 @@ -1737,15 +1737,13 @@ packages: '@composio/client@0.1.0-alpha.76': resolution: {integrity: sha512-MXC5JGRVdiQ4EgLricy9o/mqBa1+1T7wHFZ6Q4ZJkrjzZqOMvxTgy21Zlb5J/1oGkB2bg9UDzpH8PkXCp/D4zA==} - '@composio/core@0.18.0': - resolution: {integrity: sha512-u8dfSDJV+BYiQYJBxbk6xRI8o8f7O6nb8xE0rJmvC+8zKNJdY/M119R8a30grLh07a6fcLeVFYCJlTRpCjwKyg==} - engines: {node: '>=22.22.3'} + '@composio/core@0.16.0': + resolution: {integrity: sha512-m4LEjOaH4LdhP3rz3NyNXZwh3AI915morLV1vrjscW2ACGqL+fTHACiuem0sz45tPoW5DkPYP8k/Z5D4Zk3KLA==} peerDependencies: zod: '>=3.25.76 <5' - '@composio/json-schema-to-zod@0.3.1': - resolution: {integrity: sha512-ZmDksfKZrkJZj6XCgR+AmjsxqvFnfc4XDEQPfJJZMYxIqhE3MBZ2KimLuPT+YGahO4h9uzT2EYi9asMF4qR4+g==} - engines: {node: '>=22.22.3'} + '@composio/json-schema-to-zod@0.3.0': + resolution: {integrity: sha512-Lgz5QlclPsd9mo8FGn/12h3cwTpN6A9aYvucO1umjnuvyetb6nC//nHEu9TDU+BN5Sa0n+qmUUv4Tzd4wI4xeA==} peerDependencies: zod: '>=3.25.76 <5' @@ -11026,17 +11024,16 @@ snapshots: '@composio/client@0.1.0-alpha.76': {} - '@composio/core@0.18.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3)': + '@composio/core@0.16.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3)': dependencies: '@composio/client': 0.1.0-alpha.76 - '@composio/json-schema-to-zod': 0.3.1(zod@4.4.3) + '@composio/json-schema-to-zod': 0.3.0(zod@4.4.3) '@types/json-schema': 7.0.15 is-fs-case-sensitive: 2.0.0 openai: 7.4.0(@aws-sdk/credential-provider-node@3.972.80)(@smithy/signature-v4@5.7.0)(ws@8.21.3)(zod@4.4.3) picocolors: 1.1.1 pusher-js: 8.6.0 semver: 7.8.5 - undici: 7.29.0 zod: 4.4.3 zod-to-json-schema: 3.25.2(zod@4.4.3) transitivePeerDependencies: @@ -11045,7 +11042,7 @@ snapshots: - '@smithy/signature-v4' - ws - '@composio/json-schema-to-zod@0.3.1(zod@4.4.3)': + '@composio/json-schema-to-zod@0.3.0(zod@4.4.3)': dependencies: zod: 4.4.3 From a4ef816935c874749326ad57bfbc595587540caf Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 3 Sep 2026 18:39:48 +0000 Subject: [PATCH 04/22] fix(chat): forward peer waiting and ask events on subscribe Open threads dropped peer-run waiting_input and ask message events, so a delegated ask that paused while the chat was already open stayed invisible until an unrelated refresh. Co-authored-by: Elie Steinbock --- apps/api/src/router.ts | 27 +++++-------------- apps/api/src/thread-message-pages.test.ts | 32 ++++++++++++++++++++++- apps/api/src/thread-message-pages.ts | 32 +++++++++++++++++++++++ 3 files changed, 70 insertions(+), 21 deletions(-) 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..df0161a2e 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 and ask 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(false); + 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 () => [ { diff --git a/apps/api/src/thread-message-pages.ts b/apps/api/src/thread-message-pages.ts index 88aad2440..350453edc 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -133,6 +133,38 @@ export async function isPeerRun( return peerRun; } +/** Peer-run SSE events that must still reach an open thread (terminals, waits, receipts, asks). */ +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"), + ) + ); +} + function toThreadMessage(row: { id: string; threadId: string; From d86c86696f2810afc973c1be1af205d5c6b37a0e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 3 Sep 2026 18:59:35 +0000 Subject: [PATCH 05/22] style: format sendThreadMessage waiting CONFLICT test Co-authored-by: Elie Steinbock --- apps/api/src/thread-target.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/api/src/thread-target.test.ts b/apps/api/src/thread-target.test.ts index ff4978cdc..85dd736a7 100644 --- a/apps/api/src/thread-target.test.ts +++ b/apps/api/src/thread-target.test.ts @@ -918,9 +918,9 @@ describe("sendThreadMessage", () => { update: vi.fn(), }, run: { - findMany: vi.fn().mockResolvedValue([ - { id: "run-waiting", taskId: "task-1", status: "waiting_input" }, - ]), + findMany: vi + .fn() + .mockResolvedValue([{ id: "run-waiting", taskId: "task-1", status: "waiting_input" }]), }, steeringMessage: { create: vi.fn() }, event: { create: vi.fn() }, From 69ee84297a9cb709d7770c89bc5d4525c931c6bd Mon Sep 17 00:00:00 2001 From: Joseph Harper <8pgg4z5yvz@privaterelay.appleid.com> Date: Fri, 4 Sep 2026 00:50:13 +0000 Subject: [PATCH 06/22] fix(chat): surface a delegating bot's own reply to the user MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run started by another bot's message_bot call (trigger: "bot_message") has its activity hidden from the thread transcript except for receipt chips and, since the previous fix, ask cards. But that same run is also how the delegating bot answers the human: the delegate wake prompt tells it to "summarize this result to the user now" as part of that turn, and that reply is a plain `text` block in the same run — so it was caught by the identical blanket filter and silently dropped, even though it saved to the database correctly. The delegating bot looked like it never responded. Extend the allowlist in both the shared client/server filter (message-visibility.ts) and its server-side mirror (thread-message-pages.ts) to also keep `kind: "text"` blocks. Verified live against a running instance: an already-saved reply that a coordinator bot had sent after delegating to a teammate now renders in the thread without needing a new run. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01YCwT37ZV6mCY6BzSpjgcKi --- apps/api/src/thread-message-pages.ts | 8 +++++++- packages/core/src/message-visibility.ts | 9 ++++++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/apps/api/src/thread-message-pages.ts b/apps/api/src/thread-message-pages.ts index 6e3bcd264..3ef849934 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -111,12 +111,18 @@ async function withoutPeerRunMessages block.kind === "bot_message_sent" || block.kind === "bot_message_received" || - block.kind === "ask", + block.kind === "ask" || + block.kind === "text", ); }); } diff --git a/packages/core/src/message-visibility.ts b/packages/core/src/message-visibility.ts index a72d12fc2..03db5448b 100644 --- a/packages/core/src/message-visibility.ts +++ b/packages/core/src/message-visibility.ts @@ -41,6 +41,13 @@ export function userVisibleMessages( // answer (approval, secret, or multiple-choice ask) partway through. That ask // card has nowhere else to render, so it must survive this filter even though // the rest of that peer run's activity stays hidden from the transcript. - return message.blocks.some((block) => block.kind === "ask"); + // + // The same run also ends its turn by writing the bot's own reply to the human + // (the delegate wake prompt tells it to "summarize this result to the user + // now"). That reply is a plain `text` block in the same run/thread as the + // `bot_message_received` row, so without this it was caught by the same + // blanket hide and silently dropped: the delegating bot looked like it never + // answered, even though it did and the message was saved correctly. + return message.blocks.some((block) => block.kind === "ask" || block.kind === "text"); }); } From 8272895b41ca5f9d254cf556bb2871317d454e00 Mon Sep 17 00:00:00 2001 From: Joseph Harper <8pgg4z5yvz@privaterelay.appleid.com> Date: Fri, 4 Sep 2026 02:16:20 +0000 Subject: [PATCH 07/22] fix(messaging): poll Telegram from the process that owns the inbound handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telegram was hardcoded to webhook mode, which requires a public HTTPS endpoint for Telegram to push to. Local/self-hosted deployments have none (api is bound to 127.0.0.1 with no tunnel), so linking a bot or messaging it got no response at all. Switch to "auto" mode: the adapter checks getWebhookInfo and falls back to long-polling getUpdates when no webhook is registered, which is exactly what a local deployment needs. Auto/polling mode only starts once something calls initialize() proactively, so add that as an optional method on MessagingSurface / ChatSdkMessagingSurface. That call has to happen in apps/api/src/app.ts specifically: that's the process that registers the inbound sink (createMessagingInboundHandler / messaging.onInbound), and Telegram allows only one live getUpdates connection per bot token. An earlier attempt called initialize() from the worker instead (it also constructs its own ChatSdkMessagingSurface, for outbound delivery jobs) — that poller successfully held Telegram's single getUpdates slot with no sink registered, so every inbound message vanished silently. Confirmed via a manual getUpdates call from inside the worker container returning "409 Conflict: terminated by other getUpdates request" — proof a poller was live there with nothing listening. Verified live: linking a Telegram bot and messaging it now completes end-to-end. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01YCwT37ZV6mCY6BzSpjgcKi --- apps/api/src/app.ts | 10 ++++++++++ packages/adapter-kit/src/interfaces.ts | 9 +++++++++ packages/adapters/src/chat-sdk-surface.ts | 12 ++++++++++++ packages/adapters/src/messaging-platforms.ts | 15 ++++++++++++--- 4 files changed, 43 insertions(+), 3 deletions(-) diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 678d010bd..701f30932 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -429,6 +429,16 @@ 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. + void messaging.initialize?.().catch((error) => { + console.error("messaging surface initialize failed", error); + }); } app.get("/health", (c) => diff --git a/packages/adapter-kit/src/interfaces.ts b/packages/adapter-kit/src/interfaces.ts index bc58fbece..58771d612 100644 --- a/packages/adapter-kit/src/interfaces.ts +++ b/packages/adapter-kit/src/interfaces.ts @@ -323,6 +323,15 @@ 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; } /** diff --git a/packages/adapters/src/chat-sdk-surface.ts b/packages/adapters/src/chat-sdk-surface.ts index b4b54f6ed..c595e16fe 100644 --- a/packages/adapters/src/chat-sdk-surface.ts +++ b/packages/adapters/src/chat-sdk-surface.ts @@ -154,6 +154,18 @@ 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(); + } + private ensureInitialized(): Promise { // Webhook handling initializes lazily inside the Chat SDK; proactive // sends from job runners need the explicit call. diff --git a/packages/adapters/src/messaging-platforms.ts b/packages/adapters/src/messaging-platforms.ts index 5ad698e8e..3ae3184ab 100644 --- a/packages/adapters/src/messaging-platforms.ts +++ b/packages/adapters/src/messaging-platforms.ts @@ -122,12 +122,21 @@ export function messagingPlatformsFromEnv(env: MessagingEnvironmentValues): Mess platforms.push({ provider: "telegram", capabilities: { direct: true, groups: false, typing: false }, - // Webhook-only: auto mode can long-poll getUpdates from the worker on - // initialize() and consume updates so the HTTP webhook never sees them. + // Auto mode: uses the webhook route when Telegram has one registered + // (checked via getWebhookInfo), and otherwise falls back to + // long-polling getUpdates. Self-hosted/local deployments typically + // have no public HTTPS endpoint for Telegram to push to, so the API + // process calls initialize() at startup (apps/api/src/app.ts) to + // start that polling loop immediately rather than waiting for the + // first inbound webhook or outbound send. It must be the API + // process specifically: that's where the inbound sink is registered, + // and Telegram allows only one live getUpdates connection per bot — + // a second poller elsewhere would just steal that slot and drop + // every message into the void. adapter: createTelegramAdapter({ botToken: env.telegramBotToken, secretToken: env.telegramWebhookSecret, - mode: "webhook", + mode: "auto", }), }); } From 8017c3a9594eb877a4a84646f51a35ce0e9206e4 Mon Sep 17 00:00:00 2001 From: Joseph Harper <8pgg4z5yvz@privaterelay.appleid.com> Date: Fri, 4 Sep 2026 02:16:27 +0000 Subject: [PATCH 08/22] fix(messaging): mirror a delegated bot's reply out to the linked chat app MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit deliverMessagingOutbound only mirrored a run's text replies to the vendor (Telegram/Slack/etc.) when the run's trigger was "messaging" — i.e. a reply generated directly from an inbound vendor message. But when a bot delegates work via message_bot, its own summary back to the human is generated in the run that receives the delegate's result, which is trigger "bot_message" — the same shape of run apps/web's message-visibility fix already had to special-case for in-app rendering. The reply saved to the database correctly and even rendered in the web UI, but never reached the vendor at all. Mirror a bot_message-triggered run's text blocks too, but only when the run landed on that bot's actual linked DM thread (identity.dmThreadId) — otherwise it's inter-bot chatter happening on a different bot's private thread, which must never leak out to the vendor. extractText only ever pulls `kind: "text"` blocks (never bot_message_sent/received), so this can't leak delegation traffic even on the DM thread itself. Verified live: asking a bot to delegate a task (e.g. a calendar lookup) now delivers the delegating bot's reply over Telegram, not just in-app. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01YCwT37ZV6mCY6BzSpjgcKi --- packages/adapters/src/messaging-delivery.ts | 33 +++++++++++++++------ 1 file changed, 24 insertions(+), 9 deletions(-) diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index 3edc85977..41821a452 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -71,15 +71,18 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise => - block.kind === "channel_message", - ); - if (channelBlock) { - await mirrorChannelRun(deps, run, channelBlock); - return; + 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; + } } const identity = await deps.prisma.messagingIdentity.findUnique({ @@ -87,6 +90,18 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise Date: Fri, 4 Sep 2026 03:23:12 +0000 Subject: [PATCH 09/22] fix(messaging): correct the dmThreadId check that blocked delegated replies from mirroring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous fix (8017c3a) gated mirroring of a bot_message-triggered reply on run.threadId === identity.dmThreadId, but dmThreadId is the vendor's own opaque chat id (learned on inbound), not Rakazo's internal Thread.id — these two ID namespaces can never be equal, so the check always failed and delegated final replies never reached the vendor, even though interim messaging-triggered replies worked fine. Removed the check: it's unnecessary as well as broken, since Thread.botId is @unique (each bot has at most one thread), and extractText() already only pulls kind: "text" blocks, which is sufficient on its own to keep inter-bot delegation chatter from leaking to the vendor. Confirmed live: Commander's delegated calendar-summary reply now reaches Telegram verbatim, matching what's shown in the web UI. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01YCwT37ZV6mCY6BzSpjgcKi --- packages/adapters/src/messaging-delivery.ts | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index 41821a452..000248678 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -94,13 +94,15 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise Date: Fri, 4 Sep 2026 03:23:19 +0000 Subject: [PATCH 10/22] fix(messaging): stop worker from also polling Telegram MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Once the previous fix started mirroring more replies out through the worker's own outbound delivery path, a lazy ensureInitialized() call on its Telegram adapter (mode: "auto", no webhook registered) started a second getUpdates poller — competing with api's poller for Telegram's single allowed polling connection per bot token. Both sides spent every cycle losing a 409 Conflict to the other and inbound messages stopped arriving. Added a pollInboundMessages option to messagingPlatformsFromEnv: Telegram now defaults to passive "webhook" mode (resolves identity, sends fine, never polls) unless the caller opts in. Only apps/api/src/app.ts opts in, since it's the process that registers the inbound sink; apps/worker stays passive. Confirmed live: Telegram receiving works again with no 409s in either process's logs. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01YCwT37ZV6mCY6BzSpjgcKi --- apps/api/src/app.ts | 6 +++++- apps/worker/src/index.ts | 4 ++++ packages/adapters/src/messaging-platforms.ts | 22 ++++++++++++++++++-- 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 701f30932..75f2e3a3e 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, { 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/adapters/src/messaging-platforms.ts b/packages/adapters/src/messaging-platforms.ts index 3ae3184ab..5816db843 100644 --- a/packages/adapters/src/messaging-platforms.ts +++ b/packages/adapters/src/messaging-platforms.ts @@ -51,8 +51,26 @@ export function messagingEnvFromProcess( * Build the platform list for every fully configured provider. Group * conversations stay sendblue-only until channel semantics are mapped for * the other platforms, so their capabilities say so instead of half-working. + * + * `pollInboundMessages` must be true only in the one process that also + * registers the inbound sink (messaging.onInbound — apps/api/src/app.ts). + * Telegram's "auto" mode starts a long-poll the moment anything calls + * chat.initialize() when no webhook is registered — and that includes a + * process that only ever meant to *send*: outbound delivery + * (sendToThread) lazily initializes too. A second process polling with no + * inbound sink attached doesn't just do nothing — it actively steals + * Telegram's single getUpdates slot away from the process that IS + * listening, so both sides spend every cycle losing a 409 Conflict to the + * other and messages stop arriving at all. Any caller that only sends + * (e.g. apps/worker/src/index.ts, for messaging.deliver jobs) must leave + * this false so Telegram mode resolves to "webhook" (passive — resolves + * bot identity for outbound calls, never polls, and no webhook route is + * mounted there for it to receive on anyway). */ -export function messagingPlatformsFromEnv(env: MessagingEnvironmentValues): MessagingPlatform[] { +export function messagingPlatformsFromEnv( + env: MessagingEnvironmentValues, + options: { pollInboundMessages?: boolean } = {}, +): MessagingPlatform[] { const platforms: MessagingPlatform[] = []; if ( @@ -136,7 +154,7 @@ export function messagingPlatformsFromEnv(env: MessagingEnvironmentValues): Mess adapter: createTelegramAdapter({ botToken: env.telegramBotToken, secretToken: env.telegramWebhookSecret, - mode: "auto", + mode: options.pollInboundMessages ? "auto" : "webhook", }), }); } From 1e2c010ce934047b0a6979e467f064c71d8a9e28 Mon Sep 17 00:00:00 2001 From: Joseph Harper <8pgg4z5yvz@privaterelay.appleid.com> Date: Fri, 4 Sep 2026 03:23:24 +0000 Subject: [PATCH 11/22] fix(chat): make the delegated-result wake prompt explicit about relaying content MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A delegating bot's wake prompt for a result/status intent already said "do not merely acknowledge it," but the model sometimes replied with a vague confirmation ("the summary came through") without restating any of the actual content it had just received, even though that content was right there in its context. Reworded the instruction to be explicit and give a concrete bad example: it must include the real names, dates, numbers, and details the sender sent, not just a note that a result arrived. This is a prompt-wording mitigation, not a hard guarantee — it depends on the model following instructions — but confirmed live it resolved the reported case. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01YCwT37ZV6mCY6BzSpjgcKi --- packages/core/src/bot-messages.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/core/src/bot-messages.ts b/packages/core/src/bot-messages.ts index 1ba9f8093..976768452 100644 --- a/packages/core/src/bot-messages.ts +++ b/packages/core/src/bot-messages.ts @@ -161,7 +161,7 @@ export function buildBotMessageWakePrompt(args: { const intent = args.intent ?? "request"; const action = intent === "result" || intent === "status" - ? `This is a ${intent} for work you delegated. Concisely summarize this result to the user now. Do not stay silent and do not merely acknowledge it.` + ? `This is a ${intent} for work you delegated. Relay it to the user now, and include the actual substance — the real names, dates, numbers, and details ${safeName} sent — not just a note that a ${intent} arrived. A reply like "the summary came through" or "it's done" without repeating what it says is not acceptable. Do not stay silent and do not merely acknowledge it.` : intent === "question" ? `This is a question about delegated work. Answer it if you can, then continue the coordination and keep the user informed.` : intent === "fyi" From 9adf510e0c6d6227ef6b2f4760b79576c01c4467 Mon Sep 17 00:00:00 2001 From: Joseph Harper <8pgg4z5yvz@privaterelay.appleid.com> Date: Fri, 4 Sep 2026 03:23:30 +0000 Subject: [PATCH 12/22] fix(chat): warn message_bot's own description against narrating instead of calling it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Observed a bot reply to the user consisting of a single line describing a delegation it never actually made ("Assistant: [to Comms] Please send..."), with no message_bot tool invocation anywhere in that run's steps — confirmed via the run's message blocks, which had no "Message bot" step, unlike every working delegation in the same conversation. The recipient bot never received anything. Working theory: after several rounds of near-identical delegation in one thread, the model pattern-matched onto the "[to X] " notation blocksToAgentHistoryText uses to render its own past bot_message_sent blocks back into context, and reproduced that notation as plain reply text instead of invoking the tool. Strengthened the message_bot tool's own description to explicitly warn against this: writing the message in reply text does not send it, and the recipient never sees it. Same caveat as the previous commit: this nudges the model, it does not guarantee compliance. Confirmed live it resolved the reported case. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01YCwT37ZV6mCY6BzSpjgcKi --- packages/adapters/src/builtin-tools.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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: { From b5982b5bd2f85817ec3f472398f9194fcd26898f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 03:35:56 +0000 Subject: [PATCH 13/22] fix(chat): align peer text visibility tests and DM mirror scope Keep peer-run text replies visible alongside ask cards, update page and visibility tests for that rule, and mirror only messaging and bot_message runs to linked DMs so in-app user runs are not pushed outbound. Co-authored-by: Elie Steinbock --- apps/api/src/thread-message-pages.test.ts | 30 +++++++++++++------ apps/api/src/thread-message-pages.ts | 11 +------ .../adapters/src/messaging-delivery.test.ts | 8 ++++- packages/adapters/src/messaging-delivery.ts | 19 ++++-------- packages/core/src/message-visibility.test.ts | 19 ++++++++---- packages/core/src/message-visibility.ts | 12 +------- 6 files changed, 49 insertions(+), 50 deletions(-) diff --git a/apps/api/src/thread-message-pages.test.ts b/apps/api/src/thread-message-pages.test.ts index df0161a2e..0d2471b38 100644 --- a/apps/api/src/thread-message-pages.test.ts +++ b/apps/api/src/thread-message-pages.test.ts @@ -96,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", @@ -134,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", @@ -182,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", @@ -243,6 +247,7 @@ describe("thread message pages", () => { expect(page.messages.map((message) => message.id)).toEqual([ "message-user", "message-peer-receipt", + "message-peer-text", ]); }); @@ -276,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, @@ -290,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 be3908b02..07bd4a330 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -106,16 +106,7 @@ 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. Also keep any - // ask card a peer-triggered run created — it needs the user's own answer (approval, - // secret, or choice) and has nowhere else to render. Hiding it here left bots that - // were delegated a task by another bot (the normal way work gets assigned) stuck in - // waiting_input forever with no visible way to unblock them. - // - // Also keep plain `text` replies: the same peer-triggered run is how the delegating - // bot writes its own answer back to the human (the delegate wake prompt tells it to - // summarize the result "to the user now"), and that reply was being caught by this - // same filter and silently dropped even though it saved correctly. + // Keep peer receipts (chips), ask cards, and the bot's own text reply. const blocks = row.blocks as MessageBlock[]; return blocks.some( (block) => diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index a22e231c3..2419a4815 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,12 @@ 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).toHaveBeenCalled(); + }); + 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 000248678..ab640db5f 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -83,6 +83,10 @@ 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 visible while hiding other peer activity", () => { + it("keeps a peer-run ask card and text reply while hiding other peer activity", () => { const messages = [ message("ask", "run-peer", [ { @@ -71,6 +78,6 @@ describe("user-visible messages", () => { expect( userVisibleMessages(messages, { knownPeerRunIds: ["run-peer"] }).map((item) => item.id), - ).toEqual(["ask", "answer"]); + ).toEqual(["ask", "reply", "answer"]); }); }); diff --git a/packages/core/src/message-visibility.ts b/packages/core/src/message-visibility.ts index 03db5448b..ce2a93475 100644 --- a/packages/core/src/message-visibility.ts +++ b/packages/core/src/message-visibility.ts @@ -37,17 +37,7 @@ export function userVisibleMessages( return messages.filter((message) => { if (isPeerReceiptBlocks(message.blocks)) return includePeerReceipts; if (!message.runId || !peerRunIds.has(message.runId)) return true; - // A run started by another bot's message_bot call can still pause for a human - // answer (approval, secret, or multiple-choice ask) partway through. That ask - // card has nowhere else to render, so it must survive this filter even though - // the rest of that peer run's activity stays hidden from the transcript. - // - // The same run also ends its turn by writing the bot's own reply to the human - // (the delegate wake prompt tells it to "summarize this result to the user - // now"). That reply is a plain `text` block in the same run/thread as the - // `bot_message_received` row, so without this it was caught by the same - // blanket hide and silently dropped: the delegating bot looked like it never - // answered, even though it did and the message was saved correctly. + // 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"); }); } From 39f8b2d0f1d501541b3aa465473df35ab6051bcc Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 03:37:40 +0000 Subject: [PATCH 14/22] fix(mobile): apply peer computer takeover as waiting_takeover Server already forwards computer.takeover.requested for peer runs; mobile was filtering it out and never flipping run status, so takeover UI never appeared. Match web event reduction. Co-authored-by: Elie Steinbock --- apps/mobile/app/thread.tsx | 1 + apps/mobile/lib/api.test.ts | 17 +++++++++++++++++ apps/mobile/lib/api.ts | 17 +++++++++++------ 3 files changed, 29 insertions(+), 6 deletions(-) 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..3397ebf8e 100644 --- a/apps/mobile/lib/api.test.ts +++ b/apps/mobile/lib/api.test.ts @@ -1062,6 +1062,23 @@ describe("mobile thread event reduction", () => { expect(repeated?.run).toBe(waiting?.run); }); + it("applies computer takeover requests as waiting_takeover", () => { + const initial: MobileSnapshot = { + ...snapshot(), + 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?.cursor).toBe(10); + }); + 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..207719273 100644 --- a/apps/mobile/lib/api.ts +++ b/apps/mobile/lib/api.ts @@ -652,24 +652,29 @@ 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); - const messages = prev.messages.filter((message) => message.id !== progressId); + // Waiting-input pauses drop live progress server-side; clear a leftover bubble. + const messages = + event.type === "run.waiting_input" + ? prev.messages.filter((message) => message.id !== progressId) + : prev.messages; const progressCleared = messages.length !== prev.messages.length; const runChanged = Boolean( - prev.run && prev.run.id === event.runId && prev.run.status !== "waiting_input", + prev.run && prev.run.id === event.runId && prev.run.status !== status, ); const activeRunChanged = prev.activeRuns?.some( - (candidate) => candidate.id === event.runId && candidate.status !== "waiting_input", + (candidate) => candidate.id === event.runId && candidate.status !== status, ); const cursor = event.seq ?? prev.cursor; if (!runChanged && !activeRunChanged && !progressCleared) { return cursor === prev.cursor ? prev : { ...prev, cursor }; } - const run = runChanged && prev.run ? { ...prev.run, status: "waiting_input" } : prev.run; + 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 === event.runId ? { ...candidate, status } : candidate, ) : prev.activeRuns; return { ...prev, cursor, run, activeRuns, messages }; From 17973adedcfab76f4483fb68cef848f124f78548 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 03:52:59 +0000 Subject: [PATCH 15/22] fix(messaging): retry failed init and stop Telegram polling on shutdown Chat caches a rejected initPromise until shutdown, so a transient Telegram startup failure permanently blocked retries. Clear that on failure, expose MessagingSurface.shutdown that calls stopPolling, and invoke it from API stop so a replacement process can reclaim getUpdates. Co-authored-by: Elie Steinbock --- apps/api/src/app.ts | 1 + packages/adapter-kit/src/interfaces.ts | 5 +++ .../adapters/src/chat-sdk-surface.test.ts | 19 +++++++++++ packages/adapters/src/chat-sdk-surface.ts | 32 +++++++++++++++++-- 4 files changed, 55 insertions(+), 2 deletions(-) diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 75f2e3a3e..e22675705 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -473,6 +473,7 @@ export async function createApp( executor, stop: async () => { oauthLogins.abortAll(); + await messaging?.shutdown?.(); await email?.drain?.(); await reconciler?.stop(); await jobs.close(); diff --git a/packages/adapter-kit/src/interfaces.ts b/packages/adapter-kit/src/interfaces.ts index 58771d612..946353706 100644 --- a/packages/adapter-kit/src/interfaces.ts +++ b/packages/adapter-kit/src/interfaces.ts @@ -332,6 +332,11 @@ export interface MessagingSurface { * 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/chat-sdk-surface.test.ts b/packages/adapters/src/chat-sdk-surface.test.ts index ffb18087b..9cf9650ab 100644 --- a/packages/adapters/src/chat-sdk-surface.test.ts +++ b/packages/adapters/src/chat-sdk-surface.test.ts @@ -376,4 +376,23 @@ 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/); + await expect(surface.initialize()).resolves.toBeUndefined(); + expect(adapter.initialize).toHaveBeenCalledTimes(2); + + await surface.shutdown(); + expect(stopPolling).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/adapters/src/chat-sdk-surface.ts b/packages/adapters/src/chat-sdk-surface.ts index c595e16fe..aa2a7364b 100644 --- a/packages/adapters/src/chat-sdk-surface.ts +++ b/packages/adapters/src/chat-sdk-surface.ts @@ -166,10 +166,38 @@ export class ChatSdkMessagingSurface implements MessagingSurface { 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); + for (const platform of this.byProvider.values()) { + const adapter = platform.adapter as { stopPolling?: () => Promise }; + await adapter.stopPolling?.().catch(() => undefined); + } + 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 { + 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; } From 3c4c397bc9de2297d422cd3bee6b1524448c3b1b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 03:57:33 +0000 Subject: [PATCH 16/22] fix(mobile): clear live progress on computer takeover waiting_takeover kept the progress bubble, which hid the waiting footer behind a stale Working row. Clear it the same way as waiting_input. Co-authored-by: Elie Steinbock --- apps/mobile/lib/api.test.ts | 4 +++- apps/mobile/lib/api.ts | 8 +++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/apps/mobile/lib/api.test.ts b/apps/mobile/lib/api.test.ts index 3397ebf8e..5d6ab453d 100644 --- a/apps/mobile/lib/api.test.ts +++ b/apps/mobile/lib/api.test.ts @@ -1063,8 +1063,9 @@ describe("mobile thread event reduction", () => { }); it("applies computer takeover requests as waiting_takeover", () => { + const progress = mobileMessage("progress:run-1", [{ kind: "progress", text: "working…" }], 1); const initial: MobileSnapshot = { - ...snapshot(), + ...snapshot([progress]), run: { id: "run-1", status: "running" }, activeRuns: [{ id: "run-1", status: "running" }], }; @@ -1076,6 +1077,7 @@ describe("mobile thread event reduction", () => { 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); }); diff --git a/apps/mobile/lib/api.ts b/apps/mobile/lib/api.ts index 207719273..8fab281a4 100644 --- a/apps/mobile/lib/api.ts +++ b/apps/mobile/lib/api.ts @@ -655,11 +655,9 @@ export function applyMobileThreadEvent( 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-input pauses drop live progress server-side; clear a leftover bubble. - const messages = - event.type === "run.waiting_input" - ? prev.messages.filter((message) => message.id !== progressId) - : prev.messages; + // 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 !== status, From a83f1daaa410949ae923c50ddd26b3a6de615e22 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 03:58:39 +0000 Subject: [PATCH 17/22] test(messaging): assert bot_message mirror destination and body Strengthen the delegated DM mirror test so it checks the linked thread and outbound text, not only that sendToThread was invoked. Co-authored-by: Elie Steinbock --- packages/adapters/src/messaging-delivery.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index 2419a4815..46900cded 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -263,7 +263,10 @@ describe("deliverMessagingOutbound", () => { 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).toHaveBeenCalled(); + 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 () => { From be9ee96f416e4dc4e739bead4f6d1471c8ebc1fa Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 04:03:34 +0000 Subject: [PATCH 18/22] fix(mobile): insert absent peer runs on takeover wait events Peer bot_message runs are hidden from snapshots while busy, so an open thread never had them in run/activeRuns when takeover arrived. Upsert the waiting run and clear busyBotName so takeover UI appears without a refresh. Co-authored-by: Elie Steinbock --- apps/mobile/lib/api.test.ts | 31 +++++++++++++++++++++++++ apps/mobile/lib/api.ts | 46 ++++++++++++++++++++++++++++++------- 2 files changed, 69 insertions(+), 8 deletions(-) diff --git a/apps/mobile/lib/api.test.ts b/apps/mobile/lib/api.test.ts index 5d6ab453d..48a3eb6e0 100644 --- a/apps/mobile/lib/api.test.ts +++ b/apps/mobile/lib/api.test.ts @@ -1081,6 +1081,37 @@ describe("mobile thread event reduction", () => { 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 8fab281a4..daf38b32f 100644 --- a/apps/mobile/lib/api.ts +++ b/apps/mobile/lib/api.ts @@ -659,23 +659,53 @@ export function applyMobileThreadEvent( // 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 !== status, - ); - const activeRunChanged = prev.activeRuns?.some( - (candidate) => candidate.id === event.runId && candidate.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)); + // 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 }; } + 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 } : 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); From f776b2cbe79a20ee97445692d97f9a189018db09 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 04:07:42 +0000 Subject: [PATCH 19/22] fix(api): forward peer text message events on open threads Snapshot pages already kept peer text replies, but the live peer-event allowlist dropped them, so delegated replies stayed hidden until refresh. Co-authored-by: Elie Steinbock --- apps/api/src/thread-message-pages.test.ts | 4 ++-- apps/api/src/thread-message-pages.ts | 5 +++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/apps/api/src/thread-message-pages.test.ts b/apps/api/src/thread-message-pages.test.ts index 0d2471b38..b735bfeed 100644 --- a/apps/api/src/thread-message-pages.test.ts +++ b/apps/api/src/thread-message-pages.test.ts @@ -18,7 +18,7 @@ describe("thread message pages", () => { expect(findUnique).toHaveBeenCalledTimes(1); }); - it("forwards peer waiting and ask events on an open thread", () => { + 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, @@ -34,7 +34,7 @@ describe("thread message pages", () => { type: "thread.message.created", payload: { blocks: [{ kind: "text", text: "peer body" }] }, }), - ).toBe(false); + ).toBe(true); expect( shouldForwardPeerThreadEvent({ type: "thread.progress", diff --git a/apps/api/src/thread-message-pages.ts b/apps/api/src/thread-message-pages.ts index 07bd4a330..c40d3bd48 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -134,7 +134,7 @@ export async function isPeerRun( return peerRun; } -/** Peer-run SSE events that must still reach an open thread (terminals, waits, receipts, asks). */ +/** 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 }; @@ -161,7 +161,8 @@ export function shouldForwardPeerThreadEvent(event: { "kind" in block && (block.kind === "bot_message_received" || block.kind === "bot_message_sent" || - block.kind === "ask"), + block.kind === "ask" || + block.kind === "text"), ) ); } From be352139d6be1e28cf5746b0517f39c3d1b2b647 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 04:15:07 +0000 Subject: [PATCH 20/22] fix(messaging): stop Telegram polling on init failure and retry startup Partial multi-adapter initialize can leave Telegram getUpdates running when Chat.shutdown only calls disconnect. Clear polling on failure, and retry API messaging init a few times so polling-only bots recover from transient startup errors without waiting for an outbound send. Co-authored-by: Elie Steinbock --- apps/api/src/app.ts | 29 +++++++++++++++++-- .../adapters/src/chat-sdk-surface.test.ts | 4 ++- packages/adapters/src/chat-sdk-surface.ts | 15 +++++++--- 3 files changed, 40 insertions(+), 8 deletions(-) diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index e22675705..ecf2ea0b5 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -400,6 +400,8 @@ 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; // Messaging webhooks only exist when the surface is enabled. if (messaging) { const inbound = createMessagingInboundHandler({ @@ -440,9 +442,29 @@ export async function createApp( // 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. - void messaging.initialize?.().catch((error) => { - console.error("messaging surface initialize failed", error); - }); + // Bounded retries cover transient Telegram startup failures; polling-only + // bots otherwise stay dark until an unrelated outbound send re-inits. + void (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) => setTimeout(resolve, delayMs[attempt])); + } + 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) => @@ -473,6 +495,7 @@ export async function createApp( executor, stop: async () => { oauthLogins.abortAll(); + messagingStopped = true; await messaging?.shutdown?.(); await email?.drain?.(); await reconciler?.stop(); diff --git a/packages/adapters/src/chat-sdk-surface.test.ts b/packages/adapters/src/chat-sdk-surface.test.ts index 9cf9650ab..4da7ed86d 100644 --- a/packages/adapters/src/chat-sdk-surface.test.ts +++ b/packages/adapters/src/chat-sdk-surface.test.ts @@ -389,10 +389,12 @@ describe("ChatSdkMessagingSurface shape", () => { } 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(1); + expect(stopPolling).toHaveBeenCalledTimes(2); }); }); diff --git a/packages/adapters/src/chat-sdk-surface.ts b/packages/adapters/src/chat-sdk-surface.ts index aa2a7364b..31d648164 100644 --- a/packages/adapters/src/chat-sdk-surface.ts +++ b/packages/adapters/src/chat-sdk-surface.ts @@ -174,10 +174,7 @@ export class ChatSdkMessagingSurface implements MessagingSurface { const pending = this.initialized; this.initialized = undefined; if (pending) await pending.catch(() => undefined); - for (const platform of this.byProvider.values()) { - const adapter = platform.adapter as { stopPolling?: () => Promise }; - await adapter.stopPolling?.().catch(() => undefined); - } + await this.stopPollingAdapters(); const chat = this.chat as { shutdown?: () => Promise }; await chat.shutdown?.().catch(() => undefined); } @@ -189,6 +186,9 @@ export class ChatSdkMessagingSurface implements MessagingSurface { 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. @@ -201,6 +201,13 @@ export class ChatSdkMessagingSurface implements MessagingSurface { 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) { From 9a24c383ff5712c857b89ad35ee98f71463782c7 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 04:27:08 +0000 Subject: [PATCH 21/22] fix(api): cancel messaging init retry timer on shutdown The startup retry setTimeout stayed referenced after stop(), which could keep the process alive until the delay fired. Clear the timer and await the retry task during shutdown. Co-authored-by: Elie Steinbock --- apps/api/src/app.ts | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index ecf2ea0b5..1f05a6356 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -402,6 +402,8 @@ 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({ @@ -444,12 +446,20 @@ export async function createApp( // 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. - void (async () => { + 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) => setTimeout(resolve, delayMs[attempt])); + await new Promise((resolve) => { + const timer = setTimeout(resolve, delayMs[attempt]); + clearMessagingRetryDelay = () => { + clearTimeout(timer); + clearMessagingRetryDelay = undefined; + resolve(); + }; + }); + clearMessagingRetryDelay = undefined; } if (messagingStopped) return; try { @@ -496,6 +506,8 @@ export async function createApp( stop: async () => { oauthLogins.abortAll(); messagingStopped = true; + clearMessagingRetryDelay?.(); + await messagingInitTask?.catch(() => undefined); await messaging?.shutdown?.(); await email?.drain?.(); await reconciler?.stop(); From 3211b6b11c75798107a5ec2b242bfe6599ab09f3 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 04:32:14 +0000 Subject: [PATCH 22/22] fix(web): insert absent peer runs on takeover wait events Peer bot_message runs are hidden from snapshots while busy, so an open web thread never had them in run/activeRuns when takeover arrived. Upsert the waiting run so takeover UI appears without a refresh. Co-authored-by: Elie Steinbock --- apps/web/src/lib/thread-events.test.ts | 30 ++++++++++++++++ apps/web/src/lib/thread-events.ts | 48 +++++++++++++++++++++++--- 2 files changed, 73 insertions(+), 5 deletions(-) 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, };