Improving the extraction prompt with Agent-sourced facts, event-time capture

CHANGELOG.md

@@ -5,13 +5,18 @@
 #### Features Added
 * Write-time in-place deduplication: near-duplicate memories fold into the existing record (same id, newer content) instead of creating a new doc. See [PR:#26](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/26)
 * Reconciliation now resolves contradictions only, soft-deleting the loser with `superseded_by`; `get_memory_history()` walks that chain. See [PR:#26](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/26)
+* Agent-sourced facts: extraction can now capture the assistant's own actions and recommendations (not just user statements), tagged `source=agent` / `sys:agent-fact` so retrieval can include or exclude them. See [PR:#31](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/31)
+* Event time on write: `add_cosmos(..., created_at=...)` sets a memory's event time (falls back to ingestion time), enabling time-aware retrieval and temporal reasoning over backfilled conversations. See [PR:#31](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/31)
+* Unified retrieval: `search_cosmos(include_turns=True)` blends raw conversation turns into results alongside extracted memories. See [PR:#31](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/31)
+* `DEDUP_VECTOR_ENABLED` is now an environment knob (default `false` = add-only) instead of a fixed internal constant. See [PR:#31](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/31)
 
 #### Bugs Fixed
 * Fixed a re-extraction loop that re-extracted the whole conversation every cycle (turns were never stamped `extracted_at` when vector dedup was on). See [PR:#26](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/26)
 
 #### Other Changes
 * Reworked the extraction prompt (anti-inference, preserve specifics, topic-grouped memories) and simplified the schema to `fact`/`episodic` with fixed fact categories. See [PR:#26](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/26)
 * Token-bounded extraction batches with per-batch failure isolation; embedding inputs truncated to the model token budget. See [PR:#26](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/26)
+* Extraction now sees per-turn timestamps and resolves relative dates ("3 weeks ago") to absolute dates in the fact text (time-range filtering uses the memory's `created_at`). See [PR:#31](https://github.com/AzureCosmosDB/AgentMemoryToolkit/pull/31)
 
 ## [0.2.0b3] (2026-07-08)
 

Docs/concepts.md

@@ -96,6 +96,10 @@ Facts work especially well for vector search because each fact is stored as a sm
 
 By default raw conversation turns are *not* embedded — only derived memories (facts, episodic, procedural, summaries) carry vectors. Set `enable_turn_embeddings=True` (env `ENABLE_TURN_EMBEDDINGS`) to also embed turns on write, then call `search_turns()` to vector-search the raw conversation log. The turns container is always provisioned with a `quantizedFlat` vector index, so this flag only toggles embedding generation and can be turned on or off at any time without recreating the container.
 
+### Unified retrieval (`include_turns`)
+
+`search_cosmos(..., include_turns=True)` returns extracted memories plus raw conversation turns in one call — useful for recovering detail that extraction dropped. Up to `turn_top_k` turns (default `top_k`) are **appended after** the memory hits, so memory results keep priority; the two sets are not score-fused. A turn is skipped only when its content is an **exact** string match of a returned memory — paraphrased overlaps are not de-duplicated. The turn search requires `enable_turn_embeddings` and is best-effort: if it fails, the memory results are returned unchanged.
+
 ---
 
 ## Processing Pipeline
@@ -159,12 +163,13 @@ Extraction only ever reads turns not yet stamped `extracted_at`; `persist` stamp
 
 ### Tunable
 
-Only two reconcile knobs are operator-configurable:
+Only three reconcile knobs are operator-configurable:
 
 - `DEDUP_EVERY_N` (default `5`) — how often reconcile runs in the auto-trigger path (every Nth **extract**, not every Nth turn). Set to `0` to disable.
 - `DEDUP_POOL_SIZE` (default `50`, hard cap `500`) — the pool size `n` passed to `reconcile_memories`; also overridable per call. Larger values give the LLM a wider view at higher token cost.
+- `DEDUP_VECTOR_ENABLED` (default `false`) — write-time in-place near-duplicate folding. Default off = **add-only**. Set to `true` to fold near-duplicate restatements into their canonical record at write time.
+
 
-The similarity threshold (`DEDUP_SIM_HIGH`, `0.97`) and the vector-fold on/off switch (`DEDUP_VECTOR_ENABLED`) ship as **fixed internal constants** in `azure.cosmos.agent_memory.thresholds` — they have no env plumbing and ignore any environment variable, so the write-time fold behavior is not operator-tunable today.
 
 > **Indexing note.** The reconcile pool query orders by `created_at` (matching the prompt's "more recent first" tiebreaker). Cosmos's default indexing policy includes every property, so this works out of the box. If you customize the indexing policy to reduce write RU, ensure `/created_at/?` remains indexed or the query will fail with a 400 (`Order-by over a non-indexed path`).
 

azure/cosmos/agent_memory/_utils.py

@@ -108,6 +108,29 @@ def _coerce_datetime_iso(value: Optional[str | datetime]) -> Optional[str]:
     return value
 
 
+def normalize_created_at_iso(value: Optional[str | datetime]) -> Optional[str]:
+    """Normalize a caller-supplied event time to a tz-aware **UTC** ISO-8601 string."""
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        dt = value
+    else:
+        text = str(value).strip()
+        if not text:
+            raise ValidationError("created_at must be a non-empty ISO-8601 string or datetime")
+        try:
+            # ``fromisoformat`` doesn't accept a trailing 'Z' before 3.11 in all
+            # forms; normalize it to an explicit UTC offset first.
+            dt = datetime.fromisoformat(text.replace("Z", "+00:00"))
+        except ValueError as exc:
+            raise ValidationError(f"created_at is not a valid ISO-8601 timestamp: {value!r}") from exc
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    else:
+        dt = dt.astimezone(timezone.utc)
+    return dt.isoformat()
+
+
 def _normalize_for_hash(text: str) -> str:
     """Lowercase + collapse whitespace for write-time exact-dedup.
 

azure/cosmos/agent_memory/aio/cosmos_memory_client.py

@@ -585,6 +585,7 @@ async def add_cosmos(
         salience: Optional[float] = None,
         embedding: Optional[list[float]] = None,
         embed: Optional[bool] = None,
+        created_at: Optional[str | datetime] = None,
     ) -> str:
         """Add a memory directly to Cosmos DB, bypassing the local buffer.
 
@@ -612,6 +613,7 @@ async def add_cosmos(
             salience,
             embedding,
             embed,
+            created_at,
         )
         if memory_type == "turn" and thread_id:
             task = asyncio.create_task(self._maybe_auto_trigger({(user_id, thread_id): 1}))
@@ -717,8 +719,12 @@ async def search_cosmos(
         min_confidence: Optional[float] = None,
         created_after: Optional[str | datetime] = None,
         created_before: Optional[str | datetime] = None,
+        include_turns: bool = False,
+        turn_top_k: Optional[int] = None,
     ) -> list[dict[str, Any]]:
-        return await self._get_store().search(
+        """Search memories using vector similarity; optionally blend raw turns."""
+        store = self._get_store()
+        results = await store.search(
             search_terms=search_terms,
             memory_id=memory_id,
             user_id=user_id,
@@ -735,6 +741,30 @@ async def search_cosmos(
             created_after=created_after,
             created_before=created_before,
         )
+        if not include_turns or not user_id:
+            return results
+
+        seen_content = {str(r.get("content") or "").strip() for r in results}
+        try:
+            turns = await store.search_turns(
+                search_terms=search_terms,
+                user_id=user_id,
+                thread_id=thread_id,
+                role=role,
+                top_k=turn_top_k if turn_top_k is not None else top_k,
+                exclude_tags=exclude_tags,
+                created_after=created_after,
+                created_before=created_before,
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("search_cosmos: include_turns turn search failed (%s); returning memories only", exc)
+            return results
+        for turn in turns:
+            content = str(turn.get("content") or "").strip()
+            if content and content not in seen_content:
+                seen_content.add(content)
+                results.append(turn)
+        return results
 
     async def search_turns(
         self,

azure/cosmos/agent_memory/aio/services/pipeline.py

@@ -398,13 +398,15 @@ def _build_transcript(
         items: list[dict[str, Any]],
         *,
         group_by_thread: bool = False,
+        include_timestamp: bool = False,
     ) -> str:
         # getattr fallback covers unit tests that build AsyncPipelineService
         # via __new__ to bypass __init__ (and therefore the metadata-keys stash).
         return build_transcript(
             items,
             group_by_thread=group_by_thread,
             metadata_keys=getattr(self, "_transcript_metadata_keys", None),
+            include_timestamp=include_timestamp,
         )
 
     async def _load_existing_memories(
@@ -547,7 +549,7 @@ async def extract_memories_dry(
         deferred_turn_count = 0
         quarantined_turn_count = 0
         for batch in batches:
-            batch_transcript = self._build_transcript(batch)
+            batch_transcript = self._build_transcript(batch, include_timestamp=True)
             try:
                 response_text = await self._run_prompty(
                     "extract_memories.prompty", inputs={"transcript": batch_transcript}
@@ -607,6 +609,15 @@ async def extract_memories_dry(
             seed = _ID_SEED_SEP.join((user_id, thread_id, new_content_hash))
             det_id = f"fact_{hashlib.sha256(seed.encode()).hexdigest()[:32]}"
             topic_tags = build_topic_tags(fact.get("tags", []))
+            raw_source = fact.get("source")
+            if raw_source is not None and raw_source not in ("user", "agent"):
+                logger.debug(
+                    "extract_memories: coercing invalid fact source=%r to 'user' user_id=%s",
+                    raw_source,
+                    user_id,
+                )
+            fact_source = raw_source if raw_source in ("user", "agent") else "user"
+            source_tags = ["sys:agent-fact"] if fact_source == "agent" else []
             confidence = fact.get("confidence")
             doc: dict[str, Any] = {
                 "id": det_id,
@@ -621,9 +632,10 @@ async def extract_memories_dry(
                 "metadata": {
                     "category": fact.get("category") or "other",
                     "temporal_context": fact.get("temporal_context"),
+                    "source": fact_source,
                 },
                 "salience": fact.get("salience") if fact.get("salience") is not None else 0.5,
-                "tags": ["sys:fact", "sys:auto-extracted"] + topic_tags,
+                "tags": ["sys:fact", "sys:auto-extracted"] + source_tags + topic_tags,
                 "created_at": doc_timestamp,
                 "updated_at": doc_timestamp,
             }
@@ -860,10 +872,27 @@ async def _nearest_active_full(
         return None, 0.0
 
     async def _apply_inplace_update(self, neighbor: dict[str, Any], new_doc: dict[str, Any]) -> bool:
-        """Async mirror of the sync in-place refresh (recency-wins content+embedding)."""
+        """Async mirror of the sync in-place refresh (recency-wins content+embedding).
+
+        Folds only within the same ``metadata.source`` (user vs agent); a
+        cross-source pair returns False so the caller keeps it as a novel ADD,
+        preventing tag/source desync.
+        """
         from azure.core import MatchConditions
         from azure.cosmos.exceptions import CosmosAccessConditionFailedError
 
+        neighbor_source = (neighbor.get("metadata") or {}).get("source") or "user"
+        new_source = (new_doc.get("metadata") or {}).get("source") or "user"
+        if neighbor_source != new_source:
+            logger.info(
+                "in-place dedup update skipped (source mismatch neighbor=%s new=%s) "
+                "target_id=%s; keeping new doc as novel",
+                neighbor_source,
+                new_source,
+                neighbor.get("id"),
+            )
+            return False
+
         try:
             old_etag = neighbor.get("_etag")
             updated = dict(neighbor)
@@ -1563,11 +1592,15 @@ def _emit_reconcile_outcome(
 
     async def _active_memories_for_reconcile(self, user_id: str, memory_type: str, n: int) -> list[dict[str, Any]]:
         capped_n = top_literal(n, name="reconcile_memories.n")
+        # Agent-sourced facts (sys:agent-fact) are excluded: they record what the
+        # agent did/recommended (historical events), not mutable user state, so
+        # they must never be contradiction-superseded by a later user statement.
         query = (
             f"SELECT TOP {capped_n} * FROM c "
             "WHERE c.user_id = @user_id "
             "AND c.type = @memory_type "
             f"AND {_ACTIVE_DOC_FILTER} "
+            "AND NOT ARRAY_CONTAINS(c.tags, 'sys:agent-fact') "
             "ORDER BY c.created_at DESC"
         )
         return await self._query_items(

azure/cosmos/agent_memory/aio/store/memory_store.py

@@ -20,6 +20,7 @@
     compute_content_hash,
     extract_keywords,
     new_id,
+    normalize_created_at_iso,
 )
 from azure.cosmos.agent_memory.exceptions import (
     ConfigurationError,
@@ -168,6 +169,7 @@ async def add(
         salience: Optional[float] = None,
         embedding: Optional[list[float]] = None,
         embed: Optional[bool] = None,
+        created_at: Optional[str | datetime] = None,
     ) -> str:
         """Add a memory document to Cosmos DB and return its id."""
         kwargs: dict[str, Any] = {
@@ -185,6 +187,8 @@ async def add(
             kwargs["ttl"] = ttl
         if salience is not None:
             kwargs["salience"] = salience
+        if created_at is not None:
+            kwargs["created_at"] = normalize_created_at_iso(created_at)
         if memory_type != "turn":
             kwargs.setdefault("content_hash", compute_content_hash(content))
             kwargs.setdefault("prompt_id", "manual:add")

azure/cosmos/agent_memory/cosmos_memory_client.py

@@ -537,6 +537,7 @@ def add_cosmos(
         salience: Optional[float] = None,
         embedding: Optional[list[float]] = None,
         embed: Optional[bool] = None,
+        created_at: Optional[str | datetime] = None,
     ) -> str:
         """Add a memory directly to Cosmos DB, bypassing the local buffer.
 
@@ -565,6 +566,7 @@ def add_cosmos(
             salience,
             embedding,
             embed,
+            created_at,
         )
         if memory_type == "turn" and thread_id:
             try:
@@ -672,13 +674,12 @@ def search_cosmos(
         min_confidence: Optional[float] = None,
         created_after: Optional[str | datetime] = None,
         created_before: Optional[str | datetime] = None,
+        include_turns: bool = False,
+        turn_top_k: Optional[int] = None,
     ) -> list[dict[str, Any]]:
-        """Search memories in Cosmos DB using vector similarity.
-
-        Searches the derived memories container (facts/episodic/procedural). Use
-        :meth:`search_turns` to vector-search the raw conversation log instead.
-        """
-        return self._get_store().search(
+        """Search memories using vector similarity; optionally blend raw turns."""
+        store = self._get_store()
+        results = store.search(
             search_terms=search_terms,
             memory_id=memory_id,
             user_id=user_id,
@@ -695,6 +696,30 @@ def search_cosmos(
             created_after=created_after,
             created_before=created_before,
         )
+        if not include_turns or not user_id:
+            return results
+
+        seen_content = {str(r.get("content") or "").strip() for r in results}
+        try:
+            turns = store.search_turns(
+                search_terms=search_terms,
+                user_id=user_id,
+                thread_id=thread_id,
+                role=role,
+                top_k=turn_top_k if turn_top_k is not None else top_k,
+                exclude_tags=exclude_tags,
+                created_after=created_after,
+                created_before=created_before,
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("search_cosmos: include_turns turn search failed (%s); returning memories only", exc)
+            return results
+        for turn in turns:
+            content = str(turn.get("content") or "").strip()
+            if content and content not in seen_content:
+                seen_content.add(content)
+                results.append(turn)
+        return results
 
     def search_turns(
         self,

azure/cosmos/agent_memory/prompts/_schemas.py

@@ -52,7 +52,7 @@
 
 
 # ---------------------------------------------------------------------------
-# extract_memories.prompty - extract facts + episodic + unclassified
+# extract_memories.prompty - extract facts (user- or agent-sourced) + episodic
 # ---------------------------------------------------------------------------
 _FACT_ITEM = {
     "type": "object",
@@ -67,6 +67,10 @@
                 "other",
             ],
         },
+        "source": {
+            "type": "string",
+            "enum": ["user", "agent"],
+        },
         "confidence": {"type": "number"},
         "salience": {"type": "number"},
         "temporal_context": {"type": ["string", "null"]},
@@ -75,6 +79,7 @@
     "required": [
         "text",
         "category",
+        "source",
         "confidence",
         "salience",
         "temporal_context",