feat: add max_observations_per_scope bank config (#729)

* feat: add max_observations_per_scope bank config

Adds a configurable limit on the number of observations per tag scope.
When the limit is reached, consolidation only updates/deletes existing
observations — no new ones are created. Enforcement is done via a
constrained Pydantic response model (max_length on creates list) so the
LLM structurally cannot exceed the limit, plus prompt guidance.

- Config: HINDSIGHT_API_MAX_OBSERVATIONS_PER_SCOPE (-1 = unlimited)
- Reorder action execution: deletes → updates → creates
- Dynamic _ConsolidationBatchResponse with max_length constraint
- Prompt CAPACITY CONSTRAINT section when near/at limit
- Observations with no tags skip the limit entirely
- Control plane UI field + docs

* fix: strengthen max_observations tests with mock LLM + defensive truncation

- Rewrite integration tests to use MockLLM with deterministic responses
  (one observation per fact) instead of relying on real LLM behavior
- Add defensive truncation in _consolidate_batch_with_llm as belt-and-
  suspenders — catches LLM providers that ignore JSON schema max_length
- Tests now assert exact counts, not just upper bounds

hindsight-api-slim/hindsight_api/config.py

@@ -305,6 +305,7 @@ ENV_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS_PER_OBSERVATION = (
     "HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS_PER_OBSERVATION"
 )
 ENV_OBSERVATIONS_MISSION = "HINDSIGHT_API_OBSERVATIONS_MISSION"
+ENV_MAX_OBSERVATIONS_PER_SCOPE = "HINDSIGHT_API_MAX_OBSERVATIONS_PER_SCOPE"
 ENV_ENABLE_OBSERVATION_HISTORY = "HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY"
 ENV_ENABLE_MENTAL_MODEL_HISTORY = "HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY"
 
@@ -489,6 +490,7 @@ DEFAULT_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS_PER_OBSERVATION = (
     256  # Max tokens of source facts per observation in consolidation prompt (-1 = unlimited)
 )
 DEFAULT_OBSERVATIONS_MISSION = None  # Declarative spec of what observations are for this bank
+DEFAULT_MAX_OBSERVATIONS_PER_SCOPE = -1  # Max observations per tag scope (-1 = unlimited)
 
 # Database migrations
 DEFAULT_RUN_MIGRATIONS_ON_STARTUP = True
@@ -780,6 +782,7 @@ class HindsightConfig:
     consolidation_source_facts_max_tokens: int
     consolidation_source_facts_max_tokens_per_observation: int
     observations_mission: str | None
+    max_observations_per_scope: int
 
     # Entity labels (controlled vocabulary of key:value classification labels extracted at retain time)
     # List of label group dicts: [{key, description, type, optional, values: [{value, description}]}]
@@ -892,6 +895,7 @@ class HindsightConfig:
         "consolidation_source_facts_max_tokens",
         "consolidation_source_facts_max_tokens_per_observation",
         "observations_mission",
+        "max_observations_per_scope",
         # Reflect settings
         "reflect_mission",
         "reflect_source_facts_max_tokens",
@@ -1289,6 +1293,9 @@ class HindsightConfig:
                 )
             ),
             observations_mission=os.getenv(ENV_OBSERVATIONS_MISSION) or DEFAULT_OBSERVATIONS_MISSION,
+            max_observations_per_scope=int(
+                os.getenv(ENV_MAX_OBSERVATIONS_PER_SCOPE, str(DEFAULT_MAX_OBSERVATIONS_PER_SCOPE))
+            ),
             entity_labels=None,
             entities_allow_free_form=True,
             # Database migrations

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

@@ -119,6 +119,39 @@ def _aggregate_source_fields(source_mems: list[dict[str, Any]], tags: list[str]
     )
 
 
+async def _count_observations_for_scope(
+    conn: "Connection",
+    bank_id: str,
+    tags: list[str],
+) -> int:
+    """Count existing observations matching the given tag scope.
+
+    Returns the count of observations whose tags contain all specified tags.
+    Observations with no tags are not counted (the limit does not apply to them).
+    """
+    return await conn.fetchval(
+        f"SELECT COUNT(*) FROM {fq_table('memory_units')} "
+        f"WHERE bank_id = $1 AND fact_type = 'observation' AND tags @> $2::varchar[]",
+        bank_id,
+        tags,
+    )
+
+
+def _build_response_model(max_creates: int | None = None) -> type[_ConsolidationBatchResponse]:
+    """Build a response model, optionally constraining max creates via JSON schema."""
+    if max_creates is None or max_creates < 0:
+        return _ConsolidationBatchResponse
+
+    from pydantic import Field as PydanticField
+
+    clamped = max(max_creates, 0)
+
+    class _ConstrainedConsolidationBatchResponse(_ConsolidationBatchResponse):
+        creates: list[_CreateAction] = PydanticField(default=[], max_length=clamped)
+
+    return _ConstrainedConsolidationBatchResponse
+
+
 class ConsolidationPerfLog:
     """Performance logging for consolidation operations."""
 
@@ -698,24 +731,6 @@ async def _process_memory_batch(
         if recall_result.source_facts:
             union_source_facts.update(recall_result.source_facts)
 
-    # 3. Single LLM call
-    t0 = time.time()
-    llm_result = await _consolidate_batch_with_llm(
-        llm_config=llm_config,
-        memories=memories,
-        union_observations=union_observations,
-        union_source_facts=union_source_facts,
-        config=config,
-    )
-    if perf:
-        perf.record_timing("llm", time.time() - t0)
-        perf.record_llm_call(llm_result.obs_count, llm_result.prompt_chars)
-
-    # 4. Sequential execution of creates / updates / deletes
-    # Track which memory indices participated so we can build per-memory results for stats
-    per_memory_created: set[str] = set()
-    per_memory_updated: set[str] = set()
-
     # Determine effective tag scope for observations.
     # When obs_tags_override is set, use it; otherwise use the memory's own tags.
     if obs_tags_override is not None:
@@ -724,28 +739,52 @@ async def _process_memory_batch(
         # All memories in the batch share the same tag set (enforced by batching)
         fact_tags = memories[0].get("tags") or [] if memories else []
 
+    # 2b. Compute remaining observation slots for this scope (if limit configured)
+    max_obs = config.max_observations_per_scope if config is not None else -1
+    remaining_observation_slots: int | None = None
+    if max_obs > 0 and fact_tags:
+        current_count = await _count_observations_for_scope(conn, bank_id, fact_tags)
+        remaining_observation_slots = max(max_obs - current_count, 0)
+        if remaining_observation_slots == 0:
+            logger.info(
+                f"[CONSOLIDATION] bank={bank_id} scope={fact_tags} at observation limit "
+                f"({current_count}/{max_obs}), only updates/deletes allowed"
+            )
+
+    # 3. Single LLM call
+    t0 = time.time()
+    llm_result = await _consolidate_batch_with_llm(
+        llm_config=llm_config,
+        memories=memories,
+        union_observations=union_observations,
+        union_source_facts=union_source_facts,
+        config=config,
+        remaining_observation_slots=remaining_observation_slots,
+        max_observations_per_scope=max_obs,
+    )
+    if perf:
+        perf.record_timing("llm", time.time() - t0)
+        perf.record_llm_call(llm_result.obs_count, llm_result.prompt_chars)
+
+    # 4. Sequential execution of deletes / updates / creates
+    # Deletes run first to free observation slots before creates consume them.
+    # Track which memory indices participated so we can build per-memory results for stats
+    per_memory_created: set[str] = set()
+    per_memory_updated: set[str] = set()
+
     mem_by_id = {str(m["id"]): m for m in memories}
 
-    for create in llm_result.creates:
-        source_mems = [mem_by_id[fid] for fid in create.source_fact_ids if fid in mem_by_id]
-        if not source_mems:
+    # Execute deletes first to free observation slots before creates consume them
+    deleted_count = 0
+    for delete in llm_result.deletes:
+        # Security: the observation must be present in the unioned recall
+        if not any(str(obs.id) == delete.observation_id for obs in union_observations):
+            logger.debug(
+                f"Batch consolidation: rejected delete — observation {delete.observation_id} not in unioned recall"
+            )
             continue
-        agg = _aggregate_source_fields(source_mems, tags=fact_tags)
-        await _execute_create_action(
-            conn=conn,
-            memory_engine=memory_engine,
-            bank_id=bank_id,
-            source_memory_ids=[m["id"] for m in source_mems],
-            text=create.text,
-            source_fact_tags=agg.tags,
-            event_date=agg.event_date,
-            occurred_start=agg.occurred_start,
-            occurred_end=agg.occurred_end,
-            mentioned_at=agg.mentioned_at,
-            perf=perf,
-        )
-        for m in source_mems:
-            per_memory_created.add(str(m["id"]))
+        await _execute_delete_action(conn=conn, bank_id=bank_id, observation_id=delete.observation_id)
+        deleted_count += 1
 
     for update in llm_result.updates:
         source_mems = [mem_by_id[fid] for fid in update.source_fact_ids if fid in mem_by_id]
@@ -776,16 +815,26 @@ async def _process_memory_batch(
         for m in source_mems:
             per_memory_updated.add(str(m["id"]))
 
-    deleted_count = 0
-    for delete in llm_result.deletes:
-        # Security: the observation must be present in the unioned recall
-        if not any(str(obs.id) == delete.observation_id for obs in union_observations):
-            logger.debug(
-                f"Batch consolidation: rejected delete — observation {delete.observation_id} not in unioned recall"
-            )
+    for create in llm_result.creates:
+        source_mems = [mem_by_id[fid] for fid in create.source_fact_ids if fid in mem_by_id]
+        if not source_mems:
             continue
-        await _execute_delete_action(conn=conn, bank_id=bank_id, observation_id=delete.observation_id)
-        deleted_count += 1
+        agg = _aggregate_source_fields(source_mems, tags=fact_tags)
+        await _execute_create_action(
+            conn=conn,
+            memory_engine=memory_engine,
+            bank_id=bank_id,
+            source_memory_ids=[m["id"] for m in source_mems],
+            text=create.text,
+            source_fact_tags=agg.tags,
+            event_date=agg.event_date,
+            occurred_start=agg.occurred_start,
+            occurred_end=agg.occurred_end,
+            mentioned_at=agg.mentioned_at,
+            perf=perf,
+        )
+        for m in source_mems:
+            per_memory_created.add(str(m["id"]))
 
     # Build per-memory result dicts for the stats tracker in the outer loop
     results: list[dict[str, Any]] = []
@@ -1083,6 +1132,8 @@ async def _consolidate_batch_with_llm(
     union_observations: "list[MemoryFact]",
     union_source_facts: "dict[str, MemoryFact]",
     config: Any = None,
+    remaining_observation_slots: int | None = None,
+    max_observations_per_scope: int = -1,
 ) -> _BatchLLMResult:
     """Single LLM call for a batch of facts against a pooled set of observations."""
     if union_observations:
@@ -1106,24 +1157,51 @@ async def _consolidate_batch_with_llm(
 
     facts_lines = "\n".join(_fact_line(m) for m in memories)
 
+    # Build capacity note for the prompt when observation limit is configured
+    observation_capacity_note: str | None = None
+    if remaining_observation_slots is not None and max_observations_per_scope > 0:
+        if remaining_observation_slots == 0:
+            observation_capacity_note = (
+                f"OBSERVATION LIMIT REACHED ({max_observations_per_scope}/{max_observations_per_scope}). "
+                "Only UPDATE or DELETE existing observations. Do NOT create new ones — "
+                "merge new knowledge into existing observations via UPDATE."
+            )
+        elif remaining_observation_slots <= len(memories):
+            observation_capacity_note = (
+                f"This scope has {remaining_observation_slots} observation slot(s) remaining "
+                f"(out of {max_observations_per_scope}). Prefer UPDATE over CREATE when possible."
+            )
+
     observations_mission = config.observations_mission if config is not None else None
-    prompt_template = build_batch_consolidation_prompt(observations_mission)
+    prompt_template = build_batch_consolidation_prompt(observations_mission, observation_capacity_note)
     prompt = prompt_template.format(
         facts_text=facts_lines,
         observations_text=observations_text,
     )
 
+    # Use a constrained response model when observation limit is active
+    response_model = _build_response_model(max_creates=remaining_observation_slots)
+
     max_attempts = 3
     last_exc: Exception | None = None
     for attempt in range(1, max_attempts + 1):
         try:
             response: _ConsolidationBatchResponse = await llm_config.call(
                 messages=[{"role": "user", "content": prompt}],
-                response_format=_ConsolidationBatchResponse,
+                response_format=response_model,
                 scope="consolidation",
             )
+            # Defensive truncation: some LLM providers may not enforce JSON schema max_length
+            creates = response.creates
+            if remaining_observation_slots is not None and remaining_observation_slots >= 0:
+                if len(creates) > remaining_observation_slots:
+                    logger.info(
+                        f"[CONSOLIDATION] Truncating {len(creates)} creates to {remaining_observation_slots} "
+                        f"(max_observations_per_scope={max_observations_per_scope})"
+                    )
+                    creates = creates[:remaining_observation_slots]
             return _BatchLLMResult(
-                creates=response.creates,
+                creates=creates,
                 updates=response.updates,
                 deletes=response.deletes,
                 obs_count=len(union_observations),

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

@@ -66,7 +66,10 @@ Rules:
 - Return {{"creates": [], "updates": [], "deletes": []}} if nothing durable is found."""
 
 
-def build_batch_consolidation_prompt(observations_mission: str | None = None) -> str:
+def build_batch_consolidation_prompt(
+    observations_mission: str | None = None,
+    observation_capacity_note: str | None = None,
+) -> str:
     """
     Build the consolidation prompt for batch mode (multiple facts per LLM call).
 
@@ -75,9 +78,13 @@ def build_batch_consolidation_prompt(observations_mission: str | None = None) ->
     """
     mission = observations_mission or _DEFAULT_MISSION
 
+    capacity_section = ""
+    if observation_capacity_note:
+        capacity_section = f"\n\n## CAPACITY CONSTRAINT\n{observation_capacity_note}"
+
     return (
         "You are a memory consolidation system. Synthesize facts into observations "
         "and merge with existing observations when appropriate.\n\n"
-        f"## MISSION\n{mission}\n\n"
+        f"## MISSION\n{mission}{capacity_section}\n\n"
         f"{_PROCESSING_RULES}" + _BATCH_DATA_SECTION + _BATCH_OUTPUT_FORMAT
     )

hindsight-control-plane/src/components/bank-config-view.tsx

@@ -60,6 +60,7 @@ type ObservationsEdits = {
   consolidation_source_facts_max_tokens: number | null;
   consolidation_source_facts_max_tokens_per_observation: number | null;
   observations_mission: string | null;
+  max_observations_per_scope: number | null;
 };
 
 type LabelValue = { value: string; description: string };
@@ -179,6 +180,7 @@ function observationsSlice(config: Record<string, any>): ObservationsEdits {
     consolidation_source_facts_max_tokens_per_observation:
       config.consolidation_source_facts_max_tokens_per_observation ?? null,
     observations_mission: config.observations_mission ?? null,
+    max_observations_per_scope: config.max_observations_per_scope ?? null,
   };
 }
 
@@ -542,6 +544,23 @@ export function BankConfigView() {
               placeholder="Server default"
             />
           </FieldRow>
+          <FieldRow
+            label="Max Observations Per Scope"
+            description="Maximum number of observations allowed per tag scope. When the limit is reached, only updates and deletes are allowed. Observations with no tags are not subject to this limit. -1 = unlimited."
+          >
+            <Input
+              type="number"
+              min={-1}
+              value={observationsEdits.max_observations_per_scope ?? ""}
+              onChange={(e) =>
+                setObservationsEdits((prev) => ({
+                  ...prev,
+                  max_observations_per_scope: e.target.value ? parseInt(e.target.value, 10) : null,
+                }))
+              }
+              placeholder="Server default"
+            />
+          </FieldRow>
         </ConfigSection>
 
         {/* Reflect Section */}

hindsight-docs/docs/developer/configuration.md

@@ -878,6 +878,7 @@ Observations are consolidated knowledge synthesized from facts.
 | `HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS` | Total token budget for source facts included with observations in the consolidation prompt. `-1` = unlimited. Configurable per bank. | `-1` |
 | `HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS_PER_OBSERVATION` | Per-observation token cap for source facts in the consolidation prompt. Each observation independently gets at most this many tokens of source facts. `-1` = unlimited. Configurable per bank. | `256` |
 | `HINDSIGHT_API_OBSERVATIONS_MISSION` | What this bank should synthesise into durable observations. Replaces the built-in consolidation rules — leave unset to use the server default. | - |
+| `HINDSIGHT_API_MAX_OBSERVATIONS_PER_SCOPE` | Maximum number of observations allowed per tag scope. When the limit is reached, consolidation will only update or delete existing observations — no new ones are created. Applies per tag scope (e.g., per-tag when using `per_tag` observation scopes). Observations with no tags are not subject to this limit. `-1` = unlimited. Configurable per bank. | `-1` |
 
 #### Customizing observations: when to use what
 
@@ -1154,7 +1155,7 @@ Configuration fields are categorized for security:
 
 1. **Configurable Fields** - Safe behavioral settings that can be customized per-bank:
    - Retention: `retain_chunk_size`, `retain_extraction_mode`, `retain_mission`, `retain_custom_instructions`
-   - Observations: `enable_observations`, `observations_mission`
+   - Observations: `enable_observations`, `observations_mission`, `max_observations_per_scope`
    - MCP access control: `mcp_enabled_tools`
 
 2. **Credential Fields** - NEVER exposed or configurable via API: