> ## Documentation Index
> Fetch the complete documentation index at: https://dimensionalos.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Architecture

Observation storage and streaming layer for DimOS. Pull-based, lazy, composable.

## Architecture

```
             Live Sensor Data
                    ↓
Store → Stream → [filters / transforms / terminals] → Stream  → [filters / transforms / terminals] → Stream → Live hooks
                    ↓                                              ↓                                             ↓
    Backend (ObservationStore + BlobStore + VectorStore + Notifier)     Backend                                      In Memory
```

**Store** owns a storage location (file, in-memory) and directly manages named streams. **Stream** is the query/iteration surface - lazy until a terminal is called. **Backend** is a concrete composite that orchestrates ObservationStore + BlobStore + VectorStore + Notifier for each stream.

Supporting Systems:

* BlobStore - separates large payloads from metadata. FileBlobStore (files on disk) and SqliteBlobStore (blob table per stream). Supports lazy loading.
* Codecs - codec\_for() auto-selects: JpegCodec for images (TurboJPEG, \~10-20x compression), LcmCodec for DimOS messages, PickleCodec fallback.
* Transformers - Transformer\[T,R] ABC wrapping iterator-to-iterator. EmbedImages/EmbedText enrich observations with embeddings. QualityWindow keeps best per time window.
* Backpressure Buffers - KeepLast, Bounded, DropNew, Unbounded - bridge push/pull for live mode.

## Modules

| Module         | What                                                                 |
| -------------- | -------------------------------------------------------------------- |
| `stream.py`    | Stream node - filters, transforms, terminals                         |
| `backend.py`   | Concrete Backend composite (ObservationStore + Blob + Vector + Live) |
| `store.py`     | Store, StoreConfig                                                   |
| `transform.py` | Transformer ABC, FnTransformer, FnIterTransformer, QualityWindow     |
| `buffer.py`    | Backpressure buffers for live mode (KeepLast, Bounded, Unbounded)    |
| `embed.py`     | EmbedImages / EmbedText transformers                                 |

## Subpackages

| Package             | What                                                   | Docs                                   |
| ------------------- | ------------------------------------------------------ | -------------------------------------- |
| `type/`             | Observation, EmbeddedObservation, Filter/StreamQuery   |                                        |
| `store/`            | Store ABC + implementations (MemoryStore, SqliteStore) | `dimos/memory2/store/README.md`        |
| `notifier/`         | Notifier ABC + SubjectNotifier                         |                                        |
| `blobstore/`        | BlobStore ABC + implementations (file, sqlite)         | `dimos/memory2/blobstore/blobstore.md` |
| `codecs/`           | Encode/decode for storage (pickle, JPEG, LCM)          | `dimos/memory2/codecs/README.md`       |
| `vectorstore/`      | VectorStore ABC + implementations (memory, sqlite)     |                                        |
| `observationstore/` | ObservationStore Protocol + implementations            |                                        |

## Docs

| Doc                                                         | What                                                                               |
| ----------------------------------------------------------- | ---------------------------------------------------------------------------------- |
| [streaming.mdx](/capabilities/memory/reference/streaming)   | Lazy vs materializing vs terminal - evaluation model, live safety                  |
| [embeddings.mdx](/capabilities/memory/reference/embeddings) | Embedding layer design - EmbeddedObservation, vector search, EmbedImages/EmbedText |
| `dimos/memory2/blobstore/blobstore.md`                      | BlobStore architecture - separate payload storage from metadata                    |

## Query execution

`StreamQuery` holds the full query spec (filters, text search, vector search, ordering, offset/limit). It also provides `apply(iterator)` - a Python-side execution path that runs all operations as in-memory predicates, brute-force cosine, and list sorts.

This is the **default fallback**. ObservationStore implementations are free to push down operations using store-specific strategies instead:

| Operation      | Python fallback (`StreamQuery.apply`) | Store push-down (example) |
| -------------- | ------------------------------------- | ------------------------- |
| Filters        | `filter.matches()` predicates         | SQL WHERE clauses         |
| Text search    | Case-insensitive substring            | FTS5 full-text index      |
| Vector search  | Brute-force cosine similarity         | vec0 / FAISS ANN index    |
| Ordering       | `sorted()` materialization            | SQL ORDER BY              |
| Offset / limit | `islice()`                            | SQL OFFSET / LIMIT        |

`ListObservationStore` delegates entirely to `StreamQuery.apply()`. `SqliteObservationStore` translates the query into SQL and only falls back to Python for operations it can't express natively.

Transform-sourced streams (post `.transform()`) always use `StreamQuery.apply()` since there's no index to push down to.

## Quick start

```python theme={null}
from dimos.memory2 import MemoryStore

store = MemoryStore()
images = store.stream("images")

# Write
images.append(frame, ts=time.time(), pose=(x, y, z), tags={"camera": "front"})

# Query
recent = images.after(t).limit(10).to_list()
nearest = images.near(pose, radius=2.0).to_list()
latest = images.last()

# Transform (class or bare generator function)
edges = images.transform(Canny()).save(store.stream("edges"))
edges.drain()  # actually run the pipeline; .save() is lazy

def running_avg(upstream):
    total, n = 0.0, 0
    for obs in upstream:
        total += obs.data; n += 1
        yield obs.derive(data=total / n)
avgs = stream.transform(running_avg).to_list()

# Live
for obs in images.live().transform(process):
    handle(obs)

# Embed + search
images.transform(EmbedImages(clip)).save(store.stream("embedded")).drain()
results = store.stream("embedded").search(query_vec, k=5).to_list()
```

## Implementations

| ObservationStore         | Status   | Storage                                |
| ------------------------ | -------- | -------------------------------------- |
| `ListObservationStore`   | Complete | In-memory (lists + brute-force search) |
| `SqliteObservationStore` | Complete | SQLite (WAL, FTS5, vec0)               |