apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-07-21 21:31:12 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge pull request #750 from rowboatlabs/email_improvements

Browse source 
feat(x): unified email classifier with labels, triage sections, and user instructions

Replace the two disjoint email classification systems (inline importance
classifier + batch labeling agent) with a single LLM pass that emits
importance, category, knowledge verdict, summary, and a pre-drafted reply.

Classification:
- classify_thread.ts now returns {importance, category, knowledge}; the
  category enum and prompt are built at call time from a label registry
  (7 built-ins named after Superhuman's vocabulary — Direct, Calendar,
  News, Marketing, Pitch, Notification, Receipt — plus up to 12 custom)
- Users create labels in plain text via the new "Email agent instructions"
  dialog; an extraction pass syncs them into config/email_labels.json.
  Instructions are also injected into every classify/draft call and the
  composer's Write-with-AI bar (seeded: Investor / Candidate / Customer)
- Per-thread category corrections are sticky and feed few-shot learning
  (email_category_feedback.ts), same pattern as importance corrections

Knowledge graph:
- Delete the labeling agent pipeline (label_emails, labeling_agent,
  labeling_state); admission is now the stamped `knowledge: extract|skip`
  frontmatter, with legacy noise-tag fallback and a budgeted sweep that
  backfills unclassified markdown (build_graph holds unstamped files)
- Guardrails: user-participated threads always extract; user importance
  flips never touch the knowledge verdict; classify failures stamp nothing

Inbox UI:
- Sections become Needs you / Waiting on them / Everything else; waiting
  state is derived per render (user sent last, others participated), never
  stored, so it can't go stale
- Category chips on rows (Direct stays chipless), filter pills with counts,
  one-click bulk archive per category, "Reply ready" indicator
- Fix: Everything else section could never load/render under sustained
  sync writes (fetch was chained on hasReachedEnd, which live reloads reset)
This commit is contained in:
arkml 2026-07-14 11:54:44 +05:30 • committed by GitHub
commit 7cfc2200ba
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 1596 additions and 698 deletions
Download patch file Download diff file Expand all files Collapse all files

1
apps/x/ANALYTICS.md
View file

@ -53,7 +53,6 @@ Every `llm_usage` emit point in the codebase:
| `knowledge_sync` | `agent_notes` | yes | Agent notes learning service | `packages/core/src/knowledge/agent_notes.ts:309` (createRun) |
| `knowledge_sync` | `tag_notes` | yes | Note tagging | `packages/core/src/knowledge/tag_notes.ts:86` (createRun) |
| `knowledge_sync` | `build_graph` | yes | Knowledge graph note creation | `packages/core/src/knowledge/build_graph.ts:253` (createRun) |
| `knowledge_sync` | `label_emails` | yes | Email labeling | `packages/core/src/knowledge/label_emails.ts:73` (createRun) |
| `knowledge_sync` | `inline_task_run` | yes | Inline `@rowboat` task execution (two call sites) | `packages/core/src/knowledge/inline_tasks.ts:471, 552` (createRun) |
| `knowledge_sync` | `inline_task_classify` | no | Inline task scheduling classifier (`generateText`) | `packages/core/src/knowledge/inline_tasks.ts:673` |
| `knowledge_sync` | `pre_built` | yes | Pre-built scheduled agents | `packages/core/src/pre_built/runner.ts:43` (createRun) |

33
apps/x/apps/main/src/ipc.ts
View file

@ -98,7 +98,9 @@ import { summarizeMeeting } from '@x/core/dist/knowledge/summarize_meeting.js';
import { getAccessToken } from '@x/core/dist/auth/tokens.js';
import { getRowboatConfig } from '@x/core/dist/config/rowboat.js';
import { runLiveNoteAgent } from '@x/core/dist/knowledge/live-note/runner.js';
import { listImportantThreads, listEverythingElseThreads, saveMessageBodyHeight, triggerSync as triggerGmailSync, sendThreadReply, saveThreadDraft, deleteThreadDraft, listDraftThreads, searchThreads, archiveThread, trashThread, markThreadRead, downloadAttachment, getAccountEmail, getAccountName, getConnectionStatus as getGmailConnectionStatus, setThreadImportance } from '@x/core/dist/knowledge/sync_gmail.js';
import { listImportantThreads, listEverythingElseThreads, saveMessageBodyHeight, triggerSync as triggerGmailSync, sendThreadReply, saveThreadDraft, deleteThreadDraft, listDraftThreads, searchThreads, archiveThread, archiveCategoryThreads, trashThread, markThreadRead, downloadAttachment, getAccountEmail, getAccountName, getConnectionStatus as getGmailConnectionStatus, setThreadImportance, setThreadCategory } from '@x/core/dist/knowledge/sync_gmail.js';
import { loadEmailInstructions, saveEmailInstructions } from '@x/core/dist/knowledge/email_instructions.js';
import { getEmailLabels, syncCustomLabelsFromInstructions } from '@x/core/dist/knowledge/email_labels.js';
import { searchContacts as searchGmailContacts, warmContactIndex } from '@x/core/dist/knowledge/gmail_contacts.js';
import { searchSentContacts, warmSentContacts } from '@x/core/dist/knowledge/gmail_sent_contacts.js';
import { getGoogleDocsConnectionStatus, importGoogleDoc, syncGoogleDocDown, syncGoogleDocUp, getGoogleDocLink } from '@x/core/dist/knowledge/google_docs.js';
@ -880,7 +882,7 @@ export function setupIpcHandlers() {
return listImportantThreads({ cursor: args.cursor, limit: args.limit });
},
'gmail:getEverythingElse': async (_event, args) => {
return listEverythingElseThreads({ cursor: args.cursor, limit: args.limit });
return listEverythingElseThreads({ cursor: args.cursor, limit: args.limit, category: args.category });
},
'gmail:triggerSync': async () => {
triggerGmailSync();
@ -914,6 +916,33 @@ export function setupIpcHandlers() {
const result = setThreadImportance(args.threadId, args.importance);
return { ok: result.success, previous: result.previous, error: result.error };
},
'gmail:setCategory': async (_event, args) => {
const result = setThreadCategory(args.threadId, args.category);
return { ok: result.success, error: result.error };
},
'gmail:archiveCategory': async (_event, args) => {
return archiveCategoryThreads(args.category);
},
'gmail:getEmailInstructions': async () => {
return { instructions: loadEmailInstructions() };
},
'gmail:setEmailInstructions': async (_event, args) => {
const saved = saveEmailInstructions(args.instructions);
if (!saved.ok) return saved;
// Extract any custom labels the instructions define so they become
// valid classifier outputs immediately. Extraction failure shouldn't
// fail the save — the instructions themselves are already persisted
// and still steer classification as free text.
try {
await syncCustomLabelsFromInstructions(args.instructions);
} catch (err) {
console.warn('[EmailLabels] custom label extraction failed:', err);
}
return saved;
},
'gmail:getEmailLabels': async () => {
return { labels: getEmailLabels().map(({ id, name, kind }) => ({ id, name, kind })) };
},
'gmail:archiveThread': async (_event, args) => {
return archiveThread(args.threadId);
},

4
apps/x/apps/main/src/main.ts
View file

@ -23,7 +23,6 @@ import { init as initCalendarSync } from "@x/core/dist/knowledge/sync_calendar.j
import { init as initFirefliesSync } from "@x/core/dist/knowledge/sync_fireflies.js";
import { init as initGranolaSync } from "@x/core/dist/knowledge/granola/sync.js";
import { init as initGraphBuilder } from "@x/core/dist/knowledge/build_graph.js";
import { init as initEmailLabeling } from "@x/core/dist/knowledge/label_emails.js";
import { init as initNoteTagging } from "@x/core/dist/knowledge/tag_notes.js";
import { init as initInlineTasks } from "@x/core/dist/knowledge/inline_tasks.js";
import { init as initAgentRunner } from "@x/core/dist/agent-schedule/runner.js";
@ -522,9 +521,6 @@ app.whenReady().then(async () => {
// start knowledge graph builder
initGraphBuilder();
// start email labeling service
initEmailLabeling();
// start note tagging service
initNoteTagging();

122
apps/x/apps/renderer/src/App.css
View file

@ -460,12 +460,24 @@
font-size: 13px;
}
/* The bold lead (summary or subject) may shrink — with ellipsis — but only
after the snippet span (huge shrink factor below) has fully collapsed.
Without this, a long summary plus the never-shrinking category chip
overflows the clipped row and the chip is what gets cut off. */
.gmail-row-content strong {
flex-shrink: 0;
flex-shrink: 1;
min-width: 0;
overflow: hidden;
text-overflow: ellipsis;
color: inherit;
font-weight: 400;
}
/* Snippet gives way first (chips carry .gmail-row-chip and keep shrink 0). */
.gmail-row-content > span:not(.gmail-row-chip) {
flex-shrink: 1000;
}
.gmail-row-date {
justify-self: end;
color: var(--gm-text-faint);
@ -473,6 +485,114 @@
font-variant-numeric: tabular-nums;
}
/* Category / waiting-age pill at the right edge of the content column. */
.gmail-row-chip {
flex-shrink: 0;
margin-left: auto;
padding: 1px 8px;
border: 1px solid var(--gm-border);
border-radius: 999px;
color: var(--gm-text-faint);
font-size: 11px;
line-height: 16px;
white-space: nowrap;
}
/* Two chips on one row (category + status): only the first takes the auto
margin; the second sits right next to it, so the pair clusters at the
right edge instead of spreading across the free space. */
.gmail-row-chip + .gmail-row-chip {
margin-left: 6px;
}
.gmail-row-chip-waiting {
border-color: transparent;
background: var(--gm-bg-pill-hover);
color: var(--gm-text-muted);
font-variant-numeric: tabular-nums;
}
.gmail-row-chip-ready {
border-color: transparent;
background: var(--gm-bg-row-selected);
color: var(--gm-accent);
}
/* Shown in place of the "Needs you" section when it's empty. */
.gmail-caughtup {
padding: 18px 24px;
color: var(--gm-text-faint);
font-size: 13px;
}
/* Category filter pills under the "Everything else" header. */
.gmail-category-pills {
display: flex;
flex-wrap: wrap;
align-items: center;
gap: 6px;
padding: 8px 24px;
border-bottom: 1px solid var(--gm-border);
}
.gmail-category-pill {
display: inline-flex;
align-items: center;
gap: 5px;
padding: 2px 10px;
border: 1px solid var(--gm-border);
border-radius: 999px;
background: transparent;
color: var(--gm-text-muted);
font-family: inherit;
font-size: 12px;
line-height: 18px;
cursor: pointer;
transition: background 120ms ease, color 120ms ease, border-color 120ms ease;
}
.gmail-category-pill:hover {
background: var(--gm-bg-pill-hover);
color: var(--gm-text-strong);
}
.gmail-category-pill-active {
background: var(--gm-bg-row-selected);
border-color: var(--gm-accent);
color: var(--gm-text-strong);
}
.gmail-category-pill-count {
color: var(--gm-text-faint);
font-variant-numeric: tabular-nums;
}
.gmail-category-pill-archive {
margin-left: auto;
border-color: transparent;
background: var(--gm-bg-pill-hover);
color: var(--gm-text-strong);
}
.gmail-category-pill-archive:disabled {
opacity: 0.6;
cursor: default;
}
/* The category chip in the thread-detail toolbar is a button (dropdown trigger). */
.gmail-category-chip {
flex-shrink: 0;
background: transparent;
cursor: pointer;
font-family: inherit;
transition: background 120ms ease, color 120ms ease;
}
.gmail-category-chip:hover {
background: var(--gm-bg-pill-hover);
color: var(--gm-text-strong);
}
.gmail-detail {
display: flex;
min-width: 0;

479
apps/x/apps/renderer/src/components/email-view.tsx
View file

@ -1,5 +1,5 @@
import { memo, useCallback, useEffect, useMemo, useRef, useState } from 'react'
import { Archive, Bold, CheckCheck, Forward, Italic, Link as LinkIcon, List, ListOrdered, LoaderIcon, Mail, Paperclip, Quote, Redo2, RefreshCw, Reply, ReplyAll, Search, Send, Sparkles, SquarePen, Star, StarOff, Strikethrough, Trash2, Undo2, X } from 'lucide-react'
import { Archive, Bold, CheckCheck, Forward, Italic, Link as LinkIcon, List, ListOrdered, LoaderIcon, Mail, Paperclip, Quote, Redo2, RefreshCw, Reply, ReplyAll, Search, Send, SlidersHorizontal, Sparkles, SquarePen, Star, StarOff, Strikethrough, Trash2, Undo2, X } from 'lucide-react'
import { useEditor, EditorContent, type Editor } from '@tiptap/react'
import StarterKit from '@tiptap/starter-kit'
import Link from '@tiptap/extension-link'
@ -12,7 +12,14 @@ import { useTheme } from '@/contexts/theme-context'
import { SettingsDialog } from '@/components/settings-dialog'
import { Button } from '@/components/ui/button'
import { Input } from '@/components/ui/input'
import { Textarea } from '@/components/ui/textarea'
import { Dialog, DialogContent, DialogTitle } from '@/components/ui/dialog'
import {
DropdownMenu,
DropdownMenuContent,
DropdownMenuItem,
DropdownMenuTrigger,
} from '@/components/ui/dropdown-menu'
type GmailThread = blocks.GmailThread
type GmailThreadMessage = blocks.GmailThreadMessage
@ -90,6 +97,70 @@ function latestMessage(thread: GmailThread): GmailThreadMessage | undefined {
return thread.messages[thread.messages.length - 1]
}
// The label set (chips, filter pills, correction dropdown) comes from the
// backend label registry: built-ins (display names follow Superhuman's Auto
// Label vocabulary — Marketing / Pitch / News / Calendar) plus any labels the
// user defined in their agent instructions. This static copy of the built-ins
// is the fallback used until the registry loads, and for ids the registry no
// longer knows (a custom label the user later removed).
type EmailCategory = string
interface EmailLabelInfo {
id: string
name: string
kind: 'builtin' | 'custom'
}
const BUILTIN_LABELS: EmailLabelInfo[] = [
{ id: 'correspondence', name: 'Direct', kind: 'builtin' },
{ id: 'meeting', name: 'Calendar', kind: 'builtin' },
{ id: 'notification', name: 'Notification', kind: 'builtin' },
{ id: 'newsletter', name: 'News', kind: 'builtin' },
{ id: 'promotion', name: 'Marketing', kind: 'builtin' },
{ id: 'cold_outreach', name: 'Pitch', kind: 'builtin' },
{ id: 'receipt', name: 'Receipt', kind: 'builtin' },
]
// Pill order in the "Everything else" filter row: noise first (that's what
// gets bulk-archived), then the rest of the built-ins; custom labels are
// appended in registry order at render time. 'unclassified' (threads the
// classifier hasn't reached yet) is always last and unarchivable.
const BUILTIN_PILL_ORDER = ['newsletter', 'promotion', 'notification', 'cold_outreach', 'receipt', 'meeting', 'correspondence']
// Fallback for ids the registry doesn't know: "portfolio_updates" → "Portfolio updates".
function prettifyLabelId(id: string): string {
const words = id.replace(/_/g, ' ').trim()
return words ? words[0].toUpperCase() + words.slice(1) : id
}
function labelNameFor(labels: EmailLabelInfo[], category: string): string {
if (category === 'unclassified') return 'Uncategorized'
return labels.find((l) => l.id === category)?.name ?? prettifyLabelId(category)
}
// The user sent the latest message and someone else is in the conversation —
// the ball is in their court. Derived from the messages on every render (never
// stored) so it can't go stale the way an arrival-time label would.
function isAwaitingThem(thread: GmailThread, selfEmail: string | null | undefined): boolean {
const self = (selfEmail || '').trim().toLowerCase()
if (!self) return false
const latest = latestMessage(thread)
if (!(latest?.from || '').toLowerCase().includes(self)) return false
return thread.messages.some((m) => m.from && !m.from.toLowerCase().includes(self))
}
function daysSince(value?: string): number | null {
if (!value) return null
const ms = Date.parse(value)
if (!Number.isFinite(ms)) return null
return Math.max(0, Math.floor((Date.now() - ms) / 86_400_000))
}
function waitingChip(thread: GmailThread): string {
const days = daysSince(latestMessage(thread)?.date || thread.date)
return days && days > 0 ? `Waiting ${days}d` : 'Waiting'
}
// Split a raw header recipient string (e.g. `"Jo Bloggs" <jo@x.com>, b@y.com`) into
// individual address tokens, respecting commas inside quotes/angle brackets.
function splitAddresses(raw?: string): string[] {
@ -1244,6 +1315,15 @@ const ComposeBox = memo(function ComposeBox({
setGenerating(true)
try {
// The user's standing email-agent instructions steer this writer too —
// otherwise "keep drafts short, sign off with X" would apply to the
// auto-drafted replies but silently not to the Write-with-AI bar.
try {
const { instructions } = await window.ipc.invoke('gmail:getEmailInstructions', {})
if (instructions.trim()) {
system = `${system}\n\nThe user's standing email preferences — follow any that concern how emails should be written (tone, length, sign-off, what not to write):\n${instructions.trim()}`
}
} catch { /* draft without them */ }
// Draft through Copilot: no model override, so the backend resolves the
// same default model/provider the Copilot chat uses (models.json).
const res = await window.ipc.invoke('llm:generate', { prompt, system })
@ -1910,6 +1990,8 @@ function ThreadDetail({
hidden,
keysDisabled,
onComposingChange,
onSetCategory,
labels = BUILTIN_LABELS,
}: {
thread: GmailThread
onClose: () => void
@ -1918,6 +2000,10 @@ function ThreadDetail({
keysDisabled?: boolean
/** Reports whether the inline composer is open, so list shortcuts pause. */
onComposingChange?: (composing: boolean) => void
/** Present for inbox threads only — search results aren't in the cache the correction writes to. */
onSetCategory?: (threadId: string, category: EmailCategory) => Promise<void>
/** The label registry (built-ins + user-defined) for the chip and correction dropdown. */
labels?: EmailLabelInfo[]
}) {
const [composeMode, setComposeMode] = useState<ComposeMode | null>(null)
const [selfEmail, setSelfEmail] = useState<string>('')
@ -1999,6 +2085,30 @@ function ThreadDetail({
<div className={cn('gmail-detail gmail-detail-inline', hidden && 'gmail-detail-hidden')}>
<div className="gmail-detail-toolbar">
<div className="gmail-thread-subject-inline">{thread.subject || '(No subject)'}</div>
{onSetCategory && (
<DropdownMenu>
<DropdownMenuTrigger asChild>
<button
type="button"
className="gmail-row-chip gmail-category-chip"
title="Correct the category — the classifier learns from this"
>
{thread.category ? labelNameFor(labels, thread.category) : 'Categorize'}
</button>
</DropdownMenuTrigger>
<DropdownMenuContent align="end" className="font-sans">
{labels.map((l) => (
<DropdownMenuItem
key={l.id}
onSelect={() => { void onSetCategory(thread.threadId, l.id) }}
className={cn(l.id === thread.category && 'font-semibold')}
>
{l.name}
</DropdownMenuItem>
))}
</DropdownMenuContent>
</DropdownMenu>
)}
<button type="button" className="gmail-icon-button" onClick={onClose} aria-label="Close thread">
<span>×</span>
</button>
@ -2090,6 +2200,10 @@ const ThreadRow = memo(function ThreadRow({
isLeaving,
keysDisabled,
section,
chip,
chipWaiting,
onSetCategory,
labels,
onToggle,
onMarkRead,
onArchive,
@ -2108,6 +2222,12 @@ const ThreadRow = memo(function ThreadRow({
keysDisabled: boolean
/** Which inbox section the row is rendered in; null hides the importance toggle (e.g. search results). */
section: 'important' | 'other' | null
/** Status pill after the subject — "Reply ready" or waiting age ("Waiting 3d"). The category chip renders automatically from thread.category. */
chip?: string | null
chipWaiting?: boolean
onSetCategory?: (threadId: string, category: EmailCategory) => Promise<void>
/** Label registry — stable reference from EmailView state (this row is memoized). */
labels: EmailLabelInfo[]
onToggle: (thread: GmailThread) => void
onMarkRead: (threadId: string, read?: boolean) => Promise<void>
onArchive: (threadId: string) => Promise<void>
@ -2120,6 +2240,14 @@ const ThreadRow = memo(function ThreadRow({
}) {
const latest = latestMessage(thread)
const isUnread = thread.unread === true
// Category chip on classified rows so the list itself answers "worth
// opening?". Plain correspondence gets none — with granular labels
// (Investor, Customer, …) available, "Direct" is the default case and
// absence reads cleaner than a chip stating it. It stays available in the
// correction dropdown and the Everything-else filter pills.
const categoryChip = thread.category && thread.category !== 'correspondence'
? labelNameFor(labels, thread.category)
: null
const stop = (e: React.MouseEvent | React.KeyboardEvent) => {
e.stopPropagation()
}
@ -2141,6 +2269,8 @@ const ThreadRow = memo(function ThreadRow({
<span className="gmail-row-content">
<strong>{thread.summary || thread.subject || '(No subject)'}</strong>
<span>{thread.summary ? thread.subject : snippet(latest?.body || thread.latest_email)}</span>
{categoryChip && <span className="gmail-row-chip">{categoryChip}</span>}
{chip && <span className={cn('gmail-row-chip', chipWaiting ? 'gmail-row-chip-waiting' : 'gmail-row-chip-ready')}>{chip}</span>}
</span>
<span className="gmail-row-date">{formatInboxTime(latest?.date || thread.date)}</span>
</button>
@ -2194,12 +2324,88 @@ const ThreadRow = memo(function ThreadRow({
hidden={!isSelected}
keysDisabled={keysDisabled}
onComposingChange={onComposingChange}
onSetCategory={onSetCategory}
labels={labels}
/>
)}
</div>
)
})
// Free-text standing instructions for the email agent. Saved to
// config/email_instructions.md and injected into every classification /
// draft call as the highest-priority section of the system prompt.
function EmailInstructionsDialog({ open, onOpenChange, onSaved }: { open: boolean; onOpenChange: (open: boolean) => void; onSaved?: () => void }) {
const [text, setText] = useState('')
const [loading, setLoading] = useState(false)
const [saving, setSaving] = useState(false)
useEffect(() => {
if (!open) return
let cancelled = false
setLoading(true)
window.ipc.invoke('gmail:getEmailInstructions', {})
.then((res) => { if (!cancelled) setText(res.instructions) })
.catch(() => {})
.finally(() => { if (!cancelled) setLoading(false) })
return () => { cancelled = true }
}, [open])
const save = async () => {
setSaving(true)
try {
const result = await window.ipc.invoke('gmail:setEmailInstructions', { instructions: text })
if (result.ok) {
toast('Instructions saved — they apply to every email from now on.', 'success')
onOpenChange(false)
onSaved?.()
} else {
toast(`Could not save: ${result.error ?? 'unknown error'}`, 'error')
}
} catch (err) {
toast(`Could not save: ${err instanceof Error ? err.message : String(err)}`, 'error')
} finally {
setSaving(false)
}
}
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="font-sans sm:max-w-[560px]">
<DialogTitle className="text-base font-semibold">Email agent instructions</DialogTitle>
<p className="text-sm text-muted-foreground">
Standing instructions for how your email is labeled, prioritized, and how draft replies are written.
The agent follows these over its defaults, on every new email. You can also define your own labels
here — e.g. “Create a label Clients for emails from active client accounts.”
</p>
<Textarea
value={text}
onChange={(e) => setText(e.target.value)}
disabled={loading}
rows={10}
maxLength={8000}
placeholder={
'Examples:\n' +
'- Emails from my investors are always important.\n' +
'- Investor updates I receive are direct emails, not news.\n' +
'- Never draft replies to recruiters or cold outreach.\n' +
'- Keep drafts short and casual; sign off with my first name.\n' +
'- Payment failure alerts are important; other billing emails are receipts.\n' +
'- Create a label "Customers" for emails from paying customers.'
}
className="min-h-[220px] font-sans text-sm leading-relaxed"
/>
<div className="flex justify-end gap-2">
<Button variant="outline" onClick={() => onOpenChange(false)}>Cancel</Button>
<Button onClick={() => void save()} disabled={saving || loading}>
{saving ? 'Saving…' : 'Save'}
</Button>
</div>
</DialogContent>
</Dialog>
)
}
function ShortcutKey({ children }: { children: React.ReactNode }) {
return (
<kbd className="inline-flex h-5 min-w-5 items-center justify-center rounded border border-border bg-muted px-1.5 font-mono text-[11px] font-medium text-foreground">
@ -2360,6 +2566,13 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
const closeCompose = useCallback(() => setComposeOpen(false), [])
// Inbox vs Drafts. Drafts are fetched live (they're not in the inbox cache).
const [view, setView] = useState<'inbox' | 'drafts'>('inbox')
// Category filter for "Everything else" (null = all) + whole-section counts
// from the last backend response, which drive the filter pills.
const [otherCategory, setOtherCategory] = useState<string | null>(null)
const otherCategoryRef = useRef<string | null>(null)
otherCategoryRef.current = otherCategory
const [categoryCounts, setCategoryCounts] = useState<Record<string, number>>({})
const [bulkArchiving, setBulkArchiving] = useState(false)
const [drafts, setDrafts] = useState<GmailThread[]>(() => persistedDrafts ?? [])
const [draftsLoading, setDraftsLoading] = useState(false)
const [draftsError, setDraftsError] = useState<string | null>(null)
@ -2372,6 +2585,17 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
// Gmail sync uses the native Google OAuth connection.
const [emailConnection, setEmailConnection] = useState<GmailConnectionStatus | null>(null)
const [settingsOpen, setSettingsOpen] = useState(false)
const [instructionsOpen, setInstructionsOpen] = useState(false)
// Label registry (built-ins + user-defined). Fetched on mount and again
// after the instructions dialog saves, since saving can create labels.
const [emailLabels, setEmailLabels] = useState<EmailLabelInfo[]>(BUILTIN_LABELS)
const refreshLabels = useCallback(async () => {
try {
const res = await window.ipc.invoke('gmail:getEmailLabels', {})
if (res.labels?.length) setEmailLabels(res.labels)
} catch { /* keep built-ins */ }
}, [])
useEffect(() => { void refreshLabels() }, [refreshLabels])
// Keyboard navigation: the j/k focus cursor over the visible rows, plus the
// "?" shortcuts overlay. lastFocusedIndexRef remembers the cursor's position
// so it can re-anchor when the focused row disappears (archive/trash/reload).
@ -2589,6 +2813,33 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
}
}, [important.threads, other.threads, markLeaving, setSection])
// User corrects a thread's category: sticky on the thread + recorded as a
// correction the classifier learns from (few-shot).
const setCategoryAction = useCallback(async (threadId: string, category: EmailCategory) => {
try {
const result = await window.ipc.invoke('gmail:setCategory', { threadId, category })
if (result.ok) {
updateThreadInState(threadId, (t) => ({ ...t, category }))
setCategoryCounts((prev) => {
const next = { ...prev }
// Only 'other'-section threads are counted; adjust optimistically and
// let the next backend response correct any drift.
const prevCat = [...important.threads, ...other.threads].find((t) => t.threadId === threadId)?.category ?? 'unclassified'
if (other.threads.some((t) => t.threadId === threadId)) {
next[prevCat] = Math.max(0, (next[prevCat] ?? 1) - 1)
next[category] = (next[category] ?? 0) + 1
}
return next
})
toast(`Filed as ${labelNameFor(emailLabels, category)} — noted for similar emails.`, 'success')
} else if (result.error) {
toast(`Could not update category: ${result.error}`, 'error')
}
} catch (err) {
toast(`Could not update category: ${err instanceof Error ? err.message : String(err)}`, 'error')
}
}, [updateThreadInState, important.threads, other.threads, emailLabels])
const trashThreadAction = useCallback(async (threadId: string) => {
markLeaving(threadId, true)
try {
@ -2654,8 +2905,21 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
// reload to be discarded whenever Other was reloaded in the same tick.)
const epochsRef = useRef<Record<InboxSection, number>>({ important: 0, other: 0 })
const sectionChannel = (section: InboxSection) =>
section === 'important' ? 'gmail:getImportant' as const : 'gmail:getEverythingElse' as const
const fetchSectionPage = useCallback(async (section: InboxSection, cursor?: string) => {
const result = section === 'important'
? await window.ipc.invoke('gmail:getImportant', { cursor, limit: PAGE_SIZE })
: await window.ipc.invoke('gmail:getEverythingElse', {
cursor,
limit: PAGE_SIZE,
category: otherCategoryRef.current ?? undefined,
})
// Counts describe the whole 'other' section regardless of filter/page —
// keep the pills fresh on every response that carries them.
if (section === 'other' && result.categoryCounts) {
setCategoryCounts(result.categoryCounts)
}
return result
}, [])
const loadNextPage = useCallback(async (section: InboxSection) => {
const current = section === 'important' ? important : other
@ -2664,10 +2928,7 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
const epoch = epochsRef.current[section]
setSection(section, (prev) => ({ ...prev, loadingPage: true }))
try {
const result = await window.ipc.invoke(sectionChannel(section), {
cursor: current.nextCursor ?? undefined,
limit: PAGE_SIZE,
})
const result = await fetchSectionPage(section, current.nextCursor ?? undefined)
if (epoch !== epochsRef.current[section]) return
setSection(section, (prev) => ({
threads: [...prev.threads, ...result.threads],
@ -2680,7 +2941,7 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
console.warn(`[Gmail] page load failed for ${section}:`, err)
setSection(section, (prev) => ({ ...prev, loadingPage: false }))
}
}, [important, other, setSection])
}, [important, other, setSection, fetchSectionPage])
const reloadFirstPage = useCallback(async (section: InboxSection, options: { silent?: boolean } = {}) => {
const epoch = ++epochsRef.current[section]
@ -2690,9 +2951,7 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
setSection(section, () => ({ ...initialSectionState, loadingPage: true }))
}
try {
const result = await window.ipc.invoke(sectionChannel(section), {
limit: PAGE_SIZE,
})
const result = await fetchSectionPage(section)
if (epoch !== epochsRef.current[section]) return
setSection(section, () => ({
threads: result.threads,
@ -2705,33 +2964,59 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
console.warn(`[Gmail] initial page load failed for ${section}:`, err)
setSection(section, (prev) => ({ ...prev, loadingPage: false }))
}
}, [setSection])
}, [setSection, fetchSectionPage])
// Initial load — fetch page 1 of Important. On first-ever mount we do a
// non-silent load (shows loading state). On re-mount with persisted state we
// do a silent reconcile against the cache — necessary because the watcher
// subscription only runs while mounted, so any cache changes that happened
// while the panel was unmounted would otherwise stay invisible.
// Initial load — fetch page 1 of BOTH sections. Everything else used to
// load lazily once Important was exhausted, but that chained on
// hasReachedEnd, which every silent live reload resets (back to page 1) —
// under sustained sync writes the chain never fired and the section never
// appeared at all. Both lists are cheap local reads (mtime-cached file
// scan), so eager is fine. On first-ever mount we do a non-silent load
// (shows loading state); on re-mount with persisted state we do a silent
// reconcile against the cache — necessary because the watcher subscription
// only runs while mounted, so any cache changes that happened while the
// panel was unmounted would otherwise stay invisible.
useEffect(() => {
if (hadPersistedDataOnMount.current) {
void reloadFirstPage('important', { silent: true })
// Reconcile Other too if it had been loaded before the unmount.
if (other.threads.length > 0) {
void reloadFirstPage('other', { silent: true })
}
} else {
void reloadFirstPage('important')
}
const silent = hadPersistedDataOnMount.current
void reloadFirstPage('important', { silent })
void reloadFirstPage('other', { silent })
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [])
// Once Important is exhausted, kick off page 1 of Everything else.
// Changing the category filter refetches Everything else from page 1
// (fetchSectionPage reads the filter from otherCategoryRef). Skipped on
// mount — the initial-load effect above covers it.
const otherCategoryMounted = useRef(false)
useEffect(() => {
if (!important.hasReachedEnd) return
if (other.threads.length > 0) return
if (other.loadingPage) return
if (!otherCategoryMounted.current) {
otherCategoryMounted.current = true
return
}
void reloadFirstPage('other')
}, [important.hasReachedEnd, other.threads.length, other.loadingPage, reloadFirstPage])
}, [otherCategory, reloadFirstPage])
// Bulk gesture behind the filter pills: archive the whole category at once.
const archiveCategoryAction = useCallback(async (category: string) => {
setBulkArchiving(true)
try {
const result = await window.ipc.invoke('gmail:archiveCategory', { category })
if (result.error) {
toast(`Bulk archive failed: ${result.error}`, 'error')
} else {
toast(
`Archived ${result.archived} thread${result.archived === 1 ? '' : 's'}${result.failed ? ` (${result.failed} failed)` : ''}. They stay searchable in Gmail.`,
result.failed ? 'error' : 'success',
)
}
setOtherCategory(null) // triggers the Everything else reload above
void reloadFirstPage('important', { silent: true })
} catch (err) {
toast(`Bulk archive failed: ${err instanceof Error ? err.message : String(err)}`, 'error')
} finally {
setBulkArchiving(false)
}
}, [reloadFirstPage])
// Live updates: watcher on inbox_lists/ → silently refresh visible sections
// when files change. Throttled to at most one reload per ~3s so a burst of
@ -2751,8 +3036,6 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
composeOpenRef.current = composeOpen
const isRefreshingRef = useRef(false)
isRefreshingRef.current = refreshing
const otherHasThreadsRef = useRef(false)
otherHasThreadsRef.current = other.threads.length > 0
const RELOAD_THROTTLE_MS = 3000
@ -2764,11 +3047,7 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
}
lastReloadAtRef.current = Date.now()
void reloadFirstPage('important', { silent: true })
// Only refresh Other if it had been loaded — otherwise the chained
// effect handles it once Important hits hasReachedEnd.
if (otherHasThreadsRef.current) {
void reloadFirstPage('other', { silent: true })
}
void reloadFirstPage('other', { silent: true })
}, [reloadFirstPage])
// Leading-edge throttle:
@ -2821,9 +3100,7 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
pendingReloadRef.current = false
lastReloadAtRef.current = Date.now()
void reloadFirstPage('important', { silent: true })
if (otherHasThreadsRef.current) {
void reloadFirstPage('other', { silent: true })
}
void reloadFirstPage('other', { silent: true })
}, [selectedThreadId, composeOpen, reloadFirstPage])
// Manual refresh: wake the background sync loop. It updates inbox_lists/,
@ -2876,13 +3153,26 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
const visibleOther = useMemo(() => filterThreads(other.threads), [other.threads, filterThreads])
const visibleDrafts = useMemo(() => filterThreads(drafts), [drafts, filterThreads])
// Split "Important" into what still needs the user vs. what they've already
// answered and are waiting on. Derived per render from the messages — thread
// state, unlike a label, can't be allowed to go stale.
const selfEmail = emailConnection?.email ?? null
const visibleNeedsYou = useMemo(
() => visibleImportant.filter((t) => !isAwaitingThem(t, selfEmail)),
[visibleImportant, selfEmail],
)
const visibleWaiting = useMemo(
() => visibleImportant.filter((t) => isAwaitingThem(t, selfEmail)),
[visibleImportant, selfEmail],
)
// ── Keyboard shortcuts (Superhuman-style) ───────────────────────────────────
// EmailView only mounts while the email tab is open, so these are naturally
// scoped to that view. Single-letter keys stay inert while typing in any
// field, while a dialog is up, or while the inline reply composer is open.
const listMode: 'search' | 'drafts' | 'inbox' = query.trim() ? 'search' : view === 'drafts' ? 'drafts' : 'inbox'
const anyModalOpen = composeOpen || settingsOpen || Boolean(editingDraft) || helpOpen
const anyModalOpen = composeOpen || settingsOpen || Boolean(editingDraft) || helpOpen || instructionsOpen
// Row identity for the focus cursor — must match each row's data-thread-id.
const rowIdOf = useCallback((thread: GmailThread) => (
@ -2896,9 +3186,9 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
const visibleList = useMemo<GmailThread[]>(() => {
if (query.trim()) return searchResults
if (view === 'drafts') return visibleDrafts
if (important.hasReachedEnd && other.threads.length > 0) return [...visibleImportant, ...visibleOther]
return visibleImportant
}, [query, searchResults, view, visibleDrafts, visibleImportant, visibleOther, important.hasReachedEnd, other.threads.length])
if (other.threads.length > 0) return [...visibleNeedsYou, ...visibleWaiting, ...visibleOther]
return [...visibleNeedsYou, ...visibleWaiting]
}, [query, searchResults, view, visibleDrafts, visibleNeedsYou, visibleWaiting, visibleOther, other.threads.length])
// Keep the cursor valid as the list changes: switching between inbox,
// search, and drafts resets it; if the focused row vanished (archived,
@ -3135,7 +3425,12 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
const closeThread = useCallback(() => setSelectedThreadId(null), [])
const renderRow = (thread: GmailThread, section: 'important' | 'other' | null = null) => {
const renderRow = (
thread: GmailThread,
section: 'important' | 'other' | null = null,
chip: string | null = null,
chipWaiting: boolean = false,
) => {
const isMounted = openedThreadIds.includes(thread.threadId)
return (
<ThreadRow
@ -3147,6 +3442,10 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
isLeaving={leavingThreadIds.has(thread.threadId)}
keysDisabled={isMounted && anyModalOpen}
section={section}
chip={chip}
chipWaiting={chipWaiting}
onSetCategory={section ? setCategoryAction : undefined}
labels={emailLabels}
onToggle={toggleThread}
onMarkRead={markThreadReadAction}
onArchive={archiveThreadAction}
@ -3242,6 +3541,15 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
onClick={() => setView('drafts')}
>Drafts{drafts.length > 0 ? ` (${drafts.length})` : ''}</button>
</div>
<button
type="button"
className="gmail-icon-button"
onClick={() => setInstructionsOpen(true)}
aria-label="Email agent instructions"
title="Email agent instructions"
>
<SlidersHorizontal size={18} />
</button>
<button
type="button"
className="gmail-icon-button"
@ -3296,25 +3604,44 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
<div className="gmail-empty-state">Could not load mail: {error}</div>
) : hasAny ? (
<div className="gmail-list" aria-label="Recent emails">
{important.threads.length > 0 && (
{visibleNeedsYou.length > 0 ? (
<section className="gmail-section">
<div className="gmail-list-header">
<span>Important</span>
<span>Needs you</span>
<span>
{important.threads.length}{important.hasReachedEnd ? '' : '+'} thread{important.threads.length === 1 ? '' : 's'}
{visibleNeedsYou.length}{important.hasReachedEnd ? '' : '+'} thread{visibleNeedsYou.length === 1 ? '' : 's'}
</span>
</div>
{visibleImportant.map((t) => renderRow(t, 'important'))}
{!important.hasReachedEnd && (
<SectionSentinel
disabled={important.loadingPage || important.hasReachedEnd}
onIntersect={() => loadNextPage('important')}
loading={important.loadingPage}
/>
)}
{visibleNeedsYou.map((t) => renderRow(t, 'important', t.draft_response ? 'Reply ready' : null))}
</section>
) : important.hasReachedEnd && !important.loadingPage ? (
<div className="gmail-caughtup">You’re caught up — nothing needs a reply.</div>
) : null}
{visibleWaiting.length > 0 && (
<section className="gmail-section">
<div className="gmail-list-header">
<span>Waiting on them</span>
<span>
{visibleWaiting.length}{important.hasReachedEnd ? '' : '+'} thread{visibleWaiting.length === 1 ? '' : 's'}
</span>
</div>
{visibleWaiting.map((t) => renderRow(t, 'important', waitingChip(t), true))}
</section>
)}
{important.hasReachedEnd && other.threads.length > 0 && (
{/* Pages of "Important" feed both sections above, so the sentinel
lives after them rather than inside either one. */}
{important.threads.length > 0 && !important.hasReachedEnd && (
<SectionSentinel
disabled={important.loadingPage || important.hasReachedEnd}
onIntersect={() => loadNextPage('important')}
loading={important.loadingPage}
/>
)}
{/* Loading stays lazy (chained on Important exhausting), but once
loaded the section never unmounts: silent live reloads reset
Important to page 1 (hasReachedEnd → false), and gating the
render on it made this whole section vanish on every sync. */}
{(other.threads.length > 0 || otherCategory !== null) && (
<section className="gmail-section">
<div className="gmail-list-header">
<span>Everything else</span>
@ -3322,6 +3649,41 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
{other.threads.length}{other.hasReachedEnd ? '' : '+'} thread{other.threads.length === 1 ? '' : 's'}
</span>
</div>
{Object.keys(categoryCounts).length > 0 && (
<div className="gmail-category-pills">
{[
...BUILTIN_PILL_ORDER,
...emailLabels.filter((l) => l.kind === 'custom').map((l) => l.id),
// Stale custom ids still present in counts render too.
...Object.keys(categoryCounts).filter((c) => c !== 'unclassified' && !BUILTIN_PILL_ORDER.includes(c) && !emailLabels.some((l) => l.id === c)),
'unclassified',
].filter((c) => (categoryCounts[c] ?? 0) > 0).map((cat) => (
<button
key={cat}
type="button"
className={cn('gmail-category-pill', otherCategory === cat && 'gmail-category-pill-active')}
onClick={() => setOtherCategory((prev) => (prev === cat ? null : cat))}
>
{labelNameFor(emailLabels, cat)} <span className="gmail-category-pill-count">{categoryCounts[cat]}</span>
</button>
))}
{otherCategory && otherCategory !== 'unclassified' && (
<button
type="button"
className="gmail-category-pill gmail-category-pill-archive"
disabled={bulkArchiving}
onClick={() => void archiveCategoryAction(otherCategory)}
>
{bulkArchiving
? 'Archiving…'
: `Archive all ${categoryCounts[otherCategory] ?? ''} ${labelNameFor(emailLabels, otherCategory).toLowerCase()}`}
</button>
)}
</div>
)}
{other.threads.length === 0 && otherCategory !== null && !other.loadingPage && (
<div className="gmail-caughtup">Nothing filed as {labelNameFor(emailLabels, otherCategory).toLowerCase()}.</div>
)}
{visibleOther.map((t) => renderRow(t, 'other'))}
{!other.hasReachedEnd && (
<SectionSentinel
@ -3367,6 +3729,7 @@ export function EmailView({ initialThreadId, threadIdVersion, initialSearchQuery
)}
<SettingsDialog open={settingsOpen} onOpenChange={setSettingsOpen} defaultTab="connections" />
<ShortcutsHelpDialog open={helpOpen} onOpenChange={setHelpOpen} />
<EmailInstructionsDialog open={instructionsOpen} onOpenChange={setInstructionsOpen} onSaved={() => void refreshLabels()} />
</div>
)
}

2
apps/x/packages/core/src/knowledge/README.md
View file

@ -213,7 +213,7 @@ On first run, `strictness_analyzer.ts` analyzes your emails and recommends a lev
### Prompt Files
(Historical: per-strictness prompt files no longer exist.) The single prompt lives in `note_creation.ts` (`getRaw()`); email gating is label-based via `renderNoteEffectRules()` from `tag_system.ts`, layered with the Email Reply Gate, direct-interaction, transactional, weekly-importance, and ongoing-relationship tests.
(Historical: per-strictness prompt files no longer exist.) The single prompt lives in `note_creation.ts` (`getRaw()`); email gating happens upstream — the inbox classifier (`classify_thread.ts`) stamps a `knowledge: extract | skip` verdict into each email's frontmatter during Gmail sync, and `build_graph.ts` only admits `extract` files — layered with the Email Reply Gate, direct-interaction, transactional, weekly-importance, and ongoing-relationship tests.
## Other Configuration

52
apps/x/packages/core/src/knowledge/build_graph.test.ts Normal file
View file

@ -0,0 +1,52 @@
import { describe, expect, it } from "vitest";
import { emailAdmission } from "./build_graph.js";
const email = (frontmatter: string | null, body = "# Subject\n\n**Thread ID:** t1\n") =>
frontmatter === null ? body : `---\n${frontmatter}\n---\n\n${body}`;
describe("emailAdmission", () => {
it("holds files with no frontmatter until the classifier stamps a verdict", () => {
expect(emailAdmission(email(null))).toBe("wait");
});
it("admits knowledge: extract", () => {
expect(
emailAdmission(email("importance: important\ncategory: correspondence\nknowledge: extract\nclassified_at: \"2026-07-11T00:00:00Z\"")),
).toBe("process");
});
it("skips knowledge: skip", () => {
expect(
emailAdmission(email("importance: other\ncategory: newsletter\nknowledge: skip\nclassified_at: \"2026-07-11T00:00:00Z\"")),
).toBe("skip");
});
it("importance never decides admission — an unimportant thread can still carry knowledge", () => {
expect(
emailAdmission(email("importance: other\ncategory: newsletter\nknowledge: extract\nclassified_at: \"2026-07-11T00:00:00Z\"")),
).toBe("process");
});
it("falls back to noise-tag matching for legacy labeling-agent frontmatter", () => {
// `newsletter` is a noise tag in the default taxonomy → skip.
expect(
emailAdmission(email("labels:\n relationship: []\n topics: []\n type: Newsletter\n filter:\n - newsletter\n action: FYI\nprocessed: true")),
).toBe("skip");
// No noise tags → process.
expect(
emailAdmission(email("labels:\n relationship:\n - investor\n topics:\n - fundraising\n filter: []\nprocessed: true")),
).toBe("process");
});
it("matches legacy noise tags anywhere in the labels block, not just under filter:", () => {
// The old labeling agent sometimes mis-filed noise tags (observed:
// `candidate` under `relationship:`).
expect(
emailAdmission(email("labels:\n relationship:\n - candidate\n topics: []\n filter: []\nprocessed: true")),
).toBe("skip");
});
it("does not mistake a message-body '---' separator for frontmatter", () => {
expect(emailAdmission("# Subject\n\n---\n\nknowledge: skip\n")).toBe("wait");
});
});

41
apps/x/packages/core/src/knowledge/build_graph.ts
View file

@ -93,6 +93,27 @@ function hasNoiseLabels(content: string): boolean {
return false;
}
/**
* Admission decision for a gmail_sync email file:
* - 'wait' — no frontmatter yet; the inbox classifier hasn't stamped a
* verdict. Hold the file (don't mark processed) and try again
* next tick.
* - 'skip' — stamped `knowledge: skip` (or, for legacy labeling-agent
* frontmatter, carries a noise tag). Mark processed, never extract.
* - 'process' — admitted for knowledge extraction.
*
* Exported for tests.
*/
export function emailAdmission(content: string): 'wait' | 'skip' | 'process' {
if (!content.startsWith('---')) return 'wait';
const endIdx = content.indexOf('\n---', 3);
const frontmatter = endIdx === -1 ? '' : content.slice(3, endIdx);
const verdict = frontmatter.match(/^knowledge:\s*(extract|skip)\s*$/m);
if (verdict) return verdict[1] === 'skip' ? 'skip' : 'process';
// Legacy labeling-agent frontmatter (labels: block) — noise tags decide.
return hasNoiseLabels(content) ? 'skip' : 'process';
}
function ensureSuggestedTopicsFileLocation(): string {
if (fs.existsSync(SUGGESTED_TOPICS_PATH)) {
@ -537,18 +558,18 @@ export async function buildGraph(sourceDir: string): Promise<void> {
// Get files that need processing (new or changed)
let filesToProcess = getFilesToProcess(sourceDir, state);
// For gmail_sync, only process emails that have been labeled AND don't have noise filter tags
// For gmail_sync, only process emails the classifier admitted: hold files
// with no stamped verdict yet, permanently skip `knowledge: skip`.
if (sourceDir.endsWith('gmail_sync')) {
filesToProcess = filesToProcess.filter(filePath => {
try {
const content = fs.readFileSync(filePath, 'utf-8');
if (!content.startsWith('---')) return false;
if (hasNoiseLabels(content)) {
const admission = emailAdmission(content);
if (admission === 'skip') {
console.log(`[buildGraph] Skipping noise email: ${path.basename(filePath)}`);
markFileAsProcessed(filePath, state);
return false;
}
return true;
return admission === 'process';
} catch {
return false;
}
@ -756,18 +777,18 @@ export async function processAllSources(): Promise<void> {
try {
let filesToProcess = getFilesToProcess(sourceDir, state);
// For gmail_sync, only process emails that have been labeled AND don't have noise filter tags
// For gmail_sync, only process emails the classifier admitted: hold
// files with no stamped verdict yet, permanently skip `knowledge: skip`.
if (source.provider === 'gmail') {
filesToProcess = filesToProcess.filter(filePath => {
try {
const content = fs.readFileSync(filePath, 'utf-8');
if (!content.startsWith('---')) return false;
if (hasNoiseLabels(content)) {
const admission = emailAdmission(content);
if (admission === 'skip') {
console.log(`[GraphBuilder] Skipping noise email: ${path.basename(filePath)}`);
markFileAsProcessed(filePath, state);
return false;
}
return true;
return admission === 'process';
} catch {
return false;
}

91
apps/x/packages/core/src/knowledge/classification_stamp.test.ts Normal file
View file

@ -0,0 +1,91 @@
import { describe, expect, it } from "vitest";
import fs from "fs";
import os from "os";
import path from "path";
// WorkDir is read from the env at module load, so it must be set before
// sync_gmail (→ config.ts) is imported — hence dynamic imports (not hoisted).
// This is why these tests don't live in sync_gmail.test.ts, whose static
// import of sync_gmail would lock in the default WorkDir first.
process.env.ROWBOAT_WORKDIR = fs.mkdtempSync(path.join(os.tmpdir(), "x-classification-stamp-test-"));
const { stampClassificationFrontmatter } = await import("./sync_gmail.js");
const { emailAdmission } = await import("./build_graph.js");
type GmailThreadSnapshot = import("./sync_gmail.js").GmailThreadSnapshot;
const SYNC_DIR = path.join(process.env.ROWBOAT_WORKDIR, "gmail_sync");
const BODY = "# Pricing discussion\n\n**Thread ID:** t1\n\n---\n\n### From: Sarah <sarah@acme.com>\n**Date:** Fri, 10 Jul 2026\n\nHere is the proposal.\n\n---\n";
function snapshot(overrides: Partial<GmailThreadSnapshot> = {}): GmailThreadSnapshot {
return {
threadId: "t1",
threadUrl: "https://mail.google.com/mail/#inbox/t1",
importance: "important",
category: "correspondence",
knowledge: "extract",
messages: [],
...overrides,
};
}
function writeMd(threadId: string, content: string): string {
fs.mkdirSync(SYNC_DIR, { recursive: true });
const p = path.join(SYNC_DIR, `${threadId}.md`);
fs.writeFileSync(p, content);
return p;
}
describe("stampClassificationFrontmatter", () => {
it("stamps a verdict the graph builder admits, preserving the body", () => {
const p = writeMd("t1", BODY);
stampClassificationFrontmatter("t1", snapshot());
const stamped = fs.readFileSync(p, "utf-8");
expect(stamped.startsWith("---\nimportance: important\ncategory: correspondence\nknowledge: extract\n")).toBe(true);
expect(stamped.endsWith(BODY)).toBe(true);
expect(emailAdmission(stamped)).toBe("process");
});
it("a knowledge: skip stamp is what excludes the thread", () => {
const p = writeMd("t1", BODY);
stampClassificationFrontmatter("t1", snapshot({ importance: "other", category: "newsletter", knowledge: "skip" }));
expect(emailAdmission(fs.readFileSync(p, "utf-8"))).toBe("skip");
});
it("does not stamp a verdict that was never made (classify failure)", () => {
const p = writeMd("t1", BODY);
stampClassificationFrontmatter("t1", snapshot({ category: undefined, knowledge: undefined }));
const content = fs.readFileSync(p, "utf-8");
expect(content).toBe(BODY);
expect(emailAdmission(content)).toBe("wait");
});
it("replaces legacy labeling-agent frontmatter instead of stacking on top", () => {
const legacy = `---\nlabels:\n relationship:\n - investor\n filter: []\nprocessed: true\n---\n\n${BODY}`;
const p = writeMd("t1", legacy);
stampClassificationFrontmatter("t1", snapshot());
const stamped = fs.readFileSync(p, "utf-8");
expect(stamped).not.toContain("labels:");
expect(stamped.endsWith(BODY)).toBe(true);
expect(emailAdmission(stamped)).toBe("process");
});
it("is idempotent — an unchanged verdict does not rewrite the file", () => {
const p = writeMd("t1", BODY);
stampClassificationFrontmatter("t1", snapshot());
// Pin classified_at to a sentinel; a rewrite would replace it.
const pinned = fs.readFileSync(p, "utf-8").replace(/^classified_at: .*$/m, 'classified_at: "sentinel"');
fs.writeFileSync(p, pinned);
stampClassificationFrontmatter("t1", snapshot());
expect(fs.readFileSync(p, "utf-8")).toContain('classified_at: "sentinel"');
// ...but a changed verdict does restamp.
stampClassificationFrontmatter("t1", snapshot({ importance: "other" }));
const restamped = fs.readFileSync(p, "utf-8");
expect(restamped).toContain("importance: other");
expect(restamped).not.toContain('classified_at: "sentinel"');
expect(restamped.endsWith(BODY)).toBe(true);
});
it("is a no-op when the thread has no markdown mirror", () => {
expect(() => stampClassificationFrontmatter("missing-thread", snapshot())).not.toThrow();
});
});

116
apps/x/packages/core/src/knowledge/classify_thread.ts
View file

@ -14,6 +14,9 @@ import { captureLlmUsage } from '../analytics/usage.js';
import { withUseCase } from '../analytics/use_case.js';
import type { GmailThreadSnapshot } from './sync_gmail.js';
import { formatImportanceFeedbackForPrompt, maybeDistillImportanceRules } from './email_importance_feedback.js';
import { formatCategoryFeedbackForPrompt } from './email_category_feedback.js';
import { formatEmailInstructionsForPrompt } from './email_instructions.js';
import { getEmailLabels, type EmailLabelDef } from './email_labels.js';
const STYLE_GUIDE_PATH = path.join(WorkDir, 'knowledge', 'Agent Notes', 'style', 'email.md');
const CALENDAR_DIR = path.join(WorkDir, 'calendar_sync');
@ -100,25 +103,72 @@ export async function getUserEmail(auth: OAuth2Client): Promise<string | null> {
return null;
}
/**
* What kind of email this is — shown as a chip on inbox rows and stamped
* into the gmail_sync markdown for the knowledge pipeline. The value set is
* open: built-in ids plus user-defined labels from the email_labels registry
* (which is why this is a string, not a closed union). Orthogonal to
* importance: a newsletter is almost always "other", but an investor update
* arriving as a newsletter can still carry knowledge.
*/
export type EmailCategory = string;
export interface Classification {
importance: 'important' | 'other';
/** Absent when the LLM call failed (fail-open) — callers must not stamp a verdict they don't have. */
category?: EmailCategory;
/** Whether the knowledge-graph pipeline should extract from this thread. Absent on LLM failure. */
knowledge?: 'extract' | 'skip';
summary?: string;
draftResponse?: string;
}
const ClassificationSchema = z.object({
importance: z.enum(['important', 'other']).describe('important = real correspondence, action-required, or content worth referencing later. other = newsletters, marketing, automated notifications, transactional receipts, cold outreach.'),
summary: z.string().optional().describe('One or two sentences capturing what the thread is about and any implied action. Required when importance is important. Omit when other.'),
draftResponse: z.string().optional().describe('A complete draft reply the user can send as-is or edit. Plain text with real line breaks (\\n): greeting on its own line, a blank line between paragraphs, and the sign-off on its own line(s) — e.g. "Hi Tyrone,\\n\\nThanks for the follow-up.\\n\\nBest,\\nJohn". If a sign-off name is included, use only the user\'s first name. Required when importance is important AND the thread implies a response is wanted. Omit when other, or when no response is appropriate (e.g. an FYI from a colleague that does not need a reply).'),
});
// The category enum is built per call from the label registry, so labels the
// user defines in their instructions become valid classifier outputs without
// a code change.
function buildClassificationSchema(labelIds: string[]) {
return z.object({
importance: z.enum(['important', 'other']).describe('important = real correspondence, action-required, or content worth referencing later. other = newsletters, marketing, automated notifications, transactional receipts, cold outreach.'),
category: z.enum(labelIds as [string, ...string[]]).describe('Which category label fits this email best — definitions are in the system prompt. When a user-defined label and a built-in both fit, prefer the user-defined one.'),
knowledge: z.enum(['extract', 'skip']).describe('Whether this thread contains durable facts worth adding to the user\'s knowledge base about their people, companies, and projects. extract = real relationships (investor, customer, prospect, partner, vendor, team, advisor, press, personal) or substantive topics (deals, contracts, hiring, fundraising, support, incidents, real meetings, intros). skip = noise with no durable facts: marketing, newsletters, automated notifications, receipts, social/forum digests, cold outreach from strangers, job applicants and recruiters.'),
summary: z.string().optional().describe('One or two sentences capturing what the thread is about and any implied action. Required when importance is important. Omit when other.'),
draftResponse: z.string().optional().describe('A complete draft reply the user can send as-is or edit. Plain text with real line breaks (\\n): greeting on its own line, a blank line between paragraphs, and the sign-off on its own line(s) — e.g. "Hi Tyrone,\\n\\nThanks for the follow-up.\\n\\nBest,\\nJohn". If a sign-off name is included, use only the user\'s first name. Required when importance is important AND the thread implies a response is wanted. Omit when other, or when no response is appropriate (e.g. an FYI from a colleague that does not need a reply).'),
});
}
const SYSTEM_PROMPT = `You classify a Gmail thread for a personal inbox view and, when appropriate, draft a reply on behalf of the user.
function buildCategorySection(labels: EmailLabelDef[]): string {
const lines = [
`# Category`,
``,
`Pick exactly one category — it labels the email in the inbox:`,
];
for (const l of labels) {
const marker = l.kind === 'custom' ? ' [user-defined]' : '';
lines.push(`- ${l.id}${marker}: ${l.description}`);
}
const hasCustom = labels.some(l => l.kind === 'custom');
if (hasCustom) {
lines.push(``);
lines.push(`User-defined labels are the user's own filing system — when one of them fits, prefer it over a built-in that also fits.`);
}
return lines.join('\n');
}
const SYSTEM_PROMPT_HEAD = `You classify a Gmail thread for a personal inbox view and, when appropriate, draft a reply on behalf of the user.
# Importance
Decide if the thread is "important" or "other":
- important: real human correspondence the user is part of (customer, investor, team, vendor, candidate); a time-sensitive notification; a message that needs a response from the user; anything worth referencing later (contracts, pricing, deadlines, decisions).
- other: newsletters, industry digests, marketing or promotional, product tips from vendors, automated notifications (verifications, recording uploads, platform policy updates), transactional confirmations (payment receipts, GST/tax filings, salary disbursements), unsolicited cold outreach.
- other: newsletters, industry digests, marketing or promotional, product tips from vendors, automated notifications (verifications, recording uploads, platform policy updates), transactional confirmations (payment receipts, GST/tax filings, salary disbursements), unsolicited cold outreach.`;
const SYSTEM_PROMPT_TAIL = `# Knowledge
Decide whether the user's knowledge base should extract durable facts from this thread:
- extract: threads involving real relationships (investors, customers, prospects, partners, vendors under contract, team, advisors, press, friends and family, government) or substantive topics (sales and deals, support, legal, finance decisions, hiring processes the user is running, fundraising, security incidents, infrastructure issues, real meetings with named people, events the user attends, warm intros, genuine follow-ups).
- skip: nothing durable to learn — spam, promotions, cold outreach, newsletters, notifications, digests, product updates, receipts, social media, mailing lists, automated scheduling reminders, travel and shopping confirmations, and unsolicited job applicants or recruiter outreach.
Importance and knowledge are independent judgments. A board member's long FYI may need no reply yet be knowledge-rich; a quick "can we move to 3pm?" needs action but adds little durable knowledge. When a message from a real relationship arrives wrapped in a bulk format (an investor update sent as a newsletter), knowledge is still extract.
# Summary (important only)
@ -227,14 +277,18 @@ export async function classifyThread(
// first-touch outreach, self-test sends) are not inbox-important —
// when a recipient replies, the thread updates and is re-classified,
// and this shortcut then correctly marks it important.
//
// Either way the user wrote the latest message themselves, so the
// knowledge pipeline always extracts: their own words carry their
// commitments, decisions, and relationships.
const needle = (userEmail ?? '').toLowerCase();
const othersParticipated = needle
? snapshot.messages.some((m) => m.from && !m.from.toLowerCase().includes(needle))
: false;
if (othersParticipated) {
return { importance: 'important' };
return { importance: 'important', category: 'correspondence', knowledge: 'extract' };
}
return { importance: 'other' };
return { importance: 'other', category: 'correspondence', knowledge: 'extract' };
}
try {
@ -249,22 +303,37 @@ export async function classifyThread(
const config = await resolveProviderConfig(provider);
const model = createLanguageModel(config, modelId);
let systemPrompt = options.skipDraft
? `${SYSTEM_PROMPT}\n\n# Skip the draft\n\nThe user already has their own draft in progress for this thread — DO NOT generate a draftResponse. Always omit the draftResponse field.`
: SYSTEM_PROMPT;
// Assembled fresh per call so labels the user just defined apply to
// the very next classification (the registry read is mtime-cached).
const labels = getEmailLabels();
const basePrompt = `${SYSTEM_PROMPT_HEAD}\n\n${buildCategorySection(labels)}\n\n${SYSTEM_PROMPT_TAIL}`;
// The user's learned importance preferences override the generic
// criteria — appended last so they take precedence.
let systemPrompt = options.skipDraft
? `${basePrompt}\n\n# Skip the draft\n\nThe user already has their own draft in progress for this thread — DO NOT generate a draftResponse. Always omit the draftResponse field.`
: basePrompt;
// The user's learned preferences override the generic criteria —
// appended last so they take precedence.
const feedback = formatImportanceFeedbackForPrompt();
if (feedback) {
systemPrompt = `${systemPrompt}\n\n${feedback}`;
}
const categoryFeedback = formatCategoryFeedbackForPrompt();
if (categoryFeedback) {
systemPrompt = `${systemPrompt}\n\n${categoryFeedback}`;
}
// The user's explicit standing instructions go last — the final word
// over both the generic criteria and the learned feedback.
const instructions = formatEmailInstructionsForPrompt();
if (instructions) {
systemPrompt = `${systemPrompt}\n\n${instructions}`;
}
const result = await withUseCase({ useCase: 'knowledge_sync', subUseCase: 'email_classifier' }, () => generateObjectSafe({
model,
system: systemPrompt,
prompt: buildPrompt(snapshot, userEmail, styleGuide, calendar),
schema: ClassificationSchema,
schema: buildClassificationSchema(labels.map((l) => l.id)),
retry: true,
}));
@ -276,7 +345,19 @@ export async function classifyThread(
usage: result.usage,
});
const out: Classification = { importance: result.object.importance };
const out: Classification = {
importance: result.object.importance,
category: result.object.category,
knowledge: result.object.knowledge,
};
// Guardrail, enforced in code rather than the prompt: if the user ever
// wrote in this thread, the knowledge pipeline must see it — their own
// messages carry commitments and decisions regardless of how the LLM
// categorized the thread.
const needle = (userEmail ?? '').toLowerCase();
if (needle && snapshot.messages.some((m) => (m.from || '').toLowerCase().includes(needle))) {
out.knowledge = 'extract';
}
if (result.object.importance === 'important') {
if (result.object.summary) out.summary = result.object.summary;
if (!options.skipDraft && result.object.draftResponse) out.draftResponse = result.object.draftResponse;
@ -284,6 +365,9 @@ export async function classifyThread(
return out;
} catch (err) {
console.warn(`[Email classifier] LLM call failed for thread ${snapshot.threadId}:`, err);
// Fail open on importance so real mail is never hidden, but leave
// category/knowledge absent — callers must not stamp a verdict that
// was never made (the sync sweep retries these threads later).
return { importance: 'important' };
}
}

91
apps/x/packages/core/src/knowledge/email_category_feedback.ts Normal file
View file

@ -0,0 +1,91 @@
import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
import type { EmailCategory } from './classify_thread.js';
/**
* User-feedback loop for the email category classifier — the sibling of
* email_importance_feedback.ts, deliberately lighter: corrections are stored
* and injected as few-shot examples on every classification call, but there
* is no distillation pass yet (categories are far less personal than
* importance; the few-shot examples carry most of the signal).
*
* The user's explicit category on a specific thread is always sticky: it is
* stored on the inbox_lists entry (categorySource: 'user') and
* re-classification never overrides it.
*/
const FEEDBACK_PATH = path.join(WorkDir, 'config', 'email_category_feedback.json');
const MAX_CORRECTIONS = 200;
const FEW_SHOT_COUNT = 20;
export interface CategoryCorrection {
threadId: string;
subject: string;
from: string;
/** What the classifier had said before the user changed it. */
agentCategory: EmailCategory | 'unknown';
/** What the user says it actually is. */
userCategory: EmailCategory;
at: string; // ISO
}
interface CategoryFeedback {
corrections: CategoryCorrection[];
}
export function loadCategoryFeedback(): CategoryFeedback {
try {
if (!fs.existsSync(FEEDBACK_PATH)) return { corrections: [] };
const parsed = JSON.parse(fs.readFileSync(FEEDBACK_PATH, 'utf-8'));
return { corrections: Array.isArray(parsed.corrections) ? parsed.corrections : [] };
} catch (err) {
console.warn('[CategoryFeedback] Failed to load, starting fresh:', err);
return { corrections: [] };
}
}
function saveCategoryFeedback(fb: CategoryFeedback): void {
const dir = path.dirname(FEEDBACK_PATH);
if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
const tmp = FEEDBACK_PATH + '.tmp';
fs.writeFileSync(tmp, JSON.stringify(fb, null, 2));
fs.renameSync(tmp, FEEDBACK_PATH);
}
/**
* Record a user correction. One entry per thread — re-picking keeps only the
* latest choice, and picking the classifier's original verdict drops the
* correction (no disagreement left to learn).
*/
export function recordCategoryCorrection(correction: CategoryCorrection): void {
const fb = loadCategoryFeedback();
const existing = fb.corrections.find(c => c.threadId === correction.threadId);
// The verdict the agent originally produced is the stable "before".
const agentCategory = existing ? existing.agentCategory : correction.agentCategory;
fb.corrections = fb.corrections.filter(c => c.threadId !== correction.threadId);
if (correction.userCategory !== agentCategory) {
fb.corrections.push({ ...correction, agentCategory });
if (fb.corrections.length > MAX_CORRECTIONS) {
fb.corrections = fb.corrections.slice(-MAX_CORRECTIONS);
}
}
saveCategoryFeedback(fb);
}
/**
* Render recent category corrections for injection into the classifier
* prompt. Returns null when there is nothing learned yet.
*/
export function formatCategoryFeedbackForPrompt(): string | null {
const fb = loadCategoryFeedback();
if (fb.corrections.length === 0) return null;
const lines: string[] = [];
lines.push(`# This user's category corrections (ground truth — these OVERRIDE the generic category definitions above)`);
lines.push('');
for (const c of fb.corrections.slice(-FEW_SHOT_COUNT)) {
lines.push(`- From: ${c.from} | Subject: "${c.subject}" → user says ${c.userCategory}${c.agentCategory !== 'unknown' ? ` (classifier had said ${c.agentCategory})` : ''}`);
}
return lines.join('\n');
}

69
apps/x/packages/core/src/knowledge/email_instructions.ts Normal file
View file

@ -0,0 +1,69 @@
import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
/**
* Free-text user instructions for the email classifier/drafter — the direct
* channel alongside the learned feedback loops. Whatever the user writes here
* is injected verbatim into every classification call (which also writes the
* pre-drafted replies), so it steers importance, category, knowledge AND
* draft tone in one place: "never draft replies to recruiters", "investor
* updates are correspondence, not newsletters", "sign off with 'Cheers'".
*
* Stored as plain markdown so it's also editable directly on disk.
*/
const INSTRUCTIONS_PATH = path.join(WorkDir, 'config', 'email_instructions.md');
const MAX_LENGTH = 8000;
/**
* First-run seed: a starter set of granular labels for real human mail, so
* the user opens the instructions dialog to something concrete they can edit
* or delete rather than an empty box. Must stay in sync with
* DEFAULT_CUSTOM_LABELS in email_labels.ts — these lines are what the label
* extractor would produce those labels from on the next save.
*/
export const DEFAULT_EMAIL_INSTRUCTIONS = [
'Create a label "Investor" for emails from investors and VCs.',
'Create a label "Candidate" for emails from job applicants and people interviewing with us.',
'Create a label "Customer" for emails from paying customers and active pilots.',
].join('\n');
export function loadEmailInstructions(): string {
try {
return fs.readFileSync(INSTRUCTIONS_PATH, 'utf-8').trim();
} catch {
// First run: seed the file so the dialog opens populated. An existing
// file — even one the user saved as empty — is never overwritten.
try {
const dir = path.dirname(INSTRUCTIONS_PATH);
if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
fs.writeFileSync(INSTRUCTIONS_PATH, DEFAULT_EMAIL_INSTRUCTIONS + '\n', { flag: 'wx' });
} catch { /* raced or unwritable — fall through */ }
return DEFAULT_EMAIL_INSTRUCTIONS;
}
}
export function saveEmailInstructions(text: string): { ok: boolean; error?: string } {
try {
const dir = path.dirname(INSTRUCTIONS_PATH);
if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
fs.writeFileSync(INSTRUCTIONS_PATH, text.slice(0, MAX_LENGTH).trim() + '\n');
return { ok: true };
} catch (err) {
return { ok: false, error: err instanceof Error ? err.message : String(err) };
}
}
/** Render for injection into the classifier system prompt; null when empty. */
export function formatEmailInstructionsForPrompt(): string | null {
const text = loadEmailInstructions();
if (!text) return null;
return [
`# The user's standing instructions (highest priority — follow these over everything above)`,
``,
`The user wrote these instructions for you directly. They govern importance, category, knowledge, and how drafts are written:`,
``,
text,
].join('\n');
}

183
apps/x/packages/core/src/knowledge/email_labels.ts Normal file
View file

@ -0,0 +1,183 @@
import fs from 'fs';
import path from 'path';
import { z } from 'zod';
import { WorkDir } from '../config/config.js';
import { createLanguageModel } from '../models/models.js';
import { generateObjectSafe } from '../models/structured.js';
import { getKgModel, resolveProviderConfig } from '../models/defaults.js';
import { captureLlmUsage } from '../analytics/usage.js';
import { withUseCase } from '../analytics/use_case.js';
/**
* The email label registry: the built-in category set plus user-defined
* custom labels. The classifier's category enum, its prompt section, and the
* UI's chips / filter pills / correction dropdown are all built from this
* one list, so a label created here exists everywhere at once.
*
* Custom labels come from the user's free-text agent instructions: on save,
* an extraction pass (syncCustomLabelsFromInstructions) pulls out any labels
* the user asked for and persists them to config/email_labels.json. Built-in
* labels are code-defined and immutable; ids are stable slugs because they
* get stamped into snapshots, markdown frontmatter, and feedback files.
*/
export interface EmailLabelDef {
/** Stable slug — stamped into snapshots/frontmatter; never renamed. */
id: string;
/** Display name shown on chips, pills, and the correction dropdown. */
name: string;
/** Classifier guidance — what belongs under this label. */
description: string;
kind: 'builtin' | 'custom';
}
export const BUILTIN_EMAIL_LABELS: EmailLabelDef[] = [
{ id: 'correspondence', name: 'Direct', kind: 'builtin', description: 'A real person writing to (or with) the user — there is prior engagement or a genuine relationship. A cold sender bumping their own unanswered email is NOT correspondence; that stays cold_outreach.' },
{ id: 'meeting', name: 'Calendar', kind: 'builtin', description: 'Scheduling with named people — calendar invites, availability requests, reschedules. Automated meeting reminders with no human context ("your meeting starts in 10 minutes") are notification, not meeting.' },
{ id: 'notification', name: 'Notification', kind: 'builtin', description: 'Automated system messages — verifications, password resets, recording uploads, policy updates, deploy/CI alerts, social and forum digests.' },
{ id: 'newsletter', name: 'News', kind: 'builtin', description: 'Subscription content, industry reports, community digests, product tips — even from platforms the user actively uses.' },
{ id: 'promotion', name: 'Marketing', kind: 'builtin', description: 'Marketing, offers, product launches, webinar/workshop invites from companies, startup-program upsells.' },
{ id: 'cold_outreach', name: 'Pitch', kind: 'builtin', description: 'Unsolicited pitches from people with no prior engagement — agencies, dev shops, freelancers, hiring platforms — even when they mention the user\'s company by name or offer something free.' },
{ id: 'receipt', name: 'Receipt', kind: 'builtin', description: 'Completed transactions with no decision remaining — payment receipts, payroll, tax filings, order/shipping confirmations, travel bookings.' },
];
const LABELS_PATH = path.join(WorkDir, 'config', 'email_labels.json');
const MAX_CUSTOM_LABELS = 12;
const BUILTIN_IDS = new Set(BUILTIN_EMAIL_LABELS.map(l => l.id));
/**
* First-run seed, matching DEFAULT_EMAIL_INSTRUCTIONS in email_instructions.ts
* line for line — so the starter labels exist before the user ever saves, and
* the next save's extraction round-trips to the same set.
*/
const DEFAULT_CUSTOM_LABELS: EmailLabelDef[] = [
{ id: 'investor', name: 'Investor', kind: 'custom', description: 'Emails from investors and VCs — current, prospective, or in an active raise.' },
{ id: 'candidate', name: 'Candidate', kind: 'custom', description: 'Emails from job applicants and people interviewing with the user\'s company.' },
{ id: 'customer', name: 'Customer', kind: 'custom', description: 'Emails from paying customers and active pilots.' },
];
let cachedCustom: EmailLabelDef[] | null = null;
let cachedMtimeMs: number | null = null;
function loadCustomLabels(): EmailLabelDef[] {
try {
if (!fs.existsSync(LABELS_PATH)) {
// First run: seed the starter custom set (mirrors the seeded
// instructions). A user who deletes those instruction lines gets
// the labels removed by the next save's sync — this seed only
// ever fires when no labels file exists at all.
saveCustomLabels(DEFAULT_CUSTOM_LABELS);
}
const stats = fs.statSync(LABELS_PATH);
if (cachedCustom && cachedMtimeMs === stats.mtimeMs) return cachedCustom;
const parsed = JSON.parse(fs.readFileSync(LABELS_PATH, 'utf-8'));
const custom: EmailLabelDef[] = Array.isArray(parsed.custom)
? parsed.custom
.filter((l: EmailLabelDef) => l && typeof l.id === 'string' && typeof l.name === 'string' && !BUILTIN_IDS.has(l.id))
.map((l: EmailLabelDef) => ({ id: l.id, name: l.name, description: String(l.description ?? ''), kind: 'custom' as const }))
: [];
cachedCustom = custom.slice(0, MAX_CUSTOM_LABELS);
cachedMtimeMs = stats.mtimeMs;
return cachedCustom;
} catch {
cachedCustom = null;
cachedMtimeMs = null;
return [];
}
}
function saveCustomLabels(custom: EmailLabelDef[]): void {
const dir = path.dirname(LABELS_PATH);
if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
const tmp = LABELS_PATH + '.tmp';
fs.writeFileSync(tmp, JSON.stringify({ custom }, null, 2) + '\n');
fs.renameSync(tmp, LABELS_PATH);
cachedCustom = null;
cachedMtimeMs = null;
}
/** Built-ins followed by the user's custom labels. Never empty. */
export function getEmailLabels(): EmailLabelDef[] {
return [...BUILTIN_EMAIL_LABELS, ...loadCustomLabels()];
}
function slugify(name: string): string {
return name
.toLowerCase()
.replace(/[^a-z0-9]+/g, '_')
.replace(/^_+|_+$/g, '')
.slice(0, 40);
}
const ExtractedLabels = z.object({
labels: z.array(z.object({
name: z.string().describe('Short display name for the label, e.g. "Investor", "Portfolio", "Legal".'),
description: z.string().describe('One or two sentences defining what belongs under this label, written as classifier guidance, derived from the user\'s instructions.'),
})).max(MAX_CUSTOM_LABELS).describe('The custom email labels the user\'s instructions define. Empty when the instructions define none.'),
});
/**
* Extract user-defined labels from the free-text agent instructions and
* persist them as the custom label set (full replace: labels the user has
* removed from their instructions disappear from the registry; threads
* already stamped with a removed id keep it and render via a fallback).
*
* Steering rules for existing labels ("investor updates are Direct, not
* News") are NOT labels and must not create one — the instructions text
* itself is injected into every classification call and handles steering.
*/
export async function syncCustomLabelsFromInstructions(instructions: string): Promise<{ labels: EmailLabelDef[] }> {
const text = instructions.trim();
if (!text) {
saveCustomLabels([]);
return { labels: getEmailLabels() };
}
const { model: modelId, provider } = await getKgModel();
const config = await resolveProviderConfig(provider);
const model = createLanguageModel(config, modelId);
const builtinList = BUILTIN_EMAIL_LABELS.map(l => `- ${l.id} ("${l.name}"): ${l.description}`).join('\n');
const result = await withUseCase({ useCase: 'knowledge_sync', subUseCase: 'email_label_extractor' }, () => generateObjectSafe({
model,
system: [
`You read a user's written instructions for their email agent and extract any CUSTOM CATEGORY LABELS the instructions define.`,
``,
`A custom label exists only when the user clearly wants emails grouped under a name of their own — "create a label Portfolio for...", "tag anything about the Series A as Fundraise", "I want an Investors label".`,
``,
`These are NOT custom labels:`,
`- Steering rules that map mail onto an existing built-in label ("investor updates are Direct, not News").`,
`- Importance rules ("emails from my board are always important").`,
`- Drafting preferences ("keep replies short").`,
``,
`Built-in labels that already exist (never re-create these or near-synonyms of them):`,
builtinList,
``,
`Return an empty list when the instructions define no custom labels.`,
].join('\n'),
prompt: text,
schema: ExtractedLabels,
retry: true,
}));
captureLlmUsage({
useCase: 'knowledge_sync',
subUseCase: 'email_label_extractor',
model: modelId,
provider,
usage: result.usage,
});
const seen = new Set<string>();
const custom: EmailLabelDef[] = [];
for (const l of result.object.labels) {
const id = slugify(l.name);
if (!id || BUILTIN_IDS.has(id) || seen.has(id)) continue;
seen.add(id);
custom.push({ id, name: l.name.trim(), description: l.description.trim(), kind: 'custom' });
}
saveCustomLabels(custom);
return { labels: getEmailLabels() };
}

257
apps/x/packages/core/src/knowledge/label_emails.ts
View file

@ -1,257 +0,0 @@
import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
import { runWhenPossible, toolInputPaths } from '../runtime/assembly/headless-app.js';
import { getKgModel } from '../models/defaults.js';
import { getErrorDetails } from '../application/lib/errors.js';
import { serviceLogger } from '../services/service_logger.js';
import { limitEventItems } from './limit_event_items.js';
import {
loadLabelingState,
saveLabelingState,
markFileAsLabeled,
type LabelingState,
} from './labeling_state.js';
const SYNC_INTERVAL_MS = 15 * 1000; // 15 seconds
const BATCH_SIZE = 15;
const DEFAULT_CONCURRENCY = 3;
const LABELING_AGENT = 'labeling_agent';
const GMAIL_SYNC_DIR = path.join(WorkDir, 'gmail_sync');
const MAX_CONTENT_LENGTH = 8000;
/**
* Find email files that haven't been labeled yet
*/
function getUnlabeledEmails(state: LabelingState): string[] {
if (!fs.existsSync(GMAIL_SYNC_DIR)) {
return [];
}
const unlabeled: string[] = [];
function traverse(dir: string) {
const entries = fs.readdirSync(dir);
for (const entry of entries) {
const fullPath = path.join(dir, entry);
const stat = fs.statSync(fullPath);
if (stat.isDirectory()) {
traverse(fullPath);
} else if (stat.isFile() && entry.endsWith('.md')) {
// Skip if already tracked in state
if (state.processedFiles[fullPath]) {
continue;
}
// Skip if file already has frontmatter
try {
const content = fs.readFileSync(fullPath, 'utf-8');
if (content.startsWith('---')) {
continue;
}
} catch {
continue;
}
unlabeled.push(fullPath);
}
}
}
traverse(GMAIL_SYNC_DIR);
return unlabeled;
}
/**
* Label a batch of email files using the labeling agent
*/
async function labelEmailBatch(
files: { path: string; content: string }[]
): Promise<{ runId: string; filesEdited: Set<string> }> {
let message = `Label the following ${files.length} email files by prepending YAML frontmatter.\n\n`;
message += `**Important:** Use workspace-relative paths with file-editText (e.g. "gmail_sync/email.md", NOT absolute paths).\n\n`;
for (let i = 0; i < files.length; i++) {
const file = files[i];
const relativePath = path.relative(WorkDir, file.path);
const truncated = file.content.length > MAX_CONTENT_LENGTH
? file.content.slice(0, MAX_CONTENT_LENGTH) + '\n\n[... content truncated, use file-readText for full content ...]'
: file.content;
message += `## File ${i + 1}: ${relativePath}\n\n`;
message += truncated;
message += `\n\n---\n\n`;
}
const { turnId, state } = await runWhenPossible({
agentId: LABELING_AGENT,
message,
...(await getKgModel()),
throwOnError: true,
});
// Edited paths come from the durable turn state instead of streaming
// bus subscriptions.
return { runId: turnId, filesEdited: toolInputPaths(state, ['file-editText']) };
}
/**
* Process all unlabeled emails in batches
*/
export async function processUnlabeledEmails(concurrency: number = DEFAULT_CONCURRENCY): Promise<void> {
console.log('[EmailLabeling] Checking for unlabeled emails...');
const state = loadLabelingState();
const unlabeled = getUnlabeledEmails(state);
if (unlabeled.length === 0) {
console.log('[EmailLabeling] No unlabeled emails found');
return;
}
console.log(`[EmailLabeling] Found ${unlabeled.length} unlabeled emails (concurrency: ${concurrency})`);
const run = await serviceLogger.startRun({
service: 'email_labeling',
message: `Labeling ${unlabeled.length} email${unlabeled.length === 1 ? '' : 's'}`,
trigger: 'timer',
});
const relativeFiles = unlabeled.map(f => path.relative(WorkDir, f));
const limitedFiles = limitEventItems(relativeFiles);
await serviceLogger.log({
type: 'changes_identified',
service: run.service,
runId: run.runId,
level: 'info',
message: `Found ${unlabeled.length} unlabeled email${unlabeled.length === 1 ? '' : 's'}`,
counts: { emails: unlabeled.length },
items: limitedFiles.items,
truncated: limitedFiles.truncated,
});
// Build all batches upfront
const batches: { batchNumber: number; files: { path: string; content: string }[] }[] = [];
for (let i = 0; i < unlabeled.length; i += BATCH_SIZE) {
const batchPaths = unlabeled.slice(i, i + BATCH_SIZE);
const batchNumber = Math.floor(i / BATCH_SIZE) + 1;
const files: { path: string; content: string }[] = [];
for (const filePath of batchPaths) {
try {
const content = fs.readFileSync(filePath, 'utf-8');
files.push({ path: filePath, content });
} catch (error) {
console.error(`[EmailLabeling] Error reading ${filePath}:`, error);
}
}
if (files.length > 0) {
batches.push({ batchNumber, files });
}
}
const totalBatches = batches.length;
let totalEdited = 0;
let hadError = false;
let failedBatches = 0;
// Process batches with concurrency limit
for (let i = 0; i < batches.length; i += concurrency) {
const chunk = batches.slice(i, i + concurrency);
const promises = chunk.map(async ({ batchNumber, files }) => {
try {
console.log(`[EmailLabeling] Processing batch ${batchNumber}/${totalBatches} (${files.length} files)`);
await serviceLogger.log({
type: 'progress',
service: run.service,
runId: run.runId,
level: 'info',
message: `Processing batch ${batchNumber}/${totalBatches} (${files.length} files)`,
step: 'batch',
current: batchNumber,
total: totalBatches,
details: { filesInBatch: files.length },
});
const result = await labelEmailBatch(files);
// Only mark files that were actually edited by the agent
for (const file of files) {
const relativePath = path.relative(WorkDir, file.path);
if (result.filesEdited.has(relativePath)) {
markFileAsLabeled(file.path, state);
}
}
console.log(`[EmailLabeling] Batch ${batchNumber}/${totalBatches} complete, ${result.filesEdited.size} files edited`);
return result.filesEdited.size;
} catch (error) {
hadError = true;
failedBatches++;
const errorDetails = getErrorDetails(error);
console.error(`[EmailLabeling] Error processing batch ${batchNumber}:`, error);
await serviceLogger.log({
type: 'error',
service: run.service,
runId: run.runId,
level: 'error',
message: `Email labeling batch ${batchNumber}/${totalBatches} failed`,
error: errorDetails,
context: { batchNumber },
});
return 0;
}
});
const results = await Promise.all(promises);
totalEdited += results.reduce((sum, n) => sum + n, 0);
// Save state after each concurrent chunk completes
saveLabelingState(state);
}
state.lastRunTime = new Date().toISOString();
saveLabelingState(state);
await serviceLogger.log({
type: 'run_complete',
service: run.service,
runId: run.runId,
level: hadError ? 'error' : 'info',
message: hadError
? `Email labeling finished with errors: ${totalEdited} files labeled`
: `Email labeling complete: ${totalEdited} files labeled`,
durationMs: Date.now() - run.startedAt,
outcome: hadError ? 'error' : 'ok',
summary: {
totalEmails: unlabeled.length,
filesLabeled: totalEdited,
failedBatches,
},
});
console.log(`[EmailLabeling] Done. ${totalEdited} emails labeled.`);
}
/**
* Main entry point - runs as independent polling service
*/
export async function init() {
console.log('[EmailLabeling] Starting Email Labeling Service...');
console.log(`[EmailLabeling] Will check for unlabeled emails every ${SYNC_INTERVAL_MS / 1000} seconds`);
// Initial run
await processUnlabeledEmails();
// Periodic polling
while (true) {
await new Promise(resolve => setTimeout(resolve, SYNC_INTERVAL_MS));
try {
await processUnlabeledEmails();
} catch (error) {
console.error('[EmailLabeling] Error in main loop:', error);
}
}
}

195
apps/x/packages/core/src/knowledge/labeling_agent.ts
View file

@ -1,195 +0,0 @@
import { renderTagSystemForEmails } from './tag_system.js';
export function getRaw(): string {
return `---
tools:
file-readText:
type: builtin
name: file-readText
file-editText:
type: builtin
name: file-editText
file-list:
type: builtin
name: file-list
---
# Task
You are an email labeling agent. Given a batch of email files, you will classify each email and prepend YAML frontmatter with structured labels.
# Email File Format
Each email is a markdown file with this structure:
\`\`\`
# {Subject line}
**Thread ID:** {hex_id}
**Message Count:** {n}
---
### From: {Display Name} <{email@address}>
**Date:** {RFC 2822 date}
{Plain-text body of the message}
---
### From: {Another Sender} <{email@address}>
**Date:** {RFC 2822 date}
{Next message in thread}
---
\`\`\`
- The \`# Subject\` heading is always the first line.
- Multi-message threads have multiple \`### From:\` blocks in chronological order, separated by \`---\`.
- Single-message threads have \`Message Count: 1\` and one \`### From:\` block.
- The body is plain text extracted from the email (HTML converted to markdown-ish text).
Use the **Subject**, **From** addresses, **Message Count**, and **body content** to classify the email.
${renderTagSystemForEmails()}
# Instructions
1. For each email file provided in the message, read its content carefully.
2. Classify the email using the taxonomy above. Think like a **YC startup founder** triaging their inbox — your time is your scarcest resource:
- **Relationship**: Who is this from? An investor, customer, team member, vendor, etc.? (\`candidate\` is a NOISE tag, not a relationship — see Filter below.)
- **Topic**: What is this about? Legal, finance, hiring, fundraising, security, infrastructure, etc.?
- **Email Type**: Is this a warm intro or a followup on an existing conversation?
- **Filter (Noise)**: Is this email noise? **Apply ALL applicable filter tags.** If even one noise tag is present the email is skipped — noise overrides everything. Common noise:
- Cold outreach / unsolicited service pitches / "YC exclusive" deals / freelancers offering free work
- Job applications, role inquiries, and recruiter mail → \`filter: ['candidate']\` (candidate is a noise tag and MUST go under filter, never under relationship)
- Newsletters, industry reports, webinar invitations, product tips from vendors
- Promotions, marketing, event invitations you did not register for, startup program upsells
- Automated notifications (email verifications, recording uploads, platform policy changes, expired OTPs)
- Transactional confirmations (salary disbursements, tax payments, GST filings, TDS workings, invoice-sharing threads)
- Spam and spam moderation digests
- **Action**: Does this need a response (\`action-required\`), is it time-sensitive (\`urgent\`), or are you waiting on them (\`waiting\`)? Use \`""\` if none apply. **Do NOT use \`fyi\` as an action value** — it is not a valid action tag.
3. **Apply noise tags aggressively.** Noise tags can and should coexist with relationship and topic tags. A salary confirmation from your finance team should have BOTH \`relationship: ['team']\` AND \`filter: ['receipt']\`. The noise tag determines whether a note is created — it overrides relationship and topic signals.
4. Be accurate — only apply labels that clearly fit. But when an email IS noise, always add the noise tag even when other tags are present.
5. Use \`file-editText\` to prepend YAML frontmatter to the file. The oldString should be the first line of the file (the \`# Subject\` heading), and the newString should be the frontmatter followed by that same first line.
6. Always include \`processed: true\` and \`labeled_at\` with the current ISO timestamp.
7. If the email already has frontmatter (starts with \`---\`), skip it.
# The Founder Signal Test
Before finalizing labels, ask: **"Would a busy YC founder want a note about this in their knowledge system?"**
**YES — create a note** if the email:
- Requires a decision or response from the founder
- Updates an active business relationship (customer deal, investor conversation, partner integration)
- Contains information that will be referenced later (pricing, terms, deadlines, compliance requirements)
- Has action items for the team (e.g. standup notes, meeting notes with to-dos)
- Presents a genuine opportunity worth evaluating (accelerator, partnership, relevant hire)
- Flags a risk that needs attention (security vulnerability, legal issue, compliance blocker)
- Is from a vendor you are actively engaged with on an ongoing process (e.g. your compliance assessor following up after a call you participated in)
**NO — skip it** if the email:
- Confirms a transaction that already happened with no open decision (payment received, tax filed, salary disbursed, invoice shared)
- Is a system-generated notification with no decision needed (email verification, recording uploaded, policy update, expired OTP)
- Is unsolicited outreach from someone you have never engaged with — regardless of how personalized it sounds
- Is a newsletter, industry report, webinar invitation, or product tips email
- Is marketing or promotional content, including from vendors you use
- Is a spam digest or Google Groups moderation report
- Is routine operational correspondence where the transaction is complete and no follow-up remains
# Cold Outreach Detection (Critical for Precision)
Many emails disguise themselves as real relationships. Before assigning \`vendor\`, \`candidate\`, \`partner\`, or \`followup\`, apply these tests:
**It's \`cold-outreach\` (noise), NOT a real relationship, if:**
- The sender is pitching their own product or service — design agencies, compliance firms, content/copy writers, dev shops, freelancers, trademark services, company closure/winding-down services, hiring platforms, etc. — even if they reference your company by name, your YC batch, or offer something "free" or "exclusive for YC founders."
- The thread consists entirely of the same sender following up on their own unanswered messages. A real followup requires prior two-way engagement.
- A student, job-seeker, freelancer, or founder cold-emails asking for your time, feedback, or offering free work/trials. These are NOT \`candidate\` — they are \`cold-outreach\`.
- Someone invites you to an event you didn't sign up for, especially if the email has marketing formatting (tracking links, unsubscribe footers, HTML banners). This is \`promotion\`, not \`event\`.
**It IS a real relationship (not noise) if:**
- You (the inbox owner) are a participant in the thread (you sent a reply, or someone on your team did).
- The sender is from a company you are already paying, or they are providing a service under contract (e.g., your law firm, your accountant, your cloud provider support).
- The sender was introduced to you by someone you know (warm intro present in the thread).
- The sender references a specific ongoing engagement with concrete details — e.g., they are your assigned compliance assessor for an audit you initiated, or they are following up after a call you participated in. This is NOT the same as a generic "I noticed your company uses X" pitch.
**Key heuristic:** If every message in the thread is FROM the same external person and the inbox owner never replied, it's almost certainly cold outreach — regardless of how personalized it sounds. Label it \`cold-outreach\`.
# Routine Operations & Finance (Often Missed as Noise)
These emails involve real relationships (team, vendor) and real topics (finance) but are **noise** because the transaction is complete and no decision remains. They MUST get a filter tag even though they also have relationship/topic tags:
- **Salary/payroll confirmations**: "Total salary disbursement is INR X, transfer initiated" → \`filter: ['receipt']\`
- **Tax payment acknowledgements**: Income tax challan confirmations, TDS workings sent for processing → \`filter: ['receipt']\`
- **GST/compliance filing confirmations**: GSTR1 ARN generated, GST OTPs (expired or used) → \`filter: ['receipt']\`
- **Recurring invoice sharing**: Monthly cloud/SaaS invoices shared between team and finance dept → \`filter: ['receipt']\`
- **Payment transfer confirmations**: "Transfer initiated", "Payment confirmed" → \`filter: ['receipt']\`
# Automated Notifications (Often Missed as Noise)
System-generated messages that require no decision:
- **Email verifications**: "Confirm your email address on Slack" → \`filter: ['notification']\`
- **Meeting recordings**: "Your meeting recording is ready in Google Drive" → \`filter: ['notification']\`
- **Platform policy updates**: "Billing permissions are changing starting next month" → \`filter: ['notification']\`
- **Expired OTPs**: One-time passwords for completed actions → \`filter: ['notification']\`
# Meeting vs Scheduling (Critical Distinction)
- **topic: meeting** (CREATE) — A calendar invite or scheduling email for a real meeting with a **named person** you have a relationship with: an investor, customer, partner, candidate, advisor, team member. Examples: "Invitation: Zoom: Rowboat Labs <> Dalton Caldwell", "YC between Peer Richelsen and Arjun", "Rowboat <> Smash Capital". The key signal is a specific person or company in the subject/body.
- **filter: scheduling** (SKIP) — Automated reminders and scheduling tool notifications with **no named person or meaningful context**: "Reminder: your meeting is about to start", "Our meeting in an hour", generic ChiliPiper/Calendly confirmations. These are system-generated noise.
**Rule of thumb:** If the email names who you're meeting with, it's \`topic: meeting\`. If it's just a system ping about a time slot, it's \`filter: scheduling\`.
# Newsletter & Promotion Detection (Often Missed as Noise)
These are noise even from a vendor you recognize or a platform you use:
- **Industry reports**: "Report: $1.2T in combined enterprise AI value" → \`filter: ['newsletter']\`
- **Webinar/workshop invitations**: "Register for our knowledge sessions", "5 Slots Left. Pitch Tomorrow." → \`filter: ['promotion']\`
- **Product tips and tutorials**: "Discover more with your free account" → \`filter: ['newsletter']\`
- **Startup program marketing**: "Reminder - Register for AI Architecture sessions" → \`filter: ['promotion']\`
**Exception:** If a tool your team actively uses is expiring and you need to make an upgrade/cancellation decision, that is NOT noise — it requires action.
# Spam Digests Are Always Spam
If the sender is \`noreply-spamdigest\` (Google Groups spam moderation reports), label it \`filter: ['spam']\`. Google already flagged these as spam. Do not evaluate the held messages inside — the digest itself is noise.
# Filter array must only contain tags from the Noise category
Do not put topic or relationship tags into the filter array. If an email is an event promotion, use \`promotion\` in filter — not \`event\`.
# Frontmatter Format
\`\`\`yaml
---
labels:
relationship:
- investor
topics:
- fundraising
- finance
type: intro
filter:
- []
action: action-required
processed: true
labeled_at: "2026-02-28T12:00:00Z"
---
\`\`\`
# Rules
- Every label category must be present in the frontmatter, even if empty (use \`[]\` for empty arrays).
- \`type\` and \`action\` are single values (strings), not arrays. Use empty string \`""\` if not applicable.
- \`relationship\`, \`topics\`, and \`filter\` are arrays.
- The \`action\` field only accepts: \`action-required\`, \`urgent\`, \`waiting\`, or \`""\`. Never use \`fyi\` as an action value.
- Use the exact label values from the taxonomy — do not invent new ones.
- The \`labeled_at\` timestamp should be the current time in ISO 8601 format.
- Process all files in the batch. Do not skip any unless they already have frontmatter.
- **Noise labels are skip signals.** If an email is clearly a newsletter, cold outreach, promotion, digest, receipt, notification, or other noise — label it in the \`filter\` array. These emails will NOT create notes.
- **Noise tags coexist with other tags.** An email from your team about salary (\`relationship: ['team']\`, \`topics: ['finance']\`) that is just a payroll confirmation should ALSO have \`filter: ['receipt']\`. The noise tag overrides — it ensures the email is skipped even when relationship/topic tags are present.
- **When in doubt, ask:** "Does this email change any decision, require any follow-up, or update a relationship I need to track?" If no, it's noise — add the appropriate filter tag.
`;
}

48
apps/x/packages/core/src/knowledge/labeling_state.ts
View file

@ -1,48 +0,0 @@
import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
const STATE_FILE = path.join(WorkDir, 'labeling_state.json');
export interface LabelingState {
processedFiles: Record<string, { labeledAt: string }>;
lastRunTime: string;
}
export function loadLabelingState(): LabelingState {
if (fs.existsSync(STATE_FILE)) {
try {
return JSON.parse(fs.readFileSync(STATE_FILE, 'utf-8'));
} catch (error) {
console.error('Error loading labeling state:', error);
}
}
return {
processedFiles: {},
lastRunTime: new Date(0).toISOString(),
};
}
export function saveLabelingState(state: LabelingState): void {
try {
fs.writeFileSync(STATE_FILE, JSON.stringify(state, null, 2));
} catch (error) {
console.error('Error saving labeling state:', error);
throw error;
}
}
export function markFileAsLabeled(filePath: string, state: LabelingState): void {
state.processedFiles[filePath] = {
labeledAt: new Date().toISOString(),
};
}
export function resetLabelingState(): void {
const emptyState: LabelingState = {
processedFiles: {},
lastRunTime: new Date().toISOString(),
};
saveLabelingState(emptyState);
}

45
apps/x/packages/core/src/knowledge/note_creation.ts
View file

@ -1,5 +1,4 @@
import { renderNoteTypesBlock } from './note_system.js';
import { renderNoteEffectRules } from './tag_system.js';
export function getRaw(): string {
return `---
@ -193,15 +192,13 @@ Either:
---
# The Core Rule: Label-Based Filtering
# The Core Rule: Pre-Filtered Emails
**Emails now have YAML frontmatter with labels.** Use these labels to decide whether to process or skip.
**Emails reaching you have already passed a noise filter.** The inbox classifier stamps every email's YAML frontmatter with a \`knowledge: extract | skip\` verdict, and \`skip\` emails (marketing, newsletters, notifications, receipts, cold outreach) never reach this agent. Do not re-litigate that decision — an email in front of you is presumed worth reading.
**Meetings and voice memos always create notes** — no label check needed.
**Meetings and voice memos always create notes** — no check needed.
**For emails, read the YAML frontmatter labels and apply these rules:**
${renderNoteEffectRules()}
**For emails, you still apply judgment about WHAT to extract:** an admitted email can still contain nothing durable (a bare "thanks!", a scheduling ping with no new facts). In that case output SKIP with a reason rather than inventing a note.
---
@ -284,7 +281,7 @@ Emails containing calendar invites (\`.ics\` attachments or inline calendar data
---
# Step 1: Source Filtering (Label-Based)
# Step 1: Source Filtering
## For Meetings and Voice Memos
Always process — no filtering needed.
@ -307,33 +304,31 @@ Skip routine metadata churn and duplicated notifications unless they change dura
## For Emails — Read YAML Frontmatter
Emails have YAML frontmatter with labels prepended by the labeling agent:
Emails carry YAML frontmatter stamped by the inbox classifier:
\`\`\`yaml
---
labels:
relationship:
- Investor
topics:
- Fundraising
type: Intro
filter: []
action: FYI
processed: true
labeled_at: "2026-02-28T12:00:00Z"
importance: important
category: correspondence
knowledge: extract
classified_at: "2026-07-11T12:00:00Z"
---
\`\`\`
- \`category\` (correspondence, meeting, notification, newsletter, promotion, cold_outreach, receipt) tells you what kind of email this is — use it as context, e.g. \`meeting\` emails usually yield a person/meeting note.
- \`knowledge: extract\` is why the file reached you; \`skip\` emails are filtered out upstream.
- Older files may instead carry a legacy \`labels:\` block from the retired labeling agent — treat those the same way: the file reached you, so it passed filtering.
## Decision Rules
Apply the label rules from "The Core Rule: Label-Based Filtering" above.
Apply "The Core Rule: Pre-Filtered Emails" above: process the email, but SKIP if it genuinely contains nothing durable.
## Filter Decision Output
If skipping:
\`\`\`
SKIP
Reason: Labels indicate skip-only categories: {list the labels}
Reason: {why nothing durable — e.g. bare acknowledgement, no new facts}
\`\`\`
If processing, continue to Step 2.
@ -1303,15 +1298,15 @@ ${renderNoteTypesBlock()}
---
# Summary: Label-Based Rules
# Summary: Filtering Rules
| Source Type | Creates Notes? | Updates Notes? | Detects State Changes? |
|-------------|---------------|----------------|------------------------|
| Meeting | Yes | Yes | Yes |
| Voice memo | Yes | Yes | Yes |
| Email (create label + user replied in thread) | Yes | Yes | Yes |
| Email (create label, purely inbound — no user reply) | Update-only (no new People/Org notes) | Yes | Yes |
| Email (only skip labels) | No (SKIP) | No | No |
| Email (user replied in thread) | Yes | Yes | Yes |
| Email (purely inbound — no user reply) | Update-only (no new People/Org notes) | Yes | Yes |
| Email (nothing durable in it) | No (SKIP) | No | No |
**Email Reply Gate:** New canonical People/Organization notes from an email require the user to have replied at least once in the thread (a \`### From:\` matching \`user.email\` or \`@user.domain\`). Purely inbound threads update existing notes only. Calendar invites for a scheduled meeting are exempt.

50
apps/x/packages/core/src/knowledge/run_pipeline.ts
View file

@ -1,22 +1,23 @@
#!/usr/bin/env node
/**
* Standalone pipeline runner for email labeling, graph building, and note tagging.
* Standalone pipeline runner for graph building and note tagging.
*
* Usage:
* npx tsx packages/core/src/knowledge/run_pipeline.ts --workdir /path/to/workdir
* npx tsx packages/core/src/knowledge/run_pipeline.ts --workdir /path/to/workdir --steps label,graph,tag
* npx tsx packages/core/src/knowledge/run_pipeline.ts --workdir /path/to/workdir --steps label
* npx tsx packages/core/src/knowledge/run_pipeline.ts --workdir /path/to/workdir --steps graph,tag
* npx tsx packages/core/src/knowledge/run_pipeline.ts --workdir /path/to/workdir --steps graph
*
* The workdir should contain a gmail_sync/ folder with email markdown files.
* Email classification frontmatter is stamped by the Gmail sync (see
* sync_gmail.ts / classify_thread.ts) — files without it are held by the
* graph step until the app's sync stamps them.
* Output notes are written to workdir/knowledge/.
*
* Steps:
* label - Classify emails with YAML frontmatter labels
* graph - Extract entities and create/update knowledge notes
* tag - Add YAML frontmatter tags to knowledge notes
*
* If --steps is omitted, all three steps run in order: label → graph → tag
* If --steps is omitted, both steps run in order: graph → tag
*/
import fs from 'fs';
@ -24,40 +25,31 @@ import path from 'path';
// --- Parse CLI args before any core imports (WorkDir reads env at import time) ---
const VALID_STEPS = ['label', 'graph', 'tag'] as const;
const VALID_STEPS = ['graph', 'tag'] as const;
type Step = typeof VALID_STEPS[number];
function parseArgs(): { workdir: string; steps: Step[]; concurrency: number } {
function parseArgs(): { workdir: string; steps: Step[] } {
const args = process.argv.slice(2);
let workdir: string | undefined;
let stepsRaw: string | undefined;
let concurrency = 3;
for (let i = 0; i < args.length; i++) {
if (args[i] === '--workdir' && args[i + 1]) {
workdir = args[++i];
} else if (args[i] === '--steps' && args[i + 1]) {
stepsRaw = args[++i];
} else if (args[i] === '--concurrency' && args[i + 1]) {
concurrency = parseInt(args[++i], 10);
if (isNaN(concurrency) || concurrency < 1) {
console.error('Error: --concurrency must be a positive integer');
process.exit(1);
}
} else if (args[i] === '--help' || args[i] === '-h') {
console.log(`
Usage: run_pipeline --workdir <path> [--steps label,graph,tag] [--concurrency N]
Usage: run_pipeline --workdir <path> [--steps graph,tag]
Options:
--workdir <path> Working directory containing gmail_sync/ folder (required)
--steps <list> Comma-separated steps to run: label, graph, tag (default: all)
--concurrency <N> Number of parallel batches for labeling (default: 3)
--steps <list> Comma-separated steps to run: graph, tag (default: all)
--help, -h Show this help message
Examples:
run_pipeline --workdir ./my-emails
run_pipeline --workdir ./my-emails --steps label --concurrency 5
run_pipeline --workdir ./my-emails --steps label,graph
run_pipeline --workdir ./my-emails --steps graph
run_pipeline --workdir ./my-emails --steps graph,tag
`);
process.exit(0);
@ -91,10 +83,10 @@ Examples:
steps = [...VALID_STEPS];
}
return { workdir, steps, concurrency };
return { workdir, steps };
}
const { workdir, steps, concurrency } = parseArgs();
const { workdir, steps } = parseArgs();
// Set env BEFORE importing core modules (WorkDir is read at module load time)
process.env.ROWBOAT_WORKDIR = workdir;
@ -104,33 +96,25 @@ process.env.ROWBOAT_WORKDIR = workdir;
async function main() {
console.log(`[Pipeline] Working directory: ${workdir}`);
console.log(`[Pipeline] Steps to run: ${steps.join(', ')}`);
console.log(`[Pipeline] Concurrency: ${concurrency}`);
console.log();
// Verify gmail_sync exists if label or graph step is requested
// Verify gmail_sync exists if the graph step is requested
const gmailSyncDir = path.join(workdir, 'gmail_sync');
if ((steps.includes('label') || steps.includes('graph')) && !fs.existsSync(gmailSyncDir)) {
if (steps.includes('graph') && !fs.existsSync(gmailSyncDir)) {
console.warn(`[Pipeline] Warning: gmail_sync/ folder not found in ${workdir}`);
}
const startTime = Date.now();
if (steps.includes('label')) {
console.log('[Pipeline] === Step 1: Email Labeling ===');
const { processUnlabeledEmails } = await import('./label_emails.js');
await processUnlabeledEmails(concurrency);
console.log();
}
if (steps.includes('graph')) {
console.log('[Pipeline] === Step 2: Graph Building ===');
console.log('[Pipeline] === Step 1: Graph Building ===');
const { processAllSources } = await import('./build_graph.js');
await processAllSources();
console.log();
}
if (steps.includes('tag')) {
console.log('[Pipeline] === Step 3: Note Tagging ===');
console.log('[Pipeline] === Step 2: Note Tagging ===');
const { processUntaggedNotes } = await import('./tag_notes.js');
await processUntaggedNotes();
console.log();

317
apps/x/packages/core/src/knowledge/sync_gmail.ts
View file

@ -9,8 +9,9 @@ import { GoogleClientFactory } from './google-client-factory.js';
import { serviceLogger, type ServiceRunContext } from '../services/service_logger.js';
import { limitEventItems } from './limit_event_items.js';
import { createEvent } from '../events/producer.js';
import { classifyThread, getUserEmail } from './classify_thread.js';
import { classifyThread, getUserEmail, type EmailCategory } from './classify_thread.js';
import { recordImportanceCorrection } from './email_importance_feedback.js';
import { recordCategoryCorrection } from './email_category_feedback.js';
import { notifyIfEnabled } from '../application/notification/notifier.js';
// Configuration
@ -110,9 +111,47 @@ export function setThreadImportance(
userVerdict: importance,
at: new Date().toISOString(),
});
// Keep the markdown mirror's stamped importance truthful. The knowledge
// verdict is deliberately untouched — "stop showing me this" must never
// silently starve the knowledge graph.
stampClassificationFrontmatter(threadId, cached.snapshot);
return { success: true, previous };
}
/**
* User explicitly picks a thread's category in the UI. Sticky on the thread
* (categorySource: 'user'; re-classification never overrides it) and recorded
* as a correction the classifier learns from as few-shot examples. Like
* importance flips, this never touches the knowledge verdict.
*/
export function setThreadCategory(
threadId: string,
category: EmailCategory,
): { success: boolean; error?: string } {
const cached = readCachedSnapshot(threadId);
if (!cached) {
return { success: false, error: `No inbox entry found for thread ${threadId}` };
}
const previous = cached.snapshot.category;
cached.snapshot.category = category;
cached.snapshot.categorySource = 'user';
try {
fs.writeFileSync(cachePath(threadId), JSON.stringify(cached), 'utf-8');
} catch (err) {
return { success: false, error: err instanceof Error ? err.message : String(err) };
}
recordCategoryCorrection({
threadId,
subject: cached.snapshot.subject || '(no subject)',
from: cached.snapshot.from || 'unknown',
agentCategory: previous ?? 'unknown',
userCategory: category,
at: new Date().toISOString(),
});
stampClassificationFrontmatter(threadId, cached.snapshot);
return { success: true };
}
export function saveMessageBodyHeight(threadId: string, messageId: string, height: number): void {
const cached = readCachedSnapshot(threadId);
if (!cached) return;
@ -245,6 +284,12 @@ export interface GmailThreadSnapshot {
importance?: 'important' | 'other';
/** 'user' when the user explicitly set importance in the UI — sticky; re-classification never overrides it. */
importanceSource?: 'user';
/** What kind of email this is (chip in the inbox UI, context for the knowledge pipeline). */
category?: EmailCategory;
/** 'user' when the user explicitly picked the category in the UI — sticky; re-classification never overrides it. */
categorySource?: 'user';
/** Knowledge-graph admission verdict, stamped into the gmail_sync markdown frontmatter. */
knowledge?: 'extract' | 'skip';
draft_response?: string;
gmail_draft?: string;
/** Gmail-side draft id, present on entries from listDraftThreads. */
@ -659,11 +704,19 @@ export interface InboxPageOptions {
section: InboxSection;
cursor?: string;
limit?: number;
/** Restrict the page to threads of this category (used by the "Everything else" filter pills). */
category?: string;
}
export interface InboxPageResult {
threads: GmailThreadSnapshot[];
nextCursor: string | null;
/**
* Threads per category across the WHOLE section (ignoring `category` and
* pagination) — drives the filter pills. Threads with no verdict yet count
* under 'unclassified'.
*/
categoryCounts: Record<string, number>;
}
interface IndexedEntry {
@ -702,7 +755,7 @@ export function listImportantThreads(opts: { cursor?: string; limit?: number } =
return listInboxPage({ section: 'important', ...opts });
}
export function listEverythingElseThreads(opts: { cursor?: string; limit?: number } = {}): InboxPageResult {
export function listEverythingElseThreads(opts: { cursor?: string; limit?: number; category?: string } = {}): InboxPageResult {
return listInboxPage({ section: 'other', ...opts });
}
@ -724,17 +777,18 @@ const listCache = new Map<string, ListCacheEntry>();
export function listInboxPage(opts: InboxPageOptions): InboxPageResult {
const limit = Math.max(1, Math.min(100, opts.limit ?? 25));
const cursor = parseCursor(opts.cursor);
const categoryCounts: Record<string, number> = {};
if (!fs.existsSync(CACHE_DIR)) {
listCache.clear();
return { threads: [], nextCursor: null };
return { threads: [], nextCursor: null, categoryCounts };
}
let names: string[];
try {
names = fs.readdirSync(CACHE_DIR);
} catch {
return { threads: [], nextCursor: null };
return { threads: [], nextCursor: null, categoryCounts };
}
const seen = new Set<string>();
@ -777,6 +831,9 @@ export function listInboxPage(opts: InboxPageOptions): InboxPageResult {
}
if (cached.section !== opts.section) continue;
const cat = cached.snapshot.category ?? 'unclassified';
categoryCounts[cat] = (categoryCounts[cat] ?? 0) + 1;
if (opts.category && cat !== opts.category) continue;
entries.push({
threadId: cached.snapshot.threadId,
dateMs: cached.dateMs,
@ -812,9 +869,48 @@ export function listInboxPage(opts: InboxPageOptions): InboxPageResult {
return {
threads: slice.map((e) => e.snapshot),
nextCursor: hasMore && last ? encodeCursor({ dateMs: last.dateMs, threadId: last.threadId }) : null,
categoryCounts,
};
}
/**
* Archive every "Everything else" thread of the given category in one sweep —
* the bulk gesture behind the category filter pills. Walks the local cache
* (never the live mailbox), so it can only touch threads the user can see in
* the section. Threads whose category was user-corrected are matched on the
* corrected value, like everywhere else.
*
* Returns per-thread failures rather than aborting: one 404 (thread deleted
* elsewhere) shouldn't strand the other 80 promotions in the inbox.
*/
export async function archiveCategoryThreads(category: string): Promise<{ archived: number; failed: number; error?: string }> {
if (!fs.existsSync(CACHE_DIR)) return { archived: 0, failed: 0 };
let names: string[];
try {
names = fs.readdirSync(CACHE_DIR);
} catch (err) {
return { archived: 0, failed: 0, error: err instanceof Error ? err.message : String(err) };
}
const targets: string[] = [];
for (const name of names) {
if (!name.endsWith('.json')) continue;
const threadId = decodeURIComponent(name.replace(/\.json$/, ''));
const snapshot = readCachedSnapshot(threadId)?.snapshot;
if (!snapshot) continue;
if (snapshotImportance(snapshot) !== 'other') continue;
if ((snapshot.category ?? 'unclassified') !== category) continue;
targets.push(threadId);
}
let archived = 0;
let failed = 0;
for (const threadId of targets) {
const result = await archiveThread(threadId);
if (result.ok) archived += 1;
else failed += 1;
}
return { archived, failed };
}
export async function listRecentThreadIds(daysAgo: number = 2): Promise<RecentThreadInfo[]> {
const auth = await GoogleClientFactory.getClient();
if (!auth) {
@ -850,6 +946,61 @@ export async function listRecentThreadIds(daysAgo: number = 2): Promise<RecentTh
return results;
}
/** Drop a leading YAML frontmatter block, returning just the document body. */
function stripLeadingFrontmatter(content: string): string {
if (!content.startsWith('---\n')) return content;
const end = content.indexOf('\n---', 4);
if (end === -1) return content;
const afterClose = content.indexOf('\n', end + 4);
if (afterClose === -1) return '';
return content.slice(afterClose + 1).replace(/^\n+/, '');
}
/**
* Stamp the classification verdict into the thread's gmail_sync markdown as
* YAML frontmatter. The knowledge-graph builder holds any email file without
* frontmatter (it can't tell noise from signal yet), so this stamp is what
* admits — or permanently excludes — the thread from knowledge extraction.
*
* No-op when the snapshot carries no verdict (LLM failure → the sweep in
* performSync retries later) or when the file already carries this verdict.
*
* Exported for tests.
*/
export function stampClassificationFrontmatter(threadId: string, snapshot: GmailThreadSnapshot): void {
if (!snapshot.category || !snapshot.knowledge) return;
const mdPath = path.join(SYNC_DIR, `${threadId}.md`);
let content: string;
try {
content = fs.readFileSync(mdPath, 'utf-8');
} catch {
return; // no markdown mirror for this thread (e.g. draft-only)
}
const importance = snapshot.importance === 'other' ? 'other' : 'important';
const frontmatter = [
'---',
`importance: ${importance}`,
`category: ${snapshot.category}`,
`knowledge: ${snapshot.knowledge}`,
`classified_at: "${new Date().toISOString()}"`,
'---',
'',
].join('\n');
if (
content.startsWith('---') &&
content.includes(`\nimportance: ${importance}\n`) &&
content.includes(`\ncategory: ${snapshot.category}\n`) &&
content.includes(`\nknowledge: ${snapshot.knowledge}\n`)
) {
return; // already stamped with this exact verdict
}
try {
fs.writeFileSync(mdPath, frontmatter + stripLeadingFrontmatter(content));
} catch (err) {
console.warn(`[Gmail] frontmatter stamp failed for ${threadId}:`, err);
}
}
/**
* Build a GmailThreadSnapshot from an already-fetched threads.get response,
* classify it, and write to inbox_lists/. Called by the background sync
@ -873,7 +1024,11 @@ async function buildAndCacheSnapshot(
// call per unchanged thread (matters most during fullSync after a
// historyId expiry, where the whole window is re-walked).
// We require `importance` to be present too — pre-classifier cache files
// would otherwise stick around forever uncategorised.
// would otherwise stick around forever uncategorised. Deliberately NOT
// requiring `category` here: a fullSync rewalk re-visits every cached
// thread, and requiring it would turn that walk into one sequential LLM
// call per pre-existing thread. Old caches gain category/knowledge via
// the budgeted sweep instead (sweepUnclassifiedMarkdown).
if (
threadData.historyId &&
cached &&
@ -881,22 +1036,30 @@ async function buildAndCacheSnapshot(
cached.parserVersion === SNAPSHOT_PARSER_VERSION &&
cached.snapshot.importance
) {
// processThread may have just rewritten the markdown mirror (which
// drops its frontmatter) — re-stamp from the cached verdict.
stampClassificationFrontmatter(threadId, cached.snapshot);
return cached.snapshot;
}
const snapshot = await parseThreadSnapshot(threadId, threadData, gmailClient);
if (!snapshot) return null;
// The user's explicit verdict on this thread is sticky — carry it over and
// skip nothing else (summary/draft still refresh below).
// The user's explicit verdicts on this thread are sticky — carry them over
// and skip nothing else (summary/draft still refresh below).
const userOverride = cached?.snapshot.importanceSource === 'user'
? cached.snapshot.importance
: undefined;
const userCategoryOverride = cached?.snapshot.categorySource === 'user'
? cached.snapshot.category
: undefined;
try {
const userEmail = await getUserEmail(auth);
const skipDraft = (snapshot.gmail_draft?.length ?? 0) > 0;
const classification = await classifyThread(snapshot, userEmail, { skipDraft });
snapshot.importance = classification.importance;
if (classification.category) snapshot.category = classification.category;
if (classification.knowledge) snapshot.knowledge = classification.knowledge;
if (classification.summary) snapshot.summary = classification.summary;
if (classification.draftResponse) {
const draftResponse = stripGmailQuotedReplyText(classification.draftResponse);
@ -910,10 +1073,15 @@ async function buildAndCacheSnapshot(
snapshot.importance = userOverride;
snapshot.importanceSource = 'user';
}
if (userCategoryOverride) {
snapshot.category = userCategoryOverride;
snapshot.categorySource = 'user';
}
if (threadData.historyId) {
writeCachedSnapshot(threadId, threadData.historyId, snapshot);
}
stampClassificationFrontmatter(threadId, snapshot);
return snapshot;
}
@ -1634,6 +1802,137 @@ async function partialSync(auth: OAuth2Client, startHistoryId: string, syncDir:
}
}
// Threads the sweep failed on (deleted in Gmail, persistent classify failure).
// In-memory so they retry on the next app launch but don't burn an LLM call
// every 30s tick in the meantime.
const sweepSkip = new Set<string>();
/**
* Classify-and-stamp any gmail_sync markdown that has no frontmatter yet.
* Covers three cases the main sync path can miss:
* - a classify call failed mid-sync (fail-open importance, no verdict)
* - files from before the unified classifier that the old labeling agent
* never got to
* - threads whose inbox cache was pruned (archived) before a verdict landed
*
* The knowledge-graph builder holds unstamped files indefinitely, so without
* this sweep those emails would silently never produce knowledge notes.
* LLM work is bounded per tick; cache-backed stamps are free and unbounded.
*/
async function sweepUnclassifiedMarkdown(auth: OAuth2Client, llmBudget: number = 15): Promise<void> {
if (!fs.existsSync(SYNC_DIR)) return;
let names: string[];
try {
names = fs.readdirSync(SYNC_DIR);
} catch {
return;
}
const gmailClient = google.gmail({ version: 'v1', auth });
let classified = 0;
let stamped = 0;
for (const name of names) {
if (!name.endsWith('.md')) continue;
const threadId = name.slice(0, -3);
if (sweepSkip.has(threadId)) continue;
let content: string;
try {
content = fs.readFileSync(path.join(SYNC_DIR, name), 'utf-8');
} catch {
continue;
}
// Anything with frontmatter is either already stamped or carries the
// legacy labeling-agent verdict the graph builder still understands.
if (content.startsWith('---')) continue;
const cached = readCachedSnapshot(threadId)?.snapshot;
if (cached?.category && cached.knowledge) {
stampClassificationFrontmatter(threadId, cached);
stamped += 1;
continue;
}
if (classified >= llmBudget) continue;
classified += 1;
try {
const res = await gmailClient.users.threads.get({ userId: 'me', id: threadId });
if (cached) {
// In the inbox — reuse the full path (classify, cache, stamp).
await buildAndCacheSnapshot(threadId, res.data, gmailClient, auth);
if (!readCachedSnapshot(threadId)?.snapshot.category) sweepSkip.add(threadId);
} else {
// Not in the inbox cache (archived/pruned). Classify and stamp
// only — writing the cache would resurrect the thread in the
// inbox UI.
const snapshot = await parseThreadSnapshot(threadId, res.data, gmailClient);
if (!snapshot) {
sweepSkip.add(threadId);
continue;
}
const userEmail = await getUserEmail(auth);
const classification = await classifyThread(snapshot, userEmail, { skipDraft: true });
snapshot.importance = classification.importance;
if (classification.category) snapshot.category = classification.category;
if (classification.knowledge) snapshot.knowledge = classification.knowledge;
stampClassificationFrontmatter(threadId, snapshot);
if (!classification.category) sweepSkip.add(threadId);
}
} catch (err) {
console.warn(`[Gmail] classification sweep failed for ${threadId}:`, err);
sweepSkip.add(threadId);
}
}
// Enrich pre-upgrade inbox caches (importance but no category) so old
// threads get chips and a knowledge verdict. The cached snapshot already
// holds the messages — no Gmail fetch needed. Deliberately does NOT
// re-stamp markdown that carries legacy labels: frontmatter — rewriting
// those files would make the graph builder reprocess emails it already
// extracted (mtime+hash change detection).
if (classified < llmBudget && fs.existsSync(CACHE_DIR)) {
let cacheNames: string[] = [];
try {
cacheNames = fs.readdirSync(CACHE_DIR);
} catch { /* ignore */ }
const userEmail = await getUserEmail(auth);
for (const name of cacheNames) {
if (classified >= llmBudget) break;
if (!name.endsWith('.json')) continue;
const threadId = decodeURIComponent(name.replace(/\.json$/, ''));
if (sweepSkip.has(threadId)) continue;
const entry = readCachedSnapshot(threadId);
if (!entry || !entry.snapshot.importance || entry.snapshot.category) continue;
classified += 1;
try {
const classification = await classifyThread(entry.snapshot, userEmail, { skipDraft: true });
if (!classification.category || !classification.knowledge) {
sweepSkip.add(threadId);
continue;
}
entry.snapshot.category = classification.category;
entry.snapshot.knowledge = classification.knowledge;
// The user's sticky verdict (and the previously shown one) win —
// this pass is about adding the missing fields, not re-judging.
writeCachedSnapshot(threadId, entry.historyId, entry.snapshot);
const mdPath = path.join(SYNC_DIR, `${threadId}.md`);
let mdContent: string | null = null;
try {
mdContent = fs.readFileSync(mdPath, 'utf-8');
} catch { /* no markdown mirror */ }
if (mdContent !== null && !mdContent.startsWith('---')) {
stampClassificationFrontmatter(threadId, entry.snapshot);
}
} catch (err) {
console.warn(`[Gmail] cache enrichment failed for ${threadId}:`, err);
sweepSkip.add(threadId);
}
}
}
if (classified > 0 || stamped > 0) {
console.log(`[Gmail] classification sweep: ${classified} classified, ${stamped} stamped from cache`);
}
}
async function performSync() {
const LOOKBACK_DAYS = 7; // Default to 1 week
const ATTACHMENTS_DIR = path.join(SYNC_DIR, 'attachments');
@ -1685,6 +1984,10 @@ async function performSync() {
// remove cache files for threads that were archived/trashed elsewhere.
await pruneInboxCache(auth);
// Backfill classification verdicts onto any markdown the main sync
// paths missed — the knowledge graph holds unstamped files forever.
await sweepUnclassifiedMarkdown(auth);
console.log("Sync completed.");
} catch (error) {
console.error("Error during sync:", error);

34
apps/x/packages/core/src/knowledge/tag_system.ts
View file

@ -194,41 +194,7 @@ function renderTagGroups(tags: TagDefinition[]): string {
return `# Tag System Reference\n\n${sections.join('\n\n')}`;
}
export function renderNoteEffectRules(): string {
const tags = getTagDefinitions();
const skipByType = new Map<string, string[]>();
const createByType = new Map<string, string[]>();
for (const t of tags) {
const effect = t.noteEffect ?? 'none';
if (effect === 'none') continue;
const label = TYPE_LABELS[t.type] ?? t.type;
const map = effect === 'skip' ? skipByType : createByType;
const list = map.get(label) ?? [];
list.push(t.tag.split('-').map(w => w[0].toUpperCase() + w.slice(1)).join(' '));
map.set(label, list);
}
const formatList = (map: Map<string, string[]>) =>
Array.from(map.entries()).map(([type, tags]) => `- **${type}:** ${tags.join(', ')}`).join('\n');
return [
`**SKIP if the email has ANY of these labels (skip labels override everything):**`,
formatList(skipByType),
``,
`**CREATE/UPDATE notes if the email has ANY of these labels (and no skip labels present):**`,
formatList(createByType),
``,
`**Logic:** If even one label falls in the "skip" list, skip the email — skip labels are hard filters that override create labels.`,
].join('\n');
}
export function renderTagSystemForNotes(): string {
const tags = getTagDefinitions().filter(t => t.applicability !== 'email');
return renderTagGroups(tags);
}
export function renderTagSystemForEmails(): string {
const tags = getTagDefinitions().filter(t => t.applicability !== 'notes');
return renderTagGroups(tags);
}

4
apps/x/packages/core/src/models/defaults.ts
View file

@ -128,8 +128,8 @@ async function getCategoryModel(
}
/**
* Model used by knowledge-graph agents (note_creation, labeling_agent, etc.)
* when they're the top-level of a run.
* Model used by knowledge-graph agents (note_creation, the email classifier,
* etc.) when they're the top-level of a run.
*/
export async function getKgModel(): Promise<ModelSelection> {
return getCategoryModel("knowledgeGraphModel", SIGNED_IN_KG_MODEL);

2
apps/x/packages/core/src/runtime/assembly/registry.test.ts
View file

@ -16,7 +16,6 @@ describe("agent registry", () => {
"background-task-agent",
"note_creation",
"note_curation",
"labeling_agent",
"note_tagging_agent",
"inline_task_agent",
"agent_notes_agent",
@ -41,7 +40,6 @@ describe("agent registry", () => {
for (const id of [
"note_creation",
"note_curation",
"labeling_agent",
"note_tagging_agent",
"inline_task_agent",
"agent_notes_agent",

2
apps/x/packages/core/src/runtime/assembly/registry.ts
View file

@ -6,7 +6,6 @@ import { buildBackgroundTaskAgent } from "../../background-tasks/agent.js";
import { buildLiveNoteAgent } from "../../knowledge/live-note/agent.js";
import { getRaw as getNoteCreationRaw } from "../../knowledge/note_creation.js";
import { getRaw as getNoteCurationRaw } from "../../knowledge/note_curation.js";
import { getRaw as getLabelingAgentRaw } from "../../knowledge/labeling_agent.js";
import { getRaw as getNoteTaggingAgentRaw } from "../../knowledge/note_tagging_agent.js";
import { getRaw as getInlineTaskAgentRaw } from "../../knowledge/inline_task_agent.js";
import { getRaw as getAgentNotesAgentRaw } from "../../knowledge/agent_notes_agent.js";
@ -60,7 +59,6 @@ const builtinAgents: Record<string, BuiltinAgentDefinition> = {
"background-task-agent": { build: buildBackgroundTaskAgent },
note_creation: promptFileAgent("note_creation", getNoteCreationRaw),
note_curation: promptFileAgent("note_curation", getNoteCurationRaw),
labeling_agent: promptFileAgent("labeling_agent", getLabelingAgentRaw),
note_tagging_agent: promptFileAgent("note_tagging_agent", getNoteTaggingAgentRaw),
inline_task_agent: promptFileAgent("inline_task_agent", getInlineTaskAgentRaw),
agent_notes_agent: promptFileAgent("agent_notes_agent", getAgentNotesAgentRaw),

4
apps/x/packages/shared/src/blocks.ts
View file

@ -145,6 +145,10 @@ export const GmailThreadSchema = EmailBlockSchema.extend({
threadUrl: z.string().url(),
unread: z.boolean().optional(),
importance: z.enum(['important', 'other']).optional(),
// What kind of email this is (correspondence, meeting, notification,
// newsletter, promotion, cold_outreach, receipt). Loose string so the
// classifier's taxonomy can evolve without a lockstep schema change.
category: z.string().optional(),
gmail_draft: z.string().optional(),
// Gmail-side draft id, present on entries returned by the Drafts list so the
// composer can update/delete that exact draft.

52
apps/x/packages/shared/src/ipc.ts
View file

@ -167,16 +167,21 @@ const ipcSchemas = {
res: z.object({
threads: z.array(GmailThreadSchema),
nextCursor: z.string().nullable(),
categoryCounts: z.record(z.string(), z.number()).optional(),
}),
},
'gmail:getEverythingElse': {
req: z.object({
cursor: z.string().optional(),
limit: z.number().int().min(1).max(100).optional(),
// Restrict to one category (filter pills). Whole-section categoryCounts
// are returned regardless, so the pills stay populated while filtered.
category: z.string().optional(),
}),
res: z.object({
threads: z.array(GmailThreadSchema),
nextCursor: z.string().nullable(),
categoryCounts: z.record(z.string(), z.number()).optional(),
}),
},
'gmail:triggerSync': {
@ -294,6 +299,53 @@ const ipcSchemas = {
error: z.string().optional(),
}),
},
// User explicitly picks a thread's category. Sticky on the thread
// (re-classification never overrides) and recorded as a correction the
// classifier learns from. Never affects the knowledge-graph verdict.
'gmail:setCategory': {
req: z.object({
threadId: z.string().min(1),
// Open string: valid ids come from the email label registry (built-ins
// plus user-defined labels), which the renderer fetches at runtime.
category: z.string().min(1).max(40),
}),
res: z.object({
ok: z.boolean(),
error: z.string().optional(),
}),
},
// The label registry: built-in categories plus labels the user defined in
// their agent instructions. Chips, filter pills, and the correction
// dropdown all render from this list.
'gmail:getEmailLabels': {
req: z.object({}),
res: z.object({
labels: z.array(z.object({
id: z.string(),
name: z.string(),
kind: z.enum(['builtin', 'custom']),
})),
}),
},
// Free-text standing instructions injected into every classification /
// draft call. Stored at config/email_instructions.md.
'gmail:getEmailInstructions': {
req: z.object({}),
res: z.object({ instructions: z.string() }),
},
'gmail:setEmailInstructions': {
req: z.object({ instructions: z.string().max(8000) }),
res: z.object({ ok: z.boolean(), error: z.string().optional() }),
},
// Archive every "Everything else" thread of one category in a single sweep.
'gmail:archiveCategory': {
req: z.object({ category: z.string().min(1) }),
res: z.object({
archived: z.number(),
failed: z.number(),
error: z.string().optional(),
}),
},
'gmail:archiveThread': {
req: z.object({ threadId: z.string().min(1) }),
res: z.object({ ok: z.boolean(), error: z.string().optional() }),