Feat/win ai clone beeper

desktop/windows/README.md

@@ -34,6 +34,30 @@ keys to obtain.
   local `.env` set `MAIN_VITE_GOOGLE_CLIENT_ID`, `MAIN_VITE_GOOGLE_CLIENT_SECRET`, and
   `VITE_ENABLE_GOOGLE_INTEGRATION=1`. Keep these in your local `.env` only — never commit them.
 
+## AI Clone (Beeper)
+
+Omi can answer your chat apps (WhatsApp, Telegram, Signal, …) on your behalf. It connects
+to [Beeper Desktop](https://www.beeper.com)'s local API — Beeper bundles all your chat
+networks into one inbox on your PC, and Omi talks to it on `localhost:23373` only.
+
+Setup:
+
+1. Install Beeper Desktop, sign in, and connect your chat networks inside Beeper.
+2. In Beeper: **Settings → Integrations → Approved connections → “+”** to create an
+   access token.
+3. In Omi: **AI Clone** in the sidebar → paste the token → Connect → turn on
+   **Respond on my behalf**.
+
+Per chat you choose **Off**, **Draft** (Omi writes the reply; you approve/edit it in the
+AI Clone inbox), or **Auto** (Omi sends immediately — opt-in per chat, never for group
+chats, capped per hour). Replies are generated by Omi's memory-grounded chat
+(`/v2/messages`), falling back to the desktop chat-completions lane when the account's
+monthly chat limit is hit.
+
+Caveats: Beeper Desktop must be running for the clone to see or send messages. iMessage
+requires a Mac somewhere in your Beeper setup — on a Windows-only setup use the other
+networks.
+
 ## Optional keys
 
 Everything below is blank in `.env.example` and safe to leave unset:

desktop/windows/src/main/aiClone/beeperClient.test.ts

@@ -0,0 +1,74 @@
+import { describe, it, expect, vi, afterEach } from 'vitest'
+import { BeeperClient, beeperTimestampMs } from './beeperClient'
+
+const jsonResponse = (body: unknown, status = 200): Response =>
+  new Response(JSON.stringify(body), {
+    status,
+    headers: { 'Content-Type': 'application/json' }
+  })
+
+afterEach(() => vi.unstubAllGlobals())
+
+describe('BeeperClient', () => {
+  it('sends the bearer token and parses {items:[…]} chat lists', async () => {
+    const fetchMock = vi.fn().mockResolvedValue(
+      jsonResponse({ items: [{ id: 'c1', title: 'Alice', network: 'WhatsApp', type: 'single' }] })
+    )
+    vi.stubGlobal('fetch', fetchMock)
+
+    const r = await new BeeperClient('tok').listChats()
+    expect(r).toEqual({
+      ok: true,
+      value: [{ id: 'c1', title: 'Alice', network: 'WhatsApp', type: 'single' }]
+    })
+    const [url, init] = fetchMock.mock.calls[0]
+    expect(url).toBe('http://localhost:23373/v1/chats')
+    expect(init.headers.Authorization).toBe('Bearer tok')
+  })
+
+  it('POSTs a send with text (and threads replyToMessageID when given)', async () => {
+    const fetchMock = vi.fn().mockResolvedValue(jsonResponse({ pendingMessageID: 'p1' }))
+    vi.stubGlobal('fetch', fetchMock)
+
+    const r = await new BeeperClient('tok').sendMessage('chat/1', 'hello', 'm9')
+    expect(r).toEqual({ ok: true, value: { pendingMessageID: 'p1' } })
+    const [url, init] = fetchMock.mock.calls[0]
+    expect(url).toBe('http://localhost:23373/v1/chats/chat%2F1/messages')
+    expect(init.method).toBe('POST')
+    expect(JSON.parse(init.body)).toEqual({ text: 'hello', replyToMessageID: 'm9' })
+  })
+
+  it('classifies 401 as unauthorized and connection failure as unreachable', async () => {
+    vi.stubGlobal('fetch', vi.fn().mockResolvedValue(jsonResponse({}, 401)))
+    expect(await new BeeperClient('bad').validateToken()).toEqual({
+      ok: false,
+      error: 'unauthorized'
+    })
+
+    vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('ECONNREFUSED')))
+    const r = await new BeeperClient('tok').validateToken()
+    expect(r.ok).toBe(false)
+    if (!r.ok) expect(r.error).toBe('unreachable')
+  })
+
+  it('classifies other HTTP failures as http_error with detail', async () => {
+    vi.stubGlobal('fetch', vi.fn().mockResolvedValue(jsonResponse({}, 500)))
+    const r = await new BeeperClient('tok').listMessages('c1')
+    expect(r.ok).toBe(false)
+    if (!r.ok) {
+      expect(r.error).toBe('http_error')
+      expect(r.detail).toContain('500')
+    }
+  })
+})
+
+describe('beeperTimestampMs', () => {
+  it('passes numbers through, parses ISO strings, rejects garbage', () => {
+    expect(beeperTimestampMs(1739320000000)).toBe(1739320000000)
+    expect(beeperTimestampMs('2025-02-12T00:00:00.000Z')).toBe(
+      Date.parse('2025-02-12T00:00:00.000Z')
+    )
+    expect(beeperTimestampMs('not-a-date')).toBeUndefined()
+    expect(beeperTimestampMs(undefined)).toBeUndefined()
+  })
+})

desktop/windows/src/main/aiClone/beeperClient.ts

@@ -0,0 +1,187 @@
+// Thin client for the Beeper Desktop API — a localhost REST + WebSocket server
+// that Beeper Desktop runs on port 23373. The user creates an access token in
+// Beeper → Settings → Integrations ("Approved connections"); everything stays
+// on-device. Docs: https://developers.beeper.com/desktop-api
+import WebSocket from 'ws'
+
+export const BEEPER_BASE_URL = 'http://localhost:23373'
+
+/** Raw Beeper wire shapes (subset of fields the clone uses). */
+export type BeeperChat = {
+  id: string
+  title?: string
+  network?: string
+  type?: 'single' | 'group'
+  lastActivity?: string | number
+  isArchived?: boolean
+  isMuted?: boolean
+}
+
+export type BeeperMessage = {
+  id: string
+  chatID?: string
+  accountID?: string
+  text?: string
+  type?: string
+  senderID?: string
+  senderName?: string
+  /** True if the authenticated user sent the message. */
+  isSender?: boolean
+  timestamp?: string | number
+  isDeleted?: boolean
+}
+
+/** message.upserted frame from ws://…/v1/ws (subscriptions.set protocol). */
+export type BeeperWsEvent = {
+  type: string
+  chatID?: string
+  ids?: string[]
+  entries?: BeeperMessage[]
+}
+
+export type BeeperResult<T> =
+  | { ok: true; value: T }
+  | { ok: false; error: 'unreachable' | 'unauthorized' | 'http_error'; detail?: string }
+
+/** Normalize a Beeper timestamp (ISO string or ms epoch) to ms, or undefined. */
+export function beeperTimestampMs(ts: string | number | undefined): number | undefined {
+  if (ts === undefined || ts === null) return undefined
+  if (typeof ts === 'number') return ts
+  const parsed = Date.parse(ts)
+  return Number.isNaN(parsed) ? undefined : parsed
+}
+
+/** Pull the item array out of a Beeper list response ({items:[…]} or bare []). */
+function listItems<T>(body: unknown): T[] {
+  if (Array.isArray(body)) return body as T[]
+  const items = (body as { items?: T[] })?.items
+  return Array.isArray(items) ? items : []
+}
+
+export class BeeperClient {
+  constructor(
+    private token: string,
+    private baseUrl: string = BEEPER_BASE_URL
+  ) {}
+
+  private async request<T>(
+    method: 'GET' | 'POST',
+    path: string,
+    body?: unknown
+  ): Promise<BeeperResult<T>> {
+    let res: Response
+    try {
+      res = await fetch(`${this.baseUrl}${path}`, {
+        method,
+        headers: {
+          Authorization: `Bearer ${this.token}`,
+          ...(body !== undefined ? { 'Content-Type': 'application/json' } : {})
+        },
+        ...(body !== undefined ? { body: JSON.stringify(body) } : {})
+      })
+    } catch (e) {
+      // Connection refused etc. — Beeper Desktop isn't running.
+      return { ok: false, error: 'unreachable', detail: (e as Error).message }
+    }
+    if (res.status === 401 || res.status === 403) return { ok: false, error: 'unauthorized' }
+    if (!res.ok) {
+      return { ok: false, error: 'http_error', detail: `HTTP ${res.status} on ${path}` }
+    }
+    try {
+      return { ok: true, value: (await res.json()) as T }
+    } catch {
+      return { ok: true, value: undefined as T }
+    }
+  }
+
+  /** Cheap probe used by Connect and by reachability checks. */
+  async validateToken(): Promise<BeeperResult<void>> {
+    const r = await this.request<unknown>('GET', '/v1/chats')
+    return r.ok ? { ok: true, value: undefined } : r
+  }
+
+  async listChats(): Promise<BeeperResult<BeeperChat[]>> {
+    const r = await this.request<unknown>('GET', '/v1/chats')
+    return r.ok ? { ok: true, value: listItems<BeeperChat>(r.value) } : r
+  }
+
+  /** Latest messages in a chat (Beeper returns newest-first pages). */
+  async listMessages(chatID: string): Promise<BeeperResult<BeeperMessage[]>> {
+    const r = await this.request<unknown>(
+      'GET',
+      `/v1/chats/${encodeURIComponent(chatID)}/messages`
+    )
+    return r.ok ? { ok: true, value: listItems<BeeperMessage>(r.value) } : r
+  }
+
+  async sendMessage(
+    chatID: string,
+    text: string,
+    replyToMessageID?: string
+  ): Promise<BeeperResult<{ pendingMessageID?: string }>> {
+    return this.request('POST', `/v1/chats/${encodeURIComponent(chatID)}/messages`, {
+      text,
+      ...(replyToMessageID ? { replyToMessageID } : {})
+    })
+  }
+
+  /**
+   * Subscribe to message.upserted events for every chat. Reconnects with capped
+   * backoff until closed; `onDown` fires when the socket drops (so the caller
+   * can surface "Beeper unreachable" and/or fall back to polling).
+   */
+  subscribe(handlers: {
+    onEvent: (e: BeeperWsEvent) => void
+    onUp?: () => void
+    onDown?: (reason: string) => void
+  }): { close: () => void } {
+    let ws: WebSocket | null = null
+    let closed = false
+    let attempt = 0
+    let retryTimer: NodeJS.Timeout | null = null
+
+    const connect = (): void => {
+      if (closed) return
+      ws = new WebSocket(`${this.baseUrl.replace(/^http/, 'ws')}/v1/ws`, {
+        headers: { Authorization: `Bearer ${this.token}` }
+      })
+      ws.on('open', () => {
+        attempt = 0
+        ws?.send(JSON.stringify({ type: 'subscriptions.set', requestID: 'r1', chatIDs: ['*'] }))
+        handlers.onUp?.()
+      })
+      ws.on('message', (data) => {
+        try {
+          const frame = JSON.parse(String(data)) as BeeperWsEvent
+          if (frame?.type) handlers.onEvent(frame)
+        } catch {
+          /* ignore non-JSON frames */
+        }
+      })
+      const scheduleRetry = (reason: string): void => {
+        if (closed || retryTimer) return
+        handlers.onDown?.(reason)
+        const delay = Math.min(30_000, 1_000 * 2 ** attempt++)
+        retryTimer = setTimeout(() => {
+          retryTimer = null
+          connect()
+        }, delay)
+      }
+      ws.on('close', () => scheduleRetry('closed'))
+      ws.on('error', (e) => scheduleRetry(e.message))
+    }
+
+    connect()
+    return {
+      close: () => {
+        closed = true
+        if (retryTimer) clearTimeout(retryTimer)
+        try {
+          ws?.close()
+        } catch {
+          /* already closed */
+        }
+      }
+    }
+  }
+}

desktop/windows/src/main/aiClone/chatTaskQueue.test.ts

@@ -0,0 +1,87 @@
+import { describe, it, expect, vi } from 'vitest'
+import { ChatTaskQueue } from './chatTaskQueue'
+
+function gate(): { promise: Promise<void>; open: () => void } {
+  let open!: () => void
+  const promise = new Promise<void>((r) => (open = r))
+  return { promise, open }
+}
+
+describe('ChatTaskQueue', () => {
+  it('runs a message arriving mid-generation after the current one (regression: two rapid events used to drop the second)', async () => {
+    const q = new ChatTaskQueue()
+    const firstReply = gate()
+    const ran: string[] = []
+
+    // Two rapid message.upserted events for the same chat: m2 lands while m1's
+    // reply is still generating.
+    q.submit('chat1', async () => {
+      await firstReply.promise
+      ran.push('m1')
+    })
+    q.submit('chat1', async () => {
+      ran.push('m2')
+    })
+
+    expect(ran).toEqual([]) // m2 parked — not run early, not dropped
+    firstReply.open()
+    await vi.waitFor(() => expect(ran).toEqual(['m1', 'm2']))
+  })
+
+  it('coalesces parked tasks — only the newest superseding message runs', async () => {
+    const q = new ChatTaskQueue()
+    const firstReply = gate()
+    const ran: string[] = []
+
+    q.submit('chat1', async () => {
+      await firstReply.promise
+      ran.push('m1')
+    })
+    q.submit('chat1', async () => {
+      ran.push('m2')
+    })
+    q.submit('chat1', async () => {
+      ran.push('m3')
+    })
+
+    firstReply.open()
+    await vi.waitFor(() => expect(ran).toEqual(['m1', 'm3'])) // m2 superseded by m3
+  })
+
+  it('does not serialize across different chats', async () => {
+    const q = new ChatTaskQueue()
+    const blocked = gate()
+    const ran: string[] = []
+
+    q.submit('chat1', async () => {
+      await blocked.promise
+      ran.push('chat1')
+    })
+    q.submit('chat2', async () => {
+      ran.push('chat2')
+    })
+
+    await vi.waitFor(() => expect(ran).toEqual(['chat2'])) // chat2 not stuck behind chat1
+    blocked.open()
+    await vi.waitFor(() => expect(ran).toEqual(['chat2', 'chat1']))
+  })
+
+  it('releases the chat and drains the parked task when a task throws', async () => {
+    const q = new ChatTaskQueue()
+    const ran: string[] = []
+
+    q.submit('chat1', async () => {
+      throw new Error('generation failed')
+    })
+    q.submit('chat1', async () => {
+      ran.push('m2')
+    })
+
+    await vi.waitFor(() => expect(ran).toEqual(['m2']))
+    // Chat is free again for the next message.
+    q.submit('chat1', async () => {
+      ran.push('m3')
+    })
+    await vi.waitFor(() => expect(ran).toEqual(['m2', 'm3']))
+  })
+})