Feat/memory2

dimos/core/resource.py

@@ -12,10 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from abc import ABC, abstractmethod
+from __future__ import annotations
 
+from abc import abstractmethod
+from typing import TYPE_CHECKING, Self
 
-class Resource(ABC):
+if TYPE_CHECKING:
+    from types import TracebackType
+
+from reactivex.abc import DisposableBase
+from reactivex.disposable import CompositeDisposable
+
+
+class Resource(DisposableBase):
     @abstractmethod
     def start(self) -> None: ...
 
@@ -43,3 +52,35 @@ def dispose(self) -> None:
 
         """
         self.stop()
+
+    def __enter__(self) -> Self:
+        self.start()
+        return self
+
+    def __exit__(
+        self,
+        exctype: type[BaseException] | None,
+        excinst: BaseException | None,
+        exctb: TracebackType | None,
+    ) -> None:
+        self.stop()
+
+
+class CompositeResource(Resource):
+    """Resource that owns child disposables, disposed on stop()."""
+
+    _disposables: CompositeDisposable
+
+    def __init__(self) -> None:
+        self._disposables = CompositeDisposable()
+
+    def register_disposables(self, *disposables: DisposableBase) -> None:
+        """Register child disposables to be disposed when this resource stops."""
+        for d in disposables:
+            self._disposables.add(d)
+
+    def start(self) -> None:
+        pass
+
+    def stop(self) -> None:
+        self._disposables.dispose()

dimos/core/worker.py

@@ -291,7 +291,19 @@ def _suppress_console_output() -> None:
         ]
 
 
-def _worker_entrypoint(conn: Connection, worker_id: int) -> None:
+def _worker_entrypoint(
+    conn: Connection,
+    worker_id: int,
+) -> None:
+    # Limit OpenCV internal threads to avoid idle thread contention.
+    # Modules that need parallel cv2 ops can call cv2.setNumThreads() in start().
+    try:
+        import cv2
+
+        cv2.setNumThreads(2)
+    except ImportError:
+        pass
+
     instances: dict[int, Any] = {}
 
     try:

dimos/mapping/occupancy/gradient.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+
 import numpy as np
 from scipy import ndimage  # type: ignore[import-untyped]
 
@@ -50,7 +52,9 @@ def gradient(
 
     # Compute distance transform (distance to nearest obstacle in cells)
     # Unknown cells are treated as if they don't exist for distance calculation
-    distance_cells = ndimage.distance_transform_edt(1 - obstacle_map)
+    distance_cells: np.ndarray[Any, np.dtype[np.float64]] = ndimage.distance_transform_edt(
+        1 - obstacle_map
+    )  # type: ignore[assignment]
 
     # Convert to meters and clip to max distance
     distance_meters = np.clip(distance_cells * occupancy_grid.resolution, 0, max_distance)

dimos/memory2/__init__.py

@@ -0,0 +1,70 @@
+from dimos.memory2.buffer import (
+    BackpressureBuffer,
+    Bounded,
+    ClosedError,
+    DropNew,
+    KeepLast,
+    Unbounded,
+)
+from dimos.memory2.embed import EmbedImages, EmbedText
+from dimos.memory2.impl.memory import ListBackend, MemorySession, MemoryStore
+from dimos.memory2.impl.sqlite import SqliteBackend, SqliteSession, SqliteStore, SqliteStoreConfig
+from dimos.memory2.livechannel import SubjectChannel
+from dimos.memory2.store import Session, SessionConfig, Store, StoreConfig, StreamNamespace
+from dimos.memory2.stream import Stream
+from dimos.memory2.transform import FnTransformer, QualityWindow, Transformer
+from dimos.memory2.type.backend import Backend, LiveChannel, VectorStore
+from dimos.memory2.type.filter import (
+    AfterFilter,
+    AtFilter,
+    BeforeFilter,
+    Filter,
+    NearFilter,
+    PredicateFilter,
+    StreamQuery,
+    TagsFilter,
+    TimeRangeFilter,
+)
+from dimos.memory2.type.observation import EmbeddedObservation, Observation
+
+__all__ = [
+    "AfterFilter",
+    "AtFilter",
+    "Backend",
+    "BackpressureBuffer",
+    "BeforeFilter",
+    "Bounded",
+    "ClosedError",
+    "DropNew",
+    "EmbedImages",
+    "EmbedText",
+    "EmbeddedObservation",
+    "Filter",
+    "FnTransformer",
+    "KeepLast",
+    "ListBackend",
+    "LiveChannel",
+    "MemorySession",
+    "MemoryStore",
+    "NearFilter",
+    "Observation",
+    "PredicateFilter",
+    "QualityWindow",
+    "Session",
+    "SessionConfig",
+    "SqliteBackend",
+    "SqliteSession",
+    "SqliteStore",
+    "SqliteStoreConfig",
+    "Store",
+    "StoreConfig",
+    "Stream",
+    "StreamNamespace",
+    "StreamQuery",
+    "SubjectChannel",
+    "TagsFilter",
+    "TimeRangeFilter",
+    "Transformer",
+    "Unbounded",
+    "VectorStore",
+]

dimos/memory2/architecture.md

@@ -0,0 +1,115 @@
+# memory
+
+Observation storage and streaming layer for DimOS. Pull-based, lazy, composable.
+
+## Architecture
+
+```
+             Live Sensor Data
+                    ↓
+Store → Session → Stream → [filters / transforms / terminals] → Stream  → [filters / transforms / terminals] → Stream → Live hooks
+                    ↓                                              ↓                                             ↓
+                 Backend (ListBackend, SqliteBackend)           Backend                                      In Memory
+```
+
+**Store** owns a storage location (file, in-memory). **Session** manages named streams over a shared connection. **Stream** is the query/iteration surface — lazy until a terminal is called.
+
+
+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`   | Backend protocol, LiveChannel / VectorStore / BlobStore ABCs     |
+| `filter.py`    | StreamQuery dataclass, filter types, Python query execution       |
+| `transform.py` | Transformer ABC, FnTransformer, FnIterTransformer, QualityWindow  |
+| `buffer.py`    | Backpressure buffers for live mode (KeepLast, Bounded, Unbounded) |
+| `store.py`     | Store / Session (Configurable), StoreConfig / SessionConfig      |
+| `type.py`      | Observation, EmbeddedObservation dataclasses                      |
+| `embed.py`     | EmbedImages / EmbedText transformers                              |
+
+## Subpackages
+
+| Package      | What                                                 | Docs                                             |
+|--------------|------------------------------------------------------|--------------------------------------------------|
+| `impl/`         | Backend implementations (ListBackend, SqliteBackend) | [impl/README.md](impl/README.md)                 |
+| `livechannel/`  | Live notification channels (SubjectChannel)          |                                                  |
+| `blobstore/`    | Pluggable blob storage (file, sqlite)                | [blobstore/blobstore.md](blobstore/blobstore.md) |
+| `codecs/`       | Encode/decode for storage (pickle, JPEG, LCM)        | [codecs/README.md](codecs/README.md)             |
+
+## Docs
+
+| Doc | What |
+|-----|------|
+| [streaming.md](streaming.md) | Lazy vs materializing vs terminal — evaluation model, live safety |
+| [embeddings.md](embeddings.md) | Embedding layer design — EmbeddedObservation, vector search, EmbedImages/EmbedText |
+| [blobstore/blobstore.md](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**. Backends 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               |
+
+`ListBackend` delegates entirely to `StreamQuery.apply()`. `SqliteBackend` 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 backend to push down to.
+
+## Quick start
+
+```python
+from dimos.memory2 import MemoryStore
+
+store = MemoryStore()
+with store.session() as session:
+    images = session.stream("images")
+
+    # Write
+    images.append(frame, ts=time.time(), pose=(x, y, z), tags={"camera": "front"})
+
+    # Query
+    recent = images.after(t).limit(10).fetch()
+    nearest = images.near(pose, radius=2.0).fetch()
+    latest = images.last()
+
+    # Transform (class or bare generator function)
+    edges = images.transform(Canny()).save(session.stream("edges"))
+
+    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).fetch()
+
+    # Live
+    for obs in images.live().transform(process):
+        handle(obs)
+
+    # Embed + search
+    images.transform(EmbedImages(clip)).save(session.stream("embedded"))
+    results = session.stream("embedded").search(query_vec, k=5).fetch()
+```
+
+## Implementations
+
+| Backend         | Status   | Storage                                |
+|-----------------|----------|----------------------------------------|
+| `ListBackend`   | Complete | In-memory (lists + brute-force search) |
+| `SqliteBackend` | Complete | SQLite (WAL, FTS5, vec0)               |

dimos/memory2/blobstore/__init__.py

@@ -0,0 +1,19 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from dimos.memory2.blobstore.file import FileBlobStore
+from dimos.memory2.blobstore.sqlite import SqliteBlobStore
+from dimos.memory2.type.backend import BlobStore
+
+__all__ = ["BlobStore", "FileBlobStore", "SqliteBlobStore"]

dimos/memory2/blobstore/blobstore.md

@@ -0,0 +1,86 @@
+# blobstore/
+
+Separates payload blob storage from metadata indexing. Observation payloads vary hugely in size — a `Vector3` is 24 bytes, a camera frame is megabytes. Storing everything inline penalizes metadata queries. BlobStore lets large payloads live elsewhere.
+
+## ABC (`backend.py`)
+
+```python
+class BlobStore(Resource, ABC):
+    def put(self, stream_name: str, key: int, data: bytes) -> None: ...
+    def get(self, stream_name: str, key: int) -> bytes: ...    # raises KeyError if missing
+    def delete(self, stream_name: str, key: int) -> None: ...  # silent if missing
+```
+
+- `stream_name` — stream name (used to organize storage: directories, tables)
+- `key` — observation id
+- `data` — encoded payload bytes (codec handles serialization, blob store handles persistence)
+- Extends `Resource` (start/stop) but does NOT own its dependencies' lifecycle
+
+## Implementations
+
+### `file.py` — FileBlobStore
+
+Stores blobs as files on disk, one directory per stream.
+
+```
+{root}/{stream}/{key}.bin
+```
+
+`__init__(root: str | os.PathLike[str])` — `start()` creates the root directory.
+
+### `sqlite.py` — SqliteBlobStore
+
+Stores blobs in a separate SQLite table per stream.
+
+```sql
+CREATE TABLE "{stream}_blob" (id INTEGER PRIMARY KEY, data BLOB NOT NULL)
+```
+
+`__init__(conn: sqlite3.Connection)` — does NOT own the connection.
+
+**Internal use** (same db as metadata): `SqliteStore.session()` creates one connection, passes it to both the metadata backend and the blob store.
+
+**External use** (separate db): user creates a separate connection and passes it. User manages that connection's lifecycle.
+
+**JOIN optimization** (future): when `lazy=False` and the blob store shares the same connection as the metadata backend, `SqliteBackend` can optimize with a JOIN instead of separate queries:
+
+```sql
+SELECT m.id, m.ts, m.pose, m.tags, b.data
+FROM "images" m JOIN "images_blob" b ON m.id = b.id
+WHERE m.ts > ?
+```
+
+## Lazy loading
+
+`lazy` is a stream-level flag, orthogonal to blob store choice. It controls WHEN data is loaded:
+
+- `lazy=False` → backend loads payload during iteration (eager)
+- `lazy=True` → backend sets `Observation._loader`, payload loaded on `.data` access
+
+| lazy | blob store | loading strategy |
+|------|-----------|-----------------|
+| False | SqliteBlobStore (same conn) | JOIN — one round trip |
+| False | any other | iterate meta, `blob_store.get()` per row |
+| True | any | iterate meta only, `_loader = lambda: codec.decode(blob_store.get(...))` |
+
+## Usage
+
+```python
+# Per-stream blob store choice
+with store.session() as session:
+    poses = session.stream("poses", PoseStamped)                           # default, eager
+    images = session.stream("images", Image, lazy=True)                    # default, lazy
+    images = session.stream("images", Image, blob_store=file_blobs)        # override
+```
+
+## Files
+
+```
+backend.py            BlobStore ABC (alongside Backend, LiveBackend)
+blobstore/
+  blobstore.md        this file
+  __init__.py          re-exports BlobStore, FileBlobStore, SqliteBlobStore
+  file.py             FileBlobStore
+  sqlite.py           SqliteBlobStore
+  test_blobstore.py   grid tests across implementations
+```