feat(core): add temporal qualifiers and time-aware search (SPEC-82)

src/basic_memory/alembic/versions/u4t5e6m7p8o9_add_memory_time_index_table.py

@@ -0,0 +1,103 @@
+"""Add memory_time_index table
+
+Revision ID: u4t5e6m7p8o9
+Revises: t3n4o5t6e7s8
+Create Date: 2026-08-31 12:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = "u4t5e6m7p8o9"
+down_revision: Union[str, None] = "t3n4o5t6e7s8"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create memory_time_index: the projection of authored valid time (SPEC-82).
+
+    Rows derive from temporal qualifiers written in canonical markdown
+    (``@effective[2026-06-10,2026-07-27)``). Like observations and sections they are
+    rebuilt under the note_content generation fence on every (re)index and removed with
+    the entity, so the table is always reproducible from the notes.
+
+    Every column is a portable scalar, and every type used here renders on both SQLite
+    and PostgreSQL, so this migration needs no dialect branching. Bounds are canonical
+    fixed-width text rather than DATE/TIMESTAMP: a date bound must never acquire a time
+    of day or a timezone (SQLAlchemy's SQLite DateTime silently drops an offset and
+    stores the wrong instant), and fixed-width canonical text makes byte-lexicographic
+    order chronological, so one identical predicate serves both backends. Native
+    PostgreSQL range columns remain a later addition generated from these columns.
+
+    ``source_id`` carries no foreign key by design: it addresses whichever table
+    ``source_type`` names (``observation`` today). Referential lifecycle rides on
+    ``entity_id``'s cascade plus the fenced replace instead.
+
+    Only ``ix_memory_time_index_lookup`` indexes the filter columns. The bound values
+    are deliberately unindexed, and this table is *not* always driven by a full-text
+    candidate set: a valid-time filter counts as criteria on its own
+    (``SearchQuery.no_criteria``), so a temporal-only search scans the bound columns
+    for every row matching project + kind + axis. That is an acceptable scan at
+    expected sizes -- one row per authored qualifier, so thousands, not millions. If
+    temporal-only queries ever become a hot path, the answer is a native PostgreSQL
+    range column with a GiST index, not a btree over these text bounds.
+    """
+    op.create_table(
+        "memory_time_index",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("project_id", sa.Integer(), nullable=False),
+        sa.Column("entity_id", sa.Integer(), nullable=False),
+        sa.Column("source_type", sa.String(length=32), nullable=False),
+        sa.Column("source_id", sa.Integer(), nullable=False),
+        sa.Column("time_kind", sa.String(length=32), nullable=False),
+        sa.Column("range_axis", sa.String(length=16), nullable=False),
+        sa.Column("lower_value", sa.String(length=32), nullable=True),
+        sa.Column("upper_value", sa.String(length=32), nullable=True),
+        sa.Column("lower_inclusive", sa.Boolean(), nullable=False),
+        sa.Column("upper_inclusive", sa.Boolean(), nullable=False),
+        sa.Column("is_empty", sa.Boolean(), nullable=False, server_default=sa.false()),
+        sa.Column("extractor", sa.String(length=32), nullable=False),
+        sa.Column("source_text", sa.Text(), nullable=False),
+        sa.Column("assertion_metadata", sa.JSON(), nullable=True),
+        sa.ForeignKeyConstraint(["project_id"], ["project.id"]),
+        sa.ForeignKeyConstraint(["entity_id"], ["entity.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.CheckConstraint(
+            "range_axis IN ('date', 'instant')",
+            name="ck_memory_time_index_range_axis",
+        ),
+        sa.CheckConstraint(
+            "NOT is_empty OR (lower_value IS NULL AND upper_value IS NULL)",
+            name="ck_memory_time_index_empty_has_no_bounds",
+        ),
+        sa.CheckConstraint(
+            "(lower_value IS NOT NULL OR NOT lower_inclusive) "
+            "AND (upper_value IS NOT NULL OR NOT upper_inclusive)",
+            name="ck_memory_time_index_unbounded_is_exclusive",
+        ),
+    )
+    op.create_index(
+        "ix_memory_time_index_lookup",
+        "memory_time_index",
+        ["project_id", "time_kind", "range_axis", "source_type", "source_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_memory_time_index_entity_id",
+        "memory_time_index",
+        ["entity_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Drop memory_time_index and its supporting indexes."""
+    op.drop_index("ix_memory_time_index_entity_id", table_name="memory_time_index")
+    op.drop_index("ix_memory_time_index_lookup", table_name="memory_time_index")
+    op.drop_table("memory_time_index")

src/basic_memory/alembic/versions/v5o6b7s8d9e0_add_observation_duplicate_index.py

@@ -0,0 +1,49 @@
+"""Add the duplicate ordinal that keeps same-identity observations addressable.
+
+An observation's synthetic permalink is built from its category and content, both of
+which survive the peel that strips a temporal qualifier and a (context) off the authored
+line. Two observations differing only in one of those therefore shared one address, and
+the permalink-keyed search index kept only the first -- so the second note's authored
+valid time addressed an observation with no search row, and queries for its interval
+found nothing (SPEC-82).
+
+This ordinal separates such twins. It defaults to 0, which is the value every existing
+row takes and the value the first observation of any identity keeps, so no permalink that
+resolves today changes. Search rows are derived state and are rebuilt from markdown, so
+the second twin becomes addressable on the next index pass for that note.
+
+Revision ID: v5o6b7s8d9e0
+Revises: u4t5e6m7p8o9
+Create Date: 2026-09-02 10:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "v5o6b7s8d9e0"
+down_revision: Union[str, None] = "u4t5e6m7p8o9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add observation.duplicate_index, defaulting every existing row to 0."""
+    with op.batch_alter_table("observation", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "duplicate_index",
+                sa.Integer(),
+                server_default=sa.text("0"),
+                nullable=False,
+            )
+        )
+
+
+def downgrade() -> None:
+    """Remove the duplicate ordinal."""
+    with op.batch_alter_table("observation", schema=None) as batch_op:
+        batch_op.drop_column("duplicate_index")

src/basic_memory/api/v2/routers/search_router.py

@@ -12,13 +12,16 @@
 from fastapi import APIRouter, Depends, HTTPException, Path, Response
 
 import logfire
-from basic_memory.api.v2.utils import to_search_results
+from basic_memory import db
+from basic_memory.api.v2.utils import load_temporal_metadata, to_search_results
 from basic_memory.deps import (
     EntityServiceV2ExternalDep,
+    MemoryTimeIndexRepositoryV2ExternalDep,
     ProjectExternalIdPathDep,
     ReadCacheDep,
     SearchReindexSchedulerDep,
     SearchServiceV2ExternalDep,
+    SessionMakerDep,
     create_model_read_cache,
 )
 from basic_memory.read_cache import (
@@ -83,6 +86,8 @@ async def search(
     query: SearchQuery,
     search_service: SearchServiceV2ExternalDep,
     entity_service: EntityServiceV2ExternalDep,
+    temporal_repository: MemoryTimeIndexRepositoryV2ExternalDep,
+    session_maker: SessionMakerDep,
     read_cache: SearchReadCacheDep,
     response: Response,
     project_id: str = Path(..., description="Project external UUID"),
@@ -98,13 +103,19 @@ async def search(
         query: Search query parameters (text, filters, etc.)
         search_service: Search service scoped to project
         entity_service: Entity service scoped to project
+        temporal_repository: Valid-time projection, read to explain temporal matches
+        session_maker: Session factory for the temporal hydration read
         page: Page number for pagination
         page_size: Number of results per page
 
     Returns:
         SearchResponse with paginated search results
     """
     response.headers["Accept-Query"] = "application/json"
+    # Read from the request rather than from the parsed filter: this is a plain field
+    # check that cannot raise, and reaching the hydration step below already proves
+    # the service accepted and executed the filter.
+    temporal_requested = query.has_temporal_filter()
     with logfire.span(
         "api.request.search",
         entrypoint="api",
@@ -125,7 +136,9 @@ async def search(
             or query.categories
             or query.metadata_filters
             or query.file_path_prefix
+            or temporal_requested
         ),
+        has_temporal_filter=temporal_requested,
     ):
         cache_key = ReadCacheKey(
             project_id=project_id,
@@ -202,7 +215,20 @@ async def search(
                 phase="hydrate_results",
                 result_count=len(results),
             ):
-                search_results = await to_search_results(entity_service, results)
+                # Trigger: the caller asked a valid-time question.
+                # Why: the assertions explain *why* each hit matched, but loading them
+                #   costs a query, and a search with no temporal filter has nothing to
+                #   explain -- so ordinary searches stay exactly as expensive as before.
+                # Outcome: temporal metadata rides along only on temporal searches.
+                temporal_by_source = {}
+                if temporal_requested:
+                    async with db.scoped_session(session_maker) as session:
+                        temporal_by_source = await load_temporal_metadata(
+                            temporal_repository, session, results
+                        )
+                search_results = await to_search_results(
+                    entity_service, results, temporal_by_source=temporal_by_source
+                )
             with logfire.span(
                 "api.search.search.build_response",
                 domain="search",
@@ -217,6 +243,9 @@ async def search(
                     total=total,
                     total_is_exact=exact_count_available,
                     has_more=has_more,
+                    # None, not False, when nothing was asked: an ordinary search
+                    # payload stays exactly what it was before valid time existed.
+                    temporal_applied=True if temporal_requested else None,
                 )
                 cached.value = result
                 return result

src/basic_memory/api/v2/utils.py

@@ -1,7 +1,10 @@
+from collections import defaultdict
+from collections.abc import Mapping
 from typing import Any, Protocol, Optional, List, Sequence
 
 import logfire
 from sqlalchemy.ext.asyncio import AsyncSession
+from basic_memory.models import MemoryTimeIndex
 from basic_memory.repository.search_repository import SearchIndexRow
 from basic_memory.schemas.memory import (
     EntitySummary,
@@ -11,11 +14,17 @@
     GraphContext,
     ContextResult,
 )
-from basic_memory.schemas.search import SearchItemType, SearchResult
+from basic_memory.schemas.search import (
+    SearchItemType,
+    SearchResult,
+    TemporalRangeValue,
+    TemporalResultMetadata,
+)
 from basic_memory.services.context_service import (
     ContextResultRow,
     ContextResult as ServiceContextResult,
 )
+from basic_memory.temporal import TemporalRange, TemporalRangeAxis
 
 
 class EntityBatchLookup(Protocol):
@@ -32,6 +41,19 @@ class EntityServiceBatchLookup(Protocol):
     async def get_entities_by_id(self, ids: List[int]) -> Sequence[Any]: ...
 
 
+class TemporalAssertionLookup(Protocol):
+    async def find_for_sources(
+        self,
+        session: AsyncSession,
+        sources: Sequence[tuple[str, int]],
+    ) -> Sequence[MemoryTimeIndex]: ...
+
+
+# One page of hits, keyed by the (search row type, search row id) pair the projection
+# addresses. Empty means "no valid-time metadata was loaded", never "none exists".
+type TemporalMetadataBySource = Mapping[tuple[str, int], list[TemporalResultMetadata]]
+
+
 async def get_entities_by_id_lookup(
     entity_service: EntityServiceBatchLookup,
     entity_ids: Sequence[int],
@@ -217,8 +239,71 @@ def to_summary(
         )
 
 
+def _temporal_result_metadata(row: MemoryTimeIndex) -> TemporalResultMetadata:
+    """Shape one projected assertion into the value a caller sees.
+
+    Rebuilding the domain range from the stored scalars re-runs its invariants, so a
+    row that somehow violated them surfaces here instead of being rendered as a
+    plausible-looking interval.
+    """
+    valid_during = TemporalRange(
+        axis=TemporalRangeAxis(row.range_axis),
+        lower=row.lower_value,
+        upper=row.upper_value,
+        lower_inclusive=row.lower_inclusive,
+        upper_inclusive=row.upper_inclusive,
+        is_empty=row.is_empty,
+    )
+    return TemporalResultMetadata(
+        kind=row.time_kind,
+        valid_during=TemporalRangeValue(
+            axis=valid_during.axis.value,
+            literal=str(valid_during),
+            lower=valid_during.lower,
+            upper=valid_during.upper,
+            lower_inclusive=valid_during.lower_inclusive,
+            upper_inclusive=valid_during.upper_inclusive,
+            is_empty=valid_during.is_empty,
+        ),
+        source_text=row.source_text,
+    )
+
+
+async def load_temporal_metadata(
+    temporal_repository: TemporalAssertionLookup,
+    session: AsyncSession,
+    results: Sequence[SearchIndexRow],
+) -> dict[tuple[str, int], list[TemporalResultMetadata]]:
+    """Load the authored valid-time assertions behind one page of search hits.
+
+    Keyed on the search row's own ``(type, id)`` pair, which is exactly the address
+    the projection stores -- so an observation hit resolves to the assertions written
+    on that observation, not to its note's other assertions.
+    """
+    sources = [(result.type, result.id) for result in results]
+    if not sources:
+        return {}
+
+    with logfire.span(
+        "search.hydrate_results.fetch_temporal",
+        domain="search",
+        action="search",
+        phase="fetch_temporal",
+        result_count=len(sources),
+    ):
+        rows = await temporal_repository.find_for_sources(session, sources)
+
+    by_source: defaultdict[tuple[str, int], list[TemporalResultMetadata]] = defaultdict(list)
+    for row in rows:
+        by_source[(row.source_type, row.source_id)].append(_temporal_result_metadata(row))
+    return dict(by_source)
+
+
 async def to_search_results(
-    entity_service: EntityServiceBatchLookup, results: List[SearchIndexRow]
+    entity_service: EntityServiceBatchLookup,
+    results: List[SearchIndexRow],
+    *,
+    temporal_by_source: TemporalMetadataBySource | None = None,
 ) -> list[SearchResult]:
     with logfire.span(
         "search.hydrate_results",
@@ -299,6 +384,11 @@ async def to_search_results(
                         from_entity=from_entity.permalink if from_entity else None,
                         to_entity=to_entity.permalink if to_entity else None,
                         relation_type=result.relation_type,
+                        temporal=(
+                            temporal_by_source.get((result.type, result.id))
+                            if temporal_by_source
+                            else None
+                        ),
                     )
                 )
         return search_results

src/basic_memory/config_models.py

@@ -586,6 +586,15 @@ def __init__(self, **data: Any) -> None: ...
         gt=0,
     )
 
+    # Spelled as a bare Literal rather than the `DateOrder` alias so `bm config set`
+    # keeps discovering it: CONFIGURABLE_FIELDS reads model_fields annotations, and a
+    # PEP 695 alias arrives there unresolved. `temporal.DateOrder` is the same union,
+    # and a test pins the two together.
+    date_order: Literal["YMD", "DMY", "MDY"] = Field(
+        default="YMD",
+        description="Component order used to read an ambiguous slash-formatted date in an authored temporal qualifier (e.g. '@10/07/2026'). YMD and DMY read that as 10 July 2026, MDY as 7 October 2026. ISO dates like '2026-07-10' are never re-guessed.",
+    )
+
     kebab_filenames: bool = Field(
         default=False,
         description="Format for generated filenames. False preserves spaces and special chars, True converts them to hyphens for consistency with permalinks",