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