Skip to content

ARCHITECTURE.md

Latest commit

 

History

History
264 lines (208 loc) · 10.4 KB

ARCHITECTURE.md

File metadata and controls

264 lines (208 loc) · 10.4 KB
Raw

Architecture

ContextBridgeAI is a small, well-factored Python application with five concerns kept deliberately separate so any one of them can be swapped out:

src/context_bridge/
├── schema.py        # normalized PromptTranscript + Message Pydantic models
├── adapters/        # one file per IDE/CLI; all yield PromptTranscripts
│   ├── base.py
│   ├── _vscdb.py    # shared helpers for VSCode-derived state.vscdb stores
│   ├── rovodev.py     vscode.py    cursor.py     antigravity.py
│   ├── windsurf.py    kiro.py      claude_code.py codex.py
│   ├── cline.py       continue_dev.py zed.py    qodo.py    gemini.py
├── storage.py       # SQLite + FTS5 + KV memory + cooperative snapshots
├── search.py        # BM25 (via FTS5) + semantic (sbert / hashed-tfidf) + RRF
├── watcher.py       # background file-poll loop (3 s, configurable)
├── cli.py           # Typer entrypoint (index, list-apps, search, ui, …)
└── webui.py         # FastAPI app + single-file vanilla-JS frontend

Data flow

  IDE / CLI writes its private session file
                │
                ▼
  ┌──────────────────────────────────────┐
  │  SessionWatcher (3 s polling loop)   │
  │  • scans each adapter's root paths   │
  │  • diffs (size, mtime) per file      │
  │  • runs only the affected adapter(s) │
  └──────────────────────────────────────┘
                │
                ▼
  ┌──────────────────────────────────────┐
  │  Adapter.extract()                   │
  │  • reads on-disk format              │
  │  • emits PromptTranscript objects    │
  └──────────────────────────────────────┘
                │
                ▼
  ┌──────────────────────────────────────┐
  │  Storage.upsert(transcript)          │
  │  • writes JSON + flat text into      │
  │    SQLite (data/index.db)            │
  │  • FTS5 trigger keeps BM25 fresh     │
  │  • derives missing updated_at from   │
  │    raw source file mtime             │
  └──────────────────────────────────────┘
                │
                ├──► Web UI / CLI list & view sessions (always live)
                │
                ▼  (manual: user clicks "Re-index")
  ┌──────────────────────────────────────┐
  │  HybridSearch.build()                │
  │  • encodes every transcript          │
  │  • saves vectors to                  │
  │    data/semantic_index.pkl           │
  └──────────────────────────────────────┘
                │
                ▼
  ┌──────────────────────────────────────┐
  │  HybridSearch.search(q)              │
  │  • BM25 ranks via FTS5               │
  │  • semantic cosine ranks             │
  │  • RRF fusion (k = 60)               │
  │  • returns SearchHit with both ranks │
  └──────────────────────────────────────┘

The strict separation between "data refresh" (auto) and "search-index rebuild" (manual) is deliberate: encoding 2 700+ transcripts takes a few seconds, and we don't want that running while the user is typing.

The normalized schema

Every adapter — regardless of how horrible the source format is — emits the same PromptTranscript object:

class Message(BaseModel):
    role: Literal["system", "user", "assistant", "tool", "tool_call",
                  "developer", "unknown"]
    content: str                  # always a plain string
    timestamp: Optional[str] = None
    name: Optional[str] = None    # tool name etc.
    metadata: dict = {}

class PromptTranscript(BaseModel):
    session_id: str               # globally unique: f"{app}:{native-id}"
    app: str                      # rovodev | cursor | vscode | …
    title: Optional[str]
    workspace_path: Optional[str]
    created_at: Optional[str]
    updated_at: Optional[str]
    model: Optional[str]
    provider: Optional[str]
    messages: list[Message]
    raw_source_path: Optional[str]   # "<path>" or "<path>#<key>" for SQLite
    metadata: dict = {}

Exporters:

Method Output
as_text() Tagged plain-text transcript: <SYSTEM>…</SYSTEM>\n<USER>…</USER>\n…
as_markdown() Human-readable markdown with role headers and fenced code blocks
concatenated_text() Flat string used for BM25 / embedding indexing
.model_dump() Raw JSON (Pydantic)

Storage layer (storage.py)

A single SQLite database at data/index.db. Three tables + one virtual FTS:

Table Purpose
transcripts One row per session. Holds the full normalized JSON blob and a text_blob used for FTS.
transcripts_fts SQLite FTS5 virtual table mirroring (session_id, title, text_blob), kept in sync by triggers. Powers the BM25 side of search.
kv_memory Cross-session KV store (save_context / get_context cooperative API).
cooperative_snapshots Bag of agent-cooperative snapshots (one-shot prompt dumps).

Important details:

  • mtime fallback: Storage.upsert() automatically fills in updated_at from the raw source file's mtime when an adapter didn't supply it. Without this, half the apps' sessions would sort to the bottom with NULL timestamps.
  • Read-only SQLite for IDE storages: every adapter that touches a state.vscdb opens it with file:<path>?mode=ro&immutable=1 so we never hold a write lock while the IDE itself might be running.

Search layer (search.py)

BM25 (FTS5)

We rely on SQLite's built-in FTS5 extension and its bm25() ranking function. No extra dependency. Queries are sanitized to OR-of-terms ("foo" OR "bar") to avoid syntax errors on user input.

Semantic

Two interchangeable backends:

  1. sentence-transformers (preferred, optional dep): all-MiniLM-L6-v2 — fast, 384-dim, normalized embeddings.
  2. Fallback: HashingVectorizer + TfidfTransformer from scikit-learn. Zero extra deps. Quality is keyword-overlap-y but works.

Vectors are dense float32 arrays persisted to data/semantic_index.pkl.

Reciprocal Rank Fusion (RRF)

score(d) = Σ_query  1 / (k + rank_query(d))   with k = 60

We get a BM25 ranking and a semantic ranking, then fuse them. Each SearchHit exposes:

  • score (float, higher = better)
  • bm25_rank (1-based, None if not in BM25 top-k)
  • sem_rank (1-based, None if not in semantic top-k)
  • snippet (window around first matched term)

Watcher (watcher.py)

A daemon thread that polls every adapter's root paths every 3 seconds (env CONTEXT_BRIDGE_WATCH_INTERVAL). For each adapter:

  1. Build the new (path → (size, mtime)) map.
  2. Compare to the previous snapshot.
  3. If anything changed, run only that adapter's extract() and upsert.

Notably the watcher does not rebuild semantic vectors — that's a manual user action. FTS5 stays current automatically via SQLite triggers.

Disable entirely with CONTEXT_BRIDGE_NO_WATCH=1.

Web UI (webui.py)

A single FastAPI module with embedded HTML/CSS/JS (no build step, no React, no node_modules). Roughly 700 lines of vanilla JS.

Endpoint Purpose
GET / The UI HTML
GET /api/status Watcher heartbeat + search-index age & staleness
GET /api/apps All detected apps with counts + last-active timestamps
GET /api/sessions?app&limit Recent sessions, most-recent first
GET /api/session?id Full normalized JSON for one session
GET /api/session/text?id as_text() output (plain text)
GET /api/session/markdown?id as_markdown() output
GET /api/search?q&limit Hybrid BM25+semantic RRF results
POST /api/reindex Manual: re-extract + rebuild semantic vectors
GET /api/icon/<app> Per-app icon (PNG from .app bundle or bundled SVG)
GET /api/snapshots /api/memory Cooperative KV/snapshot tools

Frontend polls /api/status every 2 s to detect data refreshes and to update the "search · 5m old" staleness pill on the Re-index button.

Adapters (adapters/*.py)

All adapters inherit from BaseAdapter:

class BaseAdapter(ABC):
    app_name: str = "unknown"

    @abstractmethod
    def root_paths(self) -> list[Path]:
        """Candidate roots; existence check is `any(p.exists() for p in self.root_paths())`."""

    @abstractmethod
    def extract(self) -> Iterable[PromptTranscript]:
        """Yield normalized transcripts for every session found."""

ALL_ADAPTERS in adapters/__init__.py registers each one, and discover_all() is the convenience iterator that yields from every available adapter.

For per-adapter implementation details (storage schemas, gotchas, known limitations) see ADAPTERS.md.

Why no MCP server?

An earlier prototype shipped an MCP server, but in practice it added very little: MCP exposes ContextBridge to agents, but the value of this project is in after-the-fact browsing & search of prompts that have already happened. Since MCP has no spec primitive for push from host → server on every conversation turn (yet), an MCP server can't observe what the agent is doing in real time without the agent's cooperation. We removed the MCP layer and the project is purely a file-scraper + UI today. (If/when MCP gains a turn-subscription primitive, we'll add it back.)

Why polling instead of watchdog?

Polling at 3 s costs ~5 ms of CPU per cycle (sum of os.stat calls across ~3 000 files). The watchdog package adds a heavyweight inotify backend that doesn't actually fire reliably for SQLite databases (which mutate via Write-Ahead-Logging in a separate *-wal file). Polling (size, mtime) is simpler, dependency-free, and correct.