test: update consolidation test for source_memory_ids behavior

Updated test_consolidation_creates_memory_links to test_consolidation_uses_source_memory_ids to reflect the new behavior where observations use source_memory_ids instead of memory_links for traversal. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <[email protected]>
fix: observations rely on source_memory_ids, no link copying
2026-01-28 12:20:37 +01:00 · 2026-01-28 12:08:27 +01:00 · 2026-01-28 11:12:27 +01:00 · 2026-01-28 09:54:20 +01:00
13 changed files with 362 additions and 296 deletions
@@ -173,7 +173,7 @@ RETAIN_EXTRACTION_MODES = ("concise", "verbose")  # Allowed extraction modes
 DEFAULT_RETAIN_OBSERVATIONS_ASYNC = False  # Run observation generation async (after retain completes)

 # Observations defaults (consolidated knowledge from facts)
-DEFAULT_ENABLE_OBSERVATIONS = False  # Observations disabled by default (experimental)
+DEFAULT_ENABLE_OBSERVATIONS = True  # Observations enabled by default
 DEFAULT_CONSOLIDATION_BATCH_SIZE = 50  # Memories to load per batch (internal memory optimization)

 # Database migrations
@@ -254,11 +254,79 @@ async def run_consolidation_job(
    if timing_parts:
        perf.log(f"[4] Timing breakdown: {', '.join(timing_parts)}")

+    # Trigger mental model refreshes for models with refresh_after_consolidation=true
+    mental_models_refreshed = await _trigger_mental_model_refreshes(
+        memory_engine=memory_engine,
+        bank_id=bank_id,
+        request_context=request_context,
+        perf=perf,
+    )
+    stats["mental_models_refreshed"] = mental_models_refreshed
+
    perf.flush()

    return {"status": "completed", "bank_id": bank_id, **stats}


+async def _trigger_mental_model_refreshes(
+    memory_engine: "MemoryEngine",
+    bank_id: str,
+    request_context: "RequestContext",
+    perf: ConsolidationPerfLog | None = None,
+) -> int:
+    """
+    Trigger refreshes for mental models with refresh_after_consolidation=true.
+
+    Args:
+        memory_engine: MemoryEngine instance
+        bank_id: Bank identifier
+        request_context: Request context for authentication
+        perf: Performance logging
+
+    Returns:
+        Number of mental models scheduled for refresh
+    """
+    pool = memory_engine._pool
+
+    # Find mental models with refresh_after_consolidation=true
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            f"""
+            SELECT id, name
+            FROM {fq_table("mental_models")}
+            WHERE bank_id = $1
+              AND (trigger->>'refresh_after_consolidation')::boolean = true
+            """,
+            bank_id,
+        )
+
+    if not rows:
+        return 0
+
+    if perf:
+        perf.log(f"[5] Triggering refresh for {len(rows)} mental models with refresh_after_consolidation=true")
+
+    # Submit refresh tasks for each mental model
+    refreshed_count = 0
+    for row in rows:
+        mental_model_id = row["id"]
+        try:
+            await memory_engine.submit_async_refresh_mental_model(
+                bank_id=bank_id,
+                mental_model_id=mental_model_id,
+                request_context=request_context,
+            )
+            refreshed_count += 1
+            logger.info(
+                f"[CONSOLIDATION] Triggered refresh for mental model {mental_model_id} "
+                f"(name: {row['name']}) in bank {bank_id}"
+            )
+        except Exception as e:
+            logger.warning(f"[CONSOLIDATION] Failed to trigger refresh for mental model {mental_model_id}: {e}")
+
+    return refreshed_count
+
+
 async def _process_memory(
    conn: "Connection",
    memory_engine: "MemoryEngine",
@@ -301,11 +369,11 @@ async def _process_memory(
        perf.record_timing("recall", time.time() - t0)

    # Single LLM call handles ALL cases (with or without existing observations)
+    # Note: Tags are NOT passed to LLM - they are handled algorithmically
    t0 = time.time()
    actions = await _consolidate_with_llm(
        memory_engine=memory_engine,
        fact_text=fact_text,
-        fact_tags=fact_tags,
        observations=related_observations,  # Can be empty list
        mission=mission,
    )
@@ -328,6 +396,7 @@ async def _process_memory(
                memory_id=memory_id,
                action=action,
                observations=related_observations,
+                source_fact_tags=fact_tags,  # Pass source fact's tags for security
                source_occurred_start=memory.get("occurred_start"),
                source_occurred_end=memory.get("occurred_end"),
                source_mentioned_at=memory.get("mentioned_at"),
@@ -341,6 +410,7 @@ async def _process_memory(
                bank_id=bank_id,
                memory_id=memory_id,
                action=action,
+                source_fact_tags=fact_tags,  # Pass source fact's tags for security
                event_date=memory.get("event_date"),
                occurred_start=memory.get("occurred_start"),
                occurred_end=memory.get("occurred_end"),
@@ -377,6 +447,7 @@ async def _execute_update_action(
    memory_id: uuid.UUID,
    action: dict[str, Any],
    observations: list[dict[str, Any]],
+    source_fact_tags: list[str] | None = None,
    source_occurred_start: datetime | None = None,
    source_occurred_end: datetime | None = None,
    source_mentioned_at: datetime | None = None,
@@ -390,6 +461,11 @@ async def _execute_update_action(
    - occurred_start: uses LEAST to keep the earliest start time
    - occurred_end: uses GREATEST to keep the most recent end time
    - mentioned_at: uses GREATEST to keep the most recent mention time
+
+    SECURITY: Merges source fact's tags into the observation's existing tags.
+    This ensures all contributors can see the observation they contributed to.
+    For example, if Lisa's observation (tags=['user_lisa']) is updated with
+    Mike's fact (tags=['user_mike']), the observation will have both tags.
    """
    learning_id = action.get("learning_id")
    new_text = action.get("text")
@@ -418,6 +494,17 @@ async def _execute_update_action(
    source_ids = list(model.get("source_memory_ids", []))
    source_ids.append(memory_id)

+    # SECURITY: Merge source fact's tags into existing observation tags
+    # This ensures all contributors can see the observation they contributed to
+    existing_tags = set(model.get("tags", []) or [])
+    source_tags = set(source_fact_tags or [])
+    merged_tags = list(existing_tags | source_tags)  # Union of both tag sets
+    if source_tags and source_tags != existing_tags:
+        logger.debug(
+            f"Security: Merging tags for observation {learning_id}: "
+            f"existing={list(existing_tags)}, source={list(source_tags)}, merged={merged_tags}"
+        )
+
    # Generate new embedding for updated text
    t0 = time.time()
    embeddings = await embedding_utils.generate_embeddings_batch(memory_engine.embeddings, [new_text])
@@ -429,6 +516,7 @@ async def _execute_update_action(
    # - occurred_start: LEAST keeps the earliest start time across all source facts
    # - occurred_end: GREATEST keeps the most recent end time across all source facts
    # - mentioned_at: GREATEST keeps the most recent mention time
+    # - tags: merged from existing + source fact (for visibility)
    t0 = time.time()
    await conn.execute(
        f"""
@@ -438,6 +526,7 @@ async def _execute_update_action(
            history = $3,
            source_memory_ids = $4,
            proof_count = $5,
+            tags = $10,
            updated_at = now(),
            occurred_start = LEAST(occurred_start, COALESCE($7, occurred_start)),
            occurred_end = GREATEST(occurred_end, COALESCE($8, occurred_end)),
@@ -453,6 +542,7 @@ async def _execute_update_action(
        source_occurred_start,
        source_occurred_end,
        source_mentioned_at,
+        merged_tags,
    )

    # Create links from memory to observation
@@ -471,6 +561,7 @@ async def _execute_create_action(
    bank_id: str,
    memory_id: uuid.UUID,
    action: dict[str, Any],
+    source_fact_tags: list[str] | None = None,
    event_date: datetime | None = None,
    occurred_start: datetime | None = None,
    occurred_end: datetime | None = None,
@@ -480,11 +571,18 @@ async def _execute_create_action(
    """
    Execute a create action for a new observation.

-    Creates a new observation with the specified text and tags.
+    Creates a new observation with the specified text.
    The text comes directly from the classify LLM - no second LLM call needed.
+
+    Tags are determined algorithmically (not by LLM):
+    - Observations always inherit their source fact's tags
+    - This ensures visibility scope is maintained (security)
    """
    text = action.get("text")
-    tags = action.get("tags", [])
+
+    # Tags are determined algorithmically - always use source fact's tags
+    # This ensures private memories create private observations
+    tags = source_fact_tags or []

    if not text:
        return {"action": "skipped", "reason": "missing_text"}
@@ -515,83 +613,22 @@ async def _create_memory_links(
    observation_id: uuid.UUID,
 ) -> None:
    """
-    Create links between a source memory and its observation.
+    Placeholder for observation link creation.

-    This:
-    1. Creates bidirectional semantic links between memory and observation
-    2. Copies existing memory_links from the source memory to the observation
+    Observations do NOT get any memory_links copied from their source facts.
+    Instead, retrieval uses source_memory_ids to traverse:
+    - Entity connections: observation → source_memory_ids → unit_entities
+    - Semantic similarity: observations have their own embeddings
+    - Temporal proximity: observations have their own temporal fields

-    Note: We intentionally do NOT copy entity links (unit_entities) to observations.
-    Instead, the retriever traverses through source_memory_ids to find entity
-    connections. This avoids duplicating entity data and ensures observations
-    are connected via their source facts' entity relationships.
+    This avoids data duplication and ensures observations are always
+    connected via their source facts' relationships.

-    Note: Uses EXISTS checks to handle the case where source memory was deleted
-    by a concurrent operation between fetching and link creation.
+    The memory_id and observation_id parameters are kept for interface
+    compatibility but no links are created.
    """
-    mu_table = fq_table("memory_units")
-    ml_table = fq_table("memory_links")
-
-    # 1. Bidirectional link between memory and observation
-    # Only insert if both units exist (handles concurrent deletion)
-    await conn.execute(
-        f"""
-        INSERT INTO {ml_table} (from_unit_id, to_unit_id, link_type, weight)
-        SELECT $1, $2, 'semantic', 1.0
-        WHERE EXISTS (SELECT 1 FROM {mu_table} WHERE id = $1)
-          AND EXISTS (SELECT 1 FROM {mu_table} WHERE id = $2)
-        ON CONFLICT DO NOTHING
-        """,
-        memory_id,
-        observation_id,
-    )
-    await conn.execute(
-        f"""
-        INSERT INTO {ml_table} (from_unit_id, to_unit_id, link_type, weight)
-        SELECT $1, $2, 'semantic', 1.0
-        WHERE EXISTS (SELECT 1 FROM {mu_table} WHERE id = $1)
-          AND EXISTS (SELECT 1 FROM {mu_table} WHERE id = $2)
-        ON CONFLICT DO NOTHING
-        """,
-        observation_id,
-        memory_id,
-    )
-
-    # 2. Copy outgoing memory_links from source memory to observation
-    # If source memory links to X, observation should also link to X
-    await conn.execute(
-        f"""
-        INSERT INTO {ml_table} (from_unit_id, to_unit_id, link_type, entity_id, weight)
-        SELECT $1, ml.to_unit_id, ml.link_type, ml.entity_id, ml.weight
-        FROM {ml_table} ml
-        WHERE ml.from_unit_id = $2 AND ml.to_unit_id != $1
-          AND EXISTS (SELECT 1 FROM {mu_table} WHERE id = $1)
-          AND EXISTS (SELECT 1 FROM {mu_table} WHERE id = ml.to_unit_id)
-        ON CONFLICT DO NOTHING
-        """,
-        observation_id,
-        memory_id,
-    )
-
-    # 3. Copy incoming memory_links from source memory to observation
-    # If X links to source memory, X should also link to observation
-    await conn.execute(
-        f"""
-        INSERT INTO {ml_table} (from_unit_id, to_unit_id, link_type, entity_id, weight)
-        SELECT ml.from_unit_id, $1, ml.link_type, ml.entity_id, ml.weight
-        FROM {ml_table} ml
-        WHERE ml.to_unit_id = $2 AND ml.from_unit_id != $1
-          AND EXISTS (SELECT 1 FROM {mu_table} WHERE id = $1)
-          AND EXISTS (SELECT 1 FROM {mu_table} WHERE id = ml.from_unit_id)
-        ON CONFLICT DO NOTHING
-        """,
-        observation_id,
-        memory_id,
-    )
-
-    # Note: Entity links (unit_entities) are NOT copied to observations.
-    # The retriever uses source_memory_ids to traverse through source facts'
-    # entity connections, avoiding data duplication.
+    # No links are created - observations rely on source_memory_ids for traversal
+    pass


 async def _find_related_observations(
@@ -674,7 +711,6 @@ async def _find_related_observations(
 async def _consolidate_with_llm(
    memory_engine: "MemoryEngine",
    fact_text: str,
-    fact_tags: list[str],
    observations: list[dict[str, Any]],
    mission: str,
 ) -> list[dict[str, Any]]:
@@ -686,10 +722,14 @@ async def _consolidate_with_llm(
    - Related observations exist: compares and returns update/create actions
    - Purely ephemeral fact: returns empty array

+    Note: Tags are NOT handled by the LLM. They are determined algorithmically:
+    - CREATE: observation inherits source fact's tags
+    - UPDATE: observation merges source fact's tags with existing tags
+
    Returns:
        List of actions, each being:
        - {"action": "update", "learning_id": "uuid", "text": "...", "reason": "..."}
-        - {"action": "create", "tags": [...], "text": "...", "reason": "..."}
+        - {"action": "create", "text": "...", "reason": "..."}
        - [] if fact is purely ephemeral (no durable knowledge)
    """
    # Format observations WITH their tags (or "None" if empty)
@@ -713,7 +753,6 @@ Focus on DURABLE knowledge that serves this mission, not ephemeral state.
    user_prompt = CONSOLIDATION_USER_PROMPT.format(
        mission_section=mission_section,
        fact_text=fact_text,
-        fact_tags=json.dumps(fact_tags),
        observations_text=observations_text,
    )

@@ -35,38 +35,17 @@ BAD examples:
 2. CONTRADICTION: Opposite information about same topic → update with history (e.g., "used to X, now Y")
 3. UPDATE: New state replacing old state → update with history

-## TAG ROUTING RULES:
-Tags define visibility scopes. The fact and each observation have tags (can be empty = global).
-
-| Fact Tags | Obs Tags | Action |
-|-----------|----------|--------|
-| [alice] | [alice] | UPDATE the observation (same scope) |
-| [alice] | [] | UPDATE the observation (global absorbs all scopes) |
-| [alice] | [bob] | CREATE new untagged observation (cross-scope insight) |
-| [] | [alice] | UPDATE the observation (untagged facts can update any scope) |
-| [] | [] | UPDATE the observation (global to global) |
-
-When NO existing observation matches the fact's topic: CREATE new observation with fact's tags.
-
-## MULTIPLE ACTIONS:
-One fact can trigger MULTIPLE actions. For example:
- Update a scoped observation [alice] about pizza preferences
- AND update a global observation [] about pizza in general
-
-Output an ARRAY of actions (can be empty, one, or many).
-
 ## CRITICAL RULES:
 - NEVER merge facts about DIFFERENT people
 - NEVER merge unrelated topics (food preferences vs work vs hobbies)
 - When merging contradictions, capture the CHANGE (before → after)
 - Keep observations focused on ONE specific topic per person
- Cross-scope insights (alice's fact about bob's topic) become UNTAGGED (global)
- The "text" field MUST contain durable knowledge, not ephemeral state"""
+- The "text" field MUST contain durable knowledge, not ephemeral state
+- Do NOT include "tags" in output - tags are handled automatically"""

 CONSOLIDATION_USER_PROMPT = """Analyze this new fact and consolidate into knowledge.
 {mission_section}
 NEW FACT: {fact_text}
-FACT TAGS: {fact_tags}

 EXISTING OBSERVATIONS:
 {observations_text}
@@ -76,16 +55,15 @@ Instructions:
 2. Then compare with existing observations:
   - If an observation covers the same topic: UPDATE it with the new knowledge
   - If no observation covers the topic: CREATE a new one
-   - If fact is about different scope: apply tag routing rules

 Output JSON array of actions (ALWAYS an array, even for single action):
 [
  {{"action": "update", "learning_id": "uuid", "text": "updated durable knowledge", "reason": "..."}},
-  {{"action": "create", "tags": ["tag"], "text": "new durable knowledge", "reason": "..."}}
+  {{"action": "create", "text": "new durable knowledge", "reason": "..."}}
 ]

 If NO consolidation is needed (fact is purely ephemeral with no durable knowledge):
 []

 If no observations exist and fact contains durable knowledge:
-[{{"action": "create", "tags": {fact_tags}, "text": "durable knowledge text", "reason": "new topic"}}]"""
+[{{"action": "create", "text": "durable knowledge text", "reason": "new topic"}}]"""
@@ -3580,6 +3580,16 @@ class MemoryEngine(MemoryEngineInterface):
        if directives:
            logger.info(f"[REFLECT {reflect_id}] Loaded {len(directives)} directives")

+        # Check if the bank has any mental models
+        async with pool.acquire() as conn:
+            mental_model_count = await conn.fetchval(
+                f"SELECT COUNT(*) FROM {fq_table('mental_models')} WHERE bank_id = $1",
+                bank_id,
+            )
+        has_mental_models = mental_model_count > 0
+        if has_mental_models:
+            logger.info(f"[REFLECT {reflect_id}] Bank has {mental_model_count} mental models")
+
        # Run the agent
        agent_result = await run_reflect_agent(
            llm_config=self._reflect_llm_config,
@@ -3595,6 +3605,8 @@ class MemoryEngine(MemoryEngineInterface):
            max_tokens=max_tokens,
            response_schema=response_schema,
            directives=directives,
+            has_mental_models=has_mental_models,
+            budget=effective_budget,
        )

        total_time = time.time() - reflect_start
@@ -80,6 +80,18 @@ def _is_done_tool(name: str) -> bool:
 # Pattern to match done() call as text - handles done({...}) with nested JSON
 _DONE_CALL_PATTERN = re.compile(r"done\s*\(\s*\{.*$", re.DOTALL)

+# Patterns for leaked structured output in the answer field
+_LEAKED_JSON_SUFFIX = re.compile(
+    r'\s*```(?:json)?\s*\{[^}]*(?:"(?:observation_ids|memory_ids|mental_model_ids)"|\})\s*```\s*$',
+    re.DOTALL | re.IGNORECASE,
+)
+_LEAKED_JSON_OBJECT = re.compile(
+    r'\s*\{[^{]*"(?:observation_ids|memory_ids|mental_model_ids|answer)"[^}]*\}\s*$', re.DOTALL
+)
+_TRAILING_IDS_PATTERN = re.compile(
+    r"\s*(?:observation_ids|memory_ids|mental_model_ids)\s*[=:]\s*\[.*?\]\s*$", re.DOTALL | re.IGNORECASE
+)
+

 def _clean_answer_text(text: str) -> str:
    """Clean up answer text by removing any done() tool call syntax.
@@ -92,6 +104,33 @@ def _clean_answer_text(text: str) -> str:
    return cleaned if cleaned else text


+def _clean_done_answer(text: str) -> str:
+    """Clean up the answer field from a done() tool call.
+
+    Some LLMs leak structured output patterns into the answer text, such as:
+    - JSON code blocks with observation_ids/memory_ids at the end
+    - Raw JSON objects with these fields
+    - Plain text like "observation_ids: [...]"
+
+    This cleans those patterns while preserving the actual answer content.
+    """
+    if not text:
+        return text
+
+    cleaned = text
+
+    # Remove leaked JSON in code blocks at the end
+    cleaned = _LEAKED_JSON_SUFFIX.sub("", cleaned).strip()
+
+    # Remove leaked raw JSON objects at the end
+    cleaned = _LEAKED_JSON_OBJECT.sub("", cleaned).strip()
+
+    # Remove trailing ID patterns
+    cleaned = _TRAILING_IDS_PATTERN.sub("", cleaned).strip()
+
+    return cleaned if cleaned else text
+
+
 async def _generate_structured_output(
    answer: str,
    response_schema: dict,
@@ -141,35 +180,55 @@ async def _generate_structured_output(
            fields[field_name] = (field_type, default)

        if not fields:
-            return None
+            logger.warning(f"[REFLECT {reflect_id}] No fields found in response_schema, skipping structured output")
+            return None, 0, 0

        DynamicModel = create_model("StructuredResponse", **fields)

        # Include the full schema in the prompt for better LLM guidance
        schema_str = json.dumps(response_schema, indent=2)

+        # Build field descriptions for the prompt
+        field_descriptions = []
+        for field_name, field_schema in schema_props.items():
+            field_type = field_schema.get("type", "string")
+            field_desc = field_schema.get("description", "")
+            is_required = field_name in required_fields
+            req_marker = " (REQUIRED)" if is_required else " (optional)"
+            field_descriptions.append(f"- {field_name} ({field_type}){req_marker}: {field_desc}")
+        fields_text = "\n".join(field_descriptions)
+
        # Call LLM with the answer to extract structured data
-        structured_prompt = f"""Based on this answer, extract the information into the requested structured format.
+        structured_prompt = f"""Your task is to extract specific information from the answer below and format it as JSON.

-Answer: {answer}
+ANSWER TO EXTRACT FROM:
+\"\"\"
+{answer}
+\"\"\"

-JSON Schema to follow:
+REQUIRED OUTPUT FORMAT - Extract the following fields from the answer above:
+{fields_text}
+
+JSON Schema:
 ```json
 {schema_str}
 ```

-Return ONLY a valid JSON object that matches this exact schema. Pay special attention to field types:
- "type": "array" means the value must be a JSON array/list, NOT a string
- "type": "string" means the value must be a string
- "type": "object" means the value must be a JSON object
+INSTRUCTIONS:
+1. Read the answer carefully and identify the information that matches each field
+2. Extract the ACTUAL content from the answer - do NOT leave fields empty if information is present
+3. For string fields: use the exact text or a clear summary from the answer
+4. For array fields: return a JSON array (e.g., ["item1", "item2"]), NOT a string
+5. For required fields: you MUST provide a value extracted from the answer
+6. Return ONLY the JSON object, no explanation

-Do not include any explanation, only the JSON object."""
+OUTPUT:"""

        structured_result, usage = await llm_config.call(
            messages=[
                {
                    "role": "system",
-                    "content": "Extract structured data from the given answer. Return only valid JSON matching the provided schema exactly.",
+                    "content": "You are a precise data extraction assistant. Extract information from text and return it as valid JSON matching the provided schema. Always extract actual content - never return empty strings for required fields if information is available.",
                },
                {"role": "user", "content": structured_prompt},
            ],
@@ -188,6 +247,12 @@ Do not include any explanation, only the JSON object."""
            # Try to parse as JSON
            structured_output = json.loads(str(structured_result))

+        # Validate that required fields have non-empty values
+        for field_name in required_fields:
+            value = structured_output.get(field_name)
+            if value is None or value == "" or value == []:
+                logger.warning(f"[REFLECT {reflect_id}] Required field '{field_name}' is empty in structured output")
+
        logger.info(f"[REFLECT {reflect_id}] Generated structured output with {len(structured_output)} fields")
        return structured_output, usage.input_tokens, usage.output_tokens

@@ -722,7 +787,9 @@ async def _process_done_tool(
    """Process the done tool call and return the result."""
    args = done_call.arguments

-    answer = args.get("answer", "").strip()
+    # Extract and clean the answer - some LLMs leak structured output into the answer text
+    raw_answer = args.get("answer", "").strip()
+    answer = _clean_done_answer(raw_answer) if raw_answer else ""
    if not answer:
        answer = "No answer provided."

@@ -268,8 +268,16 @@ class TestConsolidationIntegration:
        await memory.delete_bank(bank_id, request_context=request_context)

    @pytest.mark.asyncio
-    async def test_consolidation_creates_memory_links(self, memory: MemoryEngine, request_context):
-        """Test that observations get bidirectional links to their source memories."""
+    async def test_consolidation_uses_source_memory_ids(self, memory: MemoryEngine, request_context):
+        """Test that observations use source_memory_ids (not memory_links) to track source facts.
+
+        Observations rely on source_memory_ids for traversal:
+        - Entity connections: observation → source_memory_ids → unit_entities
+        - Semantic similarity: observations have their own embeddings
+        - Temporal proximity: observations have their own temporal fields
+
+        No memory_links are created between observations and their source facts.
+        """
        bank_id = f"test-consolidation-links-{uuid.uuid4().hex[:8]}"

        # Create the bank
@@ -282,7 +290,7 @@ class TestConsolidationIntegration:
            request_context=request_context,
        )

-        # Check memory_links between observation and source memory
+        # Check that observation has source_memory_ids but no memory_links
        async with memory._pool.acquire() as conn:
            observation = await conn.fetchrow(
                """
@@ -294,32 +302,35 @@ class TestConsolidationIntegration:
                bank_id,
            )

-            if observation and observation["source_memory_ids"]:
+            if observation:
+                # Observation should have source_memory_ids
+                assert observation["source_memory_ids"] is not None, "Observation should have source_memory_ids"
+                assert len(observation["source_memory_ids"]) > 0, "Observation should have at least one source memory"
+
                source_memory_id = observation["source_memory_ids"][0]

-                # Check that bidirectional links exist
-                link_from_memory = await conn.fetchrow(
+                # Verify the source memory exists
+                source_memory = await conn.fetchrow(
                    """
-                    SELECT * FROM memory_links
-                    WHERE from_unit_id = $1 AND to_unit_id = $2
+                    SELECT id, fact_type FROM memory_units WHERE id = $1
                    """,
                    source_memory_id,
-                    observation["id"],
-                )
-                link_to_memory = await conn.fetchrow(
-                    """
-                    SELECT * FROM memory_links
-                    WHERE from_unit_id = $1 AND to_unit_id = $2
-                    """,
-                    observation["id"],
-                    source_memory_id,
                )
+                assert source_memory is not None, "Source memory should exist"
+                assert source_memory["fact_type"] in ("world", "experience"), "Source should be a fact"

-                # Both directions should have links
-                assert link_from_memory is not None, "Expected link from source memory to observation"
-                assert link_to_memory is not None, "Expected link from observation to source memory"
-                assert link_from_memory["link_type"] == "semantic"
-                assert link_to_memory["link_type"] == "semantic"
+                # No memory_links should exist between observation and source
+                # (observations rely on source_memory_ids for traversal)
+                links = await conn.fetch(
+                    """
+                    SELECT * FROM memory_links
+                    WHERE (from_unit_id = $1 AND to_unit_id = $2)
+                       OR (from_unit_id = $2 AND to_unit_id = $1)
+                    """,
+                    source_memory_id,
+                    observation["id"],
+                )
+                assert len(links) == 0, "No memory_links should exist between observation and source"

        # Cleanup
        await memory.delete_bank(bank_id, request_context=request_context)
@@ -8,9 +8,20 @@ populated from the summary for backwards compatibility.
 import pytest
 from hindsight_api.engine.memory_engine import Budget
 from hindsight_api import RequestContext
+from hindsight_api.config import get_config
 from datetime import datetime, timezone


+@pytest.fixture
+def disable_observations():
+    """Disable observations for a specific test."""
+    config = get_config()
+    original_value = config.enable_observations
+    config.enable_observations = False
+    yield
+    config.enable_observations = original_value
+
+
@pytest.mark.asyncio
 async def test_entity_extraction_on_retain(memory, request_context):
    """
@@ -370,12 +381,12 @@ async def test_get_entity_state(memory, request_context):


@pytest.mark.asyncio
-async def test_observation_fact_type_in_database(memory, request_context):
+async def test_observation_fact_type_in_database(memory, request_context, disable_observations):
    """
-    Test that observations are NOT stored as memory_units with fact_type='observation'.
+    Test that when observations are disabled, no observation records are created.

-    NOTE: Observations are now handled via mental models, not as memory_units
-    or entity summaries.
+    When enable_observations=False, consolidation does not run and no
+    memory_units with fact_type='observation' should exist.
    """
    bank_id = f"test_obs_db_{datetime.now(timezone.utc).timestamp()}"

@@ -14,6 +14,7 @@ from hindsight_api.engine.reflect.agent import (
    _normalize_tool_name,
    _is_done_tool,
    _clean_answer_text,
+    _clean_done_answer,
    run_reflect_agent,
 )
 from hindsight_api.engine.response_models import LLMToolCall, LLMToolCallResult, TokenUsage
@@ -61,6 +62,79 @@ class TestCleanAnswerText:
        assert cleaned == "Summary of findings."


+class TestCleanDoneAnswer:
+    """Test cleanup of answer field from done() tool call that leaks structured output."""
+
+    def test_clean_answer_with_leaked_json_code_block(self):
+        """Answer with leaked JSON code block at the end should be cleaned."""
+        text = '''The user's favorite color is blue.
+
+```json
+{"observation_ids": ["obs-1", "obs-2"]}
+```'''
+        cleaned = _clean_done_answer(text)
+        assert cleaned == "The user's favorite color is blue."
+        assert "observation_ids" not in cleaned
+
+    def test_clean_answer_with_memory_ids_code_block(self):
+        """Answer with leaked memory_ids JSON code block should be cleaned."""
+        text = '''Here is the answer.
+
+```json
+{"memory_ids": ["mem-1"]}
+```'''
+        cleaned = _clean_done_answer(text)
+        assert cleaned == "Here is the answer."
+
+    def test_clean_answer_with_raw_json_object(self):
+        """Answer with raw JSON object containing IDs at the end should be cleaned."""
+        text = 'The answer is 42. {"observation_ids": ["obs-1"]}'
+        cleaned = _clean_done_answer(text)
+        assert cleaned == "The answer is 42."
+
+    def test_clean_answer_with_trailing_ids_pattern(self):
+        """Answer with 'observation_ids: [...]' pattern at the end should be cleaned."""
+        text = "This is the answer.\n\nobservation_ids: [\"obs-1\", \"obs-2\"]"
+        cleaned = _clean_done_answer(text)
+        assert cleaned == "This is the answer."
+
+    def test_clean_answer_with_memory_ids_equals(self):
+        """Answer with 'memory_ids = [...]' pattern at the end should be cleaned."""
+        text = "Answer text here.\nmemory_ids = [\"mem-1\"]"
+        cleaned = _clean_done_answer(text)
+        assert cleaned == "Answer text here."
+
+    def test_clean_normal_answer_unchanged(self):
+        """Normal answer without leaked output should be unchanged."""
+        text = "This is a normal answer about observation strategies."
+        cleaned = _clean_done_answer(text)
+        assert cleaned == text
+
+    def test_clean_empty_answer(self):
+        """Empty answer should return empty."""
+        assert _clean_done_answer("") == ""
+
+    def test_clean_answer_with_observation_word_in_content(self):
+        """The word 'observation' in regular text should not be stripped."""
+        text = "Based on my observation, the user prefers dark mode."
+        cleaned = _clean_done_answer(text)
+        assert cleaned == text
+
+    def test_clean_answer_multiline_with_markdown(self):
+        """Answer with markdown and leaked JSON at end should clean only the leak."""
+        text = '''Summary:
+- Point 1
+- Point 2
+
+```json
+{"mental_model_ids": ["mm-1"]}
+```'''
+        cleaned = _clean_done_answer(text)
+        assert "Point 1" in cleaned
+        assert "Point 2" in cleaned
+        assert "mental_model_ids" not in cleaned
+
+
 class TestToolNameNormalization:
    """Test tool name normalization for various LLM output formats."""

@@ -633,7 +633,12 @@ async def test_student_tracking_visibility(api_client):

@pytest.mark.asyncio
 async def test_list_tags_returns_all_tags(api_client):
-    """Test that list_tags returns all unique tags with counts."""
+    """Test that list_tags returns all unique tags with counts.
+
+    Note: list_tags counts all memory units including observations.
+    Observations inherit tags from their source facts (for visibility security),
+    so counts may be higher than the number of stored memories.
+    """
    bank_id = f"list_tags_test_{datetime.now().timestamp()}"

    # Store memories with various tags
@@ -662,18 +667,19 @@ async def test_list_tags_returns_all_tags(api_client):
    assert "limit" in result
    assert "offset" in result

-    # Verify tags and counts
+    # Verify tags exist with at least the expected counts
+    # Note: Counts may be higher due to observations inheriting source fact tags
    tags_map = {item["tag"]: item["count"] for item in result["items"]}
    assert "user:alice" in tags_map
-    assert tags_map["user:alice"] == 3  # 3 memories have this tag
+    assert tags_map["user:alice"] >= 3  # At least 3 memories have this tag
    assert "user:bob" in tags_map
-    assert tags_map["user:bob"] == 1
+    assert tags_map["user:bob"] >= 1
    assert "session:123" in tags_map
-    assert tags_map["session:123"] == 1
+    assert tags_map["session:123"] >= 1
    assert "session:456" in tags_map
-    assert tags_map["session:456"] == 1
+    assert tags_map["session:456"] >= 1

-    assert result["total"] == 4  # 4 unique tags
+    assert result["total"] >= 4  # At least 4 unique tags


@pytest.mark.asyncio
@@ -15,8 +15,6 @@ The framework supports two answer generation patterns:
 2. Integrated: Answer generator performs its own retrieval (e.g., think API)
   - Indicated by needs_external_search() returning False
   - Skips the search step for efficiency
-
-Optional --include-mental-models flag enables returning mental models in recall results.
 """

 import asyncio
@@ -536,8 +534,6 @@ class BenchmarkRunner:
        max_tokens: int = 4096,
        question_date: Optional[datetime] = None,
        question_type: Optional[str] = None,
-        include_mental_models: bool = False,
-        only_mental_models: bool = False,
    ) -> Tuple[str, str, List[Dict], Dict[str, Dict]]:
        """
        Answer a question using memory retrieval.
@@ -549,8 +545,6 @@ class BenchmarkRunner:
            max_tokens: Maximum tokens to retrieve
            question_date: Date when the question was asked (for temporal filtering)
            question_type: Question category/type (e.g., 'multi-session', 'temporal-reasoning')
-            include_mental_models: If True, include mental models in recall results
-            only_mental_models: If True, only retrieve mental models (no facts)

        Returns:
            Tuple of (answer, reasoning, retrieved_memories, chunks)
@@ -565,26 +559,16 @@ class BenchmarkRunner:
            import time

            recall_start_time = time.time()
-            # Build fact_types based on what's requested
-            if only_mental_models:
-                # Only retrieve mental models
-                fact_types = ["mental_model"]
-            elif include_mental_models:
-                # Retrieve facts AND mental models
-                fact_types = ["world", "experience", "mental_model"]
-            else:
-                # Only retrieve facts
-                fact_types = ["world", "experience"]
+            # Use default fact types (no filtering)
            search_result = await self.memory.recall_async(
                bank_id=agent_id,
                query=question,
                budget=budget,
                max_tokens=max_tokens,
-                fact_type=fact_types,
                question_date=question_date,
-                include_entities=not only_mental_models,  # Skip entities when only mental models
+                include_entities=True,
                max_entity_tokens=2048,
-                include_chunks=True,  # Always include chunks (mental models fetch from source memories)
+                include_chunks=True,
                request_context=RequestContext(),
            )
            recall_time = time.time() - recall_start_time
@@ -641,16 +625,12 @@ class BenchmarkRunner:
        max_tokens: int,
        max_questions: Optional[int] = None,
        semaphore: asyncio.Semaphore = None,
-        include_mental_models: bool = False,
-        only_mental_models: bool = False,
    ) -> List[Dict]:
        """
        Evaluate QA task with parallel question processing.

        Args:
            semaphore: Semaphore to limit concurrent question processing
-            include_mental_models: If True, include mental models in recall results
-            only_mental_models: If True, only retrieve mental models (no facts)

        Returns:
            List of QA results
@@ -695,8 +675,6 @@ class BenchmarkRunner:
                            max_tokens,
                            question_date,
                            category,
-                            include_mental_models,
-                            only_mental_models,
                        )

                        # Remove embeddings from retrieved memories to reduce file size
@@ -875,8 +853,6 @@ class BenchmarkRunner:
        question_semaphore: asyncio.Semaphore,
        eval_semaphore_size: int = 8,
        clear_this_agent: bool = True,
-        include_mental_models: bool = False,
-        only_mental_models: bool = False,
    ) -> Dict:
        """
        Process a single item (ingest + evaluate).
@@ -884,8 +860,6 @@ class BenchmarkRunner:
        Args:
            clear_this_agent: Whether to clear this agent's data before ingesting.
                             Set to False to skip clearing (e.g., when agent_id is shared and already cleared)
-            include_mental_models: If True, include mental models in recall results and wait for consolidation after ingestion
-            only_mental_models: If True, only retrieve mental models (no facts)

        Returns:
            Result dict with metrics
@@ -902,11 +876,10 @@ class BenchmarkRunner:
                await self.memory.delete_bank(agent_id, request_context=RequestContext())
                console.print(f"      [green]✓[/green] Cleared '{agent_id}' agent data")

-            # Ingest conversation (wait for consolidation if mental models are requested)
+            # Ingest conversation
            step += 1
            console.print(f"  [{step}] Ingesting conversation (batch mode)...")
-            wait_for_consolidation = include_mental_models or only_mental_models
-            num_sessions = await self.ingest_conversation(item, agent_id, wait_for_consolidation=wait_for_consolidation)
+            num_sessions = await self.ingest_conversation(item, agent_id, wait_for_consolidation=False)
            console.print(f"      [green]✓[/green] Ingested {num_sessions} sessions")
        else:
            num_sessions = -1
@@ -923,8 +896,6 @@ class BenchmarkRunner:
            max_tokens,
            max_questions_per_item,
            question_semaphore,
-            include_mental_models,
-            only_mental_models,
        )

        # Calculate metrics
@@ -956,8 +927,6 @@ class BenchmarkRunner:
        max_concurrent_items: int = 1,  # Max concurrent items (conversations) to process in parallel
        output_path: Optional[Path] = None,  # Path to save results incrementally
        merge_with_existing: bool = False,  # Whether to merge with existing results
-        include_mental_models: bool = False,  # If True, include mental models in recall results
-        only_mental_models: bool = False,  # If True, only retrieve mental models (no facts)
    ) -> Dict[str, Any]:
        """
        Run the full benchmark evaluation.
@@ -977,8 +946,6 @@ class BenchmarkRunner:
            separate_ingestion_phase: If True, ingest all data first, then evaluate all questions (single agent)
            filln: If True, only process items where the agent has no indexed data yet
            max_concurrent_items: Max concurrent items to process in parallel (requires clear_agent_per_item=True)
-            include_mental_models: If True, include mental models in recall results and wait for consolidation after ingestion.
-            only_mental_models: If True, only retrieve mental models (no facts). Implies waiting for consolidation.

        Returns:
            Dict with complete benchmark results
@@ -1020,8 +987,6 @@ class BenchmarkRunner:
                eval_semaphore_size,
                output_path,
                merge_with_existing,
-                include_mental_models,
-                only_mental_models,
            )
        else:
            # Original approach: process each item independently
@@ -1039,8 +1004,6 @@ class BenchmarkRunner:
                max_concurrent_items,
                output_path,
                merge_with_existing,
-                include_mental_models,
-                only_mental_models,
            )

    async def _run_single_phase(
@@ -1058,8 +1021,6 @@ class BenchmarkRunner:
        max_concurrent_items: int = 1,
        output_path: Optional[Path] = None,
        merge_with_existing: bool = False,
-        include_mental_models: bool = False,
-        only_mental_models: bool = False,
    ) -> Dict[str, Any]:
        """Original single-phase approach: process each item independently."""
        # Create semaphore for question processing
@@ -1081,8 +1042,6 @@ class BenchmarkRunner:
                max_concurrent_items,
                output_path,
                merge_with_existing,
-                include_mental_models,
-                only_mental_models,
            )
        else:
            # Sequential item processing (original behavior)
@@ -1099,8 +1058,6 @@ class BenchmarkRunner:
                filln,
                output_path,
                merge_with_existing,
-                include_mental_models,
-                only_mental_models,
            )

        # Calculate overall metrics
@@ -1136,8 +1093,6 @@ class BenchmarkRunner:
        filln: bool,
        output_path: Optional[Path] = None,
        merge_with_existing: bool = False,
-        include_mental_models: bool = False,
-        only_mental_models: bool = False,
    ) -> List[Dict]:
        """Process items sequentially (original behavior)."""
        all_results = []
@@ -1185,8 +1140,6 @@ class BenchmarkRunner:
                question_semaphore,
                eval_semaphore_size,
                clear_this_agent,
-                include_mental_models,
-                only_mental_models,
            )

            # Replace existing result or append new one
@@ -1218,8 +1171,6 @@ class BenchmarkRunner:
        max_concurrent_items: int,
        output_path: Optional[Path] = None,
        merge_with_existing: bool = False,
-        include_mental_models: bool = False,
-        only_mental_models: bool = False,
    ) -> List[Dict]:
        """Process items in parallel (requires unique agent IDs per item)."""
        # Load existing results if merge_with_existing is True
@@ -1264,8 +1215,6 @@ class BenchmarkRunner:
                    question_semaphore,
                    eval_semaphore_size,
                    clear_this_agent=True,  # Always clear for parallel processing
-                    include_mental_models=include_mental_models,
-                    only_mental_models=only_mental_models,
                )
                return result

@@ -1304,17 +1253,11 @@ class BenchmarkRunner:
        eval_semaphore_size: int,
        output_path: Optional[Path] = None,
        merge_with_existing: bool = False,
-        include_mental_models: bool = False,
-        only_mental_models: bool = False,
    ) -> Dict[str, Any]:
        """
        Two-phase approach: ingest all data into single agent, then evaluate all questions.

        More realistic scenario where agent accumulates memories over time.
-
-        Args:
-            include_mental_models: If True, include mental models in recall results and wait for consolidation
-            only_mental_models: If True, only retrieve mental models (no facts)
        """
        # Phase 1: Ingestion
        if not skip_ingestion:
@@ -1350,10 +1293,6 @@ class BenchmarkRunner:
            )

            console.print(f"    [green]✓[/green] Ingested {len(all_sessions)} sessions from {len(items)} items")
-
-            # Wait for consolidation if mental models are requested
-            if include_mental_models or only_mental_models:
-                await self._wait_for_consolidation(agent_id)
        else:
            console.print("\n[3] Skipping ingestion (using existing data)")

@@ -1380,8 +1319,6 @@ class BenchmarkRunner:
                max_tokens,
                max_questions_per_item,
                question_semaphore,
-                include_mental_models,
-                only_mental_models,
            )

            # Calculate metrics
@@ -278,8 +278,6 @@ async def run_benchmark(
    max_questions_per_conv: int = None,
    skip_ingestion: bool = False,
    use_think: bool = False,
-    include_mental_models: bool = False,
-    only_mental_models: bool = False,
    conversation: str = None,
    api_url: str = None,
    max_concurrent_questions_override: int = None,
@@ -294,8 +292,6 @@ async def run_benchmark(
        max_questions_per_conv: Maximum questions per conversation (None for all)
        skip_ingestion: Whether to skip ingestion and use existing data
        use_think: Whether to use the think API instead of search + LLM
-        include_mental_models: If True, include mental models in recall results and wait for consolidation after ingestion.
-        only_mental_models: If True, only retrieve mental models (no facts). Implies waiting for consolidation.
        conversation: Specific conversation ID to run (e.g., "conv-26")
        api_url: Optional API URL to connect to (default: use local memory)
        only_failed: If True, only run conversations that have failed questions (is_correct=False)
@@ -403,14 +399,7 @@ async def run_benchmark(
        dataset.load = filtered_load

    # Determine output filename based on mode
-    if use_think:
-        suffix = "_think"
-    elif only_mental_models:
-        suffix = "_only_mental_models"
-    elif include_mental_models:
-        suffix = "_mental_models"
-    else:
-        suffix = ""
+    suffix = "_think" if use_think else ""
    results_filename = f"benchmark_results{suffix}.json"
    output_path = Path(__file__).parent / "results" / results_filename

@@ -423,11 +412,7 @@ async def run_benchmark(
    # Each conversation gets its own isolated bank
    separate_ingestion = False
    clear_per_item = True  # Use unique agent ID per conversation
-    if include_mental_models or only_mental_models:
-        # Mental models requires more time due to consolidation, limit parallelism
-        concurrent_items = 2
-    else:
-        concurrent_items = 3  # Process up to 3 conversations in parallel
+    concurrent_items = 3  # Process up to 3 conversations in parallel

    # Run benchmark with parallel conversation processing
    # Each conversation gets its own agent ID (locomo_conv-26, locomo_conv-30, etc.)
@@ -448,8 +433,6 @@ async def run_benchmark(
        max_concurrent_items=concurrent_items,
        output_path=output_path,  # Save results incrementally
        merge_with_existing=merge_with_existing,
-        include_mental_models=include_mental_models,  # Include mental models in recall results
-        only_mental_models=only_mental_models,  # Only retrieve mental models (no facts)
    )

    # Display results (final save already happened incrementally)
@@ -457,16 +440,12 @@ async def run_benchmark(
    console.print(f"\n[green]✓[/green] Results saved incrementally to {output_path}")

    # Generate markdown table
-    generate_markdown_table(
-        results, use_think=use_think, include_mental_models=include_mental_models, only_mental_models=only_mental_models
-    )
+    generate_markdown_table(results, use_think=use_think)

    return results


-def generate_markdown_table(
-    results: dict, use_think: bool = False, include_mental_models: bool = False, only_mental_models: bool = False
-):
+def generate_markdown_table(results: dict, use_think: bool = False):
    """
    Generate a markdown table with benchmark results.

@@ -484,14 +463,7 @@ def generate_markdown_table(

    # Build markdown content
    lines = []
-    if use_think:
-        mode_str = " (Think Mode)"
-    elif only_mental_models:
-        mode_str = " (Only Mental Models Mode)"
-    elif include_mental_models:
-        mode_str = " (Mental Models Mode)"
-    else:
-        mode_str = ""
+    mode_str = " (Think Mode)" if use_think else ""
    lines.append(f"# LoComo Benchmark Results{mode_str}")
    lines.append("")

@@ -542,14 +514,7 @@ def generate_markdown_table(
        )

    # Write to file with suffix
-    if use_think:
-        suffix = "_think"
-    elif only_mental_models:
-        suffix = "_only_mental_models"
-    elif include_mental_models:
-        suffix = "_mental_models"
-    else:
-        suffix = ""
+    suffix = "_think" if use_think else ""
    output_file = Path(__file__).parent / "results" / f"results_table{suffix}.md"
    output_file.parent.mkdir(parents=True, exist_ok=True)
    output_file.write_text("\n".join(lines))
@@ -592,16 +557,6 @@ if __name__ == "__main__":
        action="store_true",
        help="Only run conversations that have invalid questions (is_invalid=True). Requires existing results file.",
    )
-    parser.add_argument(
-        "--include-mental-models",
-        action="store_true",
-        help="Include mental models in recall results. This waits for consolidation to complete after ingestion and includes mental models in the recall response.",
-    )
-    parser.add_argument(
-        "--only-mental-models",
-        action="store_true",
-        help="Only retrieve mental models (no facts). This waits for consolidation to complete after ingestion and only returns mental models.",
-    )

    args = parser.parse_args()

@@ -615,8 +570,6 @@ if __name__ == "__main__":
            max_questions_per_conv=args.max_questions,
            skip_ingestion=args.skip_ingestion,
            use_think=args.use_think,
-            include_mental_models=args.include_mental_models,
-            only_mental_models=args.only_mental_models,
            conversation=args.conversation,
            api_url=args.api_url,
            max_concurrent_questions_override=args.max_concurrent_questions,
@@ -433,8 +433,6 @@ async def run_benchmark(
    results_filename: str = "benchmark_results.json",
    context_format: str = "json",
    source_results: str = None,
-    include_mental_models: bool = False,
-    only_mental_models: bool = False,
 ):
    """
    Run the LongMemEval benchmark.
@@ -456,8 +454,6 @@ async def run_benchmark(
        results_filename: Filename for results (default: benchmark_results.json). Directory is fixed to results/.
        context_format: How to format context for answer generation. "json" (raw JSON) or "structured" (human-readable with facts+chunks).
        source_results: Source results file to read failed/invalid questions from (for --only-failed/--only-invalid). Defaults to benchmark_results.json.
-        include_mental_models: If True, include mental models in recall results and wait for consolidation after ingestion.
-        only_mental_models: If True, only retrieve mental models (no facts). Implies waiting for consolidation.
    """
    from rich.console import Console

@@ -629,10 +625,6 @@ async def run_benchmark(
    answer_generator = LongMemEvalAnswerGenerator(context_format=context_format)
    # Log context format being used
    console.print(f"[blue]Context format: {context_format}[/blue]")
-    if only_mental_models:
-        console.print("[blue]Mental models: ONLY (no facts)[/blue]")
-    elif include_mental_models:
-        console.print("[blue]Mental models: included in recall[/blue]")

    answer_evaluator = LLMAnswerEvaluator()

@@ -705,7 +697,7 @@ async def run_benchmark(
    # Configuration for single-phase benchmark
    separate_ingestion = False
    clear_per_item = True  # Use unique agent_id per question
-    concurrent_questions = 4 if (include_mental_models or only_mental_models) else 8
+    concurrent_questions = 8

    results = await runner.run(
        dataset_path=dataset_path,
@@ -726,8 +718,6 @@ async def run_benchmark(
        max_concurrent_items=max_concurrent_items,  # Parallel instance processing
        output_path=output_path,  # Save results incrementally
        merge_with_existing=merge_with_existing,  # Merge when using --fill, --category, --only-failed, --only-invalid flags or specific question
-        include_mental_models=include_mental_models,  # Include mental models in recall results
-        only_mental_models=only_mental_models,  # Only retrieve mental models (no facts)
    )

    # Display results (final save already happened incrementally)
@@ -978,16 +968,6 @@ if __name__ == "__main__":
        default=None,
        help="Source results file to read failed/invalid questions from (for --only-failed/--only-invalid). Defaults to benchmark_results.json if not specified.",
    )
-    parser.add_argument(
-        "--include-mental-models",
-        action="store_true",
-        help="Include mental models in recall results. This waits for consolidation to complete after ingestion and includes mental models in the recall response.",
-    )
-    parser.add_argument(
-        "--only-mental-models",
-        action="store_true",
-        help="Only retrieve mental models (no facts). This waits for consolidation to complete after ingestion and only returns mental models.",
-    )

    args = parser.parse_args()

@@ -1018,7 +998,5 @@ if __name__ == "__main__":
            results_filename=args.results_filename,
            context_format=args.context_format,
            source_results=args.source_results,
-            include_mental_models=args.include_mental_models,
-            only_mental_models=args.only_mental_models,
        )
    )
@@ -332,11 +332,11 @@ The extraction mode controls how aggressively facts are extracted from content:

 ### Observations (Experimental)

-Observations are consolidated knowledge synthesized from facts. This feature is experimental and disabled by default.
+Observations are consolidated knowledge synthesized from facts.

 | Variable | Description | Default |
 |----------|-------------|---------|
-| `HINDSIGHT_API_ENABLE_OBSERVATIONS` | Enable observation consolidation | `false` |
+| `HINDSIGHT_API_ENABLE_OBSERVATIONS` | Enable observation consolidation | `true` |
 | `HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE` | Memories to load per batch (internal optimization) | `50` |
 | `HINDSIGHT_API_RETAIN_OBSERVATIONS_ASYNC` | Run observation generation asynchronously (after retain completes) | `false` |
Author	SHA1	Message	Date
Nicolò BoschiandClaude Sonnet 4.5	6f2fa9ae1d	test: update consolidation test for source_memory_ids behavior Updated test_consolidation_creates_memory_links to test_consolidation_uses_source_memory_ids to reflect the new behavior where observations use source_memory_ids instead of memory_links for traversal. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <[email protected]>	2026-01-28 12:20:37 +01:00
Nicolò BoschiandClaude Sonnet 4.5	7039ab44a7	fix: observations rely on source_memory_ids, no link copying Observations no longer copy any memory_links from their source facts. Instead, retrieval uses source_memory_ids to traverse: - Entity connections: observation → source_memory_ids → unit_entities - Semantic similarity: observations have their own embeddings - Temporal proximity: observations have their own temporal fields This avoids data duplication and fixes bidirectionality issues with entity links being copied to observations. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <[email protected]>	2026-01-28 12:08:27 +01:00
Nicolò Boschi	1a2969724d	fix tests	2026-01-28 11:12:27 +01:00
Nicolò Boschi	3ce8e557e4	chore: cleanup benchmarks runner with old flags	2026-01-28 09:54:20 +01:00