feat(observations): enumerate + filter + visualize observation scopes

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

@@ -1280,6 +1280,33 @@ class GraphDataResponse(BaseModel):
     limit: int
 
 
+class ObservationScope(BaseModel):
+    """A distinct observation scope: an exact tag set plus its observation count."""
+
+    tags: list[str] = Field(
+        description="The exact tag set defining this scope (normalized order). Empty list is the global/untagged scope."
+    )
+    count: int = Field(description="Number of observations that live under this scope")
+
+
+class ObservationScopesResponse(BaseModel):
+    """Response model for the observation scopes enumeration endpoint."""
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "scopes": [
+                    {"tags": ["user:alice"], "count": 12},
+                    {"tags": ["user:alice", "project:apollo"], "count": 4},
+                    {"tags": [], "count": 2},
+                ]
+            }
+        }
+    )
+
+    scopes: list[ObservationScope] = Field(description="Distinct observation scopes, most populous first")
+
+
 class ListMemoryUnitsResponse(BaseModel):
     """Response model for list memory units endpoint."""
 
@@ -1405,6 +1432,12 @@ class DocumentResponse(BaseModel):
     tags: list[str] = FieldWithDefault(list, description="Tags associated with this document")
     document_metadata: dict[str, Any] | None = Field(default=None, description="Document metadata")
     retain_params: dict[str, Any] | None = Field(default=None, description="Parameters used during retain")
+    observation_scopes: str | list[list[str]] | None = Field(
+        default=None,
+        description="The observation_scopes spec configured at retain time (e.g. 'all_combinations', "
+        "'per_tag', or explicit tag-set lists), captured into retain_params. None when none was set "
+        "(default 'combined' scoping) or for documents retained before this was captured.",
+    )
 
 
 class UpdateDocumentRequest(BaseModel):
@@ -5771,6 +5804,35 @@ async def api_clear_observations(bank_id: str, request_context: RequestContext =
             logger.error(f"Error in DELETE /v1/default/banks/{bank_id}/observations: {error_detail}")
             raise HTTPException(status_code=500, detail=str(e))
 
+    @app.get(
+        "/v1/default/banks/{bank_id}/observations/scopes",
+        response_model=ObservationScopesResponse,
+        summary="List observation scopes",
+        description=(
+            "Enumerate the distinct scopes across a bank's observations. Each observation lives "
+            "under a scope: the exact set of tags it was consolidated with. Returns every distinct "
+            "scope (tag order normalized) with the number of observations in it; the empty tag list "
+            "is the global/untagged scope. Use a returned scope with the graph endpoint "
+            "(tags=<scope> & tags_match=exact) to filter observations to exactly that scope."
+        ),
+        operation_id="list_observation_scopes",
+        tags=["Memory"],
+    )
+    async def api_list_observation_scopes(bank_id: str, request_context: RequestContext = Depends(get_request_context)):
+        """List the distinct observation scopes (exact tag sets) for a bank."""
+        try:
+            return await app.state.memory.list_observation_scopes(bank_id, request_context=request_context)
+        except OperationValidationError as e:
+            raise HTTPException(status_code=e.status_code, detail=e.reason)
+        except (AuthenticationError, HTTPException):
+            raise
+        except Exception as e:
+            import traceback
+
+            error_detail = f"{str(e)}\n\nTraceback:\n{traceback.format_exc()}"
+            logger.error(f"Error in GET /v1/default/banks/{bank_id}/observations/scopes: {error_detail}")
+            raise HTTPException(status_code=500, detail=str(e))
+
     @app.post(
         "/v1/default/banks/{bank_id}/consolidation/recover",
         response_model=RecoverConsolidationResponse,

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

@@ -5153,6 +5153,12 @@ async def get_document(
             # document_metadata is sourced from retain_params.metadata
             document_metadata = retain_params_parsed.get("metadata") if retain_params_parsed else None
 
+            # observation_scopes is captured into retain_params at retain time
+            # (see _build_retain_params); surface it as a top-level field so the
+            # UI can show which scoping was requested. Only present for documents
+            # retained after this was added.
+            observation_scopes = retain_params_parsed.get("observation_scopes") if retain_params_parsed else None
+
             return {
                 "id": doc["id"],
                 "bank_id": doc["bank_id"],
@@ -5169,6 +5175,7 @@ async def get_document(
                 "tags": list(doc["tags"]) if doc["tags"] else [],
                 "document_metadata": document_metadata or None,
                 "retain_params": retain_params_parsed or None,
+                "observation_scopes": observation_scopes or None,
             }
 
     async def delete_document(
@@ -5718,6 +5725,46 @@ async def clear_observations(
 
                 return {"deleted_count": count or 0}
 
+    async def list_observation_scopes(
+        self,
+        bank_id: str,
+        *,
+        request_context: "RequestContext",
+    ) -> dict[str, Any]:
+        """List the distinct scopes across a bank's observations.
+
+        Every consolidated observation lives under a "scope": the exact set of
+        tags it was consolidated with. This enumerates each distinct scope (tag
+        order normalized so ``[a, b]`` and ``[b, a]`` collapse) together with the
+        number of observations in it. The empty list ``[]`` is the "global" scope
+        of untagged observations. Results are ordered most-populous first.
+
+        Returns:
+            Dict with ``scopes``: list of ``{"tags": list[str], "count": int}``.
+        """
+        await self._authenticate_tenant(request_context)
+        if self._operation_validator:
+            from hindsight_api.extensions import BankReadContext
+
+            ctx = BankReadContext(bank_id=bank_id, operation="list_observation_scopes", request_context=request_context)
+            await self._validate_operation(self._operation_validator.validate_bank_read(ctx))
+        backend = await self._get_backend()
+        async with acquire_with_retry(backend) as conn:
+            rows = await conn.fetch(
+                f"""
+                SELECT scope, COUNT(*) AS count
+                FROM (
+                    SELECT COALESCE(ARRAY(SELECT unnest(tags) ORDER BY 1), '{{}}'::text[]) AS scope
+                    FROM {fq_table("memory_units")}
+                    WHERE bank_id = $1 AND fact_type = 'observation'
+                ) s
+                GROUP BY scope
+                ORDER BY count DESC, scope
+                """,
+                bank_id,
+            )
+            return {"scopes": [{"tags": list(r["scope"]), "count": r["count"]} for r in rows]}
+
     async def retry_failed_consolidation(
         self,
         bank_id: str,
@@ -6297,6 +6344,10 @@ async def get_graph_data(
                     query_conditions.append(tag_clause.removeprefix("AND "))
                     param_count += 1
                     query_params.append(tags)
+            elif tags_match == "exact":
+                # Exact match with no tags is the "global" scope: rows that carry no
+                # tags at all. (Other match modes treat empty tags as "no filter".)
+                query_conditions.append("(tags IS NULL OR tags = '{}')")
 
             where_clause = "WHERE " + " AND ".join(query_conditions) if query_conditions else ""
 

hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py

@@ -273,6 +273,8 @@ def _build_retain_params(contents_dicts, document_tags=None, doc_contents=None):
             )
         if first_item.get("metadata"):
             retain_params["metadata"] = first_item["metadata"]
+        if first_item.get("observation_scopes") is not None:
+            retain_params["observation_scopes"] = first_item["observation_scopes"]
 
     return retain_params, merged_tags
 

hindsight-api-slim/hindsight_api/engine/search/tags.py

@@ -2,14 +2,18 @@
 Tags filtering utilities for retrieval.
 
 Provides SQL building functions for filtering memories by tags.
-Supports four matching modes via TagsMatch enum:
+Supports five matching modes via TagsMatch enum:
 - "any": OR matching, includes untagged memories (default, backward compatible)
 - "all": AND matching, includes untagged memories
 - "any_strict": OR matching, excludes untagged memories
 - "all_strict": AND matching, excludes untagged memories
+- "exact": set-equality matching, excludes untagged memories
 
 OR matching (any/any_strict): Memory matches if ANY of its tags overlap with request tags
 AND matching (all/all_strict): Memory matches if ALL request tags are present in its tags
+EXACT matching: Memory matches only if its tag set EQUALS the request tag set (order-
+    independent). Used for observation "scope" filtering, where each observation lives
+    under exactly one scope (its full tag set) and "scope [a]" must not match "[a, b]".
 """
 
 from __future__ import annotations
@@ -18,7 +22,7 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
-TagsMatch = Literal["any", "all", "any_strict", "all_strict"]
+TagsMatch = Literal["any", "all", "any_strict", "all_strict", "exact"]
 
 
 def _parse_tags_match(match: TagsMatch) -> tuple[str, bool]:
@@ -38,6 +42,10 @@ def _parse_tags_match(match: TagsMatch) -> tuple[str, bool]:
         return "&&", False
     elif match == "all_strict":
         return "@>", False
+    elif match == "exact":
+        # Set equality is handled by the callers via `@> AND <@`; the operator
+        # here is unused. Untagged rows never equal a non-empty scope.
+        return "@>", False
     else:
         # Default to "any" behavior
         return "&&", True
@@ -78,6 +86,13 @@ def build_tags_where_clause(
         return "", [], param_offset
 
     column = f"{table_alias}tags" if table_alias else "tags"
+
+    if match == "exact":
+        # Set equality (order-independent): superset AND subset. Untagged rows
+        # (empty array) never satisfy `@>` of a non-empty scope, so they're excluded.
+        clause = f"AND ({column} @> ${param_offset} AND {column} <@ ${param_offset})"
+        return clause, [tags], param_offset + 1
+
     operator, include_untagged = _parse_tags_match(match)
 
     if include_untagged:
@@ -115,6 +130,12 @@ def build_tags_where_clause_simple(
         return ""
 
     column = f"{table_alias}tags" if table_alias else "tags"
+
+    if match == "exact":
+        # Set equality (order-independent): superset AND subset. Untagged rows
+        # (empty array) never satisfy `@>` of a non-empty scope, so they're excluded.
+        return f"AND ({column} @> ${param_num} AND {column} <@ ${param_num})"
+
     operator, include_untagged = _parse_tags_match(match)
 
     if include_untagged:
@@ -164,7 +185,11 @@ def filter_results_by_tags(
             # else: skip untagged
         else:
             result_tags_set = set(result_tags)
-            if is_any_match:
+            if match == "exact":
+                # Set equality: tag set must match the scope exactly
+                if result_tags_set == tags_set:
+                    filtered.append(result)
+            elif is_any_match:
                 # Any overlap
                 if result_tags_set & tags_set:
                     filtered.append(result)
@@ -241,6 +266,9 @@ def _build_group_clause(
     """
     if isinstance(group, TagGroupLeaf):
         column = f"{table_alias}tags" if table_alias else "tags"
+        if group.match == "exact":
+            clause = f"({column} @> ${param_offset} AND {column} <@ ${param_offset})"
+            return clause, [group.tags], param_offset + 1
         operator, include_untagged = _parse_tags_match(group.match)
         if include_untagged:
             clause = f"({column} IS NULL OR {column} = '{{}}' OR {column} {operator} ${param_offset})"
@@ -349,6 +377,8 @@ def _match_group(result: object, group: TagGroup) -> bool:
             return include_untagged
         else:
             result_tags_set = set(result_tags)
+            if group.match == "exact":
+                return result_tags_set == tags_set
             if is_any_match:
                 return bool(result_tags_set & tags_set)
             else:

hindsight-api-slim/tests/test_document_tracking.py

@@ -206,6 +206,58 @@ async def test_document_without_metadata(memory, request_context):
         await memory.delete_bank(bank_id, request_context=request_context)
 
 
+@pytest.mark.asyncio
+async def test_document_observation_scopes_from_retain_params(memory, request_context):
+    """observation_scopes passed at retain time is captured into retain_params and surfaced by get_document."""
+    bank_id = f"test_doc_obs_scopes_{datetime.now(timezone.utc).timestamp()}"
+
+    try:
+        document_id = "doc-with-scopes"
+        await memory.retain_batch_async(
+            bank_id=bank_id,
+            contents=[
+                {
+                    "content": "Alice and Bob are friends.",
+                    "tags": ["alice", "bob"],
+                    "observation_scopes": "all_combinations",
+                }
+            ],
+            document_id=document_id,
+            request_context=request_context,
+        )
+
+        doc = await memory.get_document(document_id, bank_id, request_context=request_context)
+        assert doc is not None
+        # Surfaced as a top-level field and persisted in retain_params.
+        assert doc["observation_scopes"] == "all_combinations"
+        assert doc["retain_params"]["observation_scopes"] == "all_combinations"
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_document_observation_scopes_none_when_unset(memory, request_context):
+    """get_document returns observation_scopes None when none was configured at retain time."""
+    bank_id = f"test_doc_no_scopes_{datetime.now(timezone.utc).timestamp()}"
+
+    try:
+        document_id = "doc-no-scopes"
+        await memory.retain_async(
+            bank_id=bank_id,
+            content="Bob works at Microsoft.",
+            document_id=document_id,
+            request_context=request_context,
+        )
+
+        doc = await memory.get_document(document_id, bank_id, request_context=request_context)
+        assert doc is not None
+        assert doc["observation_scopes"] is None
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
 @pytest.mark.asyncio
 @pytest.mark.hs_llm_core
 async def test_document_persisted_with_zero_facts(memory_real_llm, request_context):