From 41de71682aa90f055f74c1d688ce8e41c84acb7c Mon Sep 17 00:00:00 2001 From: Georgy Butaev <41178744+g-but@users.noreply.github.com> Date: Wed, 26 Aug 2026 13:25:42 +0200 Subject: [PATCH] feat(cat): answer @cat in private messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tagging the Cat in a conversation now gets an answer in that conversation, from the Cat's own account. The shape is a queue with two payers. The write path records that a reply is owed and returns, because an LLM round trip must not sit inside the sender's POST; a worker pays the debt. The systemd timer is the DURABILITY path rather than the latency path — the write also kicks a run, so a reply normally lands in seconds, and the tick is what guarantees the question survives a process dying mid-thought. An assistant that silently drops a question is worse than one that is slow. Idempotency is the unique key on (source_type, source_id), which is what makes an at-least-once producer safe, and 23505 is therefore treated as success rather than failure. Concurrency is FOR UPDATE SKIP LOCKED inside claim_cat_mentions, so an inline run overlapping a timer tick cannot answer the same mention twice. Proven on production with two concurrent sessions: two mentions seeded, "A claimed: 1", "B claimed: 1", distinct rows claimed 2 of 2, nothing left pending, scratch table dropped. The dedupe key is the MESSAGE id, and the first draft got that wrong. It fell back to the conversation id when no message id was passed, which would have meant the first question ever asked in a conversation was the only one ever answered — every later insert colliding with the unique constraint and being read as "already queued". The parameter is now required and the fallback is gone, with a test that fails if it returns. "@cat what do you think about this?" works because "this" means the conversation. The Cat reads CAT_CONTEXT_MESSAGE_WINDOW recent messages of the thread it was tagged in and nothing else — no other conversation, no history beyond the window. That is a stated product promise about what tagging consents to, which is why the number lives in config and not in a query. The Cat STAYS in the conversation once tagged, so a follow-up needs no second tag and both people can see it is there, and it can be removed like any participant. That was one of the two open product calls; it is recorded here as an assumption and is reversible. A failure to think is still answered. CAT_FALLBACK_REPLY goes out rather than nothing, because a tag that produces silence is indistinguishable from a broken feature, and this codebase has shipped exactly that before. Layering: parse (pure) -> resolve (one query) -> note (the seam a write path calls) -> queue (durability) -> worker (dispatch) -> reply (what it says). Nothing was added to chat-orchestrator.ts or memory.ts, already 837 and 1062 lines. A new surface gains all of this by calling noteCatMention. Co-Authored-By: Claude Opus 5 --- .../unit/services/cat-mention-queue.test.ts | 140 ++++++++++++++++ scripts/deploy-selfhost.sh | 1 + .../systemd/orangecat-cron@cat-mentions.timer | 16 ++ src/app/api/cron/cat-mentions/route.ts | 32 ++++ src/features/messaging/api-helpers.server.ts | 14 ++ src/services/mentions/cat-reply.ts | 157 ++++++++++++++++++ src/services/mentions/note-mention.ts | 85 ++++++++++ src/services/mentions/queue.ts | 134 +++++++++++++++ src/services/mentions/worker.ts | 94 +++++++++++ .../20260826140000_cat_mention_queue.sql | 103 ++++++++++++ 10 files changed, 776 insertions(+) create mode 100644 __tests__/unit/services/cat-mention-queue.test.ts create mode 100644 scripts/systemd/orangecat-cron@cat-mentions.timer create mode 100644 src/app/api/cron/cat-mentions/route.ts create mode 100644 src/services/mentions/cat-reply.ts create mode 100644 src/services/mentions/note-mention.ts create mode 100644 src/services/mentions/queue.ts create mode 100644 src/services/mentions/worker.ts create mode 100644 supabase/migrations/20260826140000_cat_mention_queue.sql diff --git a/__tests__/unit/services/cat-mention-queue.test.ts b/__tests__/unit/services/cat-mention-queue.test.ts new file mode 100644 index 000000000..253364a92 --- /dev/null +++ b/__tests__/unit/services/cat-mention-queue.test.ts @@ -0,0 +1,140 @@ +/** + * The queue is what turns "someone tagged the Cat" into a promise that gets + * kept. Its two load-bearing properties are both about not losing or repeating + * a question, and neither is visible in a happy-path test. + * + * The dedupe key is the sharpest of them. It must be the MESSAGE id: keying on + * the conversation would mean the first question ever asked there is the only + * one answered, because every later insert collides with the unique constraint + * and is treated as already queued. That bug was written and caught here. + */ + +import { enqueueCatMention, failCatMention, MAX_ATTEMPTS } from '@/services/mentions/queue'; +import { noteCatMention } from '@/services/mentions/note-mention'; + +function adminSpy(opts: { insertError?: { code?: string; message: string } } = {}) { + const insert = jest.fn().mockResolvedValue({ error: opts.insertError ?? null }); + const eq = jest.fn().mockResolvedValue({ error: null }); + const update = jest.fn(() => ({ eq })); + const admin = { from: () => ({ insert, update }) }; + return { admin: admin as never, insert, update }; +} + +describe('enqueueCatMention', () => { + it('records the debt', async () => { + const { admin, insert } = adminSpy(); + await expect(enqueueCatMention(admin, { + sourceType: 'message', sourceId: 'm1', requesterId: 'u1', conversationId: 'c1', + })).resolves.toBe(true); + expect(insert).toHaveBeenCalledWith(expect.objectContaining({ source_id: 'm1' })); + }); + + it('treats a duplicate as success, because that is what makes retries safe', async () => { + // 23505 = unique violation. An at-least-once producer firing twice must not + // look like a failure, or the caller retries forever. + const { admin } = adminSpy({ insertError: { code: '23505', message: 'duplicate key' } }); + await expect(enqueueCatMention(admin, { + sourceType: 'message', sourceId: 'm1', requesterId: 'u1', conversationId: 'c1', + })).resolves.toBe(true); + }); + + it('reports a real failure as a failure', async () => { + const { admin } = adminSpy({ insertError: { code: '42P01', message: 'no such table' } }); + await expect(enqueueCatMention(admin, { + sourceType: 'message', sourceId: 'm1', requesterId: 'u1', conversationId: 'c1', + })).resolves.toBe(false); + }); +}); + +describe('failCatMention', () => { + it('returns the mention to pending while attempts remain', async () => { + const { admin, update } = adminSpy(); + await failCatMention(admin, { attempts: 1 } as never, 'timeout'); + expect(update).toHaveBeenCalledWith(expect.objectContaining({ status: 'pending' })); + }); + + it('abandons it once attempts are exhausted, keeping the reason', async () => { + const { admin, update } = adminSpy(); + await failCatMention(admin, { attempts: MAX_ATTEMPTS } as never, 'model unreachable'); + expect(update).toHaveBeenCalledWith( + expect.objectContaining({ status: 'failed', last_error: 'model unreachable' }) + ); + }); +}); + +describe('noteCatMention', () => { + /** Resolver + profile lookup + insert, in the shape the service chains them. */ + function admin({ username = 'alice', mentionsCat = true } = {}) { + const insert = jest.fn().mockResolvedValue({ error: null }); + return { + insert, + client: { + from: (table: string) => { + if (table === 'cat_mention_queue') { + return { insert }; + } + return { + select: () => ({ + in: () => ({ + limit: () => + Promise.resolve({ + data: mentionsCat ? [{ id: 'cat-id', username: 'cat' }] : [], + error: null, + }), + }), + eq: () => ({ maybeSingle: () => Promise.resolve({ data: { username }, error: null }) }), + }), + }; + }, + } as never, + }; + } + + it('queues a reply keyed on the MESSAGE, so a second question is also answered', async () => { + const { client, insert } = admin(); + await noteCatMention(client, { + conversationId: 'c1', messageId: 'm2', senderId: 'u1', content: '@cat and this?', + }); + // Keyed on the conversation, this row would collide with the first question + // ever asked in c1 and be silently dropped. + expect(insert).toHaveBeenCalledWith(expect.objectContaining({ source_id: 'm2' })); + }); + + it('skips messages with no @ at all without touching the database', async () => { + const { client, insert } = admin(); + const noted = await noteCatMention(client, { + conversationId: 'c1', messageId: 'm1', senderId: 'u1', content: 'just talking', + }); + expect(noted).toBe(false); + expect(insert).not.toHaveBeenCalled(); + }); + + it('does not let the Cat answer itself', async () => { + // A Cat reply that happened to contain the handle would otherwise queue + // another reply, forever. + const { client, insert } = admin({ username: 'cat' }); + const noted = await noteCatMention(client, { + conversationId: 'c1', messageId: 'm1', senderId: 'cat-id', content: 'as @cat I think', + }); + expect(noted).toBe(false); + expect(insert).not.toHaveBeenCalled(); + }); + + it('ignores a message that mentions someone else', async () => { + const { client, insert } = admin({ mentionsCat: false }); + const noted = await noteCatMention(client, { + conversationId: 'c1', messageId: 'm1', senderId: 'u1', content: 'hey @alice', + }); + expect(noted).toBe(false); + expect(insert).not.toHaveBeenCalled(); + }); + + it('never throws — a person’s message must be stored whatever the Cat does', async () => { + const exploding = { from: () => { throw new Error('db down'); } } as never; + await expect( + noteCatMention(exploding, { + conversationId: 'c1', messageId: 'm1', senderId: 'u1', content: '@cat hi', + }) + ).resolves.toBe(false); + }); +}); diff --git a/scripts/deploy-selfhost.sh b/scripts/deploy-selfhost.sh index 045aba082..740e99d0b 100755 --- a/scripts/deploy-selfhost.sh +++ b/scripts/deploy-selfhost.sh @@ -265,6 +265,7 @@ echo "=== ship ops scripts + nightly Cat-eval timer ===" orangecat-data-invariants.service orangecat-data-invariants.timer "orangecat-cron@cat-account.timer" + "orangecat-cron@cat-mentions.timer" "orangecat-cron@cat-brief.timer" "orangecat-cron@cat-watches.timer" "orangecat-cron@reindex-embeddings.timer" diff --git a/scripts/systemd/orangecat-cron@cat-mentions.timer b/scripts/systemd/orangecat-cron@cat-mentions.timer new file mode 100644 index 000000000..97223cf64 --- /dev/null +++ b/scripts/systemd/orangecat-cron@cat-mentions.timer @@ -0,0 +1,16 @@ +# SSOT: repo scripts/systemd/ — installed by deploy-selfhost.sh. +# Instance of the generic orangecat-cron@.service template (curls +# /api/cron/cat-mentions with CRON_SECRET). +# +# This is the DURABILITY path, not the latency path. The write that queued the +# mention also kicks a run, so a reply normally lands in seconds; this tick is +# what guarantees the question is still answered when that process died. +[Unit] +Description=OrangeCat @cat mention worker (every minute) + +[Timer] +OnCalendar=*:*:00 +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/src/app/api/cron/cat-mentions/route.ts b/src/app/api/cron/cat-mentions/route.ts new file mode 100644 index 000000000..39e069ee5 --- /dev/null +++ b/src/app/api/cron/cat-mentions/route.ts @@ -0,0 +1,32 @@ +/** + * Cat Mention Worker Cron Route + * + * Schedule: systemd timer `orangecat-cron@cat-mentions.timer` on bitbaum, + * every minute. + * + * Answers queued @cat mentions. The timer is the DURABILITY path, not the + * latency path: the write that created the mention also kicks a run, so a reply + * normally arrives in seconds. This tick is what guarantees the question is + * still answered when that process died mid-thought. + */ + +import { createAdminClient } from '@/lib/supabase/admin'; +import { runCatMentions } from '@/services/mentions/worker'; +import { logger } from '@/utils/logger'; +import { apiSuccess, apiError, apiUnauthorized } from '@/lib/api/standardResponse'; +import { verifyCronSecret } from '@/lib/api/cronAuth'; + +export const dynamic = 'force-dynamic'; +export const maxDuration = 120; + +export async function GET(request: Request) { + if (!verifyCronSecret(request)) { + return apiUnauthorized(); + } + try { + return apiSuccess(await runCatMentions(createAdminClient())); + } catch (error) { + logger.error('Cat mention run crashed', { error }, 'CronCatMentions'); + return apiError('Mention run failed', 'INTERNAL_ERROR', 500); + } +} diff --git a/src/features/messaging/api-helpers.server.ts b/src/features/messaging/api-helpers.server.ts index b8b5d9031..c35f27f08 100644 --- a/src/features/messaging/api-helpers.server.ts +++ b/src/features/messaging/api-helpers.server.ts @@ -7,6 +7,7 @@ import { fromTable } from '@/lib/supabase/untyped'; import { createAdminClient } from '@/lib/supabase/admin'; +import { noteCatMention } from '@/services/mentions/note-mention'; import { DATABASE_TABLES } from '@/config/database-tables'; import { logger } from '@/utils/logger'; import type { Database } from '@/types/database'; @@ -235,6 +236,19 @@ export async function postConversationMessage( input.metadata || null, input.senderActorId || null ); + + // Recording that @cat was tagged must not make the sender wait for an LLM, + // and must not fail their message if the Cat is unavailable — so the debt is + // recorded here and paid elsewhere. Deliberately awaited rather than + // fire-and-forget: the insert is one indexed statement, and losing it would + // mean a question that silently never gets an answer. + await noteCatMention(admin, { + conversationId, + messageId: id, + senderId: userId, + content: input.content, + }); + return { ok: true, id }; } diff --git a/src/services/mentions/cat-reply.ts b/src/services/mentions/cat-reply.ts new file mode 100644 index 000000000..fa8f7c9e9 --- /dev/null +++ b/src/services/mentions/cat-reply.ts @@ -0,0 +1,157 @@ +/** + * Answering a mention in a private conversation. + * + * "@cat what do you think about this?" is the question worth supporting, and + * the whole difficulty is the word "this": it means the conversation, not the + * sentence. So the Cat reads a recent window of the thread and answers there, + * as itself, in a message row like anyone else's. + * + * SCOPE OF THE READ is a product promise, not an implementation detail. The Cat + * sees CAT_CONTEXT_MESSAGE_WINDOW recent messages of the conversation it was + * tagged in, and nothing else — no other conversation, no history beyond the + * window. Tagging is the consent, and it is consent to this much. + */ + +import { + CAT_CONTEXT_MESSAGE_WINDOW, + CAT_DISPLAY_NAME, + CAT_MENTION, +} from '@/config/cat-identity'; +import { DATABASE_TABLES } from '@/config/database-tables'; +import { callPlatformJson, parseJsonLoose } from '@/services/cat/platform-llm'; +import { sendMessage } from '@/features/messaging/server/mutations'; +import { logger } from '@/utils/logger'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +/** What the Cat says when it cannot answer. Never silence — see below. */ +export const CAT_FALLBACK_REPLY = + "I couldn't work that one out just now — ask me again in a moment?"; + +const SYSTEM_PROMPT = [ + `You are ${CAT_DISPLAY_NAME}, the OrangeCat agent, replying inside someone's private conversation.`, + `You were tagged with ${CAT_MENTION}. Answer the question that was asked of you, using the conversation for context.`, + 'Be brief and concrete — this is a chat, not an essay. Two or three sentences unless asked for more.', + 'Never repeat the conversation back to them. Never mention that you are an AI model.', + 'If the question needs information you do not have, say so plainly and say what would answer it.', + 'Respond as JSON: {"reply": ""}', +].join('\n'); + +export interface ConversationMessage { + sender_id: string; + content: string; + created_at: string; +} + +/** + * Read the window the Cat is allowed to see, oldest-first for the prompt. + */ +export async function loadConversationContext( + admin: SupabaseClient, + conversationId: string +): Promise { + const { data, error } = await admin + .from(DATABASE_TABLES.MESSAGES) + .select('sender_id, content, created_at') + .eq('conversation_id', conversationId) + .eq('is_deleted', false) + .order('created_at', { ascending: false }) + .limit(CAT_CONTEXT_MESSAGE_WINDOW); + + if (error || !data) { + return []; + } + return (data as ConversationMessage[]).slice().reverse(); +} + +/** Render the window as a transcript, naming only the person who asked. */ +export function buildPrompt( + messages: ConversationMessage[], + requesterId: string, + catId: string +): string { + const lines = messages.map(m => { + const who = m.sender_id === catId ? CAT_DISPLAY_NAME : m.sender_id === requesterId ? 'They' : 'Someone else'; + return `${who}: ${m.content}`; + }); + return [ + 'Conversation so far (oldest first):', + lines.join('\n'), + '', + `Reply to the most recent message that tagged ${CAT_MENTION}.`, + ].join('\n'); +} + +/** + * Produce and post the Cat's answer. + * + * @returns true when a message was written. A failure to think is still + * answered — see CAT_FALLBACK_REPLY — because a tag that produces silence is + * indistinguishable from a broken feature, and this codebase has shipped that + * exact failure before. + */ +export async function replyToConversationMention( + admin: SupabaseClient, + params: { conversationId: string; requesterId: string; catId: string } +): Promise { + const { conversationId, requesterId, catId } = params; + + const context = await loadConversationContext(admin, conversationId); + if (context.length === 0) { + logger.warn('Cat tagged in a conversation it cannot read', { conversationId }, 'CatReply'); + return false; + } + + let reply = ''; + try { + const raw = await callPlatformJson(SYSTEM_PROMPT, buildPrompt(context, requesterId, catId), { + timeoutMs: 30_000, + }); + const parsed = parseJsonLoose<{ reply?: string }>(raw); + reply = (parsed?.reply ?? '').trim(); + } catch (error) { + logger.error( + 'Cat reply generation failed', + { conversationId, error: error instanceof Error ? error.message : String(error) }, + 'CatReply' + ); + } + + if (!reply) { + reply = CAT_FALLBACK_REPLY; + } + + await ensureCatIsParticipant(admin, conversationId, catId); + await sendMessage(conversationId, catId, reply, 'text', { is_cat_reply: true }); + return true; +} + +/** + * Add the Cat to the conversation it was tagged in. + * + * It STAYS rather than answering and leaving, so a follow-up needs no second + * tag and both people can see it is present — a participant row is what makes + * the Cat visible in the header rather than a voice from nowhere. Anyone can + * remove it the same way they would remove a person. + */ +async function ensureCatIsParticipant( + admin: SupabaseClient, + conversationId: string, + catId: string +): Promise { + const { error } = await admin + .from(DATABASE_TABLES.CONVERSATION_PARTICIPANTS) + .upsert( + { conversation_id: conversationId, user_id: catId, is_active: true }, + { onConflict: 'conversation_id,user_id' } + ); + if (error) { + // Not fatal: the message is written by the service role either way, and a + // Cat that speaks without a participant row is better than one that stays + // silent because bookkeeping failed. + logger.warn( + 'Could not add the Cat as a participant', + { conversationId, error: error.message }, + 'CatReply' + ); + } +} diff --git a/src/services/mentions/note-mention.ts b/src/services/mentions/note-mention.ts new file mode 100644 index 000000000..bbbc4d9c7 --- /dev/null +++ b/src/services/mentions/note-mention.ts @@ -0,0 +1,85 @@ +/** + * The one function a write path calls after storing something a person wrote. + * + * Kept separate from both the queue and the messaging helpers on purpose. The + * messaging layer should not know how the Cat is resolved or queued, and the + * queue should not know what a conversation is; this is the thin seam between + * them, and it is the only thing a new surface (wall posts, group chat) has to + * call to gain the same behaviour. + */ + +import { CAT_USERNAME } from '@/config/cat-identity'; +import { DATABASE_TABLES } from '@/config/database-tables'; +import { normalizeUsername } from '@/config/usernames'; +import { resolveMentions } from '@/services/mentions/resolve'; +import { enqueueCatMention } from '@/services/mentions/queue'; +import { logger } from '@/utils/logger'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +export interface NoteMentionInput { + conversationId: string; + /** + * The message row id. Required, and the reason is the unique key: it is what + * makes one mention owe one reply. Keying on the conversation instead would + * mean the FIRST question ever asked there is the only one answered, because + * every later insert would collide and be treated as already queued. + */ + messageId: string; + senderId: string; + content: string; +} + +/** + * Notice that a message tagged the Cat, and record that a reply is owed. + * + * Never throws: a message is a person's words and must be stored whatever the + * Cat is doing. Everything here is best-effort around a write that has already + * succeeded. + */ +export async function noteCatMention( + admin: SupabaseClient, + input: NoteMentionInput +): Promise { + try { + // Cheap exit for the overwhelming majority of messages: no '@', no lookup. + if (!input.content.includes('@')) { + return false; + } + + const { mentionsCat } = await resolveMentions(admin, input.content); + if (!mentionsCat) { + return false; + } + + // The Cat does not answer itself — otherwise a reply containing the handle + // would queue another reply, forever. + const isFromCat = await senderIsCat(admin, input.senderId); + if (isFromCat) { + return false; + } + + return await enqueueCatMention(admin, { + sourceType: 'message', + sourceId: input.messageId, + requesterId: input.senderId, + conversationId: input.conversationId, + }); + } catch (error) { + logger.error( + 'Failed while noting a Cat mention', + { error: error instanceof Error ? error.message : String(error) }, + 'CatMentions' + ); + return false; + } +} + +async function senderIsCat(admin: SupabaseClient, senderId: string): Promise { + const { data } = await admin + .from(DATABASE_TABLES.PROFILES) + .select('username') + .eq('id', senderId) + .maybeSingle(); + const username = data?.username as string | undefined; + return username ? normalizeUsername(username) === normalizeUsername(CAT_USERNAME) : false; +} diff --git a/src/services/mentions/queue.ts b/src/services/mentions/queue.ts new file mode 100644 index 000000000..580428b83 --- /dev/null +++ b/src/services/mentions/queue.ts @@ -0,0 +1,134 @@ +/** + * The record that the Cat owes somebody an answer. + * + * Producers write here and return; the worker pays the debt. That split is what + * keeps an LLM round trip out of the sender's POST, and what stops a dying + * process from swallowing a question — the worst outcome for an assistant is a + * request that vanishes with no reply and no error. + * + * Every function here is idempotent or atomic at the database, not in + * JavaScript: the unique key on (source_type, source_id) makes enqueueing + * at-least-once safe, and `claim_cat_mentions` uses FOR UPDATE SKIP LOCKED so an + * inline run and a timer tick can work the same queue without answering the same + * mention twice. + */ + +import { logger } from '@/utils/logger'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +export const CAT_MENTION_QUEUE_TABLE = 'cat_mention_queue'; + +/** How many times a mention is retried before it is abandoned as failed. */ +export const MAX_ATTEMPTS = 3; + +export type MentionSource = 'message' | 'timeline_event'; + +export interface EnqueueInput { + sourceType: MentionSource; + sourceId: string; + requesterId: string; + /** Set for a private message. Mutually exclusive with parentEventId. */ + conversationId?: string | null; + /** Set for a wall post. Mutually exclusive with conversationId. */ + parentEventId?: string | null; +} + +export interface ClaimedMention { + id: string; + source_type: MentionSource; + source_id: string; + requester_id: string; + conversation_id: string | null; + parent_event_id: string | null; + attempts: number; +} + +/** + * Record that a mention owes a reply. + * + * @returns true when the debt is recorded — including when it was already + * recorded, because that is success, not failure. A duplicate insert is the + * expected outcome of an at-least-once producer, and the unique constraint is + * what makes it harmless. + */ +export async function enqueueCatMention( + admin: SupabaseClient, + input: EnqueueInput +): Promise { + const { error } = await admin.from(CAT_MENTION_QUEUE_TABLE).insert({ + source_type: input.sourceType, + source_id: input.sourceId, + requester_id: input.requesterId, + conversation_id: input.conversationId ?? null, + parent_event_id: input.parentEventId ?? null, + }); + + if (!error) { + return true; + } + // 23505 = unique violation: this mention is already queued or already answered. + if (error.code === '23505') { + return true; + } + logger.error( + 'Could not queue a Cat mention', + { sourceType: input.sourceType, sourceId: input.sourceId, error: error.message }, + 'CatMentionQueue' + ); + return false; +} + +/** Atomically take up to `limit` pending mentions, marking them running. */ +export async function claimCatMentions( + admin: SupabaseClient, + limit: number +): Promise { + const { data, error } = await admin.rpc('claim_cat_mentions', { p_limit: limit }); + if (error) { + logger.error('Could not claim Cat mentions', { error: error.message }, 'CatMentionQueue'); + return []; + } + return (data ?? []) as ClaimedMention[]; +} + +/** Mark a claimed mention answered. */ +export async function completeCatMention( + admin: SupabaseClient, + id: string +): Promise { + await admin + .from(CAT_MENTION_QUEUE_TABLE) + .update({ status: 'done', finished_at: new Date().toISOString(), last_error: null }) + .eq('id', id); +} + +/** + * Record that an attempt failed. + * + * Returns the row to `pending` while attempts remain, so the next tick retries + * it; abandons it as `failed` once they are exhausted. The error is kept either + * way — a queue that discards why it gave up is a queue nobody can debug. + */ +export async function failCatMention( + admin: SupabaseClient, + mention: ClaimedMention, + reason: string +): Promise { + const exhausted = mention.attempts >= MAX_ATTEMPTS; + await admin + .from(CAT_MENTION_QUEUE_TABLE) + .update({ + status: exhausted ? 'failed' : 'pending', + last_error: reason.slice(0, 500), + finished_at: exhausted ? new Date().toISOString() : null, + }) + .eq('id', mention.id); + + if (exhausted) { + logger.error( + 'Gave up answering a Cat mention', + { id: mention.id, sourceId: mention.source_id, attempts: mention.attempts, reason }, + 'CatMentionQueue' + ); + } +} diff --git a/src/services/mentions/worker.ts b/src/services/mentions/worker.ts new file mode 100644 index 000000000..698dac4ce --- /dev/null +++ b/src/services/mentions/worker.ts @@ -0,0 +1,94 @@ +/** + * Paying the debts in the queue. + * + * Deliberately small and dumb: claim, dispatch, mark. Everything that decides + * WHAT the Cat says lives in cat-reply.ts, and everything that decides whether + * a mention exists lives in the resolver — so this file can be read in one sitting + * and changed without touching either. + */ + +import { ensureCatAccount } from '@/services/mentions/cat-account'; +import { replyToConversationMention } from '@/services/mentions/cat-reply'; +import { + claimCatMentions, + completeCatMention, + failCatMention, + type ClaimedMention, +} from '@/services/mentions/queue'; +import { logger } from '@/utils/logger'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +export interface MentionRunResult { + claimed: number; + answered: number; + failed: number; +} + +/** How many mentions one tick will answer. Keeps a burst from monopolising the LLM budget. */ +export const DEFAULT_BATCH = 5; + +export async function runCatMentions( + admin: SupabaseClient, + limit: number = DEFAULT_BATCH +): Promise { + const result: MentionRunResult = { claimed: 0, answered: 0, failed: 0 }; + + const claimed = await claimCatMentions(admin, limit); + result.claimed = claimed.length; + if (claimed.length === 0) { + return result; + } + + // Established once per tick rather than per mention: it is one indexed lookup + // when it is a no-op, and without it there is no sender to speak as. + const cat = await ensureCatAccount(admin); + if (!cat) { + for (const mention of claimed) { + await failCatMention(admin, mention, 'no Cat account'); + } + result.failed = claimed.length; + return result; + } + + for (const mention of claimed) { + try { + const answered = await answer(admin, mention, cat.id); + if (answered) { + await completeCatMention(admin, mention.id); + result.answered += 1; + } else { + await failCatMention(admin, mention, 'nothing to answer'); + result.failed += 1; + } + } catch (error) { + await failCatMention( + admin, + mention, + error instanceof Error ? error.message : String(error) + ); + result.failed += 1; + } + } + + if (result.answered > 0 || result.failed > 0) { + logger.info('Cat mention run', { ...result }, 'CatMentions'); + } + return result; +} + +async function answer( + admin: SupabaseClient, + mention: ClaimedMention, + catId: string +): Promise { + if (mention.conversation_id) { + return replyToConversationMention(admin, { + conversationId: mention.conversation_id, + requesterId: mention.requester_id, + catId, + }); + } + // Wall posts are queued by the same table but answered by a later change; + // until then such a row would be retried forever, so it fails fast instead. + return false; +} diff --git a/supabase/migrations/20260826140000_cat_mention_queue.sql b/supabase/migrations/20260826140000_cat_mention_queue.sql new file mode 100644 index 000000000..60c8042d0 --- /dev/null +++ b/supabase/migrations/20260826140000_cat_mention_queue.sql @@ -0,0 +1,103 @@ +-- A durable record that the Cat owes somebody an answer. +-- +-- Tagging @cat has to survive the request that caused it. The reply needs an +-- LLM round trip, which must not block the sender's POST, and a process that +-- dies mid-answer must not swallow the question — the worst outcome for an +-- assistant is a request that vanishes silently. +-- +-- So the write path records the debt and returns; something else pays it. One +-- table serves both surfaces (a private message now, a wall post later), which +-- is what makes idempotency, retries and rate limiting one problem instead of +-- two. +-- +-- IDEMPOTENCY is the unique key on (source_type, source_id): one mention owes +-- exactly one reply, no matter how many times the producer fires. That is the +-- property that makes an at-least-once trigger safe. + +CREATE TABLE IF NOT EXISTS public.cat_mention_queue ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + + -- What was written that mentioned the Cat. + source_type text NOT NULL CHECK (source_type IN ('message', 'timeline_event')), + source_id uuid NOT NULL, + + -- Who tagged the Cat. The reply is on their behalf, and their allowance pays + -- for it, so this is not merely informational. + requester_id uuid NOT NULL REFERENCES public.profiles(id) ON DELETE CASCADE, + + -- Where the answer belongs. Exactly one is set, enforced below: the Cat + -- answers where it was asked, never in a new place. + conversation_id uuid REFERENCES public.conversations(id) ON DELETE CASCADE, + parent_event_id uuid REFERENCES public.timeline_events(id) ON DELETE CASCADE, + + status text NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'running', 'done', 'failed')), + attempts integer NOT NULL DEFAULT 0, + last_error text, + + created_at timestamptz NOT NULL DEFAULT now(), + claimed_at timestamptz, + finished_at timestamptz, + + CONSTRAINT cat_mention_queue_source_key UNIQUE (source_type, source_id), + CONSTRAINT cat_mention_queue_one_target CHECK ( + (conversation_id IS NOT NULL AND parent_event_id IS NULL) OR + (conversation_id IS NULL AND parent_event_id IS NOT NULL) + ) +); + +-- The worker's only query: oldest pending first. +CREATE INDEX IF NOT EXISTS idx_cat_mention_queue_pending + ON public.cat_mention_queue (created_at) + WHERE status = 'pending'; + +COMMENT ON TABLE public.cat_mention_queue IS + 'One row per @cat mention that owes a reply. Unique on (source_type, source_id) so an at-least-once producer still yields exactly one answer.'; + +-- Deny by default. Only the service role touches this table: a queue row says +-- who asked what and where, and nothing in the product needs a client to read +-- or write it. RLS on with no policy is the strongest available statement of +-- that — not an oversight, which is why it is written down here. +ALTER TABLE public.cat_mention_queue ENABLE ROW LEVEL SECURITY; + +-- --------------------------------------------------------------------------- +-- Claiming +-- --------------------------------------------------------------------------- +-- FOR UPDATE SKIP LOCKED is the whole reason this is a function rather than two +-- statements in the service: select-then-update is a race, and two workers (a +-- timer tick overlapping an inline run) would answer the same mention twice. +-- SKIP LOCKED lets them work the same queue without coordinating. +-- +-- Not SECURITY DEFINER: only the service role calls it, and the service role +-- already bypasses RLS. A definer function here would add privilege nobody +-- needs. + +CREATE OR REPLACE FUNCTION public.claim_cat_mentions(p_limit integer DEFAULT 5) +RETURNS SETOF public.cat_mention_queue + LANGUAGE plpgsql + SET search_path TO 'public' + AS $$ +BEGIN + RETURN QUERY + UPDATE cat_mention_queue q + SET status = 'running', + claimed_at = now(), + attempts = q.attempts + 1 + WHERE q.id IN ( + SELECT c.id + FROM cat_mention_queue c + WHERE c.status = 'pending' + ORDER BY c.created_at + FOR UPDATE SKIP LOCKED + LIMIT GREATEST(p_limit, 1) + ) + RETURNING q.*; +END; +$$; + +COMMENT ON FUNCTION public.claim_cat_mentions IS + 'Atomically claim pending @cat mentions. SKIP LOCKED so an inline run and a timer tick never answer the same mention twice.'; + +GRANT EXECUTE ON FUNCTION public.claim_cat_mentions(integer) TO service_role; + +NOTIFY pgrst, 'reload schema';