From 27fcf3a4719c12c47eaa34adeb4e8a7a4fa6aaaf Mon Sep 17 00:00:00 2001 From: Xialie Zhuang <62231346+Lieisyourlie@users.noreply.github.com> Date: Sat, 5 Sep 2026 10:19:41 +0800 Subject: [PATCH] fix(push): an agent's reply must reach the phone too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Push dispatch lived in exactly one place — the fire-and-forget block in POST /conversations/:id/messages — and that route is gated by requireCompany(req), a human session. Agents never traverse it: `cumora reply` runs server-side through runCli, and cmdReply committed the row and enqueued the realtime broadcast, nothing more. notifyMessage had one caller in the whole server. So the exact case push exists for was the case it missed. A human asks an agent something on their phone and locks the screen; the socket drops and their status goes 'resting'. The agent answers and the phone stays silent. In an agent-first workspace that is most inbound messages. Nothing suggests this was deliberate. The in-app surface has never made the distinction — NotificationToasts fires on any message.new that is not yours and not a system row, and resolves the author out of the participants roster, which holds agents too. Desktop notified, phone did not. push.ts's own header says the design "lets the WS bridge call notifyMessage(…) unconditionally". Extract the dispatch into dispatchMessagePush and call it from both paths. Recipients are unchanged — computeMessageRecipients joins `users`, so only humans are ever notified, and the mute and "currently looking at the app" filters still apply; two of the four tests pin that, and pass either way. Author name now resolves through participants with a users fallback. Agents have no users row, so the previous users-only lookup would have titled the notification with a raw agent id. notifyMessage gains a test-only observation hook: APNs soft-disables without credentials, so the send itself is unobservable and the payload the caller built is the interesting assertion. --- .../__integration__/agent-reply-push.test.ts | 145 ++++++++++++++++++ server/src/agents/cli.ts | 13 ++ server/src/api/router.ts | 31 +--- server/src/push.ts | 63 ++++++++ 4 files changed, 229 insertions(+), 23 deletions(-) create mode 100644 server/src/__integration__/agent-reply-push.test.ts diff --git a/server/src/__integration__/agent-reply-push.test.ts b/server/src/__integration__/agent-reply-push.test.ts new file mode 100644 index 00000000..24a0fd2b --- /dev/null +++ b/server/src/__integration__/agent-reply-push.test.ts @@ -0,0 +1,145 @@ +/** + * A phone must hear an agent's reply. + * + * Push dispatch lived in exactly one place — the fire-and-forget block in + * `POST /conversations/:id/messages` — and that route is gated by + * `requireCompany(req)`, a human session. Agents never traverse it: `cumora + * reply` runs server-side through `runCli`, and `cmdReply` committed the row + * and enqueued the realtime broadcast, nothing more. + * + * So the exact case push exists for was the case it missed. A human asks an + * agent a question on their phone and locks the screen; the socket drops, their + * status goes 'resting'. The agent answers two minutes later and the phone + * stays silent. In an agent-first workspace that is most inbound messages. + * + * The in-app surface never made this distinction: NotificationToasts fires on + * any `message.new` that is not yours and not a system row, and resolves the + * author out of the participants roster, which holds agents too. Desktop + * notified, phone did not. + * + * Author name is the other half. Agents have no `users` row, so the old + * users-only lookup would push with a raw agent id as the notification title. + * + * Run: INTEGRATION_DATABASE_URL=… npm run test:integration + */ +import { test, before, beforeEach, afterEach, after } from 'node:test' +import assert from 'node:assert/strict' +import { pool } from '../db/pool.js' +import { ensureSchemaOnce, resetAllTables, seedCompanyWithAgent, teardownAll } from './_helpers.js' +import { runCli } from '../agents/cli.js' +import { __setNotifyHookForTesting } from '../push.js' + +interface Captured { + authorId: string + authorName: string + conversationId: string + body: string + recipientUserIds: string[] +} + +let sent: Captured[] = [] + +before(async () => { await ensureSchemaOnce() }) +beforeEach(async () => { + await resetAllTables() + sent = [] + __setNotifyHookForTesting((a) => { sent.push(a as unknown as Captured) }) +}) +afterEach(() => { __setNotifyHookForTesting(null) }) +after(async () => { await teardownAll() }) + +/** A human who is offline — exactly the person push is for. */ +async function seedOfflineHuman(companyId: string, userId: string): Promise { + await pool.query( + `INSERT INTO users (id, email, display_name) VALUES ($1, $2, $3)`, + [userId, `${userId}@test.local`, `Human ${userId}`], + ) + await pool.query( + `INSERT INTO company_members (company_id, user_id, role) VALUES ($1, $2, 'owner')`, + [companyId, userId], + ) + await pool.query( + `INSERT INTO participants (id, company_id, kind, name, role, initial, avatar_bg, status) + VALUES ($1, $2, 'human', $3, 'owner', 'H', '#123456', 'resting')`, + [userId, companyId, `Human ${userId}`], + ) +} + +async function seedRoom(companyId: string, convoId: string, memberIds: string[]): Promise { + await pool.query( + `INSERT INTO conversations (id, company_id, kind, title, members) + VALUES ($1, $2, 'group', 'Launch room', $3::jsonb)`, + [convoId, companyId, JSON.stringify(memberIds)], + ) + for (const [i, id] of memberIds.entries()) { + await pool.query( + `INSERT INTO conversation_members (conversation_id, company_id, participant_id, ordinal) + VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING`, + [convoId, companyId, id, i], + ) + } +} + +test('[integration] an agent reply reaches the push path', async () => { + const { companyId, agentId } = await seedCompanyWithAgent() + const humanId = 'u-offline' + await seedOfflineHuman(companyId, humanId) + await seedRoom(companyId, 'c-launch', [agentId, humanId]) + + const res = await runCli(['--as', agentId, 'reply', 'c-launch', 'Q3 numbers are up 12%']) + assert.equal(res.ok, true, res.text) + + // The dispatch is fire-and-forget; give the microtask queue a turn. + await new Promise((r) => setTimeout(r, 150)) + + assert.equal(sent.length, 1, 'the agent reply produced no push') + assert.deepEqual(sent[0].recipientUserIds, [humanId]) + assert.match(sent[0].body, /Q3 numbers are up 12%/) +}) + +test('[integration] the notification is titled with the agent name, not its id', async () => { + // Agents have no `users` row; a users-only lookup would show `a-1f2e3d4c`. + const { companyId, agentId } = await seedCompanyWithAgent() + const humanId = 'u-offline-2' + await seedOfflineHuman(companyId, humanId) + await seedRoom(companyId, 'c-launch', [agentId, humanId]) + + await runCli(['--as', agentId, 'reply', 'c-launch', 'done']) + await new Promise((r) => setTimeout(r, 150)) + + assert.equal(sent.length, 1) + assert.equal(sent[0].authorName, `Agent ${agentId}`) + assert.notEqual(sent[0].authorName, agentId) +}) + +test('[integration] a human already looking at the app is still skipped', async () => { + // The recipient filters are unchanged — this must not become a broadcast. + const { companyId, agentId } = await seedCompanyWithAgent() + const humanId = 'u-online' + await seedOfflineHuman(companyId, humanId) + await pool.query(`UPDATE participants SET status = 'avail' WHERE id = $1`, [humanId]) + await seedRoom(companyId, 'c-launch', [agentId, humanId]) + + await runCli(['--as', agentId, 'reply', 'c-launch', 'hello']) + await new Promise((r) => setTimeout(r, 150)) + + const withRecipients = sent.filter((s) => s.recipientUserIds.length > 0) + assert.equal(withRecipients.length, 0, 'pushed to someone who is on the app') +}) + +test('[integration] a muted conversation is still muted', async () => { + const { companyId, agentId } = await seedCompanyWithAgent() + const humanId = 'u-muted' + await seedOfflineHuman(companyId, humanId) + await seedRoom(companyId, 'c-launch', [agentId, humanId]) + await pool.query( + `INSERT INTO conversation_mutes (user_id, conversation_id, muted_until) VALUES ($1, 'c-launch', NULL)`, + [humanId], + ) + + await runCli(['--as', agentId, 'reply', 'c-launch', 'hello']) + await new Promise((r) => setTimeout(r, 150)) + + const withRecipients = sent.filter((s) => s.recipientUserIds.length > 0) + assert.equal(withRecipients.length, 0, 'pushed into a muted conversation') +}) diff --git a/server/src/agents/cli.ts b/server/src/agents/cli.ts index e3a5fffb..937779ad 100644 --- a/server/src/agents/cli.ts +++ b/server/src/agents/cli.ts @@ -16,6 +16,7 @@ import { env } from '../env.js' import type { CliResult, CliSideEffect } from './cli-result.js' import { fetchImageBytes } from './image-fetcher.js' import { stripLoneSurrogates } from './text-safety.js' +import { dispatchMessagePush } from '../push.js' import { asMemorySource, memoryVisibleInScope, @@ -2384,6 +2385,18 @@ async function cmdReply(parsed: ParsedArgs): Promise { } finally { txClient.release() } + // The row is durable now, so the phone can be told. Fire-and-forget for the + // same reason the HTTP route's dispatch is: a push must never hold up the + // reply. Without this an agent's answer reached the websocket and the desktop + // toast and no phone at all — the surface a human is on when they ask a + // question and lock the screen. + void dispatchMessagePush({ + conversationId: convoId, + authorId: me, + messageId, + body: finalBody, + companyId, + }) // Advance the Redis "seen" boundary to my own just-inserted seq, so the // freshness preflight on my NEXT cumora reply compares against the post- // insertion state (peer messages with seq <= mine are "things I obviously diff --git a/server/src/api/router.ts b/server/src/api/router.ts index 97c9776e..d185a2d8 100644 --- a/server/src/api/router.ts +++ b/server/src/api/router.ts @@ -18,7 +18,7 @@ import { getTriageEconomics, getWakeEconomics } from '../agents/observability.js import { resolveKanbanAssigneeChange, wakeKanbanAgents } from '../agents/kanban-wake.js' import { AgentCreationError, createAgentRecord } from '../agents/create.js' import { BUSY_STATUS_LEASE_MS } from '../status.js' -import { notifyMessage, computeMessageRecipients } from '../push.js' +import { dispatchMessagePush } from '../push.js' import { randomUUID, randomBytes, createHash, timingSafeEqual } from 'node:crypto' import { deleteSession, authMiddleware, type AuthedRequest, @@ -4223,28 +4223,13 @@ api.post('/conversations/:id/messages', async (req, res) => { // (NotificationToasts handles those). Fire-and-forget — push delivery // must never block the HTTP response. The push module soft-disables // when APNs creds aren't configured, so this is safe even in dev. - void (async () => { - try { - const [recipients, convoRow, authorRow] = await Promise.all([ - computeMessageRecipients({ conversationId: id, authorId: me }), - pool.query<{ title: string }>(`SELECT title FROM conversations WHERE id = $1`, [id]).then((r) => r.rows[0]), - pool.query<{ display_name: string }>(`SELECT display_name FROM users WHERE id = $1`, [me]).then((r) => r.rows[0]), - ]) - if (recipients.length === 0) return - await notifyMessage({ - conversationId: id, - conversationTitle: convoRow?.title ?? null, - authorId: me, - authorName: authorRow?.display_name ?? me, - messageId, - body, - companyId: tenant, - recipientUserIds: recipients, - }) - } catch (e) { - console.warn('[push] notifyMessage post-/messages failed', e) - } - })() + void dispatchMessagePush({ + conversationId: id, + authorId: me, + messageId, + body, + companyId: tenant, + }) // Climate signal: @-mentioned agents feel mildly more affinity / trust // toward the speaker (engagement is positive). Fire-and-forget so we diff --git a/server/src/push.ts b/server/src/push.ts index 7b4fd4ea..b29991f6 100644 --- a/server/src/push.ts +++ b/server/src/push.ts @@ -284,7 +284,16 @@ function trimBody(body: string): string { return `${trimmed.slice(0, 237)}…` } +/** Observe what notifyMessage was asked to send. Tests only — APNs + * soft-disables without credentials, so the send itself is unobservable and + * the interesting assertion is the payload the caller built. */ +let notifyHookForTesting: ((args: NotifyMessageArgs) => void) | null = null +export function __setNotifyHookForTesting(fn: ((args: NotifyMessageArgs) => void) | null): void { + notifyHookForTesting = fn +} + export async function notifyMessage(args: NotifyMessageArgs): Promise { + notifyHookForTesting?.(args) if (args.recipientUserIds.length === 0) return const title = args.conversationTitle ? `${args.authorName} · ${args.conversationTitle}` @@ -309,6 +318,60 @@ export async function notifyMessage(args: NotifyMessageArgs): Promise { * Kept in this file rather than in router.ts so the SQL stays next to * the send path; the publish call in router.ts is a one-liner. */ +/** Fire-and-forget push for one newly-posted message. + * + * Shared by the human HTTP route and the agent CLI's reply path. A teammate + * replying is a teammate replying whichever kind they are, and the in-app + * surface has never distinguished them: NotificationToasts fires on any + * `message.new` that is not yours and not a system row, and resolves the + * author out of the participants roster, which holds agents too. Push was + * wired only into POST /conversations/:id/messages — a human-session route + * agents never traverse — so a phone stayed silent for exactly the replies + * its owner had just asked for. + * + * Recipients are unchanged: computeMessageRecipients joins `users`, so only + * humans are ever notified, and the mute and "currently looking at the app" + * filters still apply. + * + * Author name resolves through participants first. An agent has no `users` + * row, so a users-only lookup would push with a raw agent id as the title. */ +export async function dispatchMessagePush(args: { + conversationId: string + authorId: string + messageId: string + body: string + companyId: string +}): Promise { + try { + const [recipients, convoRow, authorRow] = await Promise.all([ + computeMessageRecipients({ conversationId: args.conversationId, authorId: args.authorId }), + pool.query<{ title: string }>( + `SELECT title FROM conversations WHERE id = $1`, [args.conversationId], + ).then((r) => r.rows[0]), + pool.query<{ name: string | null }>( + `SELECT COALESCE( + (SELECT name FROM participants WHERE id = $1 AND company_id = $2), + (SELECT display_name FROM users WHERE id = $1) + ) AS name`, + [args.authorId, args.companyId], + ).then((r) => r.rows[0]), + ]) + if (recipients.length === 0) return + await notifyMessage({ + conversationId: args.conversationId, + conversationTitle: convoRow?.title ?? null, + authorId: args.authorId, + authorName: authorRow?.name ?? args.authorId, + messageId: args.messageId, + body: args.body, + companyId: args.companyId, + recipientUserIds: recipients, + }) + } catch (e) { + console.warn('[push] dispatchMessagePush failed', e) + } +} + export async function computeMessageRecipients(args: { conversationId: string authorId: string