feat(discord): ingest webview transcripts into memory (#1993)

app/src/services/__tests__/webviewAccountService.discord.test.ts

@@ -0,0 +1,113 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { store } from '../../store';
+import { addAccount, resetAccountsState } from '../../store/accountsSlice';
+import { fetchRespondQueue } from '../../store/providerSurfaceSlice';
+import { callCoreRpc } from '../coreRpcClient';
+import { startWebviewAccountService, stopWebviewAccountService } from '../webviewAccountService';
+
+type EventHandler = (evt: { payload: unknown }) => void;
+const listeners = new Map<string, EventHandler>();
+
+vi.mock('@tauri-apps/api/core', () => ({
+  invoke: vi.fn().mockResolvedValue(undefined),
+  isTauri: vi.fn().mockReturnValue(true),
+}));
+
+vi.mock('@tauri-apps/api/event', () => ({
+  listen: vi.fn(async (event: string, handler: EventHandler) => {
+    listeners.set(event, handler);
+    return () => listeners.delete(event);
+  }),
+}));
+
+vi.mock('../api/threadApi', () => ({ threadApi: { createNewThread: vi.fn() } }));
+vi.mock('../chatService', () => ({ chatSend: vi.fn() }));
+vi.mock('../coreRpcClient', () => ({ callCoreRpc: vi.fn().mockResolvedValue({}) }));
+vi.mock('../notificationService', () => ({ ingestNotification: vi.fn() }));
+vi.mock('../../utils/tauriCommands/common', () => ({ isTauri: vi.fn(() => true) }));
+vi.mock('../../store/providerSurfaceSlice', async importOriginal => {
+  const actual = await importOriginal<typeof import('../../store/providerSurfaceSlice')>();
+  return {
+    ...actual,
+    fetchRespondQueue: vi.fn(() => ({ type: 'providerSurface/fetchRespondQueue' })),
+  };
+});
+
+const ACCOUNT_ID = 'acct-discord-test';
+
+async function fireEvent(
+  kind: string,
+  payload: Record<string, unknown>,
+  provider = 'discord'
+): Promise<void> {
+  const handler = listeners.get('webview:event');
+  if (!handler) throw new Error('webview:event listener not attached');
+  handler({ payload: { account_id: ACCOUNT_ID, provider, kind, payload, ts: Date.now() } });
+  await new Promise(r => setTimeout(r, 0));
+}
+
+describe('webviewAccountService — Discord events', () => {
+  beforeEach(async () => {
+    listeners.clear();
+    vi.clearAllMocks();
+    store.dispatch(resetAccountsState());
+    store.dispatch(
+      addAccount({
+        id: ACCOUNT_ID,
+        provider: 'discord',
+        label: 'Discord',
+        createdAt: new Date().toISOString(),
+        status: 'closed',
+      })
+    );
+    stopWebviewAccountService();
+    startWebviewAccountService();
+    await new Promise(r => setTimeout(r, 0));
+  });
+
+  it('does not persist raw discord ingest transport events to memory', async () => {
+    await fireEvent('ingest', { source: 'cdp-ws', payload_data: '{"op":0}' });
+
+    expect(callCoreRpc).not.toHaveBeenCalled();
+    expect(fetchRespondQueue).not.toHaveBeenCalled();
+  });
+
+  it('persists generic non-discord ingest events through the legacy memory path', async () => {
+    await fireEvent(
+      'ingest',
+      { snapshotKey: 'snap-1', unread: 2, messages: [{ sender: 'Telegram Alice', body: 'Ping' }] },
+      'telegram'
+    );
+
+    expect(fetchRespondQueue).toHaveBeenCalledWith({ silent: true });
+    expect(callCoreRpc).toHaveBeenCalledWith(
+      expect.objectContaining({
+        method: 'openhuman.memory_doc_ingest',
+        params: expect.objectContaining({ namespace: 'webview:telegram:acct-discord-test' }),
+      })
+    );
+    expect(store.getState().accounts.messages[ACCOUNT_ID]).toEqual([
+      expect.objectContaining({ id: 'acct-discord-test:0', from: 'Telegram Alice', body: 'Ping' }),
+    ]);
+  });
+
+  it('refreshes queue for normalized discord transcript events without re-writing memory', async () => {
+    await fireEvent('discord_memory_ingest', {
+      channelId: '123',
+      channelName: 'general',
+      messages: [{ sender: 'Alice', body: 'Ship it', date: 1715000000 }],
+    });
+
+    expect(fetchRespondQueue).toHaveBeenCalledWith({ silent: true });
+    expect(callCoreRpc).not.toHaveBeenCalled();
+    expect(store.getState().accounts.messages[ACCOUNT_ID]).toEqual([
+      expect.objectContaining({
+        id: 'acct-discord-test:0',
+        from: 'Alice',
+        body: 'Ship it',
+        ts: 1715000000 * 1000,
+      }),
+    ]);
+  });
+});

app/src/services/webviewAccountService.ts

@@ -119,11 +119,13 @@ interface RecipeEventPayload {
 interface IngestMessage {
   id?: string;
   from?: string | null;
+  sender?: string | null;
   to?: string | null;
   fromMe?: boolean;
   body?: string | null;
   type?: string | null;
   timestamp?: number | null; // seconds since epoch
+  date?: number | null; // seconds since epoch
   unread?: number;
 }
 
@@ -138,6 +140,9 @@ interface IngestPayload {
   chatName?: string | null;
   day?: string; // YYYY-MM-DD UTC
   isSeed?: boolean;
+  channelId?: string;
+  channelName?: string | null;
+  guildId?: string | null;
 }
 
 interface LinkedInConversationPayload {
@@ -148,6 +153,12 @@ interface LinkedInConversationPayload {
   isSeed?: boolean;
 }
 
+interface DiscordMemoryIngestPayload extends IngestPayload {
+  channelId: string;
+  channelName?: string | null;
+  guildId?: string | null;
+}
+
 interface NotificationClickPayload {
   account_id: string;
   provider: string;
@@ -405,7 +416,7 @@ function handleRecipeEvent(evt: RecipeEventPayload) {
     const ingest = evt.payload as IngestPayload;
     const messages: IngestedMessage[] = (ingest.messages ?? []).map((m, idx) => ({
       id: m.id ?? `${accountId}:${idx}`,
-      from: m.from ?? null,
+      from: m.from ?? m.sender ?? null,
       body: m.body ?? null,
       unread: m.unread,
       ts: evt.ts ?? Date.now(),
@@ -413,13 +424,32 @@ function handleRecipeEvent(evt: RecipeEventPayload) {
 
     store.dispatch(appendMessages({ accountId, messages, unread: ingest.unread }));
 
-    // Tauri already forwarded this ingest to core; refresh queue immediately for Agent pane.
-    void store.dispatch(fetchRespondQueue({ silent: true }));
+    if (evt.provider !== 'discord') {
+      // Tauri already forwarded this ingest to core; refresh queue immediately for Agent pane.
+      void store.dispatch(fetchRespondQueue({ silent: true }));
 
-    // Fire-and-forget memory write via the existing core RPC.
-    // Namespace mirrors the skill-sync convention so the recall pipeline
-    // can find these alongside other ingested context.
-    void persistIngestToMemory(accountId, evt.provider, ingest, messages);
+      // Fire-and-forget memory write via the existing core RPC.
+      // Namespace mirrors the skill-sync convention so the recall pipeline
+      // can find these alongside other ingested context.
+      void persistIngestToMemory(accountId, evt.provider, ingest, messages);
+    }
+    return;
+  }
+
+  if (evt.kind === 'discord_memory_ingest') {
+    const ingest = evt.payload as unknown as DiscordMemoryIngestPayload;
+    const messages: IngestedMessage[] = (ingest.messages ?? []).map((m, idx) => ({
+      id: m.id ?? `${accountId}:${idx}`,
+      from: m.from ?? m.sender ?? null,
+      body: m.body ?? null,
+      unread: m.unread,
+      ts:
+        (m.date ?? m.timestamp ?? null)
+          ? (m.date ?? m.timestamp ?? 0) * 1000
+          : (evt.ts ?? Date.now()),
+    }));
+    store.dispatch(appendMessages({ accountId, messages, unread: ingest.unread }));
+    void store.dispatch(fetchRespondQueue({ silent: true }));
     return;
   }