feat(memory): validity windows on memory_units (valid_to + /invalidate) — revives #1395

hindsight-api-slim/hindsight_api/alembic/versions/39a06891cc3f_add_valid_to_to_memory_units.py

@@ -0,0 +1,64 @@
+"""Add valid_to validity-window column to memory_units table.
+
+Revision ID: 39a06891cc3f
+Revises: c1d2e3f4a5b6
+Create Date: 2026-05-31
+
+Adds a nullable ``valid_to TIMESTAMPTZ`` column on ``memory_units`` so that
+superseded facts can be soft-retired without losing their timeline. Also
+adds a partial index on the bank/fact_type prefix limited to currently
+active rows so recall keeps using a small index even as historical data
+accumulates.
+
+Recall queries filter out rows where ``valid_to <= now()`` so invalidated
+memories no longer surface in semantic / BM25 / graph-spreading search,
+while ``GET /memories/{id}`` and ``GET /memories/{id}/history`` still return
+them — preserving the audit trail.
+
+See issue #1391 for the full design rationale.
+"""
+
+from collections.abc import Sequence
+
+from alembic import context, op
+
+from hindsight_api.alembic._dialect import run_for_dialect
+
+revision: str = "39a06891cc3f"
+down_revision: str | Sequence[str] | None = "c1d2e3f4a5b6"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _get_schema_prefix() -> str:
+    """Get schema prefix for table names (required for multi-tenant support)."""
+    schema = context.config.get_main_option("target_schema")
+    return f'"{schema}".' if schema else ""
+
+
+def _pg_upgrade() -> None:
+    schema = _get_schema_prefix()
+    op.execute(f"ALTER TABLE {schema}memory_units ADD COLUMN IF NOT EXISTS valid_to TIMESTAMPTZ NULL")
+    op.execute(
+        f"COMMENT ON COLUMN {schema}memory_units.valid_to IS "
+        "'NULL = still valid; non-NULL = superseded at this timestamp. "
+        "Recall filters out memories with valid_to <= now().'"
+    )
+    op.execute(
+        f"CREATE INDEX IF NOT EXISTS idx_memory_units_active "
+        f"ON {schema}memory_units (bank_id, fact_type) WHERE valid_to IS NULL"
+    )
+
+
+def _pg_downgrade() -> None:
+    schema = _get_schema_prefix()
+    op.execute(f"DROP INDEX IF EXISTS {schema}idx_memory_units_active")
+    op.execute(f"ALTER TABLE {schema}memory_units DROP COLUMN IF EXISTS valid_to")
+
+
+def upgrade() -> None:
+    run_for_dialect(pg=_pg_upgrade)
+
+
+def downgrade() -> None:
+    run_for_dialect(pg=_pg_downgrade)

hindsight-api-slim/hindsight_api/api/http.py

@@ -1464,6 +1464,42 @@ class ClearMemoryObservationsResponse(BaseModel):
     deleted_count: int
 
 
+class InvalidateMemoryRequest(BaseModel):
+    """Request model for marking a memory unit as superseded."""
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "valid_to": "2026-05-02T17:30:00Z",
+                "reason": "Server srv-04 was decommissioned",
+            }
+        }
+    )
+
+    valid_to: str | None = None  # ISO-8601; defaults to now() if omitted
+    reason: str | None = None
+
+
+class InvalidateMemoryResponse(BaseModel):
+    """Response model for invalidating a memory unit."""
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "id": "0c14e4f1-9eb6-4dde-b0a4-c2e8b3a3e5f1",
+                "valid_to": "2026-05-02T17:30:00.123456+00:00",
+                "fact_type": "world",
+                "preview": "Server srv-04 runs PostgreSQL 17 on Debian 12...",
+            }
+        }
+    )
+
+    id: str
+    valid_to: str | None
+    fact_type: str
+    preview: str
+
+
 class RecoverConsolidationResponse(BaseModel):
     """Response model for recovering failed consolidation."""
 
@@ -5335,6 +5371,72 @@ async def api_recover_consolidation(bank_id: str, request_context: RequestContex
             logger.error(f"Error in POST /v1/default/banks/{bank_id}/consolidation/recover: {error_detail}")
             raise HTTPException(status_code=500, detail=str(e))
 
+    @app.post(
+        "/v1/default/banks/{bank_id}/memories/{memory_id}/invalidate",
+        response_model=InvalidateMemoryResponse,
+        summary="Invalidate a memory unit",
+        description=(
+            "Mark a memory unit as invalidated as of a timestamp. The row stays in the timeline "
+            "(reachable via GET /memories/{id} and GET /memories/{id}/history) but recall queries "
+            "no longer return it once valid_to <= now(). Use when a fact has been superseded "
+            "(server decommissioned, person changed roles, default value changed) — not for "
+            "deletion of accidentally retained data."
+        ),
+        operation_id="invalidate_memory",
+        tags=["Memory"],
+    )
+    @audited("invalidate_memory", request_param="payload")
+    async def api_invalidate_memory(
+        bank_id: str,
+        memory_id: str,
+        payload: InvalidateMemoryRequest | None = None,
+        request_context: RequestContext = Depends(get_request_context),
+    ):
+        """Mark a memory unit as superseded as of a timestamp (default: now()).
+
+        The memory is *not* deleted; recall filters it out, but the audit trail
+        remains accessible via the regular get / history endpoints.
+        """
+        from datetime import datetime as _datetime
+
+        valid_to_dt: _datetime | None = None
+        reason: str | None = None
+        if payload is not None:
+            reason = payload.reason
+            if payload.valid_to:
+                try:
+                    valid_to_dt = _datetime.fromisoformat(payload.valid_to.replace("Z", "+00:00"))
+                except ValueError as ve:
+                    raise HTTPException(
+                        status_code=400,
+                        detail=f"valid_to is not a valid ISO-8601 timestamp: {payload.valid_to!r}",
+                    ) from ve
+        try:
+            result = await app.state.memory.invalidate_memory_unit(
+                bank_id=bank_id,
+                memory_id=memory_id,
+                valid_to=valid_to_dt,
+                reason=reason,
+                request_context=request_context,
+            )
+            if result is None:
+                raise HTTPException(status_code=404, detail=f"Memory unit '{memory_id}' not found")
+            return InvalidateMemoryResponse(**result)
+        except OperationValidationError as e:
+            raise HTTPException(status_code=e.status_code, detail=e.reason)
+        except ValueError as e:
+            raise HTTPException(status_code=400, detail=str(e))
+        except (AuthenticationError, HTTPException):
+            raise
+        except Exception as e:
+            import traceback
+
+            error_detail = f"{str(e)}\n\nTraceback:\n{traceback.format_exc()}"
+            logger.error(
+                f"Error in POST /v1/default/banks/{bank_id}/memories/{memory_id}/invalidate: {error_detail}"
+            )
+            raise HTTPException(status_code=500, detail=str(e))
+
     @app.delete(
         "/v1/default/banks/{bank_id}/memories/{memory_id}/observations",
         response_model=ClearMemoryObservationsResponse,

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -4990,6 +4990,84 @@ async def clear_observations_for_memory(
 
         return {"deleted_count": deleted_count}
 
+    async def invalidate_memory_unit(
+        self,
+        bank_id: str,
+        memory_id: str,
+        *,
+        valid_to: "datetime | None" = None,
+        reason: str | None = None,
+        request_context: "RequestContext",
+    ) -> dict | None:
+        """
+        Mark a memory unit as invalidated as of ``valid_to``. The row remains in
+        the timeline (still reachable via ``GET /memories/{id}`` and
+        ``GET /memories/{id}/history``) but recall queries will no longer
+        surface it once ``valid_to <= now()``.
+
+        Args:
+            bank_id: Bank ID.
+            memory_id: ID of the memory unit to invalidate.
+            valid_to: Timestamp at which the fact stops being valid. Defaults to
+                ``now()`` if omitted.
+            reason: Free-text rationale, stored to ``metadata.invalidation_reason``.
+            request_context: Request context for authentication.
+
+        Returns:
+            Dict with ``id``, ``valid_to``, ``fact_type`` and a short ``preview``
+            of the row's text, or ``None`` if no matching row was found.
+
+        Raises:
+            ValueError: If ``memory_id`` is not a valid UUID.
+        """
+        import uuid as uuid_module
+        from datetime import datetime, timezone
+
+        try:
+            memory_uuid = uuid_module.UUID(memory_id)
+        except ValueError:
+            raise ValueError(f"Invalid memory_id: '{memory_id}' is not a valid UUID")
+
+        if valid_to is None:
+            valid_to = datetime.now(timezone.utc)
+
+        await self._authenticate_tenant(request_context)
+        if self._operation_validator:
+            from hindsight_api.extensions import BankWriteContext
+
+            ctx = BankWriteContext(
+                bank_id=bank_id, operation="invalidate_memory_unit", request_context=request_context
+            )
+            await self._validate_operation(self._operation_validator.validate_bank_write(ctx))
+
+        backend = await self._get_backend()
+        async with acquire_with_retry(backend) as conn:
+            row = await conn.fetchrow(
+                f"""
+                UPDATE {fq_table("memory_units")}
+                   SET valid_to = $3::timestamptz,
+                       metadata = COALESCE(metadata, '{{}}'::jsonb)
+                                  || jsonb_build_object('invalidation_reason', $4::text)
+                 WHERE id = $1::uuid
+                   AND bank_id = $2
+                RETURNING id, valid_to, fact_type, LEFT(text, 200) AS preview
+                """,
+                str(memory_uuid),
+                bank_id,
+                valid_to,
+                reason,
+            )
+
+        if row is None:
+            return None
+
+        return {
+            "id": str(row["id"]),
+            "valid_to": row["valid_to"].isoformat() if row["valid_to"] else None,
+            "fact_type": row["fact_type"],
+            "preview": row["preview"],
+        }
+
     async def run_consolidation(
         self,
         bank_id: str,
@@ -5649,7 +5727,7 @@ async def get_memory_unit(
                 f"""
                 SELECT id, text, context, event_date, occurred_start, occurred_end,
                        mentioned_at, fact_type, document_id, chunk_id, tags, source_memory_ids,
-                       observation_scopes
+                       observation_scopes, valid_to
                 FROM {fq_table("memory_units")}
                 WHERE id = $1 AND bank_id = $2
                 """,

hindsight-api-slim/hindsight_api/engine/search/retrieval.py

@@ -190,6 +190,15 @@ async def retrieve_semantic_bm25_combined(
         created_range_clause += f" AND updated_at < ${_next_idx}"
         _next_idx += 1
 
+    # --- validity-window filter ---------------------------------------------
+    # Skip memories that have been explicitly invalidated (valid_to <= now()).
+    # NULL valid_to means "still valid" — the default for every retain.
+    # The partial index ``idx_memory_units_active`` covers the hot path.
+    validity_clause = (
+        f" AND (valid_to IS NULL OR valid_to > {dialect.current_timestamp()})"
+    )
+    extra_where_clause = validity_clause + created_range_clause
+
     # --- Semantic UNION ALL arms (one per fact_type) ---
     # Each arm has its own ORDER BY ... LIMIT, enabling the partial HNSW indexes
     # per fact_type instead of forcing a full sequential scan.
@@ -203,7 +212,7 @@ async def retrieve_semantic_bm25_combined(
             fetch_limit=hnsw_fetch,
             tags_clause=tags_clause,
             groups_clause=groups_clause,
-            extra_where=created_range_clause,
+            extra_where=extra_where_clause,
         )
         for ft in fact_types
     ]
@@ -226,7 +235,7 @@ async def retrieve_semantic_bm25_combined(
                     arm_index=i,
                     text_search_extension=text_ext,
                     bm25_language=config.text_search_extension_native_language,
-                    extra_where=created_range_clause,
+                    extra_where=extra_where_clause,
                 )
             )
 
@@ -265,6 +274,7 @@ async def retrieve_semantic_bm25_combined(
             if created_before is not None:
                 fb_created_clause += f" AND updated_at < ${fb_next_idx}"
                 fb_next_idx += 1
+            fb_extra_where = validity_clause + fb_created_clause
             fb_arms = [
                 dialect.build_semantic_arm(
                     table=table,
@@ -275,7 +285,7 @@ async def retrieve_semantic_bm25_combined(
                     fetch_limit=hnsw_fetch,
                     tags_clause=fb_tags_clause,
                     groups_clause=fb_groups_clause,
-                    extra_where=fb_created_clause,
+                    extra_where=fb_extra_where,
                 )
                 for ft in fact_types
             ]
@@ -392,6 +402,7 @@ async def retrieve_temporal_combined(
             WHERE bank_id = $2
               AND fact_type = ANY($3)
               AND embedding IS NOT NULL
+              AND (valid_to IS NULL OR valid_to > now())
               AND (
                   (occurred_start IS NOT NULL AND occurred_end IS NOT NULL
                    AND occurred_start <= $5 AND occurred_end >= $4)
@@ -532,6 +543,7 @@ async def retrieve_temporal_combined(
                 WHERE mu.bank_id = $6
                   AND mu.fact_type = $3
                   AND mu.embedding IS NOT NULL
+                  AND (mu.valid_to IS NULL OR mu.valid_to > now())
                   AND (1 - (mu.embedding <=> $1::vector)) >= $4
                   {spreading_tags_clause}
                   {spreading_groups_clause}

hindsight-api-slim/hindsight_api/models.py

@@ -107,6 +107,9 @@ class MemoryUnit(Base):
     )  # User-defined metadata (str->str)
     created_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now())
     updated_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now())
+    valid_to: Mapped[datetime | None] = mapped_column(
+        TIMESTAMP(timezone=True)
+    )  # NULL = still valid; non-NULL = superseded at this timestamp (recall filters out)
 
     # Relationships
     document = relationship("Document", back_populates="memory_units")
@@ -152,6 +155,12 @@ class MemoryUnit(Base):
             postgresql_using="hnsw",
             postgresql_ops={"embedding": "vector_cosine_ops"},
         ),
+        Index(
+            "idx_memory_units_active",
+            "bank_id",
+            "fact_type",
+            postgresql_where=sql_text("valid_to IS NULL"),
+        ),
     )