feat(cat): answer @cat in private messages

__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);
+  });
+});

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"

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

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);
+  }
+}

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 };
 }
 

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": "<your message>"}',
+].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<ConversationMessage[]> {
+  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<boolean> {
+  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<void> {
+  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'
+    );
+  }
+}