Memory questions

[leshy]

1. "where was I when this log line was added?"

• pose lookup from a timestamp

2. "how long have I been observing the red socks currently in view?"

• how many times did I see them before?
• temporal duration tracking + observation frequency

3. "how many people did I see during last week?"

• assume we are generating a facial recognition db — is this matching a face detection stream, then embeddings? then we are searching over that stream?

4. "where did you see red socks during last week?"

• we query for red socks embedding similarity, then feed this data into a VLM that further filters for socks
• is this data output into some table? is it like an ObservationSet again?
• then we can create a map (costmap) of red socks?

5. "did anyone ever open this door? at what times did I see this door open? who opened it?"

• event detection + temporal querying of state changes

6. "I have a transcription log (STT) and voice embeddings, how do I figure out who is saying what?"

• cross-stream correlation: audio → identity

7. "I have parallel voice and facial recognition streams, how do I correlate voice to people?"

• I don't see all people speaking at all times
• multi-modal fusion with incomplete overlap

8. "what's different in this room compared to yesterday?"

• comparing scene snapshots across time, diffing object sets
• requires baseline modeling / temporal comparison

9. "show me everywhere the cat went today"

• continuous spatial tracking over time, not point queries
• dense pose-stream retrieval + path aggregation

10. "what happened in the 30 seconds before the vase fell?"

• event-anchored temporal window across all streams
• multi-stream temporal slicing relative to a detected event

11. "when was the last time I did NOT see the cat in the apartment?"

• finding gaps in an observation stream

12. "what time does the mailman usually come?"

• aggregation across days, extracting temporal regularity from sparse events
• cross-session pattern extraction

13. "what did robot-2 observe in the warehouse that I missed?"

• cross-agent memory diff
• session/robot-scoped queries and set difference across streams

14. "how far did I travel while carrying an object?"

• only accumulate distance when a parallel detection stream has a positive signal, or between log entries?
• cross-stream conditional joins

[leshy]

Memory2 Implementation Plan

Context

Check questions.md

File Structure

dimos/memory2/
    __init__.py              # public exports (re-exports from API + default backend)
    types.py                 # ObservationRef, ObservationRow, Lineage, StreamInfo
    store.py                 # Store ABC (Resource lifecycle)
    session.py               # Session ABC (stream factory)
    stream.py                # StreamBase, BlobStream, EmbeddingStream, TextStream ABCs
    query.py                 # Query ABC (filter/search/rank/limit → fetch/fetch_set)
    observation_set.py       # ObservationSet ABC

    impl/
        sqlite/
            __init__.py      # exports SqliteStore
            store.py         # SqliteStore (connection, WAL, extension loading)
            session.py       # SqliteSession (stream factory, _streams registry)
            stream.py        # SqliteBlobStream, SqliteEmbeddingStream, SqliteTextStream
            query.py         # SqliteQuery (SQL generation, execution)
            observation_set.py # SqliteObservationSet (predicate/ref-table backing)
            _sql.py          # SQL helpers, identifier validation, schema DDL

    test_memory2.py          # tests (against SqliteStore)

API Layer (dimos/memory2/)

types.py — Data classes

@dataclass(frozen=True)
class ObservationRef:
    stream: str
    rowid: int

@dataclass
class ObservationRow:
    ref: ObservationRef
    ts: float | None = None
    pose: PoseLike | None = None
    scores: dict[str, float] = field(default_factory=dict)  # query-time only (from rank/search), not stored
    tags: dict[str, Any] = field(default_factory=dict)

@dataclass
class Lineage:
    parent_stream: str | None = None   # from _streams registry (stream-level)
    parent_rowid: int | None = None    # per-row: which row in parent stream

@dataclass
class StreamInfo:
    name: str
    payload_type: type
    parent_stream: str | None     # lineage: all rows derive from this stream
    count: int

Poses use DimOS's existing PoseLike type alias (Pose | PoseStamped | Point | PointStamped). No custom Pose type.

store.py — Store ABC

class Store(Resource, ABC):
    def session(self) -> Session: ...
    def close(self) -> None: ...
    def start(self) -> None: pass
    def stop(self) -> None: self.close()

session.py — Session ABC

PoseProvider = Callable[[], PoseLike | None]

class Session(ABC):
    def stream(self, name: str, payload_type: type, *,
               retention: str = "run",
               pose_provider: PoseProvider | None = None) -> BlobStream: ...
    def embedding_stream(self, name: str, payload_type: type, *,
                         dim: int, retention: str = "run",
                         parent: StreamBase | None = None,
                         pose_provider: PoseProvider | None = None) -> EmbeddingStream: ...
    def text_stream(self, name: str, payload_type: type, *,
                    tokenizer: str = "unicode61",
                    retention: str = "run",
                    parent: StreamBase | None = None,
                    pose_provider: PoseProvider | None = None) -> TextStream: ...
    def list_streams(self) -> list[StreamInfo]: ...
    def close(self) -> None: ...
    def __enter__ / __exit__

stream.py — Stream hierarchy ABCs

class StreamBase(ABC, Generic[T]):
    """Abstract base. No text/vector indexes."""
    pose_provider: PoseProvider | None = None  # auto-fills pose on append if set

    # Write
    def append(self, payload: T, *,
               ts: float | None = None,      # defaults to time.time()
               pose: PoseLike | None = None,  # explicit pose overrides provider
               tags: dict[str, Any] | None = None,
               parent_rowid: int | None = None,
               ) -> ObservationRef: ...

    # Read
    def query(self) -> Query[T]: ...
    def load(self, ref: ObservationRef) -> T: ...
    def load_many(self, refs: list[ObservationRef], *, batch_size=32) -> list[T]: ...
    def iter_meta(self, *, page_size=128) -> Iterator[list[ObservationRow]]: ...
    def count(self) -> int: ...

    # Introspection
    def meta(self, ref: ObservationRef) -> ObservationRow: ...

class BlobStream(StreamBase[T]):
    """Stream for arbitrary serializable payloads. No special indexes."""

class EmbeddingStream(StreamBase[T]):
    """Stream with vector index. No payload table — the vector IS the data."""
    def vector(self, ref: ObservationRef) -> list[float] | None: ...
    # load() not supported — use vector() instead

class TextStream(StreamBase[T]):
    """Stream with FTS index."""

query.py — Query ABC

class Query(ABC, Generic[T]):
    # Hard filters
    def time_range(self, t1: float, t2: float) -> Query[T]: ...
    def before(self, t: float) -> Query[T]: ...
    def after(self, t: float) -> Query[T]: ...
    
    def filter_tags(self, **tags: Any) -> Query[T]: ...
    def filter_refs(self, refs: list[ObservationRef]) -> Query[T]: ...
    def at(self, t: float, *, tolerance: float = 1.0) -> Query[T]: ...

    # Candidate generation (raise TypeError if stream lacks the required index)
    def search_text(self, text: str, *, candidate_k: int | None = None) -> Query[T]: ...
    def search_embedding(self, vector: list[float], *, candidate_k: int) -> Query[T]: ...

    # Ranking + ordering + limit
    def rank(self, **weights: float) -> Query[T]: ...
    def order_by(self, field: str, *, desc: bool = False) -> Query[T]: ...
    def limit(self, k: int) -> Query[T]: ...

    # Terminals
    def fetch(self) -> list[ObservationRow]: ...
    def fetch_set(self) -> ObservationSet[T]: ...
    def count(self) -> int: ...
    def one(self) -> ObservationRow: ...
    def last(self) -> ObservationRow: ...

TODO: terminals that generate spatial or temporal summaries (maybe as numpy arrays).

observation_set.py — ObservationSet ABC

class ObservationSet(ABC, Generic[T]):
    # Re-query
    def query(self) -> Query[T]: ...

    # Read
    def load(self, ref: ObservationRef) -> T: ...
    def load_many(self, refs, *, batch_size=32) -> list[T]: ...
    def refs(self, *, limit=None) -> list[ObservationRef]: ...
    def rows(self, *, limit=None) -> list[ObservationRow]: ...
    def one(self) -> ObservationRow: ...
    def fetch_page(self, *, limit=128, offset=0) -> list[ObservationRow]: ...
    def count(self) -> int: ...
    def lineage(self) -> Lineage: ...

    # Cross-stream
    def project_to(self, stream: StreamBase) -> ObservationSet: ...

    # Cleanup
    def close(self) -> None: ...
    def __enter__(self) -> Self: ...
    def __exit__(self, *exc) -> None: ...
    def __del__(self) -> None: ...  # best-effort fallback

---

SQLite Implementation (dimos/memory2/impl/sqlite/)

store.py — SqliteStore

• Stores file path, creates parent dirs on connect
• _connect(): sqlite3.connect(), WAL mode, loads sqlite-vec (optional), loads FTS5
• Tracks sessions via WeakSet for cleanup
• :memory: uses file::memory:?cache=shared URI
• Thread safety: each session = one connection, no check_same_thread=False

session.py — SqliteSession

• Holds one sqlite3.Connection
• stream() / embedding_stream() / text_stream(): creates tables if needed, caches stream instances
• Registers stream metadata in a _streams registry table:

CREATE TABLE _streams (
    rowid INTEGER PRIMARY KEY,
    name TEXT UNIQUE NOT NULL,
    type TEXT NOT NULL,           -- 'blob', 'embedding', 'text'
    payload_type TEXT NOT NULL,
    parent_stream_id INTEGER,     -- FK to _streams.rowid (lineage)
    retention TEXT DEFAULT 'run'
);

stream.py — SqliteBlobStream, SqliteEmbeddingStream, SqliteTextStream

append() inserts a metadata row (SQLite auto-assigns rowid), serializes payload into _payload, and inserts an R*Tree entry if pose is provided. EmbeddingStream.append() inserts into _vec only (no _payload). TextStream.append() inserts into both _payload (as TEXT) and _fts. Returns ObservationRef(stream, rowid).

query.py — SqliteQuery

• Accumulates filter predicates, search ops, rank spec, ordering, limit
• at(t, tolerance) → sugar for filter_time(t - tol, t + tol) + ORDER BY ABS(ts - t) LIMIT 1
• order_by(field, desc) → appends ORDER BY clause; valid fields: ts
• fetch(): generates SQL, executes, returns rows
• fetch_set(): creates ObservationSet (predicate-backed or ref-table-backed)
• search_embedding → sqlite-vec MATCH, writes top-k to temp table → ref-table-backed
• search_text → FTS5 MATCH
• filter_near → R*Tree range query
• rank → computes composite score from available score columns

observation_set.py — SqliteObservationSet

Internal backing:

@dataclass
class PredicateBacking:
    """Lazy: expressible as SQL WHERE over source stream."""
    source_name: str
    query_repr: str  # serialized query filters for replay

@dataclass
class RefTableBacking:
    """Materialized: temp table of refs + scores."""
    table_name: str  # SQLite temp table
    source_streams: list[str]
    ordered: bool = False

• .query() on predicate-backed → adds more predicates
• .query() on ref-table-backed → filters within that temp table
• project_to() → joins backing refs via lineage parent_rowid to target stream
• close() drops the temp table for ref-table-backed sets; no-op for predicate-backed
• Supports context manager (with) for deterministic cleanup; __del__ as fallback
• SQLite connection close is the final safety net for any leaked temp tables

_sql.py — SQL helpers

def validate_identifier(name: str) -> str: ...  # regex check, length limit

Pose extraction: _extract_pose(p: PoseLike) -> tuple[float, ...] pulls (x, y, z, qx, qy, qz, qw). _reconstruct_pose(row) -> Pose rebuilds from stored floats.

Payload serialization: lcm_encode(payload) / lcm_decode(blob, payload_type). Non-LCM types rejected at append() with TypeError.

---

Schema (per stream)

{name}_meta — metadata for all stream types:

CREATE TABLE {name}_meta (
    rowid INTEGER PRIMARY KEY,
    ts REAL,
    pose_x REAL, pose_y REAL, pose_z REAL,
    pose_qx REAL, pose_qy REAL, pose_qz REAL, pose_qw REAL,
    tags TEXT,                     -- JSON
    parent_rowid INTEGER           -- lineage: rowid in parent stream
);
CREATE INDEX idx_{name}_meta_ts ON {name}_meta(ts);

{name}_payload — BlobStream and TextStream only (not EmbeddingStream):

CREATE TABLE {name}_payload (
    rowid INTEGER PRIMARY KEY,    -- matches _meta.rowid
    data BLOB NOT NULL            -- TextStream stores TEXT here instead of BLOB
);

{name}_rtree — all stream types (rows with pose only):

CREATE VIRTUAL TABLE {name}_rtree USING rtree(
    rowid,
    min_t, max_t,                 -- both set to ts
    min_x, max_x, min_y, max_y, min_z, max_z  -- both set to pose_xyz
);

{name}_fts — TextStream only:

CREATE VIRTUAL TABLE {name}_fts USING fts5(content);

{name}_vec — EmbeddingStream only:

CREATE VIRTUAL TABLE {name}_vec USING vec0(embedding float[{dim}]);

All virtual table rowids match _meta.rowid directly.

---

Phase 3: Later (not in first PR)

• derive() with Transform protocol
• CompositeBacking (union/intersection/difference)
• Correlator / s.correlate()
• retention enforcement / cleanup
• Full introspection (stats, spatial_bounds)

Design Decisions

API-level

• Poses: all pose params accept PoseLike (Pose | PoseStamped | Point | PointStamped). No custom pose type.
• ObservationRef identity: rowid is auto-assigned integer. ObservationRef(stream, rowid) is globally unique within a session.
• Unlocalized observations: rows without pose excluded from filter_near() by default. include_unlocalized=True to include them.
• Stream hierarchy: StreamBase (ABC) → BlobStream, EmbeddingStream, TextStream. Indexing is determined by stream type, not config.
• Lineage: parent stream defined at stream level (in _streams registry). Per-row parent_rowid links to specific row in parent.

SQLite-specific

• Separate payload table: _payload separate from _meta so queries never page in multi-MB blobs.
• EmbeddingStream has no payload table: the vector in _vec IS the data.
• R*Tree for spatio-temporal: time-only queries use B-tree index on _meta.ts (faster for 1D). Spatial/spatio-temporal queries use R*Tree.
• Payload serialization: lcm_encode() / lcm_decode(). Non-LCM types rejected at append() with TypeError.
• ObservationSet cleanup: ref-table-backed sets use SQLite temp tables. Cleaned via context manager, __del__ fallback, or connection close.

[leshy]

Query Objects — 4D Region + Soft Scoring System

Problem

We need to query observations across 4 dimensions (x, y, z, t) plus embedding space. Current API has flat filter_* methods — works for simple cases but doesn't compose. We need:

1. Regions — composable hard boundaries (include/exclude)
2. Fields — soft scoring that biases toward a point/time/embedding without hard cutoffs
3. A way to combine both in a single query

Key Insight

Hard filters and soft biases are the same thing at different extremes:

• Hard filter = step function (1 inside, 0 outside)
• Soft bias = smooth decay (gaussian, linear, etc.)

A unified Criterion type handles both. Each criterion maps an observation to a score in [0, 1]. Hard filters are just criteria with score {0, 1}.

Primitives

Temporal

# Hard boundaries
TimeRange(t1, t2)                  # 1 inside, 0 outside
Before(t)                          # sugar for TimeRange(-inf, t)
After(t)                           # sugar for TimeRange(t, inf)

# Soft — score decays with distance from target
TimeProximity(target, sigma=60.0)  # gaussian: exp(-dt²/2σ²)

Spatial

# Hard boundaries
Sphere(center: PoseLike, radius: float)    # 1 inside, 0 outside
Box(min: PoseLike, max: PoseLike)          # axis-aligned bounding box
HeightRange(z_min, z_max)                  # horizontal slice

# Soft
SpatialProximity(point: PoseLike, sigma=5.0)  # gaussian in 3D

Embedding

# Soft only (no hard boundary in embedding space makes sense)
EmbeddingSimilarity(vector, candidate_k=100)  # cosine similarity, top-k pre-filter

Tags

TagMatch(robot_id="robot1")        # hard: exact match on tag values

Composition

Criteria compose via set operators:

# Intersection — all criteria must score > 0
region = TimeRange(t1, t2) & Sphere(point, 5.0)

# Union — any criterion scoring > 0 passes
region = Sphere(p1, 3.0) | Sphere(p2, 3.0)

# Complement
region = ~TimeRange(t1, t2)  # everything outside this window

For soft criteria, composition combines scores:

• a & b → min(a.score, b.score) (conservative)
• a | b → max(a.score, b.score) (permissive)

Weighted Scoring

The interesting problem: "I care about embedding similarity, temporal proximity, AND spatial proximity" — but as soft preferences, not hard cutoffs.

Score(
    time=TimeProximity(target_t, sigma=60),
    space=SpatialProximity(point, sigma=5.0),
    embedding=EmbeddingSimilarity(vector, candidate_k=200),
    weights={"time": 0.3, "space": 0.3, "embedding": 0.4}
)

Each dimension produces a [0, 1] score. Final score = weighted sum. This replaces the vague rank(**weights) in the current API.

Integration with Query

# Current flat API (still works, sugar for simple cases)
q.after(t).near(point, 5.0).search_embedding(vec, candidate_k=100)

# Region object approach
region = After(t) & Sphere(point, 5.0)
q.where(region).search_embedding(vec, candidate_k=100)

# Full soft scoring — no hard boundaries, just preferences
q.score(
    time=TimeProximity(target_t, sigma=120),
    space=SpatialProximity(point, sigma=10.0),
    embedding=EmbeddingSimilarity(vec, candidate_k=500),
).limit(20)

# Mixed — hard boundary + soft ranking within
q.where(TimeRange(t1, t2)).score(
    space=SpatialProximity(point, sigma=5.0),
    embedding=EmbeddingSimilarity(vec, candidate_k=200),
).limit(10)

SQL Mapping (SQLite impl)

How each primitive maps to SQL:

Criterion	SQL Strategy
TimeRange(t1, t2)	WHERE ts BETWEEN ? AND ? (B-tree)
Before(t) / After(t)	WHERE ts < ? / WHERE ts > ?
Sphere(p, r)	R*Tree range query on _rtree
HeightRange(z1, z2)	WHERE pose_z BETWEEN ? AND ?
Box(min, max)	R*Tree range query
TimeProximity(t, σ)	ORDER BY ABS(ts - ?) ASC or compute score in SELECT
SpatialProximity(p, σ)	R*Tree range (pre-filter at ~3σ) + score in SELECT
EmbeddingSimilarity	sqlite-vec MATCH → temp table
TagMatch	WHERE json_extract(tags, ?) = ?

Soft scoring strategy: generous hard pre-filter in SQL, then score in Python.

• Each soft criterion auto-generates a hard pre-filter at ~3σ (captures 99.7% of relevant results)
• TimeProximity(t, σ=60) → SQL: WHERE ts BETWEEN t-180 AND t+180 (B-tree)
• SpatialProximity(p, σ=5) → SQL: R*Tree range query with 15m box
• EmbeddingSimilarity → sqlite-vec MATCH top-k (already a pre-filter)
• Python computes [0, 1] scores on the pre-filtered set, applies weights, sorts

This keeps SQL simple (range queries on indexes) and Python handles the math.

Open Questions

2. How does Score interact with search_embedding? Embedding search already returns ranked results from vec0. Should Score.embedding just re-weight those scores, or does it need a separate search pass?

3. Region objects as first-class types? Do we store/serialize regions (e.g., "the kitchen region" as a reusable spatial boundary)? Or are they always constructed in code?

4. Do we need NOT regions for exclusion zones? E.g., "everywhere except within 2m of the charging station." ~Sphere(charger, 2.0) — complement on spatial regions requires scanning all of _meta, can't use R*Tree efficiently.

5. Gradient fields? "Prefer observations taken at higher elevation" — not proximity to a point but a directional preference. HeightGradient(ascending=True) as a scorer?

Priority

• Phase 1: Keep the flat filter_* / rank() API. Implement primitives internally.
• Phase 2: Expose Criterion objects + where() + score() as the composable API.
• Phase 3: Region persistence, named regions, gradient fields.