Skip to content

ARCHITECTURE.md

Latest commit

 

History

History
226 lines (176 loc) · 7.58 KB

ARCHITECTURE.md

File metadata and controls

226 lines (176 loc) · 7.58 KB
Raw

Architecture

Overview

gbrain-import is a three-stage pipeline: index → extract → materialize. Each stage is pure and resumable. Intermediate state lives in a SQLite workspace (~/.gbrain-import/<hash>/).

archive file
     │
     ▼
┌─────────────┐
│    index    │  parse archive into SQLite, no API calls
└──────┬──────┘
       │
       ▼
┌─────────────┐
│   extract   │  LLM reads batches, stores structured results
└──────┬──────┘
       │ (resolve pass for Gmail)
       ▼
┌──────────────┐
│  materialize │  quality gate → StructuredPage[]
└──────┬───────┘
       │
       ▼
┌──────────────┐
│    adapter   │  write pages to GBrain or Obsidian
└──────────────┘

Connector Interface

src/core/types.ts defines the contract every connector must implement:

interface Connector {
  name: string;
  detect(path: string): boolean;          // returns true if this path is this connector's format
  index(path, db, opts?): Promise<void>;  // parse archive → SQLite, no API calls
  extract(db): AsyncGenerator<ExtractBatch>; // yield batches for LLM extraction
  resolve(db): Promise<void>;             // optional dedup/merge pass before materialize
  materialize(db): StructuredPage[];      // quality gate → pages (pure, no side effects)
}

extract() is an async generator. Each ExtractBatch carries the batch ID, a fully self-contained LLM prompt, the thread/conversation IDs included, and a token estimate. The batch is marked in_progress in SQLite before being yielded, and completed or failed when storeBatchResults / markBatchFailed is called.

materialize() is a pure function — it reads from the database and returns StructuredPage[]. It never writes to the filesystem or calls any API.

Connectors:

  • src/connectors/gmail/ — .mbox and Google Takeout Mail/ directories
  • src/connectors/chatgpt/ — conversations.json or conversations-*.json directories

Output Adapter Interface

interface OutputAdapter {
  write(pages: StructuredPage[]): Promise<void>;
}

Adapters:

  • src/adapters/gbrain/ — writes via GBrain engine API (peer dep, optional)
  • src/adapters/obsidian/ — writes .md files to a vault directory

Both adapters respect sourcePrefix (set to the connector name by the CLI) to namespace pages: <vault>/gmail/people/alice-example.md vs <vault>/chatgpt/people/alice-example.md. This prevents collisions when importing from multiple sources into the same vault.

SQLite Workspace

Each import session gets a workspace directory at ~/.gbrain-import/<12-char-hash>/ where the hash is sha256(connectorName + ":" + absPath).slice(0, 12). Same input path always produces the same workspace, so re-runs resume from where they left off.

Each connector creates its own database file (gmail.db or chatgpt.db). The schema is initialized on first open and is idempotent (CREATE TABLE IF NOT EXISTS).

Batch tracking table (both connectors):

batches (batch_id, status, thread_ids, created_at, error)
status: pending | in_progress | completed | failed

Extraction Modes

Agent-driven (default)

CLI stdout: { type: "batch", batchId, estimatedTokens, threadCount, prompt }
                                │
                        agent reads prompt
                                │
                        calls MCP store_batch_results
                                │
                        storeBatchResults(db, batchId, result)

The CLI prints one JSON object per batch to stdout. Any agent with access to the MCP server can read the prompt, call the LLM, and store the result. Re-submitting a batch is safe — storeBatchResults deletes prior results before inserting new ones (DELETE-before-INSERT in a transaction).

--api mode

runApiLoop(db, connector, options)
  ├── collect all pending batches from connector.extract(db)
  ├── ConcurrencyPool (default limit: 5)
  └── for each batch: callClaude → storeBatchResults

ConcurrencyPool (src/core/api-loop.ts) is a simple promise queue. It processes up to N batches in parallel and drains before returning. After extraction, the CLI calls resolve() then materialize() then the adapter's write().

Startup Reset

On every extract() call, both connectors reset in_progress batches older than 30 minutes to failed. This handles killed processes without leaking stuck batches.

UPDATE batches SET status = 'failed', error = 'stranded: reset on startup'
 WHERE status = 'in_progress' AND created_at < '<cutoff>'

Quality Gate

Gmail only. materialize() skips entities that don't meet:

thread_count >= 3  AND  event_count >= 2

Constants in src/connectors/gmail/materialize.ts:

export const DEEP_MIN_THREADS = 3;
export const DEEP_MIN_EVENTS = 2;

Entities that don't pass are kept in SQLite for future use (lower thresholds, manual review, etc.) but are not written as pages.

ChatGPT deduplicates differently: normalize by entityType::name.toLowerCase(), keep the highest-confidence extraction, emit all results (no thread-count gate).

Markdown Contract

All pages use <!-- timeline --> as the sentinel between the main body and the timeline section. Plain --- is a markdown horizontal rule and must not be used as a section separator. This is tested in test/markdown-contract.test.ts.

MCP Server

src/mcp.ts is a minimal JSON-RPC 2.0 stdio server. It handles four methods: initialize, notifications/initialized, tools/list, tools/call.

The one exposed tool (store_batch_results) accepts { batchId: string, result: object } and routes to the correct connector's storeBatchResults based on --connector.

Start with:

gbrain-import --mcp --workspace ~/.gbrain-import/<hash> --connector gmail

Key Files

src/
  cli.ts                       entry point, flag parsing, mode dispatch
  mcp.ts                       MCP stdio server
  core/
    types.ts                   Connector, OutputAdapter, StructuredPage interfaces
    api-loop.ts                ConcurrencyPool, runApiLoop
    utils.ts                   slugify
  connectors/
    gmail/
      index.ts                 GmailConnector (implements Connector)
      index-mbox.ts            mbox parser → SQLite
      extract.ts               batch building, storeBatchResults, markBatchFailed
      resolve.ts               entity dedup/merge pass
      materialize.ts           quality gate → StructuredPage[]
      types.ts                 Gmail-specific SQLite row types
    chatgpt/
      index.ts                 ChatGPTConnector (implements Connector)
      index-chatgpt.ts         conversations.json parser → SQLite
      extract.ts               batch building, storeBatchResults
      materialize.ts           dedup by confidence → StructuredPage[]
  adapters/
    gbrain/index.ts            GBrainAdapter
    obsidian/index.ts          ObsidianAdapter
test/
  cli.test.ts                  CLI flag and exit code tests
  resumability.test.ts         storeBatchResults idempotency
  collision.test.ts            source-prefix isolation
  markdown-contract.test.ts    <!-- timeline --> sentinel contract
  adapters/gbrain.test.ts
  adapters/obsidian.test.ts
  connectors/gmail.detect.test.ts
  connectors/gmail.materialize.test.ts
  connectors/chatgpt.detect.test.ts
  connectors/chatgpt.materialize.test.ts