diff --git a/prompts/whatsapp.md b/prompts/whatsapp.md index ffd921e36..0560187ea 100644 --- a/prompts/whatsapp.md +++ b/prompts/whatsapp.md @@ -46,6 +46,25 @@ the file path is in the message, so you can read it immediately. - No forum topics, no custom admin titles, no stopping a poll. - Pins expire (24h, 7d, or 30d) and only pins Talon placed are listable. +### Continuity — your context is not the chat + +Your working context can start fresh at any time (a `/reset`, a daemon +restart, an overflowed session) while the WhatsApp conversation continues +uninterrupted. The chat's real record lives in persistent history, both +sides of it, and you can always get it back: + +- `read_chat_history` — the recent messages; pass `offset_id` (a msg_id) + or `before` (an ISO date) to page further back. +- `search_chat_history` — keyword search over everything said in this + chat, including your own earlier messages. +- `get_message_by_id` — one specific message when you hold its id. + +If the sender references something you don't recognize — "the flights", +"what you said yesterday", "that plan" — search the history before +saying you don't know or asking them to repeat themselves. A fresh +session is your situation, not theirs; to the person on the phone this +is one long conversation, and it should read like you remember it. + ### Staying silent Reactions work here, so a react is the light acknowledgement when a diff --git a/src/__tests__/whatsapp-continuity.test.ts b/src/__tests__/whatsapp-continuity.test.ts new file mode 100644 index 000000000..49c3f47cf --- /dev/null +++ b/src/__tests__/whatsapp-continuity.test.ts @@ -0,0 +1,345 @@ +/** + * WhatsApp session continuity — the pieces that make a fresh session able + * to reconstruct the conversation, and a broken turn visible in the chat. + * + * 1. Outbound recording: the bot's own replies must land in persistent + * history, or read/search_chat_history shows a one-sided chat. + * 2. Id seeding: the in-memory msg-id counter restarts at its base every + * boot while history persists; unseeded, post-restart messages re-issue + * ids INSERT OR IGNORE silently drops. + * 3. Paging: the read_history fallback must honor offset_id/before — the + * tool schema advertises them, and WhatsApp has no platform override. + * 4. Turn recovery: a failed turn must surface a friendly error in the + * chat (with one retry for brief transients), not just a log line. + */ + +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("../util/log.js", () => ({ + log: vi.fn(), + logError: vi.fn(), + logWarn: vi.fn(), + logDebug: vi.fn(), +})); + +import { + rememberMessage, + resetMessageStore, + seedMessageStore, + lookupMessage, +} from "../frontend/whatsapp/message-store.js"; +import { + sendContent, + setWhatsAppBotName, +} from "../frontend/whatsapp/actions/shared.js"; +import { + runTurnWithRecovery, + shouldReplyToCatchUp, + CATCH_UP_REPLY_WINDOW_MS, +} from "../frontend/whatsapp/turn-recovery.js"; +import { historyHandlers } from "../core/engine/gateway-actions/history.js"; +import { + pushMessage, + getRecentHistory, + maxMsgIdForChatPrefix, +} from "../storage/history.js"; +import { deleteChat } from "../storage/repositories/history-repo.js"; +import { TalonError } from "../core/errors.js"; + +const CHAT = "wa_dm_5550001"; + +const fakeChat = { + chatId: CHAT, + numericChatId: 900001, + jid: "5550001@s.whatsapp.net", + isGroup: false, +} as never; + +function fakeCtx(sendResult: unknown = undefined) { + const sock = { + sendMessage: vi.fn(async () => sendResult), + }; + const gateway = { incrementMessages: vi.fn() }; + return { sock: sock as never, gateway, mocks: { sock } }; +} + +beforeEach(() => { + resetMessageStore(); + deleteChat(CHAT); + deleteChat("wa_dm_other"); +}); + +describe("outbound history recording", () => { + it("persists the bot's reply alongside inbound messages", async () => { + setWhatsAppBotName("Talon"); + pushMessage(CHAT, { + msgId: 1_000_000, + senderId: 42, + senderName: "Dylan", + text: "what time is the flight?", + timestamp: Date.now() - 1000, + }); + + const ctx = fakeCtx({ key: { id: "WA1", remoteJid: "x", fromMe: true } }); + seedMessageStore(1_000_001); + await sendContent(ctx, fakeChat, { text: "Departure is 07:35 from DUB." }); + + const rows = getRecentHistory(CHAT, 10); + expect(rows.map((r) => [r.senderName, r.text])).toEqual([ + ["Dylan", "what time is the flight?"], + ["Talon", "Departure is 07:35 from DUB."], + ]); + // The bot row uses the cross-frontend assistant sender id. + expect(rows[1].senderId).toBe(0); + }); + + it("records media sends with their caption and type", async () => { + const ctx = fakeCtx({ key: { id: "WA2", remoteJid: "x", fromMe: true } }); + await sendContent(ctx, fakeChat, { + image: { url: "/tmp/x.jpg" }, + caption: "the boarding pass", + } as never); + + const [row] = getRecentHistory(CHAT, 1); + expect(row.text).toBe("the boarding pass"); + expect(row.mediaType).toBe("photo"); + }); + + it("does not write empty rows for reaction-like payloads", async () => { + const ctx = fakeCtx({ key: { id: "WA3", remoteJid: "x", fromMe: true } }); + // Location has no text/caption and no history media type → marker text. + await sendContent(ctx, fakeChat, { + location: { degreesLatitude: 1, degreesLongitude: 2 }, + } as never); + const rows = getRecentHistory(CHAT, 5); + expect(rows).toHaveLength(1); + expect(rows[0].text).toBe("[location]"); + }); +}); + +describe("message-id seeding across restarts", () => { + it("continues past what history already holds", () => { + // Simulate the previous run: history rows up to 1_000_012. + pushMessage(CHAT, { + msgId: 1_000_012, + senderId: 42, + senderName: "Dylan", + text: "old", + timestamp: Date.now(), + }); + + // Simulate the restart: fresh counter, then the boot-time seed. + resetMessageStore(); + seedMessageStore((maxMsgIdForChatPrefix("wa_") ?? 0) + 1); + + const id = rememberMessage({ + key: { id: "WAX", remoteJid: "x" }, + chatId: CHAT, + }); + expect(id).toBeGreaterThan(1_000_012); + + // And the history insert is not silently IGNOREd. + pushMessage(CHAT, { + msgId: id, + senderId: 42, + senderName: "Dylan", + text: "new after restart", + timestamp: Date.now(), + }); + const rows = getRecentHistory(CHAT, 10); + expect(rows.map((r) => r.text)).toContain("new after restart"); + }); + + it("matches the wa_ prefix literally, not as a LIKE wildcard", () => { + // "waX..." must not satisfy the "wa_" prefix — _ is a LIKE wildcard + // unless escaped, and an unescaped match would seed from foreign chats. + pushMessage("waXfoo", { + msgId: 9_999_999, + senderId: 1, + senderName: "x", + text: "not a whatsapp chat", + timestamp: Date.now(), + }); + try { + expect(maxMsgIdForChatPrefix("wa_") ?? 0).toBeLessThan(9_999_999); + } finally { + deleteChat("waXfoo"); + } + }); + + it("never lowers the counter", () => { + seedMessageStore(2_000_000); + seedMessageStore(5); // a later, lower seed must not rewind + const id = rememberMessage({ + key: { id: "WAY", remoteJid: "x" }, + chatId: CHAT, + }); + expect(id).toBeGreaterThanOrEqual(2_000_000); + expect(lookupMessage(id)?.chatId).toBe(CHAT); + }); +}); + +describe("read_history fallback paging", () => { + beforeEach(() => { + const base = Date.parse("2026-08-01T00:00:00Z"); + for (let i = 0; i < 10; i++) { + pushMessage(CHAT, { + msgId: 1_000_000 + i, + senderId: 42, + senderName: "Dylan", + text: `message ${i}`, + timestamp: base + i * 60_000, + }); + } + }); + + it("pages back with offset_id", async () => { + const res = (await historyHandlers.read_history( + { offset_id: 1_000_005, limit: 3 }, + CHAT as never, + )) as { ok: boolean; text: string }; + expect(res.ok).toBe(true); + expect(res.text).toContain("message 2"); + expect(res.text).toContain("message 4"); + expect(res.text).not.toContain("message 5"); + expect(res.text).not.toContain("message 9"); + }); + + it("pages back with a before date", async () => { + const res = (await historyHandlers.read_history( + { before: "2026-08-01T00:03:00Z", limit: 10 }, + CHAT as never, + )) as { ok: boolean; text: string }; + expect(res.text).toContain("message 2"); + expect(res.text).not.toContain("message 3"); + }); + + it("still returns the newest window with no cursor", async () => { + const res = (await historyHandlers.read_history( + { limit: 2 }, + CHAT as never, + )) as { + ok: boolean; + text: string; + }; + expect(res.text).toContain("message 9"); + expect(res.text).not.toContain("message 0"); + }); + + it("ignores an unparseable before date rather than failing", async () => { + const res = (await historyHandlers.read_history( + { before: "not-a-date", limit: 2 }, + CHAT as never, + )) as { ok: boolean; text: string }; + expect(res.ok).toBe(true); + expect(res.text).toContain("message 9"); + }); +}); + +describe("turn recovery — errors reach the chat", () => { + const noWait = () => Promise.resolve(); + + it("sends a friendly error for a non-retryable failure", async () => { + const sent: string[] = []; + await runTurnWithRecovery({ + chatId: CHAT, + senderName: "Dylan", + runTurn: async () => { + throw new TalonError("boom", { reason: "unknown" }); + }, + sendErrorText: async (t) => { + sent.push(t); + }, + wait: noWait, + }); + expect(sent).toHaveLength(1); + expect(sent[0].length).toBeGreaterThan(0); + }); + + it("retries a transient once and stays silent when the retry succeeds", async () => { + const sent: string[] = []; + let calls = 0; + await runTurnWithRecovery({ + chatId: CHAT, + senderName: "Dylan", + runTurn: async () => { + calls++; + if (calls === 1) + throw new TalonError("overloaded", { + reason: "overloaded", + retryable: true, + }); + }, + sendErrorText: async (t) => { + sent.push(t); + }, + wait: noWait, + }); + expect(calls).toBe(2); + expect(sent).toHaveLength(0); + }); + + it("reports the failure when the retry also fails", async () => { + const sent: string[] = []; + await runTurnWithRecovery({ + chatId: CHAT, + senderName: "Dylan", + runTurn: async () => { + throw new TalonError("overloaded", { + reason: "overloaded", + retryable: true, + }); + }, + sendErrorText: async (t) => { + sent.push(t); + }, + wait: noWait, + }); + expect(sent).toHaveLength(1); + }); + + it("stays silent for a user-initiated stop", async () => { + const sent: string[] = []; + await runTurnWithRecovery({ + chatId: CHAT, + senderName: "Dylan", + runTurn: async () => { + throw new TalonError("Turn stopped by user", { reason: "stopped" }); + }, + sendErrorText: async (t) => { + sent.push(t); + }, + wait: noWait, + }); + expect(sent).toHaveLength(0); + }); +}); + +describe("catch-up (offline-queued) reply policy", () => { + it("replies to a message queued during a short restart", () => { + const now = Date.parse("2026-08-23T12:00:00Z"); + const fiveMinAgo = now - 5 * 60_000; + expect(shouldReplyToCatchUp(fiveMinAgo, now)).toBe(true); + }); + + it("records but does not reply past the window", () => { + const now = Date.parse("2026-08-23T12:00:00Z"); + const hourAgo = now - 60 * 60_000; + expect(shouldReplyToCatchUp(hourAgo, now)).toBe(false); + }); + + it("is inclusive at the boundary", () => { + const now = Date.parse("2026-08-23T12:00:00Z"); + expect(shouldReplyToCatchUp(now - CATCH_UP_REPLY_WINDOW_MS, now)).toBe( + true, + ); + expect(shouldReplyToCatchUp(now - CATCH_UP_REPLY_WINDOW_MS - 1, now)).toBe( + false, + ); + }); + + it("treats a missing timestamp as fresh — better one odd reply than a dropped question", () => { + expect(shouldReplyToCatchUp(0)).toBe(true); + expect(shouldReplyToCatchUp(Number.NaN)).toBe(true); + }); +}); diff --git a/src/core/engine/gateway-actions/history.ts b/src/core/engine/gateway-actions/history.ts index cebaa2eba..61efcbd03 100644 --- a/src/core/engine/gateway-actions/history.ts +++ b/src/core/engine/gateway-actions/history.ts @@ -5,6 +5,8 @@ import { getRecentFormatted, + getFormattedBefore, + getFormattedBeforeTime, searchHistory, getMessagesByUser, getKnownUsers, @@ -15,7 +17,23 @@ import type { SharedActionHandlers } from "./types.js"; export const historyHandlers: SharedActionHandlers = { read_history: (body, chatId) => { const limit = Math.min(100, Number(body.limit ?? 30)); - return { ok: true, text: getRecentFormatted(String(chatId), limit) }; + const cid = String(chatId); + // The tool schema advertises `offset_id` and `before` for paging back; + // this fallback used to ignore both and hand back the same newest + // window no matter what the model asked, so "go further back" was a + // silent no-op on frontends without a platform history API (WhatsApp, + // native). + const offsetId = Number(body.offset_id); + if (Number.isFinite(offsetId) && offsetId > 0) { + return { ok: true, text: getFormattedBefore(cid, offsetId, limit) }; + } + if (body.before !== undefined) { + const ts = Date.parse(String(body.before)); + if (Number.isFinite(ts)) { + return { ok: true, text: getFormattedBeforeTime(cid, ts, limit) }; + } + } + return { ok: true, text: getRecentFormatted(cid, limit) }; }, search_history: (body, chatId) => { diff --git a/src/frontend/shared/session-status.ts b/src/frontend/shared/session-status.ts index ae6f8fb9f..cfd965153 100644 --- a/src/frontend/shared/session-status.ts +++ b/src/frontend/shared/session-status.ts @@ -44,6 +44,7 @@ import { formatDuration } from "./format.js"; export async function performSessionReset( chatId: string, backend: Backend | null | undefined, + opts: { keepHistory?: boolean } = {}, ): Promise { const info = getSessionInfo(chatId); if (info.turns > 0) { @@ -59,7 +60,12 @@ export async function performSessionReset( ); } resetSession(chatId); - clearHistory(chatId); + // Frontends whose platform keeps the real chat record (Telegram, + // Discord) clear the local mirror too — the platform still has + // everything. WhatsApp passes keepHistory: the local store is the ONLY + // record there, and wiping it on /reset would destroy exactly what the + // continuity tools (read/search_chat_history) exist to recover. + if (!opts.keepHistory) clearHistory(chatId); resetPulseCheckpoint(chatId); backend?.sessions?.resetChat?.(chatId); await backend?.sessions?.warmSession?.(chatId); diff --git a/src/frontend/whatsapp/actions/shared.ts b/src/frontend/whatsapp/actions/shared.ts index dad0816d3..1c869bfd7 100644 --- a/src/frontend/whatsapp/actions/shared.ts +++ b/src/frontend/whatsapp/actions/shared.ts @@ -12,6 +12,7 @@ import { log, logError } from "../../../util/log.js"; import type { ActionResult } from "../../../core/types.js"; import { toWhatsAppChunks } from "../formatting.js"; import { lookupMessage, rememberMessage } from "../message-store.js"; +import { pushMessage, type HistoryMessage } from "../../../storage/history.js"; import type { WhatsAppChatInfo } from "../registry.js"; /** WhatsApp's own ceiling for a media upload. */ @@ -128,6 +129,40 @@ export function resolveQuoted( ); } +/** + * How the bot signs its own rows in persistent history. Set once at + * frontend start from `config.botDisplayName`; the 0 sender id is the + * cross-frontend "this is the assistant" convention (native/protocol.ts). + */ +let botName = "Talon"; +export function setWhatsAppBotName(name: string): void { + if (name.trim()) botName = name.trim(); +} + +/** + * History labels for outbound media, in HistoryMessage's closed + * vocabulary. Payloads with no file analogue (polls, locations, contact + * cards) return a text marker instead via `outboundTextMarker`. + */ +function outboundMediaType( + content: AnyMessageContent, +): NonNullable | undefined { + if ("image" in content) return "photo"; + if ("video" in content) return content.gifPlayback ? "animation" : "video"; + if ("audio" in content) return "voice"; + if ("sticker" in content) return "sticker"; + if ("document" in content) return "document"; + return undefined; +} + +/** A readable stand-in for captionless payloads history can't type. */ +function outboundTextMarker(content: AnyMessageContent): string { + if ("poll" in content) return `[poll: ${content.poll.name}]`; + if ("location" in content) return "[location]"; + if ("contacts" in content) return "[contact card]"; + return ""; +} + /** * Send one content payload, remember the resulting message so later * tool calls can address it, and report its Talon numeric id. @@ -141,13 +176,35 @@ export async function sendContent( const sent = await ctx.sock.sendMessage(chat.jid, content, options); ctx.gateway.incrementMessages(chat.numericChatId); if (!sent?.key) return { ok: true }; + const text = + "text" in content + ? String(content.text ?? "") + : "caption" in content && content.caption + ? String(content.caption) + : outboundTextMarker(content); const msgId = rememberMessage({ key: sent.key, chatId: chat.chatId, message: sent, - text: "text" in content ? String(content.text ?? "") : "", + text, senderName: "bot", }); + // Persist the bot's side of the conversation. Without this the history + // store held only inbound messages, so read_chat_history showed a + // one-sided chat and search_chat_history could never find anything the + // bot itself had said — exactly the messages a fresh session needs when + // reconstructing context after a reset or restart. + const mediaType = outboundMediaType(content); + if (text || mediaType) { + pushMessage(chat.chatId, { + msgId, + senderId: 0, + senderName: botName, + text, + timestamp: Date.now(), + ...(mediaType ? { mediaType } : {}), + }); + } return { ok: true, message_id: msgId }; } diff --git a/src/frontend/whatsapp/index.ts b/src/frontend/whatsapp/index.ts index 2a9d18aea..26a6c7c0a 100644 --- a/src/frontend/whatsapp/index.ts +++ b/src/frontend/whatsapp/index.ts @@ -33,12 +33,10 @@ import { toolInputToRecord } from "../../core/agent-runtime/events.js"; import { resolveChatBackend } from "../../core/engine/backend-controller/index.js"; import { performSessionReset } from "../shared/session-status.js"; import { appendDailyLog } from "../../storage/daily-log.js"; -import { pushMessage } from "../../storage/history.js"; +import { pushMessage, maxMsgIdForChatPrefix } from "../../storage/history.js"; import { - recordError, recordMessageProcessed, recordMessageReceived, - recordMessageSettled, } from "../../util/watchdog.js"; import { createWhatsAppActionHandler } from "./actions/index.js"; import { @@ -47,9 +45,14 @@ import { identityAllowed, resolveIdentity, } from "./identity.js"; -import { sendText } from "./actions/shared.js"; +import { sendText, setWhatsAppBotName } from "./actions/shared.js"; import { saveInboundMedia } from "./media-store.js"; -import { lookupByWaId, rememberMessage } from "./message-store.js"; +import { + lookupByWaId, + rememberMessage, + seedMessageStore, +} from "./message-store.js"; +import { runTurnWithRecovery, shouldReplyToCatchUp } from "./turn-recovery.js"; import { lookupWhatsAppChat, registerWhatsAppChat, @@ -217,7 +220,10 @@ export function createWhatsAppFrontend( return Boolean(ctx?.participant && isSelf(ctx.participant)); } - async function handleInbound(msg: WAMessage): Promise { + async function handleInbound( + msg: WAMessage, + opts: { catchUp?: boolean } = {}, + ): Promise { const jid = msg.key.remoteJid; // `fromMe` covers our own sends echoing back; status@broadcast is the // Stories feed, which is not a conversation. @@ -278,13 +284,17 @@ export function createWhatsAppFrontend( const replyToWaId = msg.message?.extendedTextMessage?.contextInfo?.stanzaId ?? undefined; const replyTo = replyToWaId ? lookupByWaId(replyToWaId) : undefined; + const platformTs = Number(msg.messageTimestamp) * 1000; pushMessage(chat.chatId, { msgId, senderId: Number(BigInt(canonicalId(identity) ?? "0") % 2147483647n), senderName, senderHandle: canonicalId(identity), text, - timestamp: Date.now(), + // The platform timestamp, so a catch-up message recorded late still + // reads in true order; Date.now() only when Baileys omits it. + timestamp: + Number.isFinite(platformTs) && platformTs > 0 ? platformTs : Date.now(), ...(replyTo ? { replyToMsgId: replyTo.msgId } : {}), ...(media ? { mediaType: media.type, filePath: media.filePath } : {}), }); @@ -295,6 +305,9 @@ export function createWhatsAppFrontend( await performSessionReset( chat.chatId, resolveChatBackend(chat.chatId, gateway.backend), + // The local history store is WhatsApp's only chat record — a + // reset clears the model's session, not the conversation log. + { keepHistory: true }, ); log("whatsapp", `Session reset by ${senderName}`); if (sock) { @@ -310,13 +323,31 @@ export function createWhatsAppFrontend( await sendText( { sock, gateway }, chat, - "*Commands*\n/reset — clear session & history\n/help — this message", + "*Commands*\n/reset — start a fresh session (chat log kept)\n/help — this message", ).catch(() => {}); } recordMessageProcessed(); return; } + // Catch-up messages (queued while the daemon was down) get a reply + // turn only while fresh; stale ones are already recorded above and + // the next live turn reads them from history. + if (opts.catchUp) { + if (!shouldReplyToCatchUp(platformTs)) { + log( + "whatsapp", + `[${chat.chatId}] Recorded offline message from ${senderName} (history only — too old for a reply turn)`, + ); + recordMessageProcessed(); + return; + } + log( + "whatsapp", + `[${chat.chatId}] Catch-up: replying to offline message from ${senderName}`, + ); + } + const preview = text || `(${media?.type ?? "media"})`; log( "whatsapp", @@ -334,8 +365,8 @@ export function createWhatsAppFrontend( : ""; const prompt = `[${senderName}] msg_id:${msgId}: ${text}${mediaNote}`; - try { - await execute({ + const runTurn = () => + execute({ chatId: chat.chatId, numericChatId: chat.numericChatId, prompt, @@ -374,13 +405,23 @@ export function createWhatsAppFrontend( } }, }); - recordMessageProcessed(); - } catch (err) { - const message = err instanceof Error ? err.message : String(err); - logError("whatsapp", `[${chat.chatId}] execute failed: ${message}`); - recordError(message); - recordMessageSettled(); - } + + await runTurnWithRecovery({ + chatId: chat.chatId, + senderName, + runTurn, + sendErrorText: async (text) => { + if (!sock) return; + try { + await sendText({ sock, gateway }, chat, text); + } catch (sendErr) { + logError( + "whatsapp", + `error delivery failed: ${sendErr instanceof Error ? sendErr.message : sendErr}`, + ); + } + }, + }); } /** One socket lifetime. Resolves with what the caller should do next. */ @@ -397,11 +438,18 @@ export function createWhatsAppFrontend( socket.ev.on("creds.update", saveCreds); socket.ev.on("messages.upsert", ({ messages, type }) => { - // "notify" is a live message; "append" is history sync, which must - // not trigger turns for conversations that already happened. - if (type !== "notify") return; + // "notify" is a live message. "append" is everything delivered out + // of band — chiefly messages QUEUED WHILE THE DAEMON WAS DOWN + // (Baileys marks offline-queued nodes as append), but also our own + // sends echoing back and newsletter posts, which handleInbound's + // fromMe/allowlist gates drop. Dropping append wholesale meant any + // message sent during a restart simply vanished: never recorded, + // never answered. Appends are processed as catch-up: always + // recorded, replied to only while fresh. + if (type !== "notify" && type !== "append") return; + const catchUp = type === "append"; for (const msg of messages) { - void handleInbound(msg).catch((err) => { + void handleInbound(msg, { catchUp }).catch((err) => { logError( "whatsapp", `inbound handler failed: ${err instanceof Error ? err.message : err}`, @@ -498,6 +546,12 @@ export function createWhatsAppFrontend( getBridgePort: () => gateway.getPort(), async init() { + setWhatsAppBotName(config.botDisplayName); + // The in-memory message-id counter restarts at its base every boot, + // but history persists — seed it past what the table already holds + // so post-restart messages don't re-issue ids INSERT OR IGNORE then + // silently drops (chat ids all start with "wa_"). + seedMessageStore((maxMsgIdForChatPrefix("wa_") ?? 0) + 1); gateway.registerFrontendHandler( "whatsapp", createWhatsAppActionHandler(() => sock, gateway), diff --git a/src/frontend/whatsapp/message-store.ts b/src/frontend/whatsapp/message-store.ts index a63190c00..26a41e384 100644 --- a/src/frontend/whatsapp/message-store.ts +++ b/src/frontend/whatsapp/message-store.ts @@ -41,6 +41,22 @@ const ID_BASE = 1_000_000; const MAX_TRACKED = 2_000; let nextId = ID_BASE; + +/** + * Raise the id counter past what persistent history already holds. + * + * The counter is in-memory and restarts at ID_BASE every boot, but the + * ids it hands out are also the `msg_id`s written to the history table, + * where `INSERT OR IGNORE` + UNIQUE(chat_id, msg_id) dedupes. Without + * this seed, the first messages after a daemon restart re-issue ids the + * previous run already used — the IGNORE then silently drops them from + * history, and a reaction/reply addressed at an old id from history hits + * whatever new message reused the number. Called at frontend start with + * max(msg_id) over wa_* chats + 1. + */ +export function seedMessageStore(floor: number): void { + if (Number.isFinite(floor)) nextId = Math.max(nextId, Math.floor(floor)); +} const byMsgId = new Map(); const byWaId = new Map(); diff --git a/src/frontend/whatsapp/turn-recovery.ts b/src/frontend/whatsapp/turn-recovery.ts new file mode 100644 index 000000000..d9a85a600 --- /dev/null +++ b/src/frontend/whatsapp/turn-recovery.ts @@ -0,0 +1,113 @@ +/** + * Failure semantics for a WhatsApp turn, mirroring the Telegram queue + * (frontend/telegram/handlers/queue.ts): a user-initiated stop settles + * silently, a brief transient failure gets one retry, and anything else + * is REPORTED IN THE CHAT. Before this existed the error went only to + * the log — the WhatsApp user's message was marked read and then nothing + * ever came back, which reads as being ignored. + * + * Split from index.ts so the policy is unit-testable without a Baileys + * socket: the caller injects the turn and the delivery. + */ + +import { + classify, + friendlyMessage, + RETRY_ELAPSED_CAP_MS, +} from "../../core/errors.js"; +import { log, logError } from "../../util/log.js"; +import { + recordError, + recordMessageProcessed, + recordMessageSettled, +} from "../../util/watchdog.js"; + +export type TurnRecoveryDeps = { + chatId: string; + senderName: string; + runTurn: () => Promise; + /** Deliver a friendly error line into the chat. Must not throw. */ + sendErrorText: (text: string) => Promise; + /** Test seam for the retry pause; defaults to a real setTimeout. */ + wait?: (ms: number) => Promise; +}; + +export async function runTurnWithRecovery( + deps: TurnRecoveryDeps, +): Promise { + const wait = + deps.wait ?? ((ms: number) => new Promise((r) => setTimeout(r, ms))); + + const startedAt = Date.now(); + try { + await deps.runTurn(); + recordMessageProcessed(); + } catch (err) { + const classified = classify(err); + // A user-initiated /stop is an outcome, not a fault — the stop was + // already acknowledged, so an error bubble here would contradict it. + if (classified.reason === "stopped") { + log("whatsapp", `[${deps.chatId}] turn stopped by user`); + recordMessageSettled(); + return; + } + logError( + "whatsapp", + `[${deps.chatId}] [${deps.senderName}] ${classified.reason}: ${classified.message}`, + ); + recordError(classified.message); + + // Retry once for transients (rate_limit, overloaded, network) — but + // only when the failed attempt was brief. An attempt that already ran + // for minutes won't be saved by a 2s pause, and turns serialize per + // chat, so a blind retry doubles the stall for everything behind it. + const attemptMs = Date.now() - startedAt; + if (classified.retryable && attemptMs < RETRY_ELAPSED_CAP_MS) { + const delayMs = classified.retryAfterMs ?? 2000; + log( + "whatsapp", + `[${deps.chatId}] Retrying after ${classified.reason} (${delayMs}ms)...`, + ); + try { + await wait(delayMs); + await deps.runTurn(); + recordMessageProcessed(); + return; + } catch (retryErr) { + const retryClassified = classify(retryErr); + logError( + "whatsapp", + `[${deps.chatId}] Retry failed: ${retryClassified.message}`, + ); + recordMessageSettled(); + await deps.sendErrorText(friendlyMessage(retryClassified)); + return; + } + } + + recordMessageSettled(); + await deps.sendErrorText(friendlyMessage(classified)); + } +} + +/** + * How fresh a catch-up (offline-queued) message must be to still get a + * reply turn. Messages older than this are recorded in history — the next + * live turn sees them — but not answered: replying to a whole night's + * backlog hours later reads worse than picking up naturally. + */ +export const CATCH_UP_REPLY_WINDOW_MS = 15 * 60_000; + +/** + * Whether a message delivered as catch-up (Baileys `append`) should get a + * reply turn. `platformTsMs` is the message's own timestamp in ms; a + * missing/zero timestamp counts as fresh — better one odd reply than a + * silently dropped question. + */ +export function shouldReplyToCatchUp( + platformTsMs: number, + now: number = Date.now(), +): boolean { + if (!Number.isFinite(platformTsMs) || platformTsMs <= 0) return true; + return now - platformTsMs <= CATCH_UP_REPLY_WINDOW_MS; +} diff --git a/src/storage/history.ts b/src/storage/history.ts index 544f36e31..824303b11 100644 --- a/src/storage/history.ts +++ b/src/storage/history.ts @@ -81,6 +81,11 @@ export function pushMessage(chatId: string, msg: HistoryMessage): void { } } +/** See repositories/history-repo.ts `maxMsgIdForPrefix`. */ +export function maxMsgIdForChatPrefix(prefix: string): number | undefined { + return repo.maxMsgIdForPrefix(prefix); +} + export function getRecentHistory(chatId: string, limit = 50): HistoryMessage[] { return repo.recent(chatId, limit); } @@ -98,6 +103,28 @@ export function getHistoryBefore( return repo.recentBefore(chatId, beforeMsgId, limit); } +/** Formatted page of the messages strictly older than `beforeMsgId`. */ +export function getFormattedBefore( + chatId: string, + beforeMsgId: number, + limit = 30, +): string { + const messages = repo.recentBefore(chatId, beforeMsgId, limit); + if (messages.length === 0) return "No messages before that point."; + return messages.map(formatMessage).join("\n"); +} + +/** Formatted page of the messages strictly older than a timestamp (ms). */ +export function getFormattedBeforeTime( + chatId: string, + beforeTs: number, + limit = 30, +): string { + const messages = repo.recentBeforeTime(chatId, beforeTs, limit); + if (messages.length === 0) return "No messages before that date."; + return messages.map(formatMessage).join("\n"); +} + /** * Raw (wire-friendly) full-text search over a chat's history. Unlike * [searchHistory] — which formats a string for the agent's tool — this diff --git a/src/storage/repositories/history-repo.ts b/src/storage/repositories/history-repo.ts index 08d5d5f42..0df898adb 100644 --- a/src/storage/repositories/history-repo.ts +++ b/src/storage/repositories/history-repo.ts @@ -108,6 +108,17 @@ export function recentBefore( return rows.reverse().map(rowToMessage); } +export function recentBeforeTime( + chatId: string, + beforeTs: number, + limit: number, +): HistoryMessage[] { + const rows = getDatabase() + .prepare(historySql.recentBeforeTime) + .all(chatId, beforeTs, limit) as Row[]; + return rows.reverse().map(rowToMessage); +} + export function setFilePath( chatId: string, msgId: number, @@ -174,6 +185,19 @@ export function latestMsgId(chatId: string): number | undefined { return row?.msg_id; } +/** + * Highest msg_id over every chat whose id begins with `prefix` (compared + * literally — LIKE wildcards in the prefix are escaped). Undefined when no + * such chat has any history. + */ +export function maxMsgIdForPrefix(prefix: string): number | undefined { + const pattern = prefix.replace(/[\\%_]/g, (ch) => `\\${ch}`) + "%"; + const row = getDatabase() + .prepare(historySql.maxMsgIdForPrefix) + .get(pattern) as { max_id: number | null } | undefined; + return row?.max_id ?? undefined; +} + export type KnownUser = { senderId: number; name: string; diff --git a/src/storage/sql/history.sql b/src/storage/sql/history.sql index 5de74c0b6..5e1233370 100644 --- a/src/storage/sql/history.sql +++ b/src/storage/sql/history.sql @@ -21,6 +21,14 @@ SELECT msg_id, sender_id, sender_name, sender_handle, text, reply_to_msg_id, FROM history_messages WHERE chat_id = ? AND msg_id < ? ORDER BY id DESC LIMIT ? +-- name: recentBeforeTime +-- Time-cursor variant of recentBefore for the read_history `before` date +-- parameter: the newest `limit` messages strictly older than a timestamp. +SELECT msg_id, sender_id, sender_name, sender_handle, text, reply_to_msg_id, + timestamp, media_type, sticker_file_id, file_path +FROM history_messages +WHERE chat_id = ? AND timestamp < ? ORDER BY id DESC LIMIT ? + -- name: setFilePath UPDATE history_messages SET file_path = ? WHERE chat_id = ? AND msg_id = ? @@ -60,6 +68,12 @@ WHERE chat_id = ? AND sender_id = ? ORDER BY id DESC LIMIT ? -- name: latestMsgId SELECT msg_id FROM history_messages WHERE chat_id = ? ORDER BY id DESC LIMIT 1 +-- name: maxMsgIdForPrefix +-- Highest msg_id across every chat whose id starts with a prefix +-- (parameter is a LIKE pattern with \ escapes). Seeds the WhatsApp +-- frontend's in-memory id counter past what history already holds. +SELECT MAX(msg_id) AS max_id FROM history_messages WHERE chat_id LIKE ? ESCAPE '\' + -- name: knownUsers SELECT sender_id, MAX(timestamp) AS last_seen, diff --git a/src/storage/sql/statements.generated.ts b/src/storage/sql/statements.generated.ts index f63c1075d..922438761 100644 --- a/src/storage/sql/statements.generated.ts +++ b/src/storage/sql/statements.generated.ts @@ -346,6 +346,12 @@ SELECT msg_id, sender_id, sender_name, sender_handle, text, reply_to_msg_id, timestamp, media_type, sticker_file_id, file_path FROM history_messages WHERE chat_id = ? AND msg_id < ? ORDER BY id DESC LIMIT ?`, + recentBeforeTime: `-- Time-cursor variant of recentBefore for the read_history \`before\` date +-- parameter: the newest \`limit\` messages strictly older than a timestamp. +SELECT msg_id, sender_id, sender_name, sender_handle, text, reply_to_msg_id, + timestamp, media_type, sticker_file_id, file_path +FROM history_messages +WHERE chat_id = ? AND timestamp < ? ORDER BY id DESC LIMIT ?`, setFilePath: `UPDATE history_messages SET file_path = ? WHERE chat_id = ? AND msg_id = ?`, deleteChat: `DELETE FROM history_messages WHERE chat_id = ?`, searchFts: `-- The match param must already be a valid FTS5 expression @@ -371,6 +377,10 @@ WHERE chat_id = ? AND msg_id = ? ORDER BY id DESC LIMIT 1`, FROM history_messages WHERE chat_id = ? AND sender_id = ? ORDER BY id DESC LIMIT ?`, latestMsgId: `SELECT msg_id FROM history_messages WHERE chat_id = ? ORDER BY id DESC LIMIT 1`, + maxMsgIdForPrefix: `-- Highest msg_id across every chat whose id starts with a prefix +-- (parameter is a LIKE pattern with \\ escapes). Seeds the WhatsApp +-- frontend's in-memory id counter past what history already holds. +SELECT MAX(msg_id) AS max_id FROM history_messages WHERE chat_id LIKE ? ESCAPE '\\'`, knownUsers: `SELECT sender_id, MAX(timestamp) AS last_seen, COUNT(*) AS message_count,