feat(consolidation): add "shared" observation_scopes keyword (#2202)

* feat(consolidation): add "shared" observation_scopes keyword

Add a "shared" value for observation_scopes that resolves to a single
global, untagged scope ([[]]). Memories consolidate into one observation
regardless of their tags, while the tags stay on the source facts for
recall filtering.

This is the supported way to deduplicate observations across volatile
per-call provenance tags (e.g. per-session ids): with combined/per_tag,
a unique session tag puts every retain in its own scope, so near-identical
facts never dedup and accumulate one observation per session. "shared"
keeps recall and the dedup probe on the same (empty) scope, fixing
consolidation quality rather than only the duplicate count.

- consolidator: _resolve_obs_tags_list -> [[]], _resolve_write_scopes -> [frozenset()]
- API/engine type literals + OpenAPI + regenerated Python/TS/Go/Rust clients
- CP client type kept in sync
- docs: retain.mdx 'shared' section (+ shared vs [[]] vs [] caveat),
  observations.mdx dedup pointer; regenerated hindsight-docs skill mirror
- tests: unit scope-resolution + e2e parallel-consolidation scope correctness

* chore(opencode): apply prettier formatting to plugin.test.ts

Pre-existing lint drift unrelated to this PR — CI's verify-generated-files
job reformats all integrations (LINT_ALL_INTEGRATIONS) and flagged this file.
Folding the one-line reflow in here to get the gate green.

Author: nicoloboschi

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

@@ -545,13 +545,16 @@ def coerce_tags(cls, v):
             return [v]
         return v
 
-    observation_scopes: Literal["per_tag", "combined", "all_combinations"] | list[list[str]] | None = Field(
+    observation_scopes: Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]] | None = Field(
         default=None,
         title="ObservationScopes",
         description=(
             "How to scope observations during consolidation. "
             "'per_tag' runs one consolidation pass per individual tag, creating separate observations for each tag. "
             "'combined' (default) runs a single pass with all tags together. "
+            "'shared' runs a single pass over one global, untagged scope, so memories consolidate together "
+            "regardless of their tags — useful for deduplicating across volatile per-call provenance tags "
+            "(e.g. per-session ids) while keeping those tags on the source facts. "
             "A list of tag lists runs one pass per inner list, giving full control over which combinations to use."
         ),
     )

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

@@ -335,7 +335,15 @@ def _resolve_obs_tags_list(memory: dict[str, Any]) -> list[list[str]] | None:
 
     Returns ``None`` for the default ``combined``-mode single pass (caller uses
     the memory's own tags). Returns a list[list[str]] when the memory requested
-    multi-pass scoping (``per_tag``, ``all_combinations``, or an explicit list).
+    multi-pass scoping (``per_tag``, ``all_combinations``, ``shared``, or an
+    explicit list).
+
+    ``shared`` resolves to ``[[]]`` — a single pass over the empty (untagged)
+    scope. The created observation carries no tags and recall/dedup match it with
+    ``tags_match="any"``, so every memory consolidates into one shared observation
+    regardless of its own tags. Use it to deduplicate across volatile per-call
+    provenance tags (e.g. per-session ids) without dropping those tags from the
+    source facts.
     """
     parsed = _parse_observation_scopes(memory)
     tags = list(memory.get("tags") or [])
@@ -346,6 +354,8 @@ def _resolve_obs_tags_list(memory: dict[str, Any]) -> list[list[str]] | None:
         if not tags:
             return None
         return [list(c) for r in range(1, len(tags) + 1) for c in combinations(tags, r)]
+    if parsed == "shared":
+        return [[]]
     if parsed == "combined" or parsed is None:
         return None
     return parsed  # explicit list[list[str]]
@@ -362,6 +372,7 @@ def _resolve_write_scopes(memory: dict[str, Any]) -> list[frozenset[str]]:
     - ``combined`` / ``None``    -> ``[frozenset(memory.tags)]``
     - ``per_tag``                -> ``[frozenset({t}) for t in memory.tags]``
     - ``all_combinations``       -> one frozenset per nonempty subset of tags
+    - ``shared``                 -> ``[frozenset()]`` (the single untagged scope)
     - explicit ``list[list[str]]`` -> one frozenset per declared scope
 
     Empty-tag memories collapse to a single ``frozenset()`` in all modes so they
@@ -376,6 +387,8 @@ def _resolve_write_scopes(memory: dict[str, Any]) -> list[frozenset[str]]:
         if not tags:
             return [frozenset()]
         return [frozenset(c) for r in range(1, len(tags) + 1) for c in combinations(tags, r)]
+    if parsed == "shared":
+        return [frozenset()]
     if parsed == "combined" or parsed is None:
         return [frozenset(tags)]
     return [frozenset(s) for s in parsed]  # explicit list[list[str]]

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

@@ -24,7 +24,9 @@ class RetainContentDict(TypedDict, total=False):
         tags: Visibility scope tags for this content item (optional)
         observation_scopes: How to scope observations for consolidation (optional).
             "per_tag" runs one pass per individual tag; "combined" (default) runs a
-            single pass with all tags; a list[list[str]] specifies exact passes.
+            single pass with all tags; "shared" runs a single pass over one global,
+            untagged scope so memories consolidate together regardless of tags;
+            a list[list[str]] specifies exact passes.
         update_mode: How to handle existing documents with the same document_id (optional).
             "replace" (default) deletes old data and reprocesses. "append" concatenates
             new content to the existing document and reprocesses.
@@ -38,7 +40,7 @@ class RetainContentDict(TypedDict, total=False):
     entities: list[dict[str, str]]  # [{"text": "...", "type": "..."}]
     tags: list[str]  # Visibility scope tags
     observation_scopes: (
-        Literal["per_tag", "combined", "all_combinations"] | list[list[str]]
+        Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]]
     )  # Observation scopes for consolidation
     update_mode: Literal["replace", "append"]
 
@@ -57,7 +59,7 @@ class RetainContent:
     metadata: dict[str, str] = field(default_factory=dict)
     entities: list[dict[str, str]] = field(default_factory=list)  # User-provided entities
     tags: list[str] = field(default_factory=list)  # Visibility scope tags
-    observation_scopes: Literal["per_tag", "combined", "all_combinations"] | list[list[str]] | None = (
+    observation_scopes: Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]] | None = (
         None  # Observation scopes
     )
 
@@ -124,7 +126,7 @@ class ExtractedFact:
     mentioned_at: datetime | None = None
     metadata: dict[str, str] = field(default_factory=dict)
     tags: list[str] = field(default_factory=list)  # Visibility scope tags
-    observation_scopes: Literal["per_tag", "combined", "all_combinations"] | list[list[str]] | None = (
+    observation_scopes: Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]] | None = (
         None  # Observation scopes
     )
 
@@ -176,7 +178,7 @@ class ProcessedFact:
     tags: list[str] = field(default_factory=list)
 
     # Observation scopes for consolidation
-    observation_scopes: Literal["per_tag", "combined", "all_combinations"] | list[list[str]] | None = None
+    observation_scopes: Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]] | None = None
 
     @property
     def is_duplicate(self) -> bool:

hindsight-api-slim/hindsight_api/engine/transfer/schema.py

@@ -22,7 +22,7 @@
 # Bump when the archive layout changes in a backward-incompatible way.
 SCHEMA_VERSION = 1
 
-ObservationScopes = Literal["per_tag", "combined", "all_combinations"] | list[list[str]]
+ObservationScopes = Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]]
 
 
 class TransferCausalRelation(BaseModel):

hindsight-api-slim/tests/test_consolidation_scope_parallelism.py

@@ -178,6 +178,42 @@ async def test_combined_mode_parallel_writes_to_memory_tag_set(memory: MemoryEng
         await memory.delete_bank(bank_id, request_context=request_context)
 
 
+@pytest.mark.asyncio
+async def test_shared_mode_parallel_writes_only_untagged_scope(memory: MemoryEngine, request_context):
+    """shared → every memory writes to the single untagged scope, ignoring its
+    own tags. Three memories with disjoint tags therefore all consolidate into
+    the same global scope (the per-session-tag dedup use case) instead of one
+    isolated observation per tag."""
+    bank_id = f"test-shared-{uuid.uuid4().hex[:8]}"
+    await memory.get_bank_profile(bank_id=bank_id, request_context=request_context)
+    try:
+        async with memory._pool.acquire() as conn:
+            await _insert_memory(conn, bank_id, "Alice likes tea", ["session:s1"], "shared")
+            await _insert_memory(conn, bank_id, "Bob bikes daily", ["session:s2"], "shared")
+            await _insert_memory(conn, bank_id, "Carol reads books", ["session:s3"], "shared")
+
+        wrapper, _ = _mock_llm_one_obs_per_fact()
+        original_llm = memory._consolidation_llm_config
+        memory._consolidation_llm_config = wrapper
+        try:
+            with (
+                _override_config(memory, consolidation_llm_parallelism=3, consolidation_llm_batch_size=1),
+                patch.object(memory, "submit_async_consolidation"),
+            ):
+                result = await run_consolidation_job(
+                    memory_engine=memory, bank_id=bank_id, request_context=request_context
+                )
+        finally:
+            memory._consolidation_llm_config = original_llm
+
+        assert result["status"] == "completed"
+        tag_sets = await _fetch_observation_tag_sets(memory, bank_id)
+        # Every observation lands at the untagged scope — none carries a session tag.
+        assert tag_sets and all(t == frozenset() for t in tag_sets), tag_sets
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
 @pytest.mark.asyncio
 async def test_per_tag_mode_parallel_writes_one_observation_per_tag(memory: MemoryEngine, request_context):
     """per_tag with tags [a, b] → two observations, tagged [a] and [b] respectively.

hindsight-api-slim/tests/test_consolidation_write_scopes.py

@@ -92,6 +92,18 @@ def test_empty_tags_collapses_to_untagged_scope(self):
         assert _resolve_write_scopes(memory) == [frozenset()]
 
 
+class TestResolveWriteScopesShared:
+    def test_collapses_to_single_untagged_scope_regardless_of_tags(self):
+        # "shared" ignores the memory's own tags and writes to one global scope,
+        # so every memory deduplicates against the same observation.
+        memory = {"tags": ["alice", "session"], "observation_scopes": _as_json_string("shared")}
+        assert _resolve_write_scopes(memory) == [frozenset()]
+
+    def test_empty_tags_also_untagged_scope(self):
+        memory = {"tags": [], "observation_scopes": _as_json_string("shared")}
+        assert _resolve_write_scopes(memory) == [frozenset()]
+
+
 class TestResolveWriteScopesExplicitList:
     def test_uses_declared_scopes_verbatim(self):
         memory = {
@@ -166,6 +178,12 @@ def test_explicit_list_passthrough(self):
         memory = {"tags": ["a", "b"], "observation_scopes": json.dumps(spec)}
         assert _resolve_obs_tags_list(memory) == spec
 
+    def test_shared_returns_single_empty_scope(self):
+        # One pass over the empty (untagged) scope; the memory's own tags are
+        # ignored so cross-tag memories consolidate into one observation.
+        memory = {"tags": ["a", "b"], "observation_scopes": json.dumps("shared")}
+        assert _resolve_obs_tags_list(memory) == [[]]
+
 
 # ---------------------------------------------------------------------------
 # Agreement between obs_tags_list (dispatch) and write_scopes (locks)
@@ -185,6 +203,7 @@ class TestDispatchLockAgreement:
             {"tags": ["a", "b", "c"], "observation_scopes": json.dumps("combined")},
             {"tags": ["a", "b"], "observation_scopes": json.dumps("per_tag")},
             {"tags": ["a", "b", "c"], "observation_scopes": json.dumps("all_combinations")},
+            {"tags": ["a", "b"], "observation_scopes": json.dumps("shared")},
             {"tags": ["a", "b"], "observation_scopes": json.dumps([["a"], ["b"], ["a", "b"]])},
             {"tags": ["a"], "observation_scopes": json.dumps([["a"], ["x"]])},
             # Pre-parsed Python shape (defensive — covers callers that hand the

hindsight-clients/go/api/openapi.yaml

@@ -8166,6 +8166,7 @@ components:
         - per_tag
         - combined
         - all_combinations
+        - shared
         type: string
       - items:
           items:
@@ -8175,8 +8176,11 @@ components:
       description: "How to scope observations during consolidation. 'per_tag' runs\
         \ one consolidation pass per individual tag, creating separate observations\
         \ for each tag. 'combined' (default) runs a single pass with all tags together.\
-        \ A list of tag lists runs one pass per inner list, giving full control over\
-        \ which combinations to use."
+        \ 'shared' runs a single pass over one global, untagged scope, so memories\
+        \ consolidate together regardless of their tags — useful for deduplicating\
+        \ across volatile per-call provenance tags (e.g. per-session ids) while keeping\
+        \ those tags on the source facts. A list of tag lists runs one pass per inner\
+        \ list, giving full control over which combinations to use."
       nullable: true
       title: ObservationScopes
     MentalModelTrigger_Input_tag_groups_inner:

hindsight-clients/go/model_observation_scopes.go

@@ -27,7 +27,7 @@
 
 class ObservationScopes(BaseModel):
     """
-    How to scope observations during consolidation. 'per_tag' runs one consolidation pass per individual tag, creating separate observations for each tag. 'combined' (default) runs a single pass with all tags together. A list of tag lists runs one pass per inner list, giving full control over which combinations to use.
+    How to scope observations during consolidation. 'per_tag' runs one consolidation pass per individual tag, creating separate observations for each tag. 'combined' (default) runs a single pass with all tags together. 'shared' runs a single pass over one global, untagged scope, so memories consolidate together regardless of their tags — useful for deduplicating across volatile per-call provenance tags (e.g. per-session ids) while keeping those tags on the source facts. A list of tag lists runs one pass per inner list, giving full control over which combinations to use.
     """
 
     # data type: str

hindsight-clients/python/hindsight_client_api/models/observation_scopes.py

@@ -2235,9 +2235,15 @@ export type MemoryItem = {
   /**
    * ObservationScopes
    *
-   * How to scope observations during consolidation. 'per_tag' runs one consolidation pass per individual tag, creating separate observations for each tag. 'combined' (default) runs a single pass with all tags together. A list of tag lists runs one pass per inner list, giving full control over which combinations to use.
-   */
-  observation_scopes?: "per_tag" | "combined" | "all_combinations" | Array<Array<string>> | null;
+   * How to scope observations during consolidation. 'per_tag' runs one consolidation pass per individual tag, creating separate observations for each tag. 'combined' (default) runs a single pass with all tags together. 'shared' runs a single pass over one global, untagged scope, so memories consolidate together regardless of their tags — useful for deduplicating across volatile per-call provenance tags (e.g. per-session ids) while keeping those tags on the source facts. A list of tag lists runs one pass per inner list, giving full control over which combinations to use.
+   */
+  observation_scopes?:
+    | "per_tag"
+    | "combined"
+    | "all_combinations"
+    | "shared"
+    | Array<Array<string>>
+    | null;
   /**
    * Strategy
    *