diff --git a/integrations/wearable-limitless-capture/README.md b/integrations/wearable-limitless-capture/README.md new file mode 100644 index 000000000..a066811a2 --- /dev/null +++ b/integrations/wearable-limitless-capture/README.md @@ -0,0 +1,361 @@ +# Limitless Wearable Capture + +> **Land your Limitless Pendant lifelogs in your Open Brain — section by +> section.** A scheduled poller pulls recent recordings from the Limitless API +> and atomizes each one into a title atom plus one atom per `##` section heading +> (its label with the rolled-up utterances), each attributed to its speakers and +> individually searchable. No per-item LLM cost. + +--- + +## What It Does + +A Supabase Edge Function runs on a schedule (every 5 minutes) and asks the +Limitless lifelog API for recordings in a rolling time window. Limitless returns +each lifelog as Markdown — a title, `##` section headings, and +`- Speaker (time): text` transcript bullets. This adapter **atomizes** that +device-native structure (no LLM call): + +- one **`title`** atom — the lifelog title, `machine`-generated; +- one **`section`** atom per heading — the section label plus its rolled-up + utterances, attributed `self` / `other` / `mixed` / `unknown` by who spoke in + it. + +There's **no cap on sections** — they're Limitless's native unit, so a long +lifelog simply yields more section atoms. Capturing at the section level (rather +than one summary per recording) means each topic becomes its own searchable row, +attributed to the people in it. + +The shared **wearable-capture-core** engine owns the write path — per-atom dedup +on a salted fingerprint, provenance metadata, embedding via OpenRouter, and the +insert. This adapter only knows how to _list_ Limitless lifelogs and _atomize_ +one. + +Because dedup keys on the device's stable lifelog id plus each atom's position +(not raw content), overlapping poll windows and re-runs are safe, and a missed +run self-heals on the next pass — no local state file. + +--- + +## Prerequisites + +- **The [wearable-capture-core](../wearable-capture-core/) engine installed + first.** This adapter imports it from `../_shared/wearable-sync.ts`; without + it, the function won't deploy. Follow that README through Step 2 (it also sets + `OPENROUTER_API_KEY`, which this adapter relies on for embeddings). (For + convenience, this folder bundles an identical copy of the engine in `_shared/` + so the function typechecks standalone — the `deno.json` import map points the + deploy path at it for local `deno check`.) +- A working Open Brain setup (Supabase project with the `thoughts` table and + pgvector). +- A **Limitless** account with a Pendant and an API key. Get the key from the + Limitless app → **Developer settings** (Settings → Developer / API). +- An [OpenRouter](https://openrouter.ai) API key (set when you install the core + — used for embeddings). +- Supabase CLI installed and logged in. + +**Cost**: Limitless API access is included with your Limitless subscription. The +only marginal cost here is OpenRouter embeddings (no per-item LLM classification +— the adapter reuses the device's own structure). Each lifelog now yields +several section atoms instead of one summary, so expect roughly +**$0.05–0.20/month** for typical personal volume — still embeddings-only. + +--- + +## Credential Tracker + +Fill these in as you go — you'll need them in Steps 2 and 5: + +| Credential | Where it comes from | Value | +| ----------------------------------- | -------------------------------------------------------------------------------------------- | ----------- | +| `LIMITLESS_API_KEY` | Limitless app → Developer settings (Step 2) | | +| `OPENROUTER_API_KEY` | Set when installing wearable-capture-core ([openrouter.ai/keys](https://openrouter.ai/keys)) | (from core) | +| `WEARABLE_SELF_LABELS` _(optional)_ | Comma-separated speaker labels that are _you_, merged with the device-generic `you/me/self` | | +| `SUPABASE_URL` | Auto-injected by Supabase | (skip) | +| `SUPABASE_SERVICE_ROLE_KEY` | Auto-injected by Supabase | (skip) | + +> [!WARNING] +> The Limitless API key is a credential. Set it with `supabase secrets set` — +> never paste it into code, commits, or screenshots. + +--- + +## Steps + +### Step 1 — Install the wearable-capture-core engine + +This adapter is built on the shared engine and can't run without it. + +Follow the [wearable-capture-core README](../wearable-capture-core/) through +**Step 2**. That gives you: + +- `supabase/functions/_shared/wearable-sync.ts` (the engine this function + imports), and +- `OPENROUTER_API_KEY` set as a Supabase secret (used for embeddings). + +✅ **Done when:** `supabase/functions/_shared/wearable-sync.ts` exists and +`supabase secrets list` shows `OPENROUTER_API_KEY`. + +--- + +### Step 2 — Get your Limitless API key and set it + +1. Open the Limitless app and go to **Settings → Developer settings** (sometimes + labelled API). +2. Create / copy your API key. +3. Set it as a Supabase secret (replace the placeholder with your real key, no + angle brackets): + +```bash +supabase secrets set LIMITLESS_API_KEY="your_limitless_api_key" +# optional: label your own speech so section atoms attribute to "self" +supabase secrets set WEARABLE_SELF_LABELS="Your Name,Nickname" +``` + +`SUPABASE_URL` and `SUPABASE_SERVICE_ROLE_KEY` are injected automatically by the +Supabase runtime, so you don't set those yourself. + +✅ **Done when:** `supabase secrets list` shows `LIMITLESS_API_KEY` (and +`OPENROUTER_API_KEY` from Step 1). + +--- + +### Step 3 — Drop the function into your Supabase project + +From the root of your Supabase project: + +```bash +mkdir -p supabase/functions/wearable-limitless-capture +``` + +Copy [`index.ts`](./index.ts) from this folder to +`supabase/functions/wearable-limitless-capture/index.ts`. It imports the engine +from `../_shared/wearable-sync.ts`, so the relative path lines up once the core +is in place from Step 1: + +```typescript +import { + type Attribution, + fetchWithRetry, + runWearableSync, + type WearableAdapter, + type WearableAtom, +} from "../_shared/wearable-sync.ts"; +``` + +The function defines a Limitless adapter (`sourceId: "limitless"`, +`sourceType: "limitless_lifelog"`) whose `recordToAtoms` parses the lifelog +Markdown into a title atom and one section atom per heading. `Deno.serve` calls +`runWearableSync(limitlessAdapter, { sinceHours: 12 })` and returns the engine's +result as JSON. It accepts `?dry_run=1` (compute, write nothing) and +`?since_hours=N` (override the 12-hour window) for manual testing. + +✅ **Done when:** The file exists at +`supabase/functions/wearable-limitless-capture/index.ts` and +`deno check index.ts _shared/wearable-sync.ts` is clean. + +--- + +### Step 4 — Deploy the edge function + +```bash +supabase functions deploy wearable-limitless-capture +``` + +Your function URL will look like: + +``` +https://YOUR_PROJECT_REF.supabase.co/functions/v1/wearable-limitless-capture +``` + +(where `YOUR_PROJECT_REF` is the subdomain of your actual Supabase project). +Keep the full URL handy for Step 5. + +You can trigger it once by hand to confirm it runs (the function reads its +credentials from secrets, so no auth header is needed for the smoke test if you +deployed with `--no-verify-jwt`; otherwise call it from the scheduled job in +Step 5): + +```bash +curl -X POST "https://YOUR_PROJECT_REF.supabase.co/functions/v1/wearable-limitless-capture?dry_run=1" +``` + +A healthy dry run returns JSON like +`{"source":"limitless","pulled":3,"recordsImported":3,"atomsInserted":18,"atomsSkipped":0,"failed":0,"attribution":{"machine":3,"self":7,"mixed":8},"dryRun":true}`. + +✅ **Done when:** `supabase functions deploy` prints a success URL and a manual +invocation returns a JSON result object. + +--- + +### Step 5 — Schedule the poller (every 5 minutes) + +Limitless has no webhook, so we poll. Use `pg_cron` + `net.http_post` to hit the +function URL on a cron. Run this SQL in the Supabase SQL editor (enable the +`pg_cron` and `pg_net` extensions first if they aren't already): + +```sql +-- Enable the extensions (no-op if already enabled) +create extension if not exists pg_cron; +create extension if not exists pg_net; + +-- Poll Limitless every 5 minutes +select cron.schedule( + 'wearable-limitless-capture', + '*/5 * * * *', + $$ + select net.http_post( + url := 'https://YOUR_PROJECT_REF.supabase.co/functions/v1/wearable-limitless-capture', + headers := jsonb_build_object( + 'Content-Type', 'application/json', + 'Authorization', 'Bearer ' || current_setting('app.settings.service_role_key', true) + ), + body := '{}'::jsonb + ); + $$ +); +``` + +> [!IMPORTANT] +> `OPENROUTER_API_KEY` must already be set (you set it while installing +> wearable-capture-core in Step 1) — the engine uses it to embed each atom at +> capture time. If it's missing, rows still insert but with a NULL embedding for +> a later backfill. + + + +> [!NOTE] +> Replace `YOUR_PROJECT_REF` with your project subdomain. The `sinceHours: 12` +> lookback in the function means the 5-minute cron has a wide overlap; the +> engine's per-atom salted-fingerprint dedup makes that overlap free of +> duplicates and lets a missed run self-heal. + +To change or remove the schedule later: + +```sql +-- Inspect +select * from cron.job where jobname = 'wearable-limitless-capture'; +-- Remove +select cron.unschedule('wearable-limitless-capture'); +``` + +✅ **Done when:** +`select * from cron.job where jobname = 'wearable-limitless-capture';` shows the +job and, after a few minutes, lifelog atoms start appearing in `thoughts`. + +--- + +### Step 6 — Verify capture + +After a cron tick (or a manual invocation), confirm rows landed: + +```sql +select count(*) from thoughts where metadata->>'wearable_source' = 'limitless'; +``` + +For a closer look at a few captured atoms, including kind and attribution: + +```sql +select + metadata->>'atom_kind' as kind, + metadata->>'attribution' as attribution, + metadata->>'section_label' as section, + left(content, 80) as preview, + created_at +from thoughts +where metadata->>'wearable_source' = 'limitless' +order by created_at desc +limit 15; +``` + +✅ **Done when:** The count is non-zero and recent rows show `title` and +`section` kinds, with section atoms carrying `self` / `other` / `mixed` +attribution. + +--- + +## Expected Outcome + +Every 5 minutes, new Limitless lifelogs become several `thoughts` rows — a +`title` atom plus one `section` atom per heading — embedded, deduplicated on a +salted per-atom fingerprint, and tagged with +`metadata.source = 'limitless_lifelog'`, +`metadata.wearable_source = 'limitless'`, `metadata.attribution`, and +`metadata.attributed_to` for retrieval and provenance. Each atom is built from +the device's own structure with no LLM call. Overlapping poll windows and +re-runs never produce duplicates, and a skipped run self-heals on the next pass. + +--- + +## Troubleshooting + +**Function won't deploy: cannot find `../_shared/wearable-sync.ts`** The +wearable-capture-core engine isn't installed. Complete Step 1 — copy +`wearable-sync.ts` into `supabase/functions/_shared/` — then redeploy. + +**`LIMITLESS_API_KEY is required` in the logs** The secret isn't set (or the +function was deployed before you set it). Run +`supabase secrets set LIMITLESS_API_KEY="..."`, then redeploy. Check with +`supabase secrets list`. + +**Limitless API returns 401 / 403** The API key is wrong, revoked, or truncated. +Regenerate it in the Limitless app → Developer settings and re-set the secret. +Note the adapter authenticates with the `X-API-Key` header (not a bearer token). + +**Section atoms aren't attributed to me (`self`)** Limitless usually labels the +wearer "You", which is recognised by default. If yours uses a different label, +set `WEARABLE_SELF_LABELS` to it (comma-separated); it's merged with the +device-generic `you` / `me` / `self`. + +**`pulled` is non-zero but `atomsInserted` is 0 (all skipped)** Those atoms are +already in the brain — dedup matched their salted fingerprints. This is the +steady state once you've caught up; the count in Step 6 still grows as new +recordings come in. + +**Atoms insert but `embedding` is null** `OPENROUTER_API_KEY` isn't set (it +comes from the core install). Set it and future captures embed at write time; a +later embedding backfill can fill the gaps. + +**Cron job runs but nothing happens** Confirm `pg_cron` and `pg_net` are +enabled, that the URL in `cron.schedule` is your real project ref, and that the +Authorization header resolves to a valid service-role key. Inspect +`select * from cron.job_run_details order by start_time desc limit 5;` for HTTP +errors, and check `supabase functions logs wearable-limitless-capture`. + +--- + +## Tool Surface Area + +This integration **registers no new MCP tools**. It is a capture-only path: a +scheduled edge function that calls the shared wearable engine to write rows into +the existing `thoughts` table. + +| Component | Type | What it does | +| ------------------------------------------ | ---------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------- | +| `wearable-limitless-capture` Edge Function | Supabase scheduled poller (not an MCP server) | Pulls recent lifelogs from the Limitless API and atomizes each into a title atom + one section atom per heading via the adapter. | +| `wearable-sync.ts` | Shared Deno module (from [wearable-capture-core](../wearable-capture-core/)) | Per-atom dedup, provenance, embedding (OpenRouter), and insert into `thoughts`. | +| `thoughts` table | Existing Open Brain primitive | No schema changes — additive rows only. | + +**External services called:** `api.limitless.ai/v1` (lifelog list, by the +adapter) and `openrouter.ai/api/v1` (embeddings, by the core). Both are outbound +HTTPS. + +**Auditing:** Because this integration adds no MCP tools, there's no MCP tool +surface to audit for it directly. If you install it alongside MCP servers that +read from `thoughts`, audit those per the +[MCP Tool Audit & Optimization Guide](../../docs/05-tool-audit.md). + +--- + +## Related + +- [Wearable Capture Core](../wearable-capture-core/) — the engine this adapter + is built on (install first). +- [Omi Wearable Capture](../wearable-omi-capture/) — sibling adapter for the Omi + pendant. +- [Smart Ingest](../smart-ingest/) — LLM extraction + dedup for raw documents + (heavier path). +- [MCP Tool Audit & Optimization Guide](../../docs/05-tool-audit.md) — + recommended reading for any integration contributor. +- [Contributing guide](../../CONTRIBUTING.md) — required reading before + submitting changes. diff --git a/integrations/wearable-limitless-capture/_shared/wearable-sync.ts b/integrations/wearable-limitless-capture/_shared/wearable-sync.ts new file mode 100644 index 000000000..b46fdb0dc --- /dev/null +++ b/integrations/wearable-limitless-capture/_shared/wearable-sync.ts @@ -0,0 +1,356 @@ +/** + * wearable-sync — generic ATOMIC capture engine for always-on wearables. + * + * A small, reusable core that turns ANY polling wearable (Omi, Limitless, and + * future devices) into Open Brain thoughts — at the granularity of ATOMS, not + * one summary per recording. A long conversation becomes many searchable rows + * (its title, each action item, each transcript chunk, …), each carrying its own + * provenance. Each device supplies a tiny `WearableAdapter`; this engine owns + * everything the adapters share: + * + * 1. pull records since a rolling time window (the adapter makes the call), + * 2. atomize each record into one or more atoms (the adapter, using the + * device's OWN structured output — no per-item LLM cost), + * 3. skip atoms already captured (idempotent dedup on a SALTED per-atom + * content fingerprint, so re-runs and overlapping windows are safe), + * 4. tag each atom with provenance (attribution / attributed_to / generator), + * 5. embed the text (OpenRouter, OB1's standard) and insert into `thoughts`. + * + * Design rules (per OB1 CONTRIBUTING): + * - Never modifies the `thoughts` schema — additive rows only. The atom + * fingerprint lives in `metadata.content_fingerprint`, deduped with a GIN- + * indexed JSONB containment query, so the engine works on the baseline + * `thoughts` schema with no migration. (If you run a schema that adds a + * UNIQUE index, a duplicate insert is also caught and treated as a skip.) + * - No secrets in code — every credential comes from Deno.env. + * - Idempotency lives in the brain, not a local file, so re-runs and + * overlapping windows are safe and the engine self-heals after outages. + * + * Deploy this file to `supabase/functions/_shared/wearable-sync.ts`; each + * per-wearable adapter (e.g. `wearable-omi-capture`) imports it. + */ +import { createClient, type SupabaseClient } from "@supabase/supabase-js"; + +/** + * Who an atom is attributed to. + * - `self` — only the brain owner speaks/authored it. + * - `other` — only other named people. + * - `mixed` — the owner and at least one other named person. + * - `machine` — the device generated it (a title, a summary, an extracted item). + * - `unknown` — speech with no resolvable speaker. + * `external` is reserved for a future cross-source backfill and is never emitted here. + */ +export type Attribution = "self" | "other" | "machine" | "mixed" | "unknown"; + +/** One atom produced from a wearable record. The adapter sets the provenance + * fields using the device's own structure; the engine merges them into + * `metadata` and computes the fingerprint. `type` maps to the thought type + * (default 'meeting'). */ +export interface WearableAtom { + /** Stable position of this atom within its record (part of the fingerprint salt). */ + atomIndex: number; + /** What kind of atom this is, e.g. 'title' | 'overview' | 'action_item' | 'event' | 'transcript_chunk' | 'section' | 'memory'. */ + atomKind: string; + content: string; + type?: string; + importance?: number; + attribution: Attribution; + /** Speaker labels / names that contributed to this atom (for `attributed_to`). */ + attributedTo?: string[]; + /** The device that machine-generated this atom (e.g. 'omi'); null for human speech. */ + generator?: string | null; + /** True when the brain owner is a speaker here — lets an optional later step self-link. */ + selfPresent?: boolean; + /** The owner's role in a self/mixed atom, for optional self-linking. */ + role?: "author" | "participant" | null; + createdAt?: string; + qualityScore?: number; + /** Atom-specific extras merged into `metadata` (e.g. section_label, speakers). */ + metadata?: Record; +} + +/** The per-wearable contract. Implement these and the engine does the rest. + * `Record` is opaque to the engine — whatever the device API returns. */ +export interface WearableAdapter { + /** Stable short id for the device, e.g. "omi", "limitless". Used for dedup + provenance. */ + sourceId: string; + /** The brain `source_type` to tag thoughts with, e.g. "omi", "limitless_lifelog". */ + sourceType: string; + /** Pull records created/started at or after `sinceISO` (UTC ISO 8601). */ + listSince(sinceISO: string): Promise; + /** The device's own stable id for a record (idempotency salt — survives content edits). */ + recordId(record: Record): string; + /** Atomize a record using the device's OWN structure (no LLM call). */ + recordToAtoms(record: Record): WearableAtom[]; +} + +export interface SyncOptions { + /** Rolling lookback window in hours (default 12). A wider window self-heals longer outages. */ + sinceHours?: number; + /** Don't write — just report what would be captured. */ + dryRun?: boolean; + /** Embed atom text via OpenRouter before insert (default true; false leaves NULL embeddings + * for a later backfill). */ + embed?: boolean; + /** Optional pre-built client (tests). Defaults to a service-role client from env. */ + client?: SupabaseClient; +} + +export interface SyncResult { + source: string; + /** Records pulled from the device this pass. */ + pulled: number; + /** Records that produced at least one NEW atom. */ + recordsImported: number; + /** Atoms written (or, in a dry run, that would be written). */ + atomsInserted: number; + /** Atoms already present (deduped) or empty. */ + atomsSkipped: number; + /** Records that errored mid-pass. */ + failed: number; + /** Count of atoms by attribution, for at-a-glance provenance. */ + attribution: Record; + dryRun: boolean; +} + +export interface FetchRetryOptions { + /** Max 429 retries before giving up and returning the 429 response (default 3). */ + maxRetries?: number; + /** Per-attempt timeout in ms (default 30000). */ + timeoutMs?: number; +} + +const OPENROUTER_BASE = "https://openrouter.ai/api/v1"; + +/** + * `fetch()` with a per-attempt timeout and Retry-After-aware, capped backoff on + * HTTP 429. Adapters use this for their device API calls so a transient rate + * limit slows a pass instead of aborting it. Non-429 responses (including other + * errors) are returned as-is for the caller to handle. + */ +export async function fetchWithRetry( + url: string | URL, + init: RequestInit = {}, + opts: FetchRetryOptions = {}, +): Promise { + const maxRetries = opts.maxRetries ?? 3; + const timeoutMs = opts.timeoutMs ?? 30000; + for (let attempt = 0;; attempt++) { + const ctrl = new AbortController(); + const timer = setTimeout(() => ctrl.abort(), timeoutMs); + try { + const resp = await fetch(url, { ...init, signal: ctrl.signal }); + if (resp.status === 429 && attempt < maxRetries) { + const retryAfter = Number(resp.headers.get("retry-after")) || 0; + const wait = Math.min( + Math.max(retryAfter * 1000, 2000 * (attempt + 1)), + 15000, + ); + await resp.body?.cancel().catch(() => {}); // free the connection before backing off + await new Promise((r) => setTimeout(r, wait)); + continue; + } + return resp; + } finally { + clearTimeout(timer); + } + } +} + +/** + * Salted per-atom identity: `sha256(source|provider_event_id|atom_index|content)`. + * Salting with the recording id and the atom's position means two atoms with + * identical text still get distinct fingerprints, while a re-run of the same + * atom is stable — which is exactly what makes overlapping windows idempotent. + */ +export async function atomFingerprint( + source: string, + providerEventId: string, + atomIndex: number, + content: string, +): Promise { + const data = new TextEncoder().encode( + `${source}|${providerEventId}|${atomIndex}|${content}`, + ); + const digest = await crypto.subtle.digest("SHA-256", data); + return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, "0")) + .join(""); +} + +/** Embed text via OpenRouter (openai/text-embedding-3-small — OB1's default). + * Returns null if no key is set, so the engine still inserts (embedding backfilled later). */ +async function embedText(text: string): Promise { + const key = Deno.env.get("OPENROUTER_API_KEY"); + if (!key) return null; + const r = await fetchWithRetry(`${OPENROUTER_BASE}/embeddings`, { + method: "POST", + headers: { + "Authorization": `Bearer ${key}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + model: "openai/text-embedding-3-small", + input: text.slice(0, 8000), + }), + }); + if (!r.ok) { + throw new Error( + `OpenRouter embeddings ${r.status}: ${(await r.text()).slice(0, 200)}`, + ); + } + const d = await r.json(); + return d?.data?.[0]?.embedding ?? null; +} + +function defaultClient(): SupabaseClient { + const url = Deno.env.get("SUPABASE_URL"); + const key = Deno.env.get("SUPABASE_SERVICE_ROLE_KEY"); + if (!url || !key) { + throw new Error("SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY are required"); + } + return createClient(url, key); +} + +/** + * Run one capture pass for a wearable. Idempotent + additive: safe to call on a + * tight schedule (e.g. every 5 minutes via cron). Each record is atomized, and + * each atom is deduped on its salted fingerprint before insert. + */ +export async function runWearableSync( + adapter: WearableAdapter, + opts: SyncOptions = {}, +): Promise { + const supabase = opts.client ?? defaultClient(); + const sinceHours = opts.sinceHours ?? 12; + const sinceISO = new Date(Date.now() - sinceHours * 3600 * 1000) + .toISOString(); + const dryRun = opts.dryRun ?? false; + const doEmbed = opts.embed ?? true; + + const records = await adapter.listSince(sinceISO); + let recordsImported = 0, atomsInserted = 0, atomsSkipped = 0, failed = 0; + const attribution: Record = {}; + + for (const record of records) { + const providerEventId = adapter.recordId(record); + if (!providerEventId) continue; + try { + const atoms = adapter.recordToAtoms(record); + if (atoms.length === 0) continue; + + // ONE indexed lookup per recording for the atoms already captured for THIS + // record (the metadata GIN index serves the containment match). Batching + // here — instead of a query per atom — follows the brain's "never per-row + // filter on a JSONB key" rule. + const seen = new Set(); + if (!dryRun) { + const { data: existing, error: selErr } = await supabase + .from("thoughts") + .select("metadata") + .contains("metadata", { + wearable_source: adapter.sourceId, + provider_event_id: providerEventId, + }); + if (selErr) throw selErr; + for (const row of existing ?? []) { + const fp = (row as { metadata?: Record }).metadata + ?.content_fingerprint; + if (typeof fp === "string") seen.add(fp); + } + } + + let newAtoms = 0; + for (const atom of atoms) { + attribution[atom.attribution] = (attribution[atom.attribution] ?? 0) + + 1; + const content = atom.content?.trim(); + if (!content) { + atomsSkipped++; + continue; + } + + const fingerprint = await atomFingerprint( + adapter.sourceType, + providerEventId, + atom.atomIndex, + content, + ); + if (seen.has(fingerprint)) { + atomsSkipped++; + continue; + } + seen.add(fingerprint); + + const metadata: Record = { + ...(atom.metadata ?? {}), + source: adapter.sourceType, + wearable_source: adapter.sourceId, + provider_event_id: providerEventId, + atom_index: atom.atomIndex, + atom_kind: atom.atomKind, + attribution: atom.attribution, + generator: atom.generator ?? null, + content_fingerprint: fingerprint, + captured_via: "wearable-atomic", + type: atom.type ?? "meeting", + importance: atom.importance ?? 3, + }; + if (atom.attributedTo?.length) { + metadata.attributed_to = atom.attributedTo; + } + if (atom.selfPresent) { + metadata.self_present = true; + if (atom.role) metadata.role = atom.role; + } + if (typeof atom.qualityScore === "number") { + metadata.quality_score = atom.qualityScore; + } + + if (dryRun) { + atomsInserted++; + newAtoms++; + continue; + } + + const row: Record = { content, metadata }; + if (atom.createdAt) row.created_at = atom.createdAt; + if (doEmbed) { + const emb = await embedText(content); + if (emb) row.embedding = emb; + } + const { error: insErr } = await supabase.from("thoughts").insert(row); + if (insErr) { + // A unique violation only happens if you run a schema with a UNIQUE + // index on the fingerprint — it means a concurrent/overlapping pass + // beat us to this atom. Treat as a skip, not a failure. + if (/duplicate key|23505/i.test(insErr.message ?? "")) { + atomsSkipped++; + continue; + } + throw insErr; + } + atomsInserted++; + newAtoms++; + } + if (newAtoms > 0) recordsImported++; + } catch (e) { + failed++; + console.error( + `[wearable-sync:${adapter.sourceId}] ${providerEventId}: ${ + (e as Error).message + }`, + ); + } + } + + return { + source: adapter.sourceId, + pulled: records.length, + recordsImported, + atomsInserted, + atomsSkipped, + failed, + attribution, + dryRun, + }; +} diff --git a/integrations/wearable-limitless-capture/deno.json b/integrations/wearable-limitless-capture/deno.json new file mode 100644 index 000000000..bfa87cb1a --- /dev/null +++ b/integrations/wearable-limitless-capture/deno.json @@ -0,0 +1,11 @@ +{ + "imports": { + "@supabase/supabase-js": "npm:@supabase/supabase-js@2.47.10", + "../_shared/wearable-sync.ts": "./_shared/wearable-sync.ts" + }, + "tasks": { + "check": "deno check index.ts _shared/wearable-sync.ts", + "fmt": "deno fmt", + "lint": "deno lint" + } +} diff --git a/integrations/wearable-limitless-capture/index.ts b/integrations/wearable-limitless-capture/index.ts new file mode 100644 index 000000000..a2b027b98 --- /dev/null +++ b/integrations/wearable-limitless-capture/index.ts @@ -0,0 +1,313 @@ +/** + * wearable-limitless-capture — Limitless Pendant adapter for wearable-capture-core. + * + * Limitless (https://limitless.ai) records spoken life as "lifelogs" and returns + * each one as Markdown: a title, `##` section headings, and `- Speaker (time): + * text` transcript bullets. This adapter atomizes that device-native structure — + * NO LLM call: + * + * - one `title` atom (machine-generated label), + * - one `section` atom per `##` heading: the heading label plus its rolled-up + * utterances, attributed to the section's speakers (self / other / mixed / + * unknown). NO cap on sections — they are Limitless's native unit. + * + * The shared core (`_shared/wearable-sync.ts`) owns the write path: per-atom + * dedup on a salted fingerprint, provenance metadata, embedding via OpenRouter, + * and the insert into `thoughts`. + * + * Deploy this file to `supabase/functions/wearable-limitless-capture/index.ts`. + * It imports the core from `../_shared/wearable-sync.ts`, so install + * wearable-capture-core FIRST (see this integration's README). The `_shared/` + * copy in this folder is a vendored copy of that same engine, present so the + * function typechecks standalone; the deno.json import map points the deploy + * path at it for local `deno check`. + * + * Limitless API facts this adapter relies on: + * - Auth: `X-API-Key: `. + * - Base: https://api.limitless.ai/v1 + * - List: GET /lifelogs?start=&timezone=&limit=&direction=asc&cursor=&includeMarkdown=true + * -> { data: { lifelogs: [{ id, title, markdown, startTime, endTime }] }, + * meta: { lifelogs: { nextCursor } } }. Has a real `start` param. + */ +import { + type Attribution, + fetchWithRetry, + runWearableSync, + type WearableAdapter, + type WearableAtom, +} from "../_shared/wearable-sync.ts"; + +const LIMITLESS_BASE = "https://api.limitless.ai/v1"; +/** Safety caps so a wide window or a runaway cursor can't fetch unbounded pages. */ +const MAX_RECORDS = 500; +const MAX_PAGES = 30; +/** Cap on a single section atom's content (sections roll up many utterances). */ +const SECTION_MAX_CHARS = 4000; + +// ── types ───────────────────────────────────────────────────────────────────── + +/** A single lifelog as returned by the Limitless API (only the fields we read). */ +interface Lifelog { + id: string; + title?: string; + markdown?: string; + startTime?: string; + endTime?: string; +} + +interface Utterance { + speaker: string; + text: string; +} +interface Section { + label: string | null; + utterances: Utterance[]; +} + +// ── speaker classification (generic — no hardcoded personal names) ───────────── + +/** Device-generic labels for the wearer. Add your own (e.g. your name) via the + * `WEARABLE_SELF_LABELS` env var (comma-separated) — never hardcode a name. + * Limitless typically labels the wearer "You". */ +const DEFAULT_SELF_LABELS = ["you", "user", "me", "self", "myself"]; +const GENERIC_SPEAKER_RE = + /^(unknown|speaker[\s_]*\d+|spk[\s_]*\d+|user\s*\d+)$/i; + +function selfLabelSet(): Set { + const extra = (Deno.env.get("WEARABLE_SELF_LABELS") ?? "") + .split(",").map((s) => s.trim().toLowerCase()).filter(Boolean); + return new Set([...DEFAULT_SELF_LABELS, ...extra]); +} +const SELF_LABELS = selfLabelSet(); + +function isSelfSpeaker(name: string): boolean { + return SELF_LABELS.has(name.trim().toLowerCase()); +} +function isGenericSpeaker(name: string): boolean { + const s = name.trim(); + return !s || GENERIC_SPEAKER_RE.test(s); +} + +/** Classify a section's speaker labels into an attribution + self presence + role. + * A section with no utterances (just a heading) is a machine-generated label. */ +function classifySpeakers( + speakers: string[], + hasUtterances: boolean, +): { + attribution: Attribution; + selfPresent: boolean; + role: "author" | "participant" | null; +} { + if (!hasUtterances) { + return { attribution: "machine", selfPresent: false, role: null }; + } + let hasSelf = false, hasNamedOther = false; + for (const s of speakers) { + if (isSelfSpeaker(s)) hasSelf = true; + else if (!isGenericSpeaker(s)) hasNamedOther = true; + } + let attribution: Attribution; + if (hasSelf && hasNamedOther) attribution = "mixed"; + else if (hasSelf) attribution = "self"; + else if (hasNamedOther) attribution = "other"; + else attribution = "unknown"; + const role = hasSelf + ? (attribution === "self" ? "author" : "participant") + : null; + return { attribution, selfPresent: hasSelf, role }; +} + +const trim = (s: unknown): string => + String(s ?? "").replace(/\s+/g, " ").trim(); + +// ── Markdown -> sections (title -> ## sections -> utterances) ─────────────────── + +/** + * Parse Limitless lifelog Markdown into sections. Any heading (`#`..`######`) + * starts a section; `- Speaker (HH:MM): text` bullets become its utterances. + * Bullets before the first heading land in a leading label-less section. + */ +function parseSections(markdown: string): Section[] { + const sections: Section[] = []; + let cur: Section | null = null; + for (const raw of markdown.split(/\r?\n/)) { + const line = raw.trim(); + const h = line.match(/^#{1,6}\s+(.*)$/); + if (h) { + cur = { label: h[1].trim(), utterances: [] }; + sections.push(cur); + continue; + } + if (!line.startsWith("- ")) continue; + const body = line.replace(/^-\s*/, ""); + // "Speaker (HH:MM[:SS] …): text" — the parenthetical holds a clock time. + const m = body.match(/^(.*?)\s*\([^)]*\d{1,2}:\d{2}[^)]*\)\s*:\s*(.*)$/); + let speaker: string; + let text: string; + if (m) { + speaker = m[1].trim(); + text = m[2].trim(); + } else { + const i = body.indexOf(": "); + if (i > 0) { + speaker = body.slice(0, i).trim(); + text = body.slice(i + 2).trim(); + } else { + speaker = "Unknown"; + text = body.trim(); + } + } + if (!text) continue; + if (!cur) { + cur = { label: null, utterances: [] }; + sections.push(cur); + } + cur.utterances.push({ speaker: speaker || "Unknown", text }); + } + return sections; +} + +/** + * Atomize one lifelog using the device's OWN structure (no LLM): a `title` atom + * plus one `section` atom per heading (label + rolled-up utterances). No cap on + * sections — they are Limitless's native unit. + */ +function atomizeLifelog(ll: Lifelog): WearableAtom[] { + const atoms: WearableAtom[] = []; + const startedAt = ll.startTime; + let idx = 0; + + const title = trim(ll.title); + if (title) { + atoms.push({ + atomIndex: idx++, + atomKind: "title", + content: title, + type: "meeting", + attribution: "machine", + generator: "limitless", + createdAt: startedAt, + qualityScore: 45, + metadata: {}, + }); + } + + const sections = parseSections(ll.markdown ?? ""); + let sectionIndex = 0; + for (const sec of sections) { + const speakers = [...new Set(sec.utterances.map((u) => u.speaker))]; + const body = sec.utterances.map((u) => `${u.speaker}: ${u.text}`).join( + "\n", + ); + const content = [sec.label, body].filter(Boolean).join("\n"); + if (!trim(content)) continue; + const cls = classifySpeakers(speakers, sec.utterances.length > 0); + atoms.push({ + atomIndex: idx++, + atomKind: "section", + content: content.slice(0, SECTION_MAX_CHARS), + type: "meeting", + attribution: cls.attribution, + attributedTo: speakers, + // A section with no utterances is just a heading Limitless generated. + generator: sec.utterances.length ? null : "limitless", + selfPresent: cls.selfPresent, + role: cls.role, + createdAt: startedAt, + qualityScore: 55, + metadata: { + section_label: sec.label, + section_index: sectionIndex++, + speakers, + utterance_count: sec.utterances.length, + }, + }); + } + + return atoms; +} + +// ── the adapter (driven by the shared core) ──────────────────────────────────── + +function limitlessKey(): string { + const key = Deno.env.get("LIMITLESS_API_KEY"); + if (!key) throw new Error("LIMITLESS_API_KEY is required"); + return key; +} + +/** + * Pull lifelogs started at/after `sinceISO`, paging forward (ascending) by + * following `meta.lifelogs.nextCursor` until it's empty, a page is empty, or a + * cap is hit. Limitless has a real `start` param, so the window is server-side. + */ +async function listSince(sinceISO: string): Promise { + const apiKey = limitlessKey(); + const out: Lifelog[] = []; + let cursor: string | undefined; + + for (let page = 0; page < MAX_PAGES; page++) { + const params = new URLSearchParams({ + start: sinceISO, + timezone: "UTC", + limit: "50", + direction: "asc", + includeMarkdown: "true", + includeHeadings: "true", + }); + if (cursor) params.set("cursor", cursor); + + const r = await fetchWithRetry( + `${LIMITLESS_BASE}/lifelogs?${params.toString()}`, + { + headers: { "X-API-Key": apiKey }, + }, + ); + if (!r.ok) { + throw new Error( + `Limitless lifelogs ${r.status}: ${(await r.text()).slice(0, 200)}`, + ); + } + + const body = await r.json(); + const lifelogs: Lifelog[] = body?.data?.lifelogs ?? []; + if (lifelogs.length === 0) break; + + out.push(...lifelogs); + if (out.length >= MAX_RECORDS) return out.slice(0, MAX_RECORDS); + + cursor = body?.meta?.lifelogs?.nextCursor ?? undefined; + if (!cursor) break; + } + + return out; +} + +const limitlessAdapter: WearableAdapter = { + sourceId: "limitless", + sourceType: "limitless_lifelog", + listSince, + recordId: (ll) => ll.id, + recordToAtoms: (ll) => atomizeLifelog(ll), +}; + +Deno.serve(async (req: Request): Promise => { + try { + const url = new URL(req.url); + const dryRun = url.searchParams.get("dry_run") === "1"; + const sinceHours = Number(url.searchParams.get("since_hours")) || 12; + const result = await runWearableSync(limitlessAdapter, { + sinceHours, + dryRun, + }); + return new Response(JSON.stringify(result), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + } catch (err) { + console.error("[wearable-limitless-capture]", (err as Error).message); + return new Response( + JSON.stringify({ error: (err as Error).message }), + { status: 500, headers: { "Content-Type": "application/json" } }, + ); + } +}); diff --git a/integrations/wearable-limitless-capture/metadata.json b/integrations/wearable-limitless-capture/metadata.json new file mode 100644 index 000000000..7a00988fe --- /dev/null +++ b/integrations/wearable-limitless-capture/metadata.json @@ -0,0 +1,29 @@ +{ + "name": "Limitless Wearable Capture", + "description": "Limitless Pendant adapter for the wearable-capture-core engine. Polls the Limitless lifelog API and atomizes each recording into a title atom plus one section atom per `##` heading (the label with its rolled-up utterances), attributed to its speakers — the device's own structure, no per-item LLM cost. Per-atom salted-fingerprint dedup and provenance via the core.", + "category": "integrations", + "author": { + "name": "Alan Shurafa", + "github": "alanshurafa" + }, + "version": "0.2.0", + "requires": { + "open_brain": true, + "services": ["Limitless", "OpenRouter"], + "tools": ["Supabase CLI"] + }, + "tags": [ + "wearable", + "limitless", + "lifelog", + "voice", + "capture", + "poller", + "atomic", + "provenance" + ], + "difficulty": "intermediate", + "estimated_time": "20 minutes", + "created": "2026-06-16", + "updated": "2026-06-16" +}