feat(lifecycle): add forget flag + TTL to v2 ingest API (#166)

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Add `forget: bool` flag to the v2 ingest API (`POST /v2/memory/ingest`): memories ingested with `forget=true` are tagged with a TTL (`expires_at`) and automatically excluded from all retrieval results after the TTL elapses (default 30 days, configurable via `MEMORY_FORGET_DEFAULT_TTL_DAYS`). No background sweeper required — enforcement is at read time.
 - Add modular Razorpay billing, credit wallets, ledger reservations, and v2 memory workflow metering.
 - Add durable Temporal-backed v2 memory and scanner workflow APIs with job status, retry, cancel, and dead-letter endpoints.
 - Add modular LoCoMo and BEAM benchmark runners for the Python XMem API.

src/api/routes/v2/activities.py

@@ -109,11 +109,14 @@ async def memory_domain_activity(payload: Dict[str, Any]) -> Dict[str, Any]:
         billing_account_id=payload.get("billing_account_id"),
         user_id=user_id,
     ):
+        lifecycle_metadata = payload.get("lifecycle_metadata")
+
         if domain == "profile":
             result = await pipeline._node_extract_profile(
                 {
                     "profile_queries": payload.get("queries", []),
                     "user_id": user_id,
+                    "lifecycle_metadata": lifecycle_metadata,
                 }
             )
             return {"domain": domain, "result": _domain_payload(result, "profile")}
@@ -124,6 +127,7 @@ async def memory_domain_activity(payload: Dict[str, Any]) -> Dict[str, Any]:
                     "temporal_queries": payload.get("queries", []),
                     "session_datetime": payload.get("session_datetime", ""),
                     "user_id": user_id,
+                    "lifecycle_metadata": lifecycle_metadata,
                 }
             )
             return {"domain": domain, "result": _domain_payload(result, "temporal")}
@@ -134,6 +138,7 @@ async def memory_domain_activity(payload: Dict[str, Any]) -> Dict[str, Any]:
                     "user_query": payload.get("user_query", ""),
                     "agent_response": payload.get("agent_response", ""),
                     "user_id": user_id,
+                    "lifecycle_metadata": lifecycle_metadata,
                 }
             )
             return {"domain": domain, "result": _domain_payload(result, "summary")}
@@ -144,6 +149,7 @@ async def memory_domain_activity(payload: Dict[str, Any]) -> Dict[str, Any]:
                     "classifier_output": payload.get("classifier_output", ""),
                     "image_url": payload.get("image_url", ""),
                     "user_id": user_id,
+                    "lifecycle_metadata": lifecycle_metadata,
                 }
             )
             return {"domain": domain, "result": _domain_payload(result, "image")}

src/api/routes/v2/memory.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+from datetime import datetime, timedelta, timezone
 import time
 from typing import Any, Dict
 
@@ -122,6 +123,18 @@ async def ingest_memory_v2(req: IngestRequest, request: Request, user: dict = De
     payload = req.model_dump()
     payload["user_id"] = user_id
     payload["timeout_seconds"] = float(settings.memory_ingest_timeout_seconds)
+
+    # When forget=true, compute lifecycle_metadata and thread it through so the
+    # weaver stamps the forget flag + TTL on every vector record it writes.
+    if req.forget:
+        now = datetime.now(timezone.utc)
+        expires_at = now + timedelta(days=settings.memory_forget_default_ttl_days)
+        payload["lifecycle_metadata"] = {
+            "forget": True,
+            "expires_at": expires_at.isoformat(),
+            "lifecycle_state": "active",
+        }
+
     idempotency_fields = {
         "user_id": user_id,
         "org_id": payload.get("org_id", "default"),
@@ -131,6 +144,9 @@ async def ingest_memory_v2(req: IngestRequest, request: Request, user: dict = De
             "session_datetime": req.session_datetime,
             "image_url": req.image_url,
             "effort_level": req.effort_level,
+            # Include forget in idempotency so forget vs non-forget of the
+            # same content are treated as distinct requests.
+            "forget": req.forget,
         }),
     }
     job_id = _durable_job_id("memory_ingest", idempotency_fields)

src/api/schemas.py

@@ -86,6 +86,14 @@ class IngestRequest(UserScopedModel):
         default="low",
         description="'low' (fast, single pass) or 'high' (chunked parallel extraction)",
     )
+    forget: bool = Field(
+        default=False,
+        description=(
+            "When true, the stored memory is tagged with a TTL and will be "
+            "automatically excluded from retrieval after it expires. "
+            "Only honoured on the v2 ingest path."
+        ),
+    )
 
     @field_validator("user_query")
     @classmethod

src/config/settings.py

@@ -506,6 +506,18 @@ class Settings(BaseSettings):
         description="Razorpay subscription plan ID for the global USD Pro plan",
     )
 
+    # =============================================================================
+    # Memory Lifecycle — Forget / TTL (default OFF — fully backward compatible)
+    # =============================================================================
+    memory_forget_default_ttl_days: float = Field(
+        default=30.0,
+        gt=0.0,
+        description=(
+            "Default TTL in days for memories ingested with forget=true via the v2 API. "
+            "Expired records are hidden from retrieval (filtered at read time; no sweeper)."
+        ),
+    )
+
     @field_validator("fallback_order")
     @classmethod
     def validate_fallback_order(cls, v: List[str]) -> List[str]:

src/pipelines/ingest.py

@@ -870,6 +870,7 @@ async def _node_extract_profile(self, state: IngestState) -> Dict[str, Any]:
             judge_result=judge_result,
             domain=JudgeDomain.PROFILE,
             user_id=user_id,
+            extra_metadata=state.get("lifecycle_metadata"),
         )
         return {
             "profile_result": result,
@@ -963,6 +964,7 @@ async def _node_extract_image(self, state: IngestState) -> Dict[str, Any]:
             judge_result=judge_result,
             domain=JudgeDomain.SUMMARY,
             user_id=user_id,
+            extra_metadata=state.get("lifecycle_metadata"),
         )
 
         return {
@@ -1090,6 +1092,7 @@ async def _node_extract_summary(self, state: IngestState) -> Dict[str, Any]:
             judge_result=judge_result,
             domain=JudgeDomain.SUMMARY,
             user_id=state.get("user_id", "default"),
+            extra_metadata=state.get("lifecycle_metadata"),
         )
         return {
             "summary_result": result,

src/pipelines/lifecycle.py

@@ -0,0 +1,71 @@
+"""
+Memory lifecycle — pure, deterministic helper functions for forget/TTL.
+
+is_retrievable() is the single retrieval-time gate. build_forget_metadata()
+stamps the lifecycle fields onto a metadata dict for storage.
+
+All functions are side-effect-free so they can be tested without live services.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Dict, Mapping, Optional
+
+
+def is_retrievable(metadata: Mapping[str, Any], now: datetime) -> bool:
+    """Return True when a record should appear in retrieval results.
+
+    Rules (applied in order):
+    1. ``lifecycle_state == "forgotten"`` → hidden (manual soft-forget).
+    2. ``forget is True`` and ``expires_at`` is present and in the past → hidden (TTL expired).
+    3. Everything else (including all legacy records with no lifecycle keys) → retrievable.
+
+    Missing keys default to the legacy-safe value so records stored before
+    lifecycle was introduced are never hidden.
+    """
+    if metadata.get("lifecycle_state", "active") == "forgotten":
+        return False
+
+    if metadata.get("forget"):
+        expires_raw = metadata.get("expires_at")
+        if expires_raw:
+            try:
+                expires_at = datetime.fromisoformat(str(expires_raw))
+                # Make both sides timezone-aware or both naive for comparison
+                if expires_at.tzinfo is None and now.tzinfo is not None:
+                    from datetime import timezone
+                    expires_at = expires_at.replace(tzinfo=timezone.utc)
+                elif expires_at.tzinfo is not None and now.tzinfo is None:
+                    expires_at = expires_at.replace(tzinfo=None)
+                if expires_at < now:
+                    return False
+            except (ValueError, TypeError):
+                pass
+
+    return True
+
+
+def build_lifecycle_metadata(
+    now: datetime,
+    ttl_days: float,
+    reason: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Return the lifecycle metadata dict to merge onto a forget=true record.
+
+    Called once at v2 ingestion time when the caller sets ``forget=true``.
+    The result is stored as part of the vector record's metadata so the
+    retrieval-time filter can enforce the TTL without any background sweeper.
+    """
+    from datetime import timedelta
+    expires_at = now + timedelta(days=ttl_days)
+    meta: Dict[str, Any] = {
+        "forget": True,
+        "expires_at": expires_at.isoformat(),
+        "lifecycle_state": "active",
+        "created_at": now.isoformat(),
+        "updated_at": now.isoformat(),
+    }
+    if reason:
+        meta["forget_reason"] = reason
+    return meta

src/pipelines/retrieval.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 import asyncio
+from datetime import datetime, timezone
 import logging
 from typing import Any, Callable, Dict, List, Optional
 
@@ -31,6 +32,7 @@
 
 from src.config import settings
 from src.graph.neo4j_client import Neo4jClient
+from src.pipelines.lifecycle import is_retrievable
 from src.prompts.retrieval import ANSWER_PROMPT, build_system_prompt
 from src.schemas.retrieval import RetrievalResult, SourceRecord
 from src.schemas.code import snippets_namespace
@@ -100,6 +102,7 @@ def __init__(
         model: Optional[BaseChatModel] = None,
         vector_store: Optional[BaseVectorStore] = None,
         neo4j_client: Optional[Neo4jClient] = None,
+        _now: Optional[Callable[[], datetime]] = None,
     ) -> None:
         # ── LLM ───────────────────────────────────────────────────────
         if model is None:
@@ -133,6 +136,7 @@ def __init__(
 
         self.embed_fn = embed_fn
         self._snippet_stores: Dict[str, BaseVectorStore] = {}
+        self._now: Callable[[], datetime] = _now or (lambda: datetime.now(timezone.utc))
 
         logger.info("RetrievalPipeline initialized")
 
@@ -413,8 +417,11 @@ async def _search_summary(
         user_id: str,
         top_k: int = 10,
     ) -> List[SourceRecord]:
-        """Semantic search over summary entries in Pinecone."""
+        """Semantic search over summary entries in Pinecone.
 
+        Records ingested with ``forget=true`` whose TTL has passed are filtered
+        out at read time. Legacy records (no lifecycle keys) always pass through.
+        """
         results = await self.vector_store.search_by_text(
             query_text=query,
             top_k=top_k,
@@ -424,8 +431,11 @@ async def _search_summary(
             },
         )
 
+        now = self._now()
         records = []
         for r in results:
+            if not is_retrievable(r.metadata, now):
+                continue
             records.append(SourceRecord(
                 domain="summary",
                 content=r.content,
@@ -503,10 +513,15 @@ def _fetch_profile_catalog(self, user_id: str):
             logger.warning("Failed to fetch profile catalog: %s", exc)
             return [], []
 
+        now = self._now()
         catalog: List[Dict[str, str]] = []
         seen = set()
+        live_results = []
 
         for r in results:
+            if not is_retrievable(r.metadata, now):
+                continue
+            live_results.append(r)
             main_content = r.metadata.get("main_content", "")
             if not main_content or main_content in seen:
                 continue
@@ -524,7 +539,7 @@ def _fetch_profile_catalog(self, user_id: str):
                     "sub_topic": "",
                 })
 
-        return catalog, results
+        return catalog, live_results
 
     def _format_catalog(self, catalog: List[Dict[str, str]]) -> str:
         """Format profile catalog for the system prompt."""

src/pipelines/weaver.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import asyncio
+from datetime import datetime, timezone
 from functools import partial
 import logging
 from typing import Any, Callable, Dict, List, Optional
@@ -63,6 +64,7 @@ def __init__(
         code_vector_store: Optional[BaseVectorStore] = None,
         graph_create_annotation: Optional[GraphCreateAnnotationFn] = None,
         snippet_vector_store: Optional[BaseVectorStore] = None,
+        _now: Optional[Callable[[], datetime]] = None,
     ) -> None:
         self.vector_store = vector_store
         self.embed_fn = embed_fn
@@ -72,6 +74,7 @@ def __init__(
         self.code_vector_store = code_vector_store
         self.graph_create_annotation = graph_create_annotation
         self.snippet_vector_store = snippet_vector_store
+        self._now: Callable[[], datetime] = _now or (lambda: datetime.now(timezone.utc))
 
     # ------------------------------------------------------------------
     # Public entry point
@@ -82,6 +85,7 @@ async def execute(
         judge_result: JudgeResult,
         domain: JudgeDomain,
         user_id: str,
+        extra_metadata: Optional[Dict[str, Any]] = None,
     ) -> WeaverResult:
         result = WeaverResult()
 
@@ -91,7 +95,9 @@ async def execute(
 
         # Optimization: Batch vector operations if possible
         if domain not in (JudgeDomain.TEMPORAL, JudgeDomain.CODE, JudgeDomain.SNIPPET) and self.vector_store:
-            batched_executed = await self._execute_batched_vector(judge_result.operations, domain, user_id)
+            batched_executed = await self._execute_batched_vector(
+                judge_result.operations, domain, user_id, extra_metadata=extra_metadata
+            )
             result.executed.extend(batched_executed)
         else:
             for op in judge_result.operations:
@@ -106,6 +112,7 @@ async def _execute_batched_vector(
         operations: List[Operation],
         domain: JudgeDomain,
         user_id: str,
+        extra_metadata: Optional[Dict[str, Any]] = None,
     ) -> List[ExecutedOp]:
         """Batch ADD and DELETE operations to reduce vector store round-trips."""
         executed_ops: List[ExecutedOp] = []
@@ -142,6 +149,8 @@ async def flush_add_batch():
                 try:
                     meta = {"user_id": user_id, "domain": domain.value}
                     meta.update(_extract_structured_metadata(op.content))
+                    if extra_metadata:
+                        meta.update(extra_metadata)
 
                     valid_ops.append(op)
                     texts.append(op.content)
@@ -371,7 +380,11 @@ async def _execute_vector(
             )
 
     async def _vector_add(
-        self, op: Operation, domain: JudgeDomain, user_id: str,
+        self,
+        op: Operation,
+        domain: JudgeDomain,
+        user_id: str,
+        extra_metadata: Optional[Dict[str, Any]] = None,
     ) -> ExecutedOp:
         if not self.embed_fn:
             return ExecutedOp(
@@ -385,6 +398,8 @@ async def _vector_add(
         # Store structured metadata for deterministic lookups
         structured = _extract_structured_metadata(op.content)
         metadata.update(structured)
+        if extra_metadata:
+            metadata.update(extra_metadata)
 
         ids = self.vector_store.add(
             texts=[op.content],