feat(api): add targeted consolidation by observation scopes

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

@@ -2188,6 +2188,19 @@ class OperationResponse(BaseModel):
     )
 
 
+class ConsolidationRequest(BaseModel):
+    """Request model for consolidation trigger endpoint."""
+
+    observation_scopes: list[list[str]] | None = Field(
+        default=None,
+        description=(
+            "Optional list of tag scopes to consolidate. Each scope is a list of tags. "
+            "Only unconsolidated memories whose tags contain all tags in at least one scope "
+            "will be processed. If omitted, all unconsolidated memories are processed."
+        ),
+    )
+
+
 class ConsolidationResponse(BaseModel):
     """Response model for consolidation trigger endpoint."""
 
@@ -5506,11 +5519,20 @@ async def api_reset_bank_config(bank_id: str, request_context: RequestContext =
         operation_id="trigger_consolidation",
         tags=["Banks"],
     )
-    @audited("consolidation", request_param=None)
-    async def api_trigger_consolidation(bank_id: str, request_context: RequestContext = Depends(get_request_context)):
+    @audited("consolidation")
+    async def api_trigger_consolidation(
+        bank_id: str,
+        request: ConsolidationRequest | None = None,
+        request_context: RequestContext = Depends(get_request_context),
+    ):
         """Trigger consolidation for a bank (async)."""
         try:
-            result = await app.state.memory.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
+            observation_scopes = request.observation_scopes if request else None
+            result = await app.state.memory.submit_async_consolidation(
+                bank_id=bank_id,
+                request_context=request_context,
+                observation_scopes=observation_scopes,
+            )
             return ConsolidationResponse(
                 operation_id=result["operation_id"],
                 deduplicated=result.get("deduplicated", False),

hindsight-api-slim/hindsight_api/config.py

@@ -365,6 +365,7 @@ def normalize_config_dict(config: dict[str, Any]) -> dict[str, Any]:
 
 # Observations settings (consolidated knowledge from facts)
 ENV_ENABLE_OBSERVATIONS = "HINDSIGHT_API_ENABLE_OBSERVATIONS"
+ENV_ENABLE_AUTO_CONSOLIDATION = "HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION"
 ENV_CONSOLIDATION_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE"
 ENV_CONSOLIDATION_MAX_MEMORIES_PER_ROUND = "HINDSIGHT_API_CONSOLIDATION_MAX_MEMORIES_PER_ROUND"
 ENV_CONSOLIDATION_LLM_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_LLM_BATCH_SIZE"
@@ -622,6 +623,7 @@ def normalize_config_dict(config: dict[str, Any]) -> dict[str, Any]:
 
 # Observations defaults (consolidated knowledge from facts)
 DEFAULT_ENABLE_OBSERVATIONS = True  # Observations enabled by default
+DEFAULT_ENABLE_AUTO_CONSOLIDATION = True  # Auto-consolidation after retain enabled by default
 DEFAULT_ENABLE_OBSERVATION_HISTORY = True  # Observation history tracking enabled by default
 DEFAULT_ENABLE_MENTAL_MODEL_HISTORY = True  # Mental model history tracking enabled by default
 # Each history entry snapshots previous_content + previous_reflect_response. Without
@@ -1095,6 +1097,7 @@ class HindsightConfig:
 
     # Observations settings (consolidated knowledge from facts)
     enable_observations: bool
+    enable_auto_consolidation: bool
     enable_observation_history: bool
     enable_mental_model_history: bool
     mental_model_history_max_entries: int
@@ -1255,6 +1258,7 @@ class HindsightConfig:
         "entities_allow_free_form",
         # Consolidation settings
         "enable_observations",
+        "enable_auto_consolidation",
         "consolidation_llm_batch_size",
         "consolidation_max_memories_per_round",
         "consolidation_source_facts_max_tokens",
@@ -1763,6 +1767,10 @@ def from_env(cls) -> "HindsightConfig":
             == "true",
             # Observations settings (consolidated knowledge from facts)
             enable_observations=os.getenv(ENV_ENABLE_OBSERVATIONS, str(DEFAULT_ENABLE_OBSERVATIONS)).lower() == "true",
+            enable_auto_consolidation=os.getenv(
+                ENV_ENABLE_AUTO_CONSOLIDATION, str(DEFAULT_ENABLE_AUTO_CONSOLIDATION)
+            ).lower()
+            == "true",
             enable_observation_history=os.getenv(
                 ENV_ENABLE_OBSERVATION_HISTORY, str(DEFAULT_ENABLE_OBSERVATION_HISTORY)
             ).lower()

hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py

@@ -230,6 +230,7 @@ async def run_consolidation_job(
     bank_id: str,
     request_context: "RequestContext",
     operation_id: str | None = None,
+    observation_scopes: list[list[str]] | None = None,
 ) -> dict[str, Any]:
     """
     Run consolidation job for a bank.
@@ -240,6 +241,10 @@ async def run_consolidation_job(
         memory_engine: MemoryEngine instance
         bank_id: Bank identifier
         request_context: Request context for authentication
+        operation_id: Optional operation ID for tracking
+        observation_scopes: Optional list of tag scopes. When provided, only
+            unconsolidated memories whose tags contain all tags in at least one
+            scope are processed.
 
     Returns:
         Dict with consolidation results
@@ -281,6 +286,18 @@ async def run_consolidation_job(
 
         perf.record_timing("fetch_bank", time.time() - t0)
 
+        # Build optional scope filter clause.  When observation_scopes is provided,
+        # only process memories whose tags contain all tags in at least one scope.
+        scope_clause = ""
+        scope_params: list[Any] = [bank_id]
+        if observation_scopes:
+            or_parts: list[str] = []
+            for scope_tags in observation_scopes:
+                idx = len(scope_params) + 1
+                or_parts.append(f"tags @> ${idx}::varchar[]")
+                scope_params.append(scope_tags)
+            scope_clause = " AND (" + " OR ".join(or_parts) + ")"
+
         # Count total unconsolidated memories for progress logging
         total_count = await conn.fetchval(
             f"""
@@ -290,8 +307,9 @@ async def run_consolidation_job(
               AND consolidated_at IS NULL
               AND consolidation_failed_at IS NULL
               AND fact_type IN ('experience', 'world')
+              {scope_clause}
             """,
-            bank_id,
+            *scope_params,
         )
 
     if total_count == 0:
@@ -330,6 +348,9 @@ async def run_consolidation_job(
         # Fetch next batch of unconsolidated memories
         async with acquire_with_retry(pool) as conn:
             t0 = time.time()
+            # scope_params[0] is bank_id; append fetch_limit after scope params
+            fetch_params = list(scope_params) + [fetch_limit]
+            limit_idx = len(fetch_params)
             memories = await conn.fetch(
                 f"""
                 SELECT id, text, fact_type, occurred_start, occurred_end, event_date, tags, mentioned_at,
@@ -339,11 +360,11 @@ async def run_consolidation_job(
                   AND consolidated_at IS NULL
                   AND consolidation_failed_at IS NULL
                   AND fact_type IN ('experience', 'world')
+                  {scope_clause}
                 ORDER BY created_at ASC
-                LIMIT $2
+                LIMIT ${limit_idx}
                 """,
-                bank_id,
-                fetch_limit,
+                *fetch_params,
             )
             perf.record_timing("fetch_memories", time.time() - t0)
 
@@ -583,7 +604,11 @@ async def run_consolidation_job(
             f" ~{remaining} remaining. Re-queuing consolidation."
         )
         try:
-            await memory_engine.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
+            await memory_engine.submit_async_consolidation(
+                bank_id=bank_id,
+                request_context=request_context,
+                observation_scopes=observation_scopes,
+            )
         except Exception as e:
             logger.warning(f"[CONSOLIDATION] bank={bank_id} failed to re-queue consolidation: {e}")
 

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

@@ -1148,6 +1148,7 @@ async def _handle_consolidation(self, task_dict: dict[str, Any]):
             bank_id=bank_id,
             request_context=internal_context,
             operation_id=task_dict.get("operation_id"),
+            observation_scopes=task_dict.get("observation_scopes"),
         )
 
         logger.info(f"[CONSOLIDATION] bank={bank_id} completed: {result.get('memories_processed', 0)} processed")
@@ -2624,7 +2625,7 @@ async def retain_batch_async(
         # Trigger consolidation as a tracked async operation if enabled
         # Resolve bank-specific config to check if observations are enabled for this bank
         config = await self._config_resolver.resolve_full_config(bank_id, request_context)
-        if config.enable_observations:
+        if config.enable_observations and config.enable_auto_consolidation:
             try:
                 await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
             except Exception as e:
@@ -4126,10 +4127,12 @@ async def delete_document(
                 }
 
         if invalidated_obs > 0:
-            try:
-                await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
-            except Exception as e:
-                logger.warning(f"Failed to submit consolidation after document deletion for bank {bank_id}: {e}")
+            config = await self._config_resolver.resolve_full_config(bank_id, request_context)
+            if config.enable_auto_consolidation:
+                try:
+                    await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
+                except Exception as e:
+                    logger.warning(f"Failed to submit consolidation after document deletion for bank {bank_id}: {e}")
 
         return result
 
@@ -4281,10 +4284,12 @@ async def update_document(
                             )
 
         if invalidated_obs > 0:
-            try:
-                await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
-            except Exception as e:
-                logger.warning(f"Failed to submit consolidation after document update for bank {bank_id}: {e}")
+            config = await self._config_resolver.resolve_full_config(bank_id, request_context)
+            if config.enable_auto_consolidation:
+                try:
+                    await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
+                except Exception as e:
+                    logger.warning(f"Failed to submit consolidation after document update for bank {bank_id}: {e}")
 
         return True
 
@@ -4357,14 +4362,17 @@ async def delete_memory_unit(
                 }
 
         if bank_id_for_consolidation:
-            try:
-                await self.submit_async_consolidation(
-                    bank_id=bank_id_for_consolidation, request_context=request_context
-                )
-            except Exception as e:
-                logger.warning(
-                    f"Failed to submit consolidation after memory deletion for bank {bank_id_for_consolidation}: {e}"
-                )
+            config = await self._config_resolver.resolve_full_config(bank_id_for_consolidation, request_context)
+            if config.enable_auto_consolidation:
+                try:
+                    await self.submit_async_consolidation(
+                        bank_id=bank_id_for_consolidation, request_context=request_context
+                    )
+                except Exception as e:
+                    logger.warning(
+                        f"Failed to submit consolidation after memory deletion"
+                        f" for bank {bank_id_for_consolidation}: {e}"
+                    )
 
         return result
 
@@ -4490,10 +4498,12 @@ async def delete_bank(
                 await bank_utils.drop_bank_vector_indexes(conn, bank_internal_id, ops=self._backend.ops)
 
         if invalidated_obs > 0:
-            try:
-                await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
-            except Exception as e:
-                logger.warning(f"Failed to submit consolidation after bank deletion for bank {bank_id}: {e}")
+            config = await self._config_resolver.resolve_full_config(bank_id, request_context)
+            if config.enable_auto_consolidation:
+                try:
+                    await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
+                except Exception as e:
+                    logger.warning(f"Failed to submit consolidation after bank deletion for bank {bank_id}: {e}")
 
         return result
 
@@ -4656,7 +4666,9 @@ async def clear_observations_for_memory(
                     )
 
         if deleted_count > 0:
-            await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
+            config = await self._config_resolver.resolve_full_config(bank_id, request_context)
+            if config.enable_auto_consolidation:
+                await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
 
         return {"deleted_count": deleted_count}
 
@@ -9706,6 +9718,7 @@ async def submit_async_consolidation(
         bank_id: str,
         *,
         request_context: "RequestContext",
+        observation_scopes: list[list[str]] | None = None,
     ) -> dict[str, Any]:
         """Submit a consolidation operation to run asynchronously.
 
@@ -9715,6 +9728,8 @@ async def submit_async_consolidation(
         Args:
             bank_id: Bank identifier
             request_context: Request context for authentication
+            observation_scopes: Optional list of tag scopes to consolidate. When provided,
+                only unconsolidated memories matching at least one scope are processed.
 
         Returns:
             Dict with operation_id
@@ -9736,13 +9751,19 @@ async def submit_async_consolidation(
             task_payload["_tenant_id"] = request_context.tenant_id
         if request_context.api_key_id:
             task_payload["_api_key_id"] = request_context.api_key_id
+        if observation_scopes is not None:
+            task_payload["observation_scopes"] = observation_scopes
+
+        # Skip bank-level deduplication when scoped — the caller wants a
+        # targeted run that should not be merged into a pending full-bank sweep.
+        dedupe = observation_scopes is None
 
         return await self._submit_async_operation(
             bank_id=bank_id,
             operation_type="consolidation",
             task_type="consolidation",
             task_payload=task_payload,
-            dedupe_by_bank=True,
+            dedupe_by_bank=dedupe,
         )
 
     async def submit_async_refresh_mental_model(