feat(agent-plugin): add portable Hindsight plugin for the Agent Plugins standard

Add a vendor-neutral Hindsight plugin conforming to Vercel's Agent Plugins 1.0.0 standard (plugin.json + mcp.json + skills/SKILL.md), so one artifact gives long-term memory to any compatible client (Codex, Cursor, GitHub Copilot, Kiro, VS Code) instead of a per-IDE integration. The plugin is a thin transport wrapper over Hindsight's existing MCP server (retain / recall / reflect); a bundled skill teaches the agent when to use it. Wiring: - CI: test-agent-plugin-integration job runs the manifest validator, gated on hindsight-integrations/agent-plugin/** changes. - Docs: integrations.json gallery entry + docs-integrations/agent-plugin.md. - Release: agent-plugin added to release-integration.sh and the changelog generator; both learn to read a root-level plugin.json and link the changelog to the source tree (git-distributed bundle, no registry package).
fix(retain): fold shared document_id items on the sync path (#3363 ) (#3386 )
2026-08-11 11:21:54 -04:00 · 2026-08-11 16:17:56 +02:00 · 2026-08-11 16:00:23 +02:00 · 2026-08-11 14:44:16 +02:00 · 2026-08-11 13:45:51 +02:00 · 2026-08-11 13:26:41 +02:00
35 changed files with 1992 additions and 182 deletions
@@ -48,6 +48,12 @@ HINDSIGHT_API_LLM_BASE_URL=https://api.openai.com/v1
 # HINDSIGHT_API_REFLECT_LLM_CACHE_AFFINITY=xai_conv_id
 # HINDSIGHT_API_CONSOLIDATION_LLM_CACHE_AFFINITY=none

+# Ask litellm/litellmrouter/bedrock for structured output via a forced tool call
+# instead of response_format. Enable it for backends that reject response_format
+# outright -- e.g. Bedrock Claude in ap-southeast-2 ("Extra inputs are not permitted");
+# the same model in us-east-1 accepts response_format and needs nothing here.
+# HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL=false
+
 # Diagnostic: on any LLM 4xx, log the exact assembled request ([LLM_4XX_DUMP]) --
 # serialized request config (message bodies stripped) + capped per-message previews.
 # For debugging otherwise-unreproducible rejected calls. Off by default.
@@ -43,6 +43,7 @@ jobs:
      integrations-continue: ${{ steps.filter.outputs.integrations-continue }}
      integrations-cursor-cli: ${{ steps.filter.outputs.integrations-cursor-cli }}
      integrations-zcode: ${{ steps.filter.outputs.integrations-zcode }}
+      integrations-agent-plugin: ${{ steps.filter.outputs.integrations-agent-plugin }}
      integrations-copilot-cli: ${{ steps.filter.outputs.integrations-copilot-cli }}
      integrations-crewai: ${{ steps.filter.outputs.integrations-crewai }}
      integrations-litellm: ${{ steps.filter.outputs.integrations-litellm }}
@@ -194,6 +195,8 @@ jobs:
            - 'hindsight-integrations/zed/**'
          integrations-zcode:
            - 'hindsight-integrations/zcode/**'
+          integrations-agent-plugin:
+            - 'hindsight-integrations/agent-plugin/**'
          integrations-n8n:
            - 'hindsight-integrations/n8n/**'
          integrations-zapier:
@@ -796,6 +799,29 @@ jobs:
      working-directory: ./hindsight-integrations/zcode
      run: uv run pytest tests -v

+  test-agent-plugin-integration:
+    needs: [detect-changes]
+    if: >-
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-agent-plugin == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+    - uses: actions/checkout@v6
+      with:
+        ref: ${{ github.event.pull_request.head.sha || '' }}
+
+    - name: Set up Python
+      uses: actions/setup-python@v6
+      with:
+        python-version-file: ".python-version"
+
+    - name: Validate Agent Plugin manifests
+      working-directory: ./hindsight-integrations/agent-plugin
+      run: python3 validate.py
+
  build-ai-sdk-integration:
    needs: [detect-changes]
    if: >-
@@ -5217,6 +5243,7 @@ jobs:
      - test-codex-integration
      - test-cursor-cli-integration
      - test-zcode-integration
+      - test-agent-plugin-integration
      - build-ai-sdk-integration
      - test-ai-sdk-integration-deno
      - test-opencode-integration
@@ -177,6 +177,11 @@ ENV_LLM_STRICT_SCHEMA_RETAIN = "HINDSIGHT_API_LLM_STRICT_SCHEMA_RETAIN"
 ENV_LLM_STRICT_SCHEMA_REFLECT = "HINDSIGHT_API_LLM_STRICT_SCHEMA_REFLECT"
 ENV_LLM_STRICT_SCHEMA_CONSOLIDATION = "HINDSIGHT_API_LLM_STRICT_SCHEMA_CONSOLIDATION"
 ENV_LLM_SUPPORTS_MAX_ITEMS = "HINDSIGHT_API_LLM_SUPPORTS_MAX_ITEMS"
+# Route structured output through a forced function tool instead of the
+# OpenAI-style ``response_format`` on the LiteLLM-backed providers (``litellm``,
+# ``litellmrouter``, ``bedrock``). Off by default; see
+# DEFAULT_LLM_STRUCTURED_OUTPUT_FORCED_TOOL.
+ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL = "HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL"
 ENV_LLM_SEND_BANK_AS_USER = "HINDSIGHT_API_LLM_SEND_BANK_AS_USER"
 ENV_LLM_OLLAMA_NUM_CTX = "HINDSIGHT_API_LLM_OLLAMA_NUM_CTX"

@@ -911,6 +916,17 @@ DEFAULT_LLAMACPP_EXTRA_ARGS = None  # Space-separated extra CLI args for llama.c
 DEFAULT_LLM_STRICT_SCHEMA = False
 DEFAULT_LLM_SUPPORTS_MAX_ITEMS = True

+# True = ask LiteLLM-backed providers for structured output via a single forced
+# function tool (the response schema becomes the tool's parameters) instead of
+# the OpenAI-style ``response_format``. Needed where the backend rejects the
+# response_format route outright — notably Bedrock Claude, whose Converse layer
+# refuses the translated ``outputConfig`` ("Extra inputs are not permitted") while
+# accepting the identical schema as a tool (issue #3300). Verified region-dependent:
+# ap-southeast-2 / au.* rejects it, us-east-1 / us.* accepts it, so this is opt-in
+# rather than keyed off the provider. Default False keeps ``response_format``, which
+# every other LiteLLM backend handles natively.
+DEFAULT_LLM_STRUCTURED_OUTPUT_FORCED_TOOL = False
+
 DEFAULT_LLM_MAX_CONCURRENT = 32
 DEFAULT_LLM_MAX_RETRIES = 3  # Max retry attempts for LLM API calls
 DEFAULT_LLM_INITIAL_BACKOFF = 1.0  # Initial backoff in seconds for retry exponential backoff
@@ -2117,6 +2133,10 @@ class HindsightConfig:
        default=DEFAULT_LLM_SUPPORTS_MAX_ITEMS,
        kw_only=True,
    )  # Whether structured-output schemas accept JSON Schema maxItems
+    llm_structured_output_forced_tool: bool = field(
+        default=DEFAULT_LLM_STRUCTURED_OUTPUT_FORCED_TOOL,
+        kw_only=True,
+    )  # LiteLLM-backed providers: structured output via a forced tool call, not response_format
    # Tags outbound OpenAI-compatible LLM + embedding calls with `user=<bank_id>` for
    # per-bank cost attribution. Downstream cost gateways (OpenRouter usage accounting,
    # LiteLLM, Helicone) key attribution on the OpenAI `user` field. Opt-in; never
@@ -3055,6 +3075,10 @@ class HindsightConfig:
                ENV_LLM_SUPPORTS_MAX_ITEMS,
                DEFAULT_LLM_SUPPORTS_MAX_ITEMS,
            ),
+            llm_structured_output_forced_tool=_parse_boolean_env(
+                ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL,
+                DEFAULT_LLM_STRUCTURED_OUTPUT_FORCED_TOOL,
+            ),
            llm_send_bank_as_user=os.getenv(ENV_LLM_SEND_BANK_AS_USER, str(DEFAULT_LLM_SEND_BANK_AS_USER)).lower()
            in ("true", "1"),
            llm_ollama_num_ctx=_parse_optional_positive_int(
@@ -303,7 +303,7 @@ async def _dedup_reconcile_create(
            live_source_ids = await _filter_live_source_memories(conn, bank_id, create_source_ids)
            if not live_source_ids:
                return None
-            if store.writes_memory_rows_in_sql:
+            if store.writes_memory_rows_in_sql_for(bank_id):
                # Oracle-safe: _native_search_vector_update emits the to_tsvector clause only for a
                # native PG tsvector column, "" otherwise (see #3021 — the raw ::regconfig cast
                # breaks Oracle). RETURNING-gate on the twin's probe-time text so a concurrent
@@ -385,7 +385,7 @@ async def _dedup_reconcile_update(
    store = get_memories()
    async with acquire_with_retry(pool) as conn:
        async with conn.transaction():
-            if store.writes_memory_rows_in_sql:
+            if store.writes_memory_rows_in_sql_for(bank_id):
                # Snapshot the updated row's sources with a PLAIN read (no FOR UPDATE). Lock order
                # must be sources-before-observation: _filter_live_source_memories below takes
                # FOR SHARE on the SOURCE rows first, then the fold UPDATE locks the observation
@@ -583,7 +583,7 @@ async def _filter_live_source_memories(
    if not source_memory_ids:
        return []
    store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
        rows = await conn.fetch(
            f"SELECT id FROM {fq_table('memory_units')} WHERE id = ANY($1::uuid[]) AND bank_id = $2 FOR SHARE",
            source_memory_ids,
@@ -612,7 +612,7 @@ async def _any_live_source_memory(
    if not source_memory_ids:
        return False
    store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
        found = await conn.fetchval(
            f"SELECT 1 FROM {fq_table('memory_units')} WHERE id = ANY($1::uuid[]) AND bank_id = $2 LIMIT 1",
            source_memory_ids,
@@ -732,7 +732,7 @@ async def _count_observations_for_scope(
    Observations with no tags are not counted (the limit does not apply to them).
    """
    store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
        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[]",
@@ -2245,7 +2245,7 @@ async def _execute_update_action(
            merged_tags = list(existing_tags | source_tags)

            t0 = time.time()
-            if store.writes_memory_rows_in_sql:
+            if store.writes_memory_rows_in_sql_for(bank_id):
                updated_rows = await conn.execute_rows_affected(
                    f"""
                    UPDATE {fq_table("memory_units")}
@@ -2396,7 +2396,7 @@ async def _execute_delete_action(
 ) -> None:
    """Delete a superseded or contradicted observation."""
    store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
        await conn.execute(
            f"DELETE FROM {fq_table('memory_units')} WHERE id = $1 AND bank_id = $2 AND fact_type = 'observation'",
            uuid.UUID(observation_id),
@@ -2782,7 +2782,7 @@ async def _create_observation_directly(
            source_memory_ids = live_source_memory_ids

            t0 = time.time()
-            if store.writes_memory_rows_in_sql:
+            if store.writes_memory_rows_in_sql_for(bank_id):
                # Query varies based on text search backend.
                from ..schema import _is_oracle  # noqa: PLC0415

@@ -313,6 +313,7 @@ def create_llm_provider(
    timeout: float | None = None,
    ollama_num_ctx: int | None = None,
    cache_affinity: str | None = None,
+    structured_output_forced_tool: bool = False,
 ) -> Any:  # Returns LLMInterface
    """
    Factory function to create the appropriate LLM provider implementation.
@@ -345,6 +346,11 @@ def create_llm_provider(
            OpenAI-compatible wire format): "none" (default), "xai_conv_id",
            "openai_prompt_cache_key", or "auto". Providers on other branches do their own
            cache work or none at all. See ``engine/cache_affinity.py``.
+        structured_output_forced_tool: Ask the LiteLLM-backed providers (``litellm``,
+            ``litellmrouter``, ``bedrock``) for structured output via a forced tool call
+            instead of ``response_format``. For backends that reject the response_format
+            route — see ``HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL``. Other
+            providers ignore it.
        vertexai_project_id: Vertex AI project ID (for VertexAI provider).
        vertexai_region: Vertex AI region (for VertexAI provider).
        vertexai_credentials: Vertex AI credentials object (for VertexAI provider).
@@ -457,6 +463,7 @@ def create_llm_provider(
            extra_body=extra_body,
            default_headers=default_headers,
            timeout=timeout,
+            structured_output_forced_tool=structured_output_forced_tool,
        )

    elif provider_lower == "litellmrouter":
@@ -477,6 +484,7 @@ def create_llm_provider(
            extra_body=extra_body,
            default_headers=default_headers,
            timeout=timeout,
+            structured_output_forced_tool=structured_output_forced_tool,
        )

    elif provider_lower == "bedrock":
@@ -492,6 +500,7 @@ def create_llm_provider(
            default_headers=default_headers,
            bedrock_service_tier=bedrock_service_tier,
            timeout=timeout,
+            structured_output_forced_tool=structured_output_forced_tool,
        )

    elif provider_lower == "llamacpp":
@@ -646,6 +655,7 @@ class LLMProvider:
        max_backoff: float | None = None,
        ollama_num_ctx: int | None = None,
        cache_affinity: str | None = None,
+        structured_output_forced_tool: bool = False,
    ):
        """
        Initialize LLM provider.
@@ -692,6 +702,9 @@ class LLMProvider:
                ``max_retries``. ``None`` keeps each method's own fallback.
            max_backoff: Default maximum retry backoff (seconds), same resolution as
                ``max_retries``. ``None`` keeps each method's own fallback.
+            structured_output_forced_tool: Structured output via a forced tool call
+                instead of ``response_format``, for the LiteLLM-backed providers - from
+                config (``HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL``).

        This constructor uses every argument as passed and does not read global
        ``HindsightConfig``: resolving the server-level default for a ``None`` argument is the
@@ -720,6 +733,9 @@ class LLMProvider:
        self.openai_service_tier = openai_service_tier
        self.bedrock_service_tier = bedrock_service_tier
        self.gemini_service_tier = gemini_service_tier
+        # Structured-output transport for the LiteLLM-backed providers. Used verbatim —
+        # the caller resolves the server-level default, like the fields above.
+        self.structured_output_forced_tool = structured_output_forced_tool
        self.ollama_num_ctx = _validate_ollama_num_ctx(ollama_num_ctx)
        # Gemini safety settings (instance default; can be overridden per-request via context var)
        self.gemini_safety_settings = gemini_safety_settings
@@ -875,6 +891,7 @@ class LLMProvider:
            ollama_num_ctx=self.ollama_num_ctx,
            timeout=self.timeout,
            cache_affinity=self.cache_affinity,
+            structured_output_forced_tool=self.structured_output_forced_tool,
        )

        # Backward compatibility: Keep mock provider properties
@@ -1420,6 +1437,7 @@ class LLMProvider:
            DEFAULT_LLM_PROMPT_CACHE_ENABLED,
            DEFAULT_LLM_PROVIDER,
            DEFAULT_LLM_REASONING_EFFORT,
+            DEFAULT_LLM_STRUCTURED_OUTPUT_FORCED_TOOL,
            DEFAULT_LLM_TIMEOUT,
            ENV_LLM_API_KEY,
            ENV_LLM_BASE_URL,
@@ -1437,11 +1455,13 @@ class LLMProvider:
            ENV_LLM_PROMPT_CACHE_ENABLED,
            ENV_LLM_PROVIDER,
            ENV_LLM_REASONING_EFFORT,
+            ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL,
            ENV_LLM_TIMEOUT,
            ENV_LLM_VERTEXAI_PROJECT_ID,
            ENV_LLM_VERTEXAI_REGION,
            ENV_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY,
            _get_default_model_for_provider,
+            _parse_boolean_env,
            _parse_llm_router_config,
            _parse_optional_positive_int,
            parse_gemini_service_tier,
@@ -1498,6 +1518,10 @@ class LLMProvider:
            vertexai_region=os.getenv(ENV_LLM_VERTEXAI_REGION) or None,
            vertexai_service_account_key=os.getenv(ENV_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY) or None,
            timeout=float(os.getenv(ENV_LLM_TIMEOUT, str(DEFAULT_LLM_TIMEOUT))),
+            structured_output_forced_tool=_parse_boolean_env(
+                ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL,
+                DEFAULT_LLM_STRUCTURED_OUTPUT_FORCED_TOOL,
+            ),
        )


@@ -138,7 +138,12 @@ class MaintenanceLoop:
    @staticmethod
    def _cross_store_recovery_enabled() -> bool:
        """True when the memories store keeps memories outside SQL and therefore has
-        cross-store write-group txns a crashed writer could leave undecided."""
+        cross-store write-group txns a crashed writer could leave undecided.
+
+        Deliberately reads the PROCESS-LEVEL class attribute, not the per-bank
+        ``writes_memory_rows_in_sql_for(bank_id)`` — this only decides whether the recovery LOOP
+        needs to run at all. A store that routes some banks outside SQL keeps the class attribute
+        False so the loop runs, then ``recover_pending_txns`` is bank-scoped inside it."""
        try:
            from .memories import get_memories

@@ -394,6 +394,20 @@ class MemoriesExtension(Extension, ABC):
    #: the inline SQL. Cold, never-searched, key-based — see docs/documents-chunks.md.
    owns_document_store: bool = False

+    def writes_memory_rows_in_sql_for(self, bank_id: str) -> bool:
+        """Per-bank form of :attr:`writes_memory_rows_in_sql`. Defaults to the class attribute, so a
+        single-store extension needs no override. A store that keeps different banks in different
+        backends (some in SQL, some not) overrides this to answer PER BANK; every *bank-scoped* call
+        site consults this instead of the class attribute, so mixed banks each take the correct path.
+        (The few process-level gates — e.g. "is cross-store txn recovery relevant at all" — keep
+        reading the class attribute.)"""
+        return self.writes_memory_rows_in_sql
+
+    def owns_document_store_for(self, bank_id: str) -> bool:
+        """Per-bank form of :attr:`owns_document_store`. Defaults to the class attribute; a store
+        that keeps some banks in a separate backend overrides it. See :meth:`writes_memory_rows_in_sql_for`."""
+        return self.owns_document_store
+
    # ------------------------------------------------------------------ lifecycle

    async def initialize(self) -> None:
@@ -523,6 +523,7 @@ def _member_to_llm(member: "LLMMemberConfig", config: HindsightConfig, defaults:
        cache_affinity=member.cache_affinity or config.llm_cache_affinity,
        ollama_num_ctx=config.llm_ollama_num_ctx,
        bedrock_service_tier=member.bedrock_service_tier,
+        structured_output_forced_tool=config.llm_structured_output_forced_tool,
        gemini_service_tier=member.gemini_service_tier or config.llm_gemini_service_tier,
        gemini_safety_settings=_get_raw_config().llm_gemini_safety_settings,
        prompt_cache_enabled=config.llm_prompt_cache_enabled,
@@ -616,6 +617,43 @@ class _SubBatchSplit:
    chunk_counts: list[int] = field(default_factory=list)


+@dataclass
+class _RetainGroup:
+    """One document's slice of a retain batch.
+
+    ``retain_batch_async`` folds items that share an explicit ``document_id``
+    into a single document. Grouping is done up front (not left to the
+    orchestrator) so the token splitter and its per-document ``chunk_index``
+    bookkeeping only ever see one document at a time — they assume a split never
+    interleaves two documents. ``origins`` records the indices the items
+    occupied in the submitted batch so per-input results merge back in order.
+    ``document_id`` is ``None`` for an item that carried no explicit id (each
+    such item is its own group and its own document).
+    """
+
+    document_id: str | None
+    origins: list[int]
+    contents: list[RetainContentDict]
+
+
+@dataclass
+class _RetainExecutionResult:
+    """Outcome of running one batch through the retain token splitter and its
+    sequential sub-batch loop (``MemoryEngine._run_retain_execution``).
+
+    ``unit_ids`` is the per-input-content list of created unit ids.
+    ``processed_content_tokens`` follows ``RetainResult.processed_content_tokens``
+    (``None`` when a sub-batch bypassed dedup). ``cancelled`` is True when the
+    operation's bank was deleted mid-flight and the loop stopped early, so the
+    caller skips the completion side effects.
+    """
+
+    unit_ids: list[list[str]]
+    usage: "TokenUsage"
+    processed_content_tokens: int | None
+    cancelled: bool
+
+
@dataclass(frozen=True)
 class _RetainChunkingConfig:
    chunk_size: int
@@ -1581,6 +1619,7 @@ class MemoryEngine(MemoryEngineInterface):
            ollama_num_ctx=config.llm_ollama_num_ctx,
            litellmrouter_config=config.llm_litellmrouter_config,
            bedrock_service_tier=config.llm_bedrock_service_tier,
+            structured_output_forced_tool=config.llm_structured_output_forced_tool,
            gemini_service_tier=config.llm_gemini_service_tier,
            groq_service_tier=config.llm_groq_service_tier,
            openai_service_tier=config.llm_openai_service_tier,
@@ -1627,6 +1666,7 @@ class MemoryEngine(MemoryEngineInterface):
            ollama_num_ctx=config.llm_ollama_num_ctx,
            litellmrouter_config=config.retain_llm_litellmrouter_config or config.llm_litellmrouter_config,
            bedrock_service_tier=config.llm_bedrock_service_tier,
+            structured_output_forced_tool=config.llm_structured_output_forced_tool,
            gemini_service_tier=config.llm_gemini_service_tier,
            groq_service_tier=config.llm_groq_service_tier,
            openai_service_tier=config.llm_openai_service_tier,
@@ -1667,6 +1707,7 @@ class MemoryEngine(MemoryEngineInterface):
            ollama_num_ctx=config.llm_ollama_num_ctx,
            litellmrouter_config=config.reflect_llm_litellmrouter_config or config.llm_litellmrouter_config,
            bedrock_service_tier=config.llm_bedrock_service_tier,
+            structured_output_forced_tool=config.llm_structured_output_forced_tool,
            gemini_service_tier=config.llm_gemini_service_tier,
            groq_service_tier=config.llm_groq_service_tier,
            openai_service_tier=config.llm_openai_service_tier,
@@ -1707,6 +1748,7 @@ class MemoryEngine(MemoryEngineInterface):
            ollama_num_ctx=config.llm_ollama_num_ctx,
            litellmrouter_config=config.consolidation_llm_litellmrouter_config or config.llm_litellmrouter_config,
            bedrock_service_tier=config.llm_bedrock_service_tier,
+            structured_output_forced_tool=config.llm_structured_output_forced_tool,
            gemini_service_tier=config.llm_gemini_service_tier,
            groq_service_tier=config.llm_groq_service_tier,
            openai_service_tier=config.llm_openai_service_tier,
@@ -4260,20 +4302,11 @@ class MemoryEngine(MemoryEngineInterface):
                if "document_id" not in item:
                    item["document_id"] = document_id

-        if outbox_callback is None and outbox_callback_factory is not None:
-            outbox_callback = outbox_callback_factory(contents)
-
-        # Validate no duplicate document_ids in the batch
-        # Having duplicate document_ids causes race conditions in document upserts during parallel processing
-        doc_ids = [item.get("document_id") for item in contents if item.get("document_id")]
-        if len(doc_ids) != len(set(doc_ids)):
-            from collections import Counter
-
-            duplicates = [doc_id for doc_id, count in Counter(doc_ids).items() if count > 1]
-            raise ValueError(
-                f"Batch contains duplicate document_ids: {duplicates}. "
-                f"Each content item in a batch must have a unique document_id to avoid race conditions."
-            )
+        # NOTE: items sharing a document_id are ALLOWED here and folded into one
+        # document (see the grouping dispatch below). The synchronous in-process
+        # path processes sub-batches sequentially, so same-document items cannot
+        # race each other — unlike the queued path, which still rejects
+        # duplicates (see submit_async_retain).

        # Validate update_mode=append requires document_id
        for item in contents:
@@ -4294,6 +4327,236 @@ class MemoryEngine(MemoryEngineInterface):
                    "document text is not stored and cannot be appended to. Use update_mode='replace' instead."
                )

+        # Fold items that share an explicit document_id into one document. On the
+        # synchronous in-process path this is safe — sub-batches run sequentially,
+        # so same-document items cannot race (unlike the queued path, which still
+        # rejects duplicates; see submit_async_retain). Each shared-document group
+        # is processed in ONE orchestrator pass rather than being token-split:
+        # splitting one document across sub-batches that carry different bodies
+        # trips the streaming pipeline's content-hash ownership check and silently
+        # drops the later sub-batches (that path is safe only for an oversized
+        # SINGLE item, whose slices all replay the same full body). The
+        # orchestrator streams a large document chunk-batch by chunk-batch on its
+        # own, so a single pass stays memory-bounded.
+        explicit_doc_ids = [item.get("document_id") for item in contents if item.get("document_id")]
+        has_shared_document = len(explicit_doc_ids) != len(set(explicit_doc_ids))
+
+        if not has_shared_document:
+            # No document is shared, so distinct-document items may be packed and
+            # token-split across sub-batches as before (the orchestrator keeps
+            # genuinely distinct per-item document_ids separate within a pass).
+            execution = await self._run_retain_execution(
+                bank_id=bank_id,
+                contents=contents,
+                request_context=request_context,
+                document_id=document_id,
+                fact_type_override=fact_type_override,
+                document_tags=document_tags,
+                operation_id=operation_id,
+                strategy=strategy,
+                outbox_callback=outbox_callback,
+                outbox_callback_factory=outbox_callback_factory,
+                start_time=start_time,
+            )
+            result = execution.unit_ids
+            total_usage = execution.usage
+            total_processed_content_tokens = execution.processed_content_tokens
+            cancelled = execution.cancelled
+        else:
+            # Group in first-appearance order: each shared document_id becomes one
+            # group, and each item without an explicit document_id becomes its own
+            # group (its own document).
+            groups: list[_RetainGroup] = []
+            groups_by_doc_id: dict[str, _RetainGroup] = {}
+            for idx, item in enumerate(contents):
+                item_doc_id = item.get("document_id")
+                existing = groups_by_doc_id.get(item_doc_id) if item_doc_id is not None else None
+                if existing is not None:
+                    existing.origins.append(idx)
+                    existing.contents.append(item)
+                    continue
+                group = _RetainGroup(document_id=item_doc_id, origins=[idx], contents=[item])
+                groups.append(group)
+                if item_doc_id is not None:
+                    groups_by_doc_id[item_doc_id] = group
+
+            result = [[] for _ in contents]
+            total_usage = TokenUsage()
+            total_processed_content_tokens = 0
+            cancelled = False
+            for group_idx, group in enumerate(groups):
+                # Checkpoint: abort if the operation was deleted (bank deleted)
+                # between documents, mirroring the sub-batch loop's checkpoint.
+                if operation_id and not await self._check_op_alive(operation_id):
+                    logger.info(
+                        f"[BATCH_RETAIN] bank={bank_id} operation {operation_id} cancelled (bank deleted), "
+                        f"stopping after {group_idx}/{len(groups)} documents"
+                    )
+                    cancelled = True
+                    break
+
+                set_stage(f"batch_retain.document.{group_idx + 1}")
+
+                # Per-document webhook rows come from the factory, rebuilt for each
+                # group's contents. A raw pre-built callback (no factory) covers the
+                # whole operation, so fire it once, on the last group.
+                is_last_group = group_idx == len(groups) - 1
+                if outbox_callback_factory is not None:
+                    group_outbox_callback = outbox_callback_factory(group.contents)
+                else:
+                    group_outbox_callback = outbox_callback if is_last_group else None
+
+                group_result, group_usage, group_processed = await self._retain_batch_async_internal(
+                    bank_id=bank_id,
+                    contents=group.contents,
+                    request_context=request_context,
+                    document_id=group.document_id,
+                    is_first_batch=True,
+                    fact_type_override=fact_type_override,
+                    document_tags=document_tags,
+                    operation_id=operation_id,
+                    strategy=strategy,
+                    outbox_callback=group_outbox_callback,
+                )
+                for local_idx, origin_idx in enumerate(group.origins):
+                    if local_idx < len(group_result):
+                        result[origin_idx] = group_result[local_idx]
+                total_usage = total_usage + group_usage
+                if total_processed_content_tokens is None or group_processed is None:
+                    total_processed_content_tokens = None
+                else:
+                    total_processed_content_tokens = total_processed_content_tokens + group_processed
+
+        # A cancelled run (bank deleted mid-flight) skips the completion side
+        # effects, mirroring the pre-grouping early return from the sub-batch loop.
+        if cancelled:
+            if return_usage:
+                return result, total_usage
+            return result
+
+        await self._write_retain_outcome_metadata(operation_id, result)
+
+        # Call post-operation hook if validator is configured
+        if self._operation_validator:
+            from hindsight_api.extensions import RetainResult
+
+            result_ctx = RetainResult(
+                bank_id=bank_id,
+                contents=contents_copy,
+                request_context=request_context,
+                document_id=document_id,
+                fact_type_override=fact_type_override,
+                unit_ids=result,
+                success=True,
+                error=None,
+                llm_input_tokens=total_usage.input_tokens,
+                llm_output_tokens=total_usage.output_tokens,
+                llm_total_tokens=total_usage.total_tokens,
+                llm_cached_input_tokens=getattr(total_usage, "cached_tokens", 0) or 0,
+                llm_thoughts_tokens=getattr(total_usage, "thoughts_tokens", 0) or 0,
+                processed_content_tokens=total_processed_content_tokens,
+            )
+            try:
+                await self._operation_validator.on_retain_complete(result_ctx)
+            except Exception as e:
+                logger.warning(f"Post-retain hook error (non-fatal): {e}")
+
+        # Same async side effects every fact insert triggers (retain or import).
+        await self._submit_post_insert_maintenance(bank_id, request_context)
+
+        if return_usage:
+            return result, total_usage
+        return result
+
+    async def _submit_post_insert_maintenance(
+        self,
+        bank_id: str,
+        request_context: "RequestContext",
+        config: HindsightConfig | None = None,
+    ) -> None:
+        """Submit the async side effects that follow any fact insert (retain or import).
+
+        Shared by the retain pipeline and the document-import pipeline so imported
+        documents aren't second-class citizens:
+          * auto-consolidation (when observations + auto-consolidation are enabled
+            for the bank) so freshly inserted facts get observations;
+          * graph maintenance, which short-circuits when no cleanup work was
+            enqueued, so a plain insert pays a single cheap indexed SELECT here.
+
+        Both are non-critical: failures are logged, never raised, so they can't
+        fail the operation that produced the facts. Pass ``config`` when the caller
+        already resolved it to avoid a redundant lookup.
+        """
+        if config is None:
+            config = await self._config_resolver.resolve_full_config(bank_id, request_context)
+        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:
+                logger.warning(f"Failed to submit consolidation task for bank {bank_id}: {e}")
+        try:
+            await self.submit_async_graph_maintenance(bank_id=bank_id, request_context=request_context)
+        except Exception as e:
+            logger.warning(f"Failed to submit graph maintenance task for bank {bank_id}: {e}")
+
+    async def _resolve_retain_config(
+        self,
+        bank_id: str,
+        request_context: "RequestContext",
+        strategy: str | None,
+    ) -> HindsightConfig:
+        """Resolve the config a retain runs under, strategy overrides applied.
+
+        Mirrors what ``_retain_batch_async_internal`` resolves before handing
+        config to the orchestrator, so anything the splitting caller derives
+        from it (chunk boundaries, Memory Defense screening) matches what the
+        orchestrator then does with each sub-batch.
+        """
+        from hindsight_api.config_resolver import apply_strategy
+
+        resolved_config = await self._config_resolver.resolve_full_config(bank_id, request_context)
+        effective_strategy = strategy or resolved_config.retain_default_strategy
+        if effective_strategy:
+            resolved_config = apply_strategy(resolved_config, effective_strategy)
+        return resolved_config
+
+    @staticmethod
+    def _retain_chunking_config(config: HindsightConfig) -> _RetainChunkingConfig:
+        """The chunk boundaries ``config`` implies, as the retain pipeline uses them."""
+        return _RetainChunkingConfig(
+            chunk_size=getattr(config, "retain_chunk_size", DEFAULT_RETAIN_CHUNK_SIZE),
+            structured_chunk_size=getattr(config, "retain_structured_chunk_size", None),
+        )
+
+    async def _run_retain_execution(
+        self,
+        *,
+        bank_id: str,
+        contents: list[RetainContentDict],
+        request_context: "RequestContext",
+        document_id: str | None,
+        fact_type_override: str | None,
+        document_tags: list[str] | None,
+        operation_id: str | None,
+        strategy: str | None,
+        outbox_callback: RetainOutboxCallback | None,
+        outbox_callback_factory: RetainOutboxCallbackFactory | None,
+        start_time: float,
+    ) -> _RetainExecutionResult:
+        """Run a batch with no shared document_id through the token splitter and
+        the sequential sub-batch loop (or a single pass for a small batch).
+
+        The orchestrator still separates genuinely distinct per-item document_ids
+        within one pass, and the only document that spans sub-batches here is an
+        oversized SINGLE item, whose slices all replay the same body — so the
+        per-document chunk_index offset stays valid. Shared-document batches are
+        handled by ``retain_batch_async`` itself (one pass per document), never
+        here, because splitting one document across differently-bodied sub-batches
+        trips the streaming pipeline's content-hash ownership check.
+        """
+        if outbox_callback is None and outbox_callback_factory is not None:
+            outbox_callback = outbox_callback_factory(contents)
+
        # Auto-chunk large batches by token count to avoid timeouts and memory issues
        # Calculate total token count
        total_tokens = sum(count_tokens(item.get("content", "")) for item in contents)
@@ -4303,6 +4566,7 @@ class MemoryEngine(MemoryEngineInterface):
        # means that sub-batch bypassed dedup, so the aggregate is None
        # (see RetainResult.processed_content_tokens).
        total_processed_content_tokens: int | None = 0
+        cancelled = False

        # Get batch size threshold from config
        config = get_config()
@@ -4398,9 +4662,8 @@ class MemoryEngine(MemoryEngineInterface):
                    logger.info(
                        f"[BATCH_RETAIN] bank={bank_id} operation {operation_id} cancelled (bank deleted), stopping after {i - 1}/{len(sub_batches)} sub-batches"
                    )
-                    if return_usage:
-                        return per_input_results, total_usage
-                    return per_input_results
+                    cancelled = True
+                    break

                sub_batch_tokens = sum(count_tokens(item.get("content", "")) for item in sub_batch)
                logger.info(
@@ -4412,10 +4675,11 @@ class MemoryEngine(MemoryEngineInterface):
                set_stage(f"batch_retain.sub_batch.{i}")

                # Resolve the document this sub-batch writes to so we can offset
-                # its chunk_index past chunks already stored by earlier
-                # sub-batches of the same document. Only the oversized-single-item
-                # split shares a document_id across sub-batches; packed multi-item
-                # sub-batches carry distinct document_ids (offset stays 0).
+                # its chunk_index past chunks already stored by earlier sub-batches
+                # of the same document. A grouped call passes ``document_id``, so
+                # every sub-batch shares it; otherwise only the oversized-single-
+                # item split shares a document_id across sub-batches (packed
+                # multi-item sub-batches carry distinct document_ids, offset 0).
                sub_doc_id = document_id or (sub_batch[0].get("document_id") if len(sub_batch) == 1 else None)
                sub_offset = chunk_offsets.get(sub_doc_id, 0) if sub_doc_id else 0

@@ -4495,98 +4759,11 @@ class MemoryEngine(MemoryEngineInterface):
            # Progress for this path is emitted by the streaming pipeline as
            # "storing N/total chunks" via progress_callback (see _retain_batch_async_internal).

-        await self._write_retain_outcome_metadata(operation_id, result)
-
-        # Call post-operation hook if validator is configured
-        if self._operation_validator:
-            from hindsight_api.extensions import RetainResult
-
-            result_ctx = RetainResult(
-                bank_id=bank_id,
-                contents=contents_copy,
-                request_context=request_context,
-                document_id=document_id,
-                fact_type_override=fact_type_override,
-                unit_ids=result,
-                success=True,
-                error=None,
-                llm_input_tokens=total_usage.input_tokens,
-                llm_output_tokens=total_usage.output_tokens,
-                llm_total_tokens=total_usage.total_tokens,
-                llm_cached_input_tokens=getattr(total_usage, "cached_tokens", 0) or 0,
-                llm_thoughts_tokens=getattr(total_usage, "thoughts_tokens", 0) or 0,
-                processed_content_tokens=total_processed_content_tokens,
-            )
-            try:
-                await self._operation_validator.on_retain_complete(result_ctx)
-            except Exception as e:
-                logger.warning(f"Post-retain hook error (non-fatal): {e}")
-
-        # Same async side effects every fact insert triggers (retain or import).
-        await self._submit_post_insert_maintenance(bank_id, request_context)
-
-        if return_usage:
-            return result, total_usage
-        return result
-
-    async def _submit_post_insert_maintenance(
-        self,
-        bank_id: str,
-        request_context: "RequestContext",
-        config: HindsightConfig | None = None,
-    ) -> None:
-        """Submit the async side effects that follow any fact insert (retain or import).
-
-        Shared by the retain pipeline and the document-import pipeline so imported
-        documents aren't second-class citizens:
-          * auto-consolidation (when observations + auto-consolidation are enabled
-            for the bank) so freshly inserted facts get observations;
-          * graph maintenance, which short-circuits when no cleanup work was
-            enqueued, so a plain insert pays a single cheap indexed SELECT here.
-
-        Both are non-critical: failures are logged, never raised, so they can't
-        fail the operation that produced the facts. Pass ``config`` when the caller
-        already resolved it to avoid a redundant lookup.
-        """
-        if config is None:
-            config = await self._config_resolver.resolve_full_config(bank_id, request_context)
-        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:
-                logger.warning(f"Failed to submit consolidation task for bank {bank_id}: {e}")
-        try:
-            await self.submit_async_graph_maintenance(bank_id=bank_id, request_context=request_context)
-        except Exception as e:
-            logger.warning(f"Failed to submit graph maintenance task for bank {bank_id}: {e}")
-
-    async def _resolve_retain_config(
-        self,
-        bank_id: str,
-        request_context: "RequestContext",
-        strategy: str | None,
-    ) -> HindsightConfig:
-        """Resolve the config a retain runs under, strategy overrides applied.
-
-        Mirrors what ``_retain_batch_async_internal`` resolves before handing
-        config to the orchestrator, so anything the splitting caller derives
-        from it (chunk boundaries, Memory Defense screening) matches what the
-        orchestrator then does with each sub-batch.
-        """
-        from hindsight_api.config_resolver import apply_strategy
-
-        resolved_config = await self._config_resolver.resolve_full_config(bank_id, request_context)
-        effective_strategy = strategy or resolved_config.retain_default_strategy
-        if effective_strategy:
-            resolved_config = apply_strategy(resolved_config, effective_strategy)
-        return resolved_config
-
-    @staticmethod
-    def _retain_chunking_config(config: HindsightConfig) -> _RetainChunkingConfig:
-        """The chunk boundaries ``config`` implies, as the retain pipeline uses them."""
-        return _RetainChunkingConfig(
-            chunk_size=getattr(config, "retain_chunk_size", DEFAULT_RETAIN_CHUNK_SIZE),
-            structured_chunk_size=getattr(config, "retain_structured_chunk_size", None),
+        return _RetainExecutionResult(
+            unit_ids=result,
+            usage=total_usage,
+            processed_content_tokens=total_processed_content_tokens,
+            cancelled=cancelled,
        )

    async def _retain_batch_async_internal(
@@ -6016,7 +6193,7 @@ class MemoryEngine(MemoryEngineInterface):
                from .memories import get_memories

                _obs_store = get_memories()
-                if observation_ids_ordered and not _obs_store.writes_memory_rows_in_sql:
+                if observation_ids_ordered and not _obs_store.writes_memory_rows_in_sql_for(bank_id):
                    # A store that keeps memories outside SQL: fetch each observation, then its
                    # source memories, for their chunk_ids — the join the SQL branch does, walked
                    # in observation-rank order so per-observation grouping is preserved.
@@ -6098,7 +6275,7 @@ class MemoryEngine(MemoryEngineInterface):
                    # row, so it selects one fewer column and keeps the asyncpg Records as-is — no
                    # per-chunk ``dict`` allocation for an overlay it never runs.
                    _chunk_store = get_memories()
-                    _owns_docs = _chunk_store.owns_document_store
+                    _owns_docs = _chunk_store.owns_document_store_for(bank_id)
                    if _owns_docs:
                        _chunk_cols = "chunk_id, chunk_text, chunk_index, document_id"
                    else:
@@ -6295,7 +6472,7 @@ class MemoryEngine(MemoryEngineInterface):
                        # Resolve each observation's sources. This is a recall hot path, so the SQL
                        # store reads only the two columns it needs rather than a full memory row; a
                        # store that owns its rows answers from its own objects via one addressed read.
-                        if store.writes_memory_rows_in_sql:
+                        if store.writes_memory_rows_in_sql_for(bank_id):
                            obs_rows = [
                                {"id": str(r["id"]), "source_memory_ids": r["source_memory_ids"]}
                                for r in await sf_conn.fetch(
@@ -6332,7 +6509,7 @@ class MemoryEngine(MemoryEngineInterface):
                        # needed, so the SQL store selects those (bank-scoped) instead of the full
                        # 17-column memory row — the difference is measurable on this hot path.
                        if source_ids_ordered:
-                            if store.writes_memory_rows_in_sql:
+                            if store.writes_memory_rows_in_sql_for(bank_id):
                                source_row_by_id = {
                                    str(r["id"]): _source_fact_dict(
                                        uid=str(r["id"]),
@@ -6705,7 +6882,7 @@ class MemoryEngine(MemoryEngineInterface):
            from .memories import get_memories

            _store = get_memories()
-            if _store.writes_memory_rows_in_sql:
+            if _store.writes_memory_rows_in_sql_for(bank_id):
                # Use a subquery for counts to avoid GROUP BY on CLOB columns
                # (Oracle cannot use CLOB types as comparison keys in GROUP BY).
                doc = await conn.fetchrow(
@@ -6767,7 +6944,7 @@ class MemoryEngine(MemoryEngineInterface):
                    # A store that owns the document store keeps the extracted text in
                    # its own store, not in documents.original_text (which is NULL here). Overlay
                    # it from the store so get_document still returns the body.
-                    if _store.owns_document_store:
+                    if _store.owns_document_store_for(bank_id):
                        _rec = await _store.get_document_record(
                            bank_id=bank_id, document_id=document_id, include_text=True
                        )
@@ -6844,7 +7021,7 @@ class MemoryEngine(MemoryEngineInterface):
                from .memories import get_memories

                _store = get_memories()
-                if _store.writes_memory_rows_in_sql:
+                if _store.writes_memory_rows_in_sql_for(bank_id):
                    unit_rows = await conn.fetch(
                        f"SELECT id FROM {fq_table('memory_units')} WHERE document_id = $1 AND fact_type IN ('experience', 'world')",
                        document_id,
@@ -6890,7 +7067,7 @@ class MemoryEngine(MemoryEngineInterface):
                # cascade to its memories (they are not SQL rows) — drop them through the store,
                # tagged with a write-group so the store tombstone commits atomically with the
                # Postgres document delete (a rolled-back delete must not orphan the memories).
-                if deleted and not _store.writes_memory_rows_in_sql:
+                if deleted and not _store.writes_memory_rows_in_sql_for(bank_id):
                    _del_txn = await _store.begin_txn(conn=conn, fq_table=fq_table, bank_id=bank_id, mutating=True)
                    await _store.delete_document(
                        conn=conn, fq_table=fq_table, bank_id=bank_id, document_id=document_id, txn=_del_txn
@@ -6899,7 +7076,7 @@ class MemoryEngine(MemoryEngineInterface):
                    # extracted text + chunk bodies; the orphan sweep reclaims the blobs), under the
                    # same write-group so it commits atomically with the Postgres document delete.
                    # This is the EXPLICIT deletion — distinct from the re-ingest facts-delete above.
-                    if _store.owns_document_store:
+                    if _store.owns_document_store_for(bank_id):
                        await _store.delete_document_record(bank_id=bank_id, document_id=document_id, txn=_del_txn)

                # Invalidate observations referencing these (now-deleted) memories
@@ -7012,7 +7189,7 @@ class MemoryEngine(MemoryEngineInterface):
                    from .memories import MemoryPatch, get_memories

                    _store = get_memories()
-                if tags is not None and not _store.writes_memory_rows_in_sql:
+                if tags is not None and not _store.writes_memory_rows_in_sql_for(bank_id):
                    # A store that keeps memories outside SQL: retag the document's memories, then
                    # invalidate the observations built on them and requeue their sources so the
                    # next consolidation rebuilds them under the new tags (the cascade the SQL
@@ -7186,7 +7363,7 @@ class MemoryEngine(MemoryEngineInterface):
                from .memories import get_memories

                _store = get_memories()
-                if _store.writes_memory_rows_in_sql:
+                if _store.writes_memory_rows_in_sql_for(bank_id):
                    row = await conn.fetchrow(
                        f"SELECT bank_id, fact_type FROM {fq_table('memory_units')} WHERE id = $1",
                        str(unit_uuid),
@@ -7215,7 +7392,7 @@ class MemoryEngine(MemoryEngineInterface):
                # observations inserted concurrently by consolidation (otherwise a
                # racing insert committed between the sweep and the delete would
                # leave an orphan referencing this just-deleted source memory).
-                if _store.writes_memory_rows_in_sql:
+                if _store.writes_memory_rows_in_sql_for(bank_id):
                    deleted = await conn.fetchval(
                        f"DELETE FROM {fq_table('memory_units')} WHERE id = $1 RETURNING id", unit_id
                    )
@@ -7529,7 +7706,7 @@ class MemoryEngine(MemoryEngineInterface):
                            from .memories import get_memories as _get_memories_for_scope

                            _scope_store = _get_memories_for_scope()
-                            if _scope_store.writes_memory_rows_in_sql:
+                            if _scope_store.writes_memory_rows_in_sql_for(bank_id):
                                unit_id_rows = await conn.fetch(
                                    f"SELECT id FROM {fq_table('memory_units')} WHERE bank_id = $1 AND fact_type = $2",
                                    bank_id,
@@ -7679,7 +7856,7 @@ class MemoryEngine(MemoryEngineInterface):
        from .memories import DeletePredicate, get_memories

        store = get_memories()
-        if not store.writes_memory_rows_in_sql:
+        if not store.writes_memory_rows_in_sql_for(bank_id):
            if fact_type:
                await store.delete_where(bank_id, DeletePredicate(fact_types=[fact_type]))
            else:
@@ -7729,7 +7906,7 @@ class MemoryEngine(MemoryEngineInterface):
        backend = await self._get_backend()
        async with acquire_with_retry(backend) as conn:
            async with conn.transaction():
-                if store.writes_memory_rows_in_sql:
+                if store.writes_memory_rows_in_sql_for(bank_id):
                    # Count observations before deletion
                    count = await conn.fetchval(
                        f"SELECT COUNT(*) FROM {fq_table('memory_units')} WHERE bank_id = $1 AND fact_type = 'observation'",
@@ -7862,7 +8039,7 @@ class MemoryEngine(MemoryEngineInterface):
        store = get_memories()
        backend = await self._get_backend()
        async with acquire_with_retry(backend) as conn:
-            if store.writes_memory_rows_in_sql:
+            if store.writes_memory_rows_in_sql_for(bank_id):
                count = await conn.fetchval(
                    f"""
                    SELECT COUNT(*) FROM {fq_table("memory_units")}
@@ -7942,7 +8119,7 @@ class MemoryEngine(MemoryEngineInterface):
                    from .memories import get_memories

                    _store = get_memories()
-                    if _store.writes_memory_rows_in_sql:
+                    if _store.writes_memory_rows_in_sql_for(bank_id):
                        await conn.execute(
                            f"""
                            UPDATE {fq_table("memory_units")}
@@ -9390,7 +9567,7 @@ class MemoryEngine(MemoryEngineInterface):
            from .memories import get_memories

            _store = get_memories()
-            if _store.owns_document_store:
+            if _store.owns_document_store_for(chunk["bank_id"]):
                _t = await _store.get_chunk_text(
                    bank_id=chunk["bank_id"],
                    document_id=chunk["document_id"],
@@ -9481,7 +9658,7 @@ class MemoryEngine(MemoryEngineInterface):
            from .memories import get_memories

            _store = get_memories()
-            if _store.owns_document_store:
+            if _store.owns_document_store_for(bank_id):
                _texts = await _store.list_chunk_texts(bank_id=bank_id, document_id=document_id)
                if _texts is not None:
                    _texts_by_index = dict(enumerate(_texts))
@@ -11631,7 +11808,7 @@ class MemoryEngine(MemoryEngineInterface):
            from .memories import get_memories

            _store = get_memories()
-            if _store.writes_memory_rows_in_sql:
+            if _store.writes_memory_rows_in_sql_for(bank_id):
                consolidation_row = await conn.fetchrow(
                    f"""
                    SELECT
@@ -15758,16 +15935,22 @@ class MemoryEngine(MemoryEngineInterface):
            if replay is not None:
                return replay

-        # Validate no duplicate document_ids in the batch
-        # Having duplicate document_ids causes race conditions in document upserts during parallel processing
+        # Reject duplicate document_ids on the QUEUED path only. Children fan out
+        # to workers that claim them in parallel with no per-document gate, and
+        # append is a non-transactional read-modify-write, so concurrent appends
+        # to one document lose updates. The synchronous path (async=false) folds
+        # shared-document items into one document safely — sub-batches there run
+        # sequentially — so a client that needs this should send async=false.
        doc_ids = [item.get("document_id") for item in contents if item.get("document_id")]
        if len(doc_ids) != len(set(doc_ids)):
            from collections import Counter

            duplicates = [doc_id for doc_id, count in Counter(doc_ids).items() if count > 1]
            raise ValueError(
-                f"Batch contains duplicate document_ids: {duplicates}. "
-                f"Each content item in a batch must have a unique document_id to avoid race conditions."
+                f"Batch contains duplicate document_ids: {duplicates}. Each content item in an "
+                f"async batch must have a unique document_id to avoid races between the parallel "
+                f"workers that process them. To fold several items into one document, send the "
+                f"batch synchronously (async=false), which processes them sequentially."
            )

        # Calculate total token count and determine if we need to split
@@ -40,6 +40,10 @@ from hindsight_api.worker.stage import set_stage

 logger = logging.getLogger(__name__)

+# Name of the single tool used when structured output is routed through a forced
+# tool call instead of ``response_format`` (see ``structured_output_forced_tool``).
+_STRUCTURED_TOOL_NAME = "structured_response"
+

 def _usage_from_litellm_response(response: Any) -> LLMResponseUsage:
    """Extract prompt/completion/cached token counts from a LiteLLM (OpenAI-shaped) usage block."""
@@ -57,6 +61,22 @@ def _usage_from_litellm_response(response: Any) -> LLMResponseUsage:
    )


+def _forced_tool_arguments(message: Any) -> str | None:
+    """Return the structured-output tool call's arguments as a JSON string.
+
+    ``None`` when the model answered with plain text instead — some gateways drop
+    ``tool_choice`` — so the caller can fall back to parsing the message content.
+    """
+    for tool_call in message.tool_calls or []:
+        if tool_call.function.name != _STRUCTURED_TOOL_NAME:
+            continue
+        # LiteLLM normalizes to the OpenAI shape (a JSON string), but some
+        # providers hand back an already-decoded object.
+        arguments = tool_call.function.arguments
+        return arguments if isinstance(arguments, str) else json.dumps(arguments)
+    return None
+
+
 class LiteLLMLLM(LLMInterface):
    """
    LLM provider using the LiteLLM SDK for universal model support.
@@ -82,6 +102,7 @@ class LiteLLMLLM(LLMInterface):
        extra_body: dict[str, Any] | None = None,
        bedrock_service_tier: str | None = None,
        default_headers: dict[str, Any] | None = None,
+        structured_output_forced_tool: bool = False,
        **kwargs: Any,
    ):
        super().__init__(provider, api_key, base_url, model, reasoning_effort, **kwargs)
@@ -103,6 +124,12 @@ class LiteLLMLLM(LLMInterface):
        # copy is handed to each call below to avoid cross-request contamination.
        self._default_headers: dict[str, Any] = dict(default_headers or {})
        self.bedrock_service_tier = bedrock_service_tier
+        # Ask for structured output via a single forced tool call instead of
+        # ``response_format``. Opt-in (HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL)
+        # for backends that reject the response_format route — Bedrock Claude's
+        # Converse layer refuses the translated ``outputConfig`` in some regions but
+        # accepts the same schema as a tool (#3300).
+        self.structured_output_forced_tool = structured_output_forced_tool

        try:
            import litellm
@@ -243,16 +270,38 @@ class LiteLLMLLM(LLMInterface):
        call_kwargs = self._build_common_kwargs(messages, max_completion_tokens, temperature)

        # Add JSON schema response format if provided
+        use_forced_tool = False
        if response_format is not None and hasattr(response_format, "model_json_schema"):
            schema = strict_json_schema(response_format) if strict_schema else response_format.model_json_schema()
-            call_kwargs["response_format"] = {
-                "type": "json_schema",
-                "json_schema": {
-                    "name": response_format.__name__ if hasattr(response_format, "__name__") else "response",
-                    "schema": schema,
-                    "strict": strict_schema,
-                },
-            }
+            schema_name = response_format.__name__ if hasattr(response_format, "__name__") else "response"
+            if self.structured_output_forced_tool:
+                # The schema travels as the tool's parameters and the model is forced
+                # to call it; the arguments are substituted for the message content
+                # below, so the parse/validate, retry and usage paths are unchanged.
+                use_forced_tool = True
+                call_kwargs["tools"] = [
+                    {
+                        "type": "function",
+                        "function": {
+                            "name": _STRUCTURED_TOOL_NAME,
+                            "description": f"Return the structured response ({schema_name}).",
+                            "parameters": schema,
+                        },
+                    }
+                ]
+                call_kwargs["tool_choice"] = {
+                    "type": "function",
+                    "function": {"name": _STRUCTURED_TOOL_NAME},
+                }
+            else:
+                call_kwargs["response_format"] = {
+                    "type": "json_schema",
+                    "json_schema": {
+                        "name": schema_name,
+                        "schema": schema,
+                        "strict": strict_schema,
+                    },
+                }

        last_exception = None

@@ -269,10 +318,19 @@ class LiteLLMLLM(LLMInterface):
                # these tokens (#2387).
                stash_response_usage(_usage_from_litellm_response(response))

-                content = response.choices[0].message.content or ""
+                message = response.choices[0].message
+                content = message.content or ""
                finish_reason = response.choices[0].finish_reason
                model_name = self._resolve_completion_model(response)

+                if use_forced_tool:
+                    # Forced tool call: its arguments ARE the structured response.
+                    # Absent (a gateway that drops tool_choice) -> keep the text
+                    # content so the existing parse path still has a chance.
+                    forced_arguments = _forced_tool_arguments(message)
+                    if forced_arguments is not None:
+                        content = forced_arguments
+
                # Check for length-limited output
                if finish_reason == "length":
                    raise OutputTooLongError("LiteLLM response was truncated due to token limit")
@@ -406,7 +406,7 @@ async def tool_expand(
    from ..memories import get_memories

    _store = get_memories()
-    if _store.writes_memory_rows_in_sql:
+    if _store.writes_memory_rows_in_sql_for(bank_id):
        memories = await conn.fetch(
            f"""
            SELECT id, text, chunk_id, document_id, fact_type, context
@@ -499,7 +499,7 @@ async def list_banks(pool) -> list:
            last_write = max(write_times) if write_times else None

            fact_count = row["fact_count"]
-            if not _store.writes_memory_rows_in_sql:
+            if not _store.writes_memory_rows_in_sql_for(row["bank_id"]):
                fact_count = sum(
                    (await _store.count_memories(conn=conn, fq_table=fq_table, bank_id=row["bank_id"])).values()
                )
@@ -14,6 +14,9 @@ from .types import ChunkMetadata

 logger = logging.getLogger(__name__)

+# Page size for walking the facts a chunk owns out of a store that keeps memories outside SQL.
+_OUTGOING_PAGE = 500
+

 def compute_chunk_hash(chunk_text: str) -> str:
    """Compute SHA256 hash of chunk text for delta comparison."""
@@ -55,7 +58,54 @@ async def load_existing_chunks(conn, bank_id: str, document_id: str) -> list[Exi
    ]


-async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None = None, txn=None) -> None:
+async def memory_ids_for_chunks(conn, bank_id: str, chunk_ids: list[str]) -> list[str]:
+    """Ids of the facts these chunks own, asked of whichever store holds them.
+
+    The SQL store keeps ``chunk_id`` as a column; a store that keeps memories outside SQL
+    carries it in the metadata bag (the same key its ``delete_where`` predicate matches on),
+    so the two are read differently. Only ``experience``/``world`` units are returned:
+    observations are not chunk-scoped, and feeding one back as a *source* id would be
+    meaningless. Paged to exhaustion — every id is about to be deleted, and a chunk whose
+    facts overflow one page must not keep half of them.
+    """
+    from ..memories import META_CHUNK_ID, get_memories
+
+    store = get_memories()
+    if store.writes_memory_rows_in_sql:
+        rows = await conn.fetch(
+            f"""
+            SELECT id
+            FROM {fq_table("memory_units")}
+            WHERE bank_id = $1
+              AND chunk_id = ANY($2::text[])
+              AND fact_type IN ('experience', 'world')
+            """,
+            bank_id,
+            chunk_ids,
+        )
+        return [str(row["id"]) for row in rows]
+
+    unit_ids: list[str] = []
+    for chunk_id in chunk_ids:
+        page_token = ""
+        while True:
+            page = await store.scan_memories(
+                conn=conn,
+                fq_table=fq_table,
+                bank_id=bank_id,
+                fact_types=["experience", "world"],
+                metadata_equals={META_CHUNK_ID: chunk_id},
+                limit=_OUTGOING_PAGE,
+                page_token=page_token,
+            )
+            unit_ids.extend(m.unit_id for m in page.memories)
+            page_token = page.next_page_token
+            if not page_token:
+                break
+    return unit_ids
+
+
+async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None = None, txn=None, ops=None) -> int:
    """
    Delete specific chunks by their IDs.

@@ -66,9 +116,29 @@ async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None =
    the store's tombstones must ride the same txn as the replacement writes so they commit
    (become visible) together — otherwise an aborted re-ingest could drop the old memories
    without landing the new ones.
+
+    ``ops`` is the backend-specific DataAccessOps the observation sweep below needs to choose
+    the PG (native array) vs Oracle (junction table) read path — pass ``pool.ops``.
+
+    Returns the number of observations invalidated by the sweep, so the caller can log it.
    """
    if not chunk_ids:
-        return
+        return 0
+
+    # Delete the observations derived from the facts these chunks own, BEFORE the facts
+    # themselves go. Nothing can reach those observations afterwards: consolidation batches
+    # are built from facts, so an observation whose sources are all deleted is never selected
+    # into a batch again, and it stays valid and recallable — stale knowledge from the previous
+    # version of the document surviving the replace (issue #3294). The full-replace path does
+    # this in ``handle_document_tracking``; the delta path deletes facts through this cascade
+    # instead, which is why the sweep has to live here rather than at one of the call sites.
+    invalidated = 0
+    if bank_id:
+        outgoing_unit_ids = await memory_ids_for_chunks(conn, bank_id, chunk_ids)
+        if outgoing_unit_ids:
+            from .fact_storage import delete_stale_observations_for_memories
+
+            invalidated = await delete_stale_observations_for_memories(conn, bank_id, outgoing_unit_ids, ops=ops)

    # The chunks->memory_units FK cascade below does not reach a store that keeps memories
    # outside SQL (its memory_units is empty), so drop the memories carrying each deleted
@@ -76,7 +146,7 @@ async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None =
    from ..memories import META_CHUNK_ID, DeletePredicate, get_memories

    _store = get_memories()
-    if bank_id and not _store.writes_memory_rows_in_sql:
+    if bank_id and not _store.writes_memory_rows_in_sql_for(bank_id):
        for _cid in chunk_ids:
            await _store.delete_where(bank_id, DeletePredicate(metadata_equals={META_CHUNK_ID: _cid}), txn=txn)

@@ -128,6 +198,7 @@ async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None =
        """,
        chunk_ids,
    )
+    return invalidated


 async def store_chunks_batch(
@@ -168,7 +239,7 @@ async def store_chunks_batch(
    # same shape as store_document_text=False, and idempotency is unaffected (content_hash stays).
    from ..memories import get_memories

-    if get_memories().owns_document_store:
+    if get_memories().owns_document_store_for(bank_id):
        store_text = False

    # Prepare chunk data for batch insert
@@ -270,6 +270,14 @@ async def handle_document_tracking(
                    f"[RETAIN] Document {document_id} re-ingested: invalidated "
                    f"{invalidated} observation(s) derived from {len(existing_unit_ids)} outgoing memory_units"
                )
+            else:
+                # Logged even at zero: "the sweep matched nothing" and "the sweep never ran"
+                # are the two candidates whenever orphan observations are reported, and
+                # without this line they look identical from the outside (issue #3294).
+                logger.debug(
+                    f"[RETAIN] Document {document_id} re-ingested: no observations derived from "
+                    f"{len(existing_unit_ids)} outgoing memory_units"
+                )
            # Capture link-recompute victims BEFORE the cascade. Same staleness
            # applies on upsert as on explicit delete: surviving units in OTHER
            # documents that linked to these doomed units are about to lose
@@ -372,7 +380,7 @@ async def _upsert_document_row(
    # the bulky body is written to the store up front (orchestrator._store_document_bodies).
    from ..memories import get_memories

-    if get_memories().owns_document_store:
+    if get_memories().owns_document_store_for(bank_id):
        original_text = None
    await conn.execute(
        f"""
@@ -414,7 +422,7 @@ async def update_memory_units_metadata_and_tags(
    from ..memories import MemoryPatch, get_memories

    store = get_memories()
-    if not store.writes_memory_rows_in_sql:
+    if not store.writes_memory_rows_in_sql_for(bank_id):
        # A store that keeps memories outside SQL: page the document's memories and patch each
        # one's tags through the store — the UPDATE below is a no-op on its empty memory_units.
        page = await store.scan_memories(
@@ -1278,7 +1278,7 @@ async def _store_document_bodies(
    from ..memories import get_memories

    store = get_memories()
-    if not store.owns_document_store:
+    if not store.owns_document_store_for(bank_id):
        return
    # The record's content_hash must equal what the SQL documents row stores, so a read is
    # consistent whichever it comes from: sanitize + sha256 the same combined_content. The
@@ -2573,10 +2573,13 @@ async def _try_delta_retain(
                    for idx in changed_indices + removed_indices
                    if idx in existing_by_index
                ]
-                await chunk_storage.delete_chunks_by_ids(conn, chunks_to_delete, bank_id, txn=_group_txn)
+                invalidated_obs = await chunk_storage.delete_chunks_by_ids(
+                    conn, chunks_to_delete, bank_id, txn=_group_txn, ops=pool.ops
+                )
                log_buffer.append(
                    f"  Deleted {len(chunks_to_delete)} chunks "
-                    f"({len(changed_indices)} changed + {len(removed_indices)} removed) "
+                    f"({len(changed_indices)} changed + {len(removed_indices)} removed), "
+                    f"invalidated {invalidated_obs} observation(s) "
                    f"in {time.time() - step_start:.3f}s"
                )

@@ -46,21 +46,134 @@ async def test_duplicate_document_ids_rejected_async(memory, request_context):


@pytest.mark.asyncio
-async def test_duplicate_document_ids_rejected_sync(memory, request_context):
-    """Test that sync retain also rejects batches with duplicate document_ids."""
-    bank_id = "test_duplicate_sync"
-    contents = [
-        {"content": "First item", "document_id": "doc1"},
-        {"content": "Second item", "document_id": "doc1"},  # Duplicate!
-    ]
+async def test_shared_document_id_folds_sync(memory, request_context):
+    """Sync retain accepts several items sharing one document_id and folds them
+    into a single document, in request order (the documented RetainRequest
+    example — see issue #3363)."""
+    bank_id = f"test_shared_doc_sync_{uuid.uuid4().hex}"
+    try:
+        contents = [
+            {"content": "Alice works at Google", "context": "work", "document_id": "conversation_123"},
+            {"content": "Bob went hiking yesterday", "document_id": "conversation_123"},
+        ]

-    # Should raise ValueError due to duplicate document_ids
-    with pytest.raises(ValueError, match="duplicate document_ids.*doc1"):
-        await memory.retain_batch_async(
+        # Must NOT raise (this used to be a 400).
+        result = await memory.retain_batch_async(
            bank_id=bank_id,
            contents=contents,
            request_context=request_context,
        )
+        # One result slot per input content is preserved.
+        assert len(result) == 2
+
+        # The items folded into exactly one document.
+        docs = await memory.list_documents(bank_id=bank_id, request_context=request_context)
+        assert docs["total"] == 1
+        assert docs["items"][0]["id"] == "conversation_123"
+
+        # Both items' content lands in that one document's body, in order.
+        doc = await memory.get_document("conversation_123", bank_id, request_context=request_context)
+        body = doc["original_text"]
+        assert "Alice works at Google" in body
+        assert "Bob went hiking yesterday" in body
+        assert body.index("Alice works at Google") < body.index("Bob went hiking yesterday")
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_batch_level_document_id_folds_sync(memory, request_context):
+    """The deprecated batch-level document_id (applied to every item without its
+    own) folds those items into one document instead of tripping the guard."""
+    bank_id = f"test_batch_doc_id_sync_{uuid.uuid4().hex}"
+    try:
+        result = await memory.retain_batch_async(
+            bank_id=bank_id,
+            contents=[
+                {"content": "Alice works at Google"},
+                {"content": "Bob loves Python"},
+            ],
+            document_id="meeting-2024-01-15",
+            request_context=request_context,
+        )
+        assert len(result) == 2
+
+        docs = await memory.list_documents(bank_id=bank_id, request_context=request_context)
+        assert docs["total"] == 1
+        assert docs["items"][0]["id"] == "meeting-2024-01-15"
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_shared_and_distinct_document_ids_sync(memory, request_context):
+    """A batch mixing a shared document_id with a distinct one folds only the
+    shared items, leaving the distinct document on its own."""
+    bank_id = f"test_mixed_doc_ids_sync_{uuid.uuid4().hex}"
+    try:
+        result = await memory.retain_batch_async(
+            bank_id=bank_id,
+            contents=[
+                {"content": "Alice works at Google", "document_id": "docA"},
+                {"content": "Bob loves Python", "document_id": "docB"},
+                {"content": "Alice also mentors interns", "document_id": "docA"},
+            ],
+            request_context=request_context,
+        )
+        assert len(result) == 3
+
+        docs = await memory.list_documents(bank_id=bank_id, request_context=request_context)
+        assert docs["total"] == 2
+
+        doc_a = await memory.get_document("docA", bank_id, request_context=request_context)
+        assert "Alice works at Google" in doc_a["original_text"]
+        assert "Alice also mentors interns" in doc_a["original_text"]
+
+        doc_b = await memory.get_document("docB", bank_id, request_context=request_context)
+        assert "Bob loves Python" in doc_b["original_text"]
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_shared_document_id_folds_sync_large_batch(memory, request_context):
+    """A shared-document batch large enough to exceed the auto-split token
+    threshold still folds into one document with NO lost content. Splitting one
+    document across sub-batches would trip the streaming pipeline's content-hash
+    ownership check and drop later sub-batches, so shared groups take a single
+    pass — this guards that decision (issue #3363)."""
+    from hindsight_api.engine.memory_engine import count_tokens
+
+    bank_id = f"test_shared_doc_large_sync_{uuid.uuid4().hex}"
+    try:
+        # Two ~5.5k-token items sharing one document_id → ~11k tokens, over the
+        # 10k default split threshold. Distinct markers pin each item's presence.
+        filler = "The quick brown fox jumps over the lazy dog. " * 500
+        contents = [
+            {"content": f"MARKER_ALPHA at the start. {filler}", "document_id": "big_conversation"},
+            {"content": f"{filler} MARKER_OMEGA at the end.", "document_id": "big_conversation"},
+        ]
+        assert sum(count_tokens(c["content"]) for c in contents) > 10_000
+
+        result = await memory.retain_batch_async(
+            bank_id=bank_id,
+            contents=contents,
+            request_context=request_context,
+        )
+        assert len(result) == 2
+
+        docs = await memory.list_documents(bank_id=bank_id, request_context=request_context)
+        assert docs["total"] == 1
+
+        # Both items survive — neither sub-batch was dropped by a takeover abort.
+        doc = await memory.get_document("big_conversation", bank_id, request_context=request_context)
+        assert "MARKER_ALPHA" in doc["original_text"]
+        assert "MARKER_OMEGA" in doc["original_text"]
+
+        chunks = await memory.list_document_chunks(bank_id, "big_conversation", request_context=request_context)
+        assert chunks["total"] > 1
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)


@pytest.mark.asyncio
@@ -0,0 +1,266 @@
+"""End-to-end regression for issue #3294: delta retain must not orphan observations.
+
+The reporter's sequence, driven through the public engine API rather than by calling
+the storage helpers directly:
+
+    retain(document) -> consolidate -> retain(same document_id, edited) -> consolidate
+
+Before the fix, the second retain took the delta path, which deletes the changed
+chunks and lets the FK cascade drop their facts — with no observation sweep in
+between. The observations derived from those facts stayed behind, still valid and
+still recallable, pointing at ``source_memory_ids`` that no longer resolved. Nothing
+could reach them afterwards: consolidation batches are built from facts, so an
+observation whose sources are all gone is never selected into a batch again.
+
+These tests assert the invariant the lifecycle documents ("removing a document: all
+observations derived from the document's memories are deleted") on the paths delta
+retain actually takes: an edit, a removal, and a no-op re-ingest. A rewrite of *every*
+chunk is deliberately not covered here — with no unchanged chunk left, delta declines
+and the full-replace path (already covered in ``test_observation_invalidation.py``)
+handles it.
+"""
+
+import uuid
+
+import pytest
+
+from hindsight_api import RequestContext
+from hindsight_api.config import _get_raw_config
+from hindsight_api.engine.memories import FactRecord, get_memories
+from hindsight_api.engine.memory_engine import MemoryEngine, fq_table
+
+# Delta retain works per chunk, so the document has to be big enough to produce several
+# (chunk size is 3000 chars) and at least one of them must come out unchanged — with no
+# unchanged chunk delta declines and falls back to a full replace. Keeping the FIRST
+# block byte-identical across a re-ingest is what guarantees that: chunking is greedy
+# from the start of the text, so an edit after chunk 0's boundary cannot move it.
+_BLOCK_A = " ".join(
+    f"Alice shipped the Alpha{i} milestone at Google in the search infrastructure group." for i in range(40)
+)
+_BLOCK_B = " ".join(f"Bob reviewed the Beta{i} rollout at Microsoft in the Azure networking group." for i in range(40))
+_BLOCK_B_EDITED = " ".join(
+    f"Bob reviewed the Beta{i} rollout at Amazon in the AWS networking group." for i in range(40)
+)
+
+_DOCUMENT_V1 = f"{_BLOCK_A} {_BLOCK_B}"
+_DOCUMENT_V2_PARTIAL_EDIT = f"{_BLOCK_A} {_BLOCK_B_EDITED}"
+
+
+@pytest.fixture(autouse=True)
+def enable_observations():
+    config = _get_raw_config()
+    original = config.enable_observations
+    config.enable_observations = True
+    yield
+    config.enable_observations = original
+
+
+async def _scan(memory: MemoryEngine, bank_id: str, fact_types: list[str]) -> list[FactRecord]:
+    """Every stored memory of these types, read through whichever store holds them."""
+    store = get_memories()
+    pool = await memory._get_pool()
+    async with pool.acquire() as conn:
+        page = await store.scan_memories(
+            conn=conn,
+            fq_table=fq_table,
+            bank_id=bank_id,
+            fact_types=fact_types,
+            limit=1_000_000,
+        )
+    return list(page.memories)
+
+
+async def _facts(memory: MemoryEngine, bank_id: str) -> list[FactRecord]:
+    return await _scan(memory, bank_id, ["experience", "world"])
+
+
+async def _observations(memory: MemoryEngine, bank_id: str) -> list[FactRecord]:
+    return await _scan(memory, bank_id, ["observation"])
+
+
+def _broken_source_refs(observations: list[FactRecord], live_fact_ids: set[str]) -> list[tuple[str, list[str]]]:
+    """The reporter's diagnostic: observations whose sources no longer resolve.
+
+    Returns ``(observation_id, unresolvable_source_ids)`` per affected row — what the
+    bug report counted as "broken references" on their bank.
+    """
+    broken = []
+    for obs in observations:
+        missing = [sid for sid in obs.source_memory_ids if sid not in live_fact_ids]
+        if missing:
+            broken.append((obs.unit_id, missing))
+    return broken
+
+
+async def _assert_no_orphans(memory: MemoryEngine, bank_id: str, when: str) -> None:
+    facts = await _facts(memory, bank_id)
+    observations = await _observations(memory, bank_id)
+    broken = _broken_source_refs(observations, {f.unit_id for f in facts})
+    assert broken == [], (
+        f"{when}: {len(broken)} of {len(observations)} observation(s) reference deleted source "
+        f"memories (issue #3294 — delta retain cascaded the facts away without sweeping the "
+        f"observations derived from them): {broken[:5]}"
+    )
+
+
+async def _retain_document(
+    memory: MemoryEngine, bank_id: str, document_id: str, content: str, request_context: RequestContext
+) -> None:
+    await memory.retain_async(
+        bank_id=bank_id,
+        content=content,
+        context="team roster",
+        document_id=document_id,
+        request_context=request_context,
+    )
+
+
+def _facts_by_chunk(facts: list[FactRecord]) -> dict[str, set[str]]:
+    by_chunk: dict[str, set[str]] = {}
+    for fact in facts:
+        if fact.chunk_id:
+            by_chunk.setdefault(fact.chunk_id, set()).add(fact.unit_id)
+    return by_chunk
+
+
+@pytest.mark.asyncio
+async def test_delta_retain_partial_edit_leaves_no_orphan_observations(
+    memory: MemoryEngine, request_context: RequestContext
+):
+    """Editing the tail of a consolidated document orphans nothing.
+
+    Also pins the precision of the sweep: the untouched first chunk keeps its facts
+    AND the observations derived only from them, so a small edit does not
+    re-consolidate the whole document — the case delta retain exists for.
+    """
+    bank_id = f"test_delta_orphan_partial_{uuid.uuid4().hex[:8]}"
+    document_id = "roster-doc"
+
+    try:
+        await _retain_document(memory, bank_id, document_id, _DOCUMENT_V1, request_context)
+        await memory.run_consolidation(bank_id=bank_id, request_context=request_context)
+
+        facts_v1 = await _facts(memory, bank_id)
+        observations_v1 = await _observations(memory, bank_id)
+        by_chunk_v1 = _facts_by_chunk(facts_v1)
+        assert len(by_chunk_v1) >= 2, f"Setup: the document should span several chunks, got {list(by_chunk_v1)}"
+        assert observations_v1, "Setup: consolidation should have produced observations to orphan"
+        await _assert_no_orphans(memory, bank_id, "after the first retain")
+
+        first_chunk = sorted(by_chunk_v1)[0]
+        kept_fact_ids = by_chunk_v1[first_chunk]
+        edited_fact_ids = {fid for chunk, ids in by_chunk_v1.items() if chunk != first_chunk for fid in ids}
+        assert kept_fact_ids and edited_fact_ids
+
+        obs_over_edited = {o.unit_id for o in observations_v1 if edited_fact_ids.intersection(o.source_memory_ids)}
+        obs_only_over_kept = {
+            o.unit_id
+            for o in observations_v1
+            if o.source_memory_ids and set(o.source_memory_ids).issubset(kept_fact_ids)
+        }
+        assert obs_over_edited, "Setup: the chunks being edited should have observations derived from them"
+        assert obs_only_over_kept, "Setup: the unchanged chunk should have observations of its own"
+
+        # Re-ingest with only the tail changed — this is the delta path.
+        await _retain_document(memory, bank_id, document_id, _DOCUMENT_V2_PARTIAL_EDIT, request_context)
+
+        surviving_fact_ids = {f.unit_id for f in await _facts(memory, bank_id)}
+        # Delta really applied: the unchanged chunk's facts were preserved rather than
+        # re-extracted under new ids (a full replace would have changed all of them).
+        assert kept_fact_ids.issubset(surviving_fact_ids), (
+            "Unchanged chunk's facts should survive the delta re-ingest — if they did not, this "
+            "test fell back to the full-replace path and no longer covers the bug"
+        )
+        assert not edited_fact_ids.intersection(surviving_fact_ids), "The edited chunks' facts should be gone"
+
+        await _assert_no_orphans(memory, bank_id, "after the delta re-ingest")
+
+        observation_ids_v2 = {o.unit_id for o in await _observations(memory, bank_id)}
+        assert not observation_ids_v2.intersection(obs_over_edited), (
+            "Observations derived from the edited chunks' facts should have been invalidated"
+        )
+        assert obs_only_over_kept.issubset(observation_ids_v2), (
+            "Observations derived only from the unchanged chunk must survive a partial edit"
+        )
+
+        # And the follow-up consolidation the reporter ran — still no orphans.
+        await memory.run_consolidation(bank_id=bank_id, request_context=request_context)
+        await _assert_no_orphans(memory, bank_id, "after re-consolidating the edited document")
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_delta_retain_removed_chunks_leave_no_orphan_observations(
+    memory: MemoryEngine, request_context: RequestContext
+):
+    """Shortening a document orphans nothing either.
+
+    Delta deletes removed chunks through the same call as changed ones, so this
+    covers the ``removed_indices`` half of that list — a document that shrinks loses
+    facts without any replacement being extracted for them.
+    """
+    bank_id = f"test_delta_orphan_shrink_{uuid.uuid4().hex[:8]}"
+    document_id = "roster-doc"
+
+    try:
+        await _retain_document(memory, bank_id, document_id, _DOCUMENT_V1, request_context)
+        await memory.run_consolidation(bank_id=bank_id, request_context=request_context)
+
+        facts_v1 = await _facts(memory, bank_id)
+        observations_v1 = await _observations(memory, bank_id)
+        by_chunk_v1 = _facts_by_chunk(facts_v1)
+        assert len(by_chunk_v1) >= 2, f"Setup: the document should span several chunks, got {list(by_chunk_v1)}"
+        assert observations_v1, "Setup: consolidation should have produced observations to orphan"
+
+        first_chunk = sorted(by_chunk_v1)[0]
+        kept_fact_ids = by_chunk_v1[first_chunk]
+        dropped_fact_ids = {fid for chunk, ids in by_chunk_v1.items() if chunk != first_chunk for fid in ids}
+        obs_over_dropped = {o.unit_id for o in observations_v1 if dropped_fact_ids.intersection(o.source_memory_ids)}
+        assert obs_over_dropped, "Setup: the chunks being dropped should have observations derived from them"
+
+        # Re-ingest only the first block: every later chunk is removed outright.
+        await _retain_document(memory, bank_id, document_id, _BLOCK_A, request_context)
+
+        surviving_fact_ids = {f.unit_id for f in await _facts(memory, bank_id)}
+        assert kept_fact_ids.issubset(surviving_fact_ids), (
+            "The retained chunk's facts should survive — if they did not, this test fell back "
+            "to the full-replace path and no longer covers the bug"
+        )
+        assert not dropped_fact_ids.intersection(surviving_fact_ids), "The removed chunks' facts should be gone"
+
+        await _assert_no_orphans(memory, bank_id, "after shrinking the document")
+        assert not {o.unit_id for o in await _observations(memory, bank_id)}.intersection(obs_over_dropped), (
+            "Observations derived from the removed chunks' facts should have been invalidated"
+        )
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_delta_retain_unchanged_content_keeps_observations(memory: MemoryEngine, request_context: RequestContext):
+    """Re-submitting identical content deletes no chunk, so it sweeps no observation
+    and requeues nothing for consolidation."""
+    bank_id = f"test_delta_orphan_noop_{uuid.uuid4().hex[:8]}"
+    document_id = "roster-doc"
+
+    try:
+        await _retain_document(memory, bank_id, document_id, _DOCUMENT_V1, request_context)
+        await memory.run_consolidation(bank_id=bank_id, request_context=request_context)
+        observation_ids_v1 = {o.unit_id for o in await _observations(memory, bank_id)}
+        assert observation_ids_v1
+
+        await _retain_document(memory, bank_id, document_id, _DOCUMENT_V1, request_context)
+
+        assert {o.unit_id for o in await _observations(memory, bank_id)} == observation_ids_v1, (
+            "A no-op delta re-ingest must not touch existing observations"
+        )
+        assert all(f.consolidated_at is not None for f in await _facts(memory, bank_id)), (
+            "A no-op delta re-ingest must not requeue facts for consolidation"
+        )
+        await _assert_no_orphans(memory, bank_id, "after a no-op re-ingest")
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
@@ -0,0 +1,67 @@
+"""Regression test: list_banks must consult the per-bank capability with the
+*current row's* bank id, and source fact_count from the store when that bank
+keeps its memories outside SQL.
+
+Bug (introduced with per-bank store capabilities, #3350): list_banks called
+``_store.writes_memory_rows_in_sql_for(bank_id)`` with a bare ``bank_id`` name
+that is not in scope inside the per-row loop (the row's id is ``row["bank_id"]``).
+Because the argument is evaluated before the call, this raised
+``NameError: name 'bank_id' is not defined`` for *every* org on the very first
+bank — i.e. GET /banks 500'd outright — regardless of the store's capability.
+
+This test swaps in a store that reports ``writes_memory_rows_in_sql_for -> False``
+(the non-SQL branch the feature added), and asserts list_banks (a) does not raise,
+(b) calls the capability + count_memories with the correct per-bank id, and
+(c) surfaces the store's live count as fact_count.
+
+Runs via: uv run pytest tests/test_list_banks_non_sql_store.py -v
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import hindsight_api.engine.memories as memories_mod
+from hindsight_api.models import RequestContext
+
+
+class _NonSqlStore:
+    """A store that keeps memory rows outside SQL: list_banks must count via the store."""
+
+    def __init__(self):
+        self.capability_calls: list[str] = []
+        self.count_calls: list[str] = []
+
+    def writes_memory_rows_in_sql_for(self, bank_id: str) -> bool:
+        self.capability_calls.append(bank_id)
+        return False
+
+    async def count_memories(self, *, conn, fq_table, bank_id: str) -> dict:
+        self.count_calls.append(bank_id)
+        return {"world": 7}
+
+
+@pytest.mark.asyncio
+async def test_list_banks_counts_via_store_for_non_sql_bank(memory, monkeypatch):
+    bank_id = "list_banks_non_sql_bank"
+    request_context = RequestContext(api_key=None, api_key_id=None, tenant_id=None, internal=False)
+
+    store = _NonSqlStore()
+    monkeypatch.setattr(memories_mod, "get_memories", lambda: store)
+
+    try:
+        await memory.get_bank_profile(bank_id, request_context=request_context)
+
+        # Must not raise NameError; must reach the store's non-SQL count path.
+        banks = await memory.list_banks(request_context=request_context)
+
+        entry = next((b for b in banks if b["bank_id"] == bank_id), None)
+        assert entry is not None, f"bank {bank_id!r} not present in list_banks output"
+
+        # The capability + count were consulted with the row's real bank id.
+        assert bank_id in store.capability_calls
+        assert bank_id in store.count_calls
+        # fact_count came from the store (sum of the per-type counts), not the empty SQL join.
+        assert entry["fact_count"] == 7
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
@@ -0,0 +1,198 @@
+"""Structured output via a forced tool call on the LiteLLM-backed providers.
+
+Covers the flag (HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL) end to end:
+config parsing, the config -> LLMProvider -> LiteLLMLLM wiring, the request shape
+it produces, and the response path that substitutes the tool call's arguments for
+the message content. Motivation: Bedrock Claude rejects the ``response_format``
+route outright (``output_config.format: Extra inputs are not permitted``, #3300)
+while accepting the identical schema as a tool.
+"""
+
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+from pydantic import BaseModel
+
+from hindsight_api.config import ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL, HindsightConfig
+from hindsight_api.engine.llm_wrapper import LLMConfig
+from hindsight_api.engine.providers.litellm_llm import LiteLLMLLM
+
+_MESSAGES = [{"role": "user", "content": "hi"}]
+
+
+class _Facts(BaseModel):
+    facts: list[str]
+
+
+def _make_litellm(*, forced_tool: bool) -> LiteLLMLLM:
+    return LiteLLMLLM(
+        provider="bedrock",
+        api_key="",
+        base_url="",
+        model="bedrock/au.anthropic.claude-haiku-4-5-20251001-v1:0",
+        structured_output_forced_tool=forced_tool,
+    )
+
+
+class _Function:
+    def __init__(self, name: str, arguments: Any):
+        self.name = name
+        self.arguments = arguments
+
+
+class _ToolCall:
+    def __init__(self, name: str, arguments: Any):
+        self.id = "call_1"
+        self.function = _Function(name, arguments)
+
+
+class _Message:
+    def __init__(self, content: str | None, tool_calls: list[_ToolCall] | None = None):
+        self.content = content
+        self.tool_calls = tool_calls
+
+
+class _Choice:
+    def __init__(self, message: _Message, finish_reason: str):
+        self.message = message
+        self.finish_reason = finish_reason
+
+
+class _Response:
+    def __init__(self, message: _Message, finish_reason: str = "tool_calls"):
+        self.choices = [_Choice(message, finish_reason)]
+        self.usage = None
+
+
+async def _call_capturing_request(llm: LiteLLMLLM, response: _Response) -> dict[str, Any]:
+    """Run ``call`` against a stubbed completion and return the request kwargs."""
+    completion = AsyncMock(return_value=response)
+    llm._acompletion = completion  # type: ignore[method-assign]
+    result = await llm.call(messages=_MESSAGES, response_format=_Facts, max_retries=0)
+    return {"kwargs": completion.await_args.kwargs, "result": result}
+
+
+# ── config ───────────────────────────────────────────────────────────────────
+
+
+def test_forced_tool_defaults_off(monkeypatch):
+    monkeypatch.delenv(ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL, raising=False)
+    assert HindsightConfig.from_env().llm_structured_output_forced_tool is False
+
+
+def test_forced_tool_can_be_enabled(monkeypatch):
+    monkeypatch.setenv(ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL, "true")
+    assert HindsightConfig.from_env().llm_structured_output_forced_tool is True
+
+
+@pytest.mark.parametrize("value", ["", "yes", "tru", "enabled"])
+def test_forced_tool_rejects_ambiguous_values(monkeypatch, value):
+    monkeypatch.setenv(ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL, value)
+
+    with pytest.raises(ValueError, match=ENV_LLM_STRUCTURED_OUTPUT_FORCED_TOOL):
+        HindsightConfig.from_env()
+
+
+def test_llm_config_threads_flag_to_provider_impl():
+    """LLMConfig -> create_llm_provider -> LiteLLMLLM carries the flag.
+
+    Without this bridge the env var is inert: the provider silently keeps the
+    default ``response_format`` transport.
+    """
+    llm = LLMConfig(
+        provider="bedrock",
+        api_key="",
+        base_url="",
+        model="au.anthropic.claude-haiku-4-5-20251001-v1:0",
+        structured_output_forced_tool=True,
+    )
+    assert llm._provider_impl.structured_output_forced_tool is True
+
+
+# ── request shape ────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_forced_tool_replaces_response_format_with_a_forced_tool():
+    llm = _make_litellm(forced_tool=True)
+    response = _Response(_Message(None, [_ToolCall("structured_response", '{"facts": ["the sky is blue"]}')]))
+
+    captured = await _call_capturing_request(llm, response)
+
+    kwargs = captured["kwargs"]
+    assert "response_format" not in kwargs
+    assert kwargs["tool_choice"] == {"type": "function", "function": {"name": "structured_response"}}
+    assert len(kwargs["tools"]) == 1
+    function = kwargs["tools"][0]["function"]
+    assert function["name"] == "structured_response"
+    assert function["parameters"] == _Facts.model_json_schema()
+    assert captured["result"] == _Facts(facts=["the sky is blue"])
+
+
+@pytest.mark.asyncio
+async def test_flag_off_keeps_response_format():
+    llm = _make_litellm(forced_tool=False)
+    response = _Response(_Message('{"facts": ["the sky is blue"]}'), finish_reason="stop")
+
+    captured = await _call_capturing_request(llm, response)
+
+    kwargs = captured["kwargs"]
+    assert "tools" not in kwargs
+    assert "tool_choice" not in kwargs
+    assert kwargs["response_format"]["json_schema"]["schema"] == _Facts.model_json_schema()
+    assert captured["result"] == _Facts(facts=["the sky is blue"])
+
+
+@pytest.mark.asyncio
+async def test_plain_calls_are_untouched_by_the_flag():
+    """No ``response_format`` -> no tool is forced, so free-text calls still work."""
+    llm = _make_litellm(forced_tool=True)
+    completion = AsyncMock(return_value=_Response(_Message("hello"), finish_reason="stop"))
+    llm._acompletion = completion  # type: ignore[method-assign]
+
+    result = await llm.call(messages=_MESSAGES, max_retries=0)
+
+    assert result == "hello"
+    assert "tools" not in completion.await_args.kwargs
+
+
+# ── response path ────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_forced_tool_accepts_already_decoded_arguments():
+    """Some providers hand back decoded arguments instead of a JSON string."""
+    llm = _make_litellm(forced_tool=True)
+    response = _Response(_Message(None, [_ToolCall("structured_response", {"facts": ["grass is green"]})]))
+
+    captured = await _call_capturing_request(llm, response)
+
+    assert captured["result"] == _Facts(facts=["grass is green"])
+
+
+@pytest.mark.asyncio
+async def test_falls_back_to_text_when_the_tool_call_is_missing():
+    """A gateway that drops ``tool_choice`` must not hard-fail the call."""
+    llm = _make_litellm(forced_tool=True)
+    response = _Response(_Message('{"facts": ["parsed from text"]}'), finish_reason="stop")
+
+    captured = await _call_capturing_request(llm, response)
+
+    assert captured["result"] == _Facts(facts=["parsed from text"])
+
+
+@pytest.mark.asyncio
+async def test_skip_validation_returns_the_raw_tool_arguments():
+    llm = _make_litellm(forced_tool=True)
+    response = _Response(_Message(None, [_ToolCall("structured_response", '{"facts": ["raw"]}')]))
+    llm._acompletion = AsyncMock(return_value=response)  # type: ignore[method-assign]
+
+    result = await llm.call(
+        messages=_MESSAGES,
+        response_format=_Facts,
+        skip_validation=True,
+        max_retries=0,
+    )
+
+    assert result == {"facts": ["raw"]}
@@ -195,6 +195,7 @@ def _make_router_provider(config: dict[str, Any], mock_router: Any) -> LiteLLMRo
        provider.reasoning_effort = "low"
        provider.timeout = 300.0
        provider._default_headers = {}
+        provider.structured_output_forced_tool = False
        provider.config = config
        provider._litellm = fake_litellm
        provider._router = mock_router
@@ -601,6 +601,59 @@ async def test_maintenance_passes_are_optional(restore_default_store):
    await store.record_unit_entities(conn=None, ops=None, fq_table=None, unit_ids=["u"], entity_ids=["e"])


+# ---------------------------------------------------------------------------
+# Per-bank store capabilities. A store may route different banks to different
+# backends, so every BANK-SCOPED call site asks per bank —
+# writes_memory_rows_in_sql_for(bank_id) / owns_document_store_for(bank_id) —
+# rather than reading the process-global class attribute. The class attribute
+# stays the single-store default the _for methods fall back to.
+# ---------------------------------------------------------------------------
+
+
+def test_per_bank_capability_defaults_to_the_class_attribute():
+    """A single-store extension needs no override: the _for methods return the class attr, so
+    every existing store keeps its exact behaviour for every bank."""
+    pg = PostgresMemories({})
+    assert (pg.writes_memory_rows_in_sql, pg.owns_document_store) == (True, False)
+    assert pg.writes_memory_rows_in_sql_for("any-bank") is True
+    assert pg.owns_document_store_for("any-bank") is False
+
+    mem = InMemoryMemories({})  # owns its rows AND its document store
+    assert mem.writes_memory_rows_in_sql_for("any-bank") is False
+    assert mem.owns_document_store_for("any-bank") is True
+
+
+def test_a_store_answers_capabilities_per_bank():
+    """The point of the _for methods: a store that keeps some banks in SQL and others in a
+    separate store answers PER BANK, so mixed banks in one process each take the right path."""
+
+    class PerBankStore(InMemoryMemories):
+        name = "per-bank"
+        # The loop-level class attr stays False so cross-store txn recovery still runs; the
+        # per-bank answer is what every bank-scoped site consults.
+        writes_memory_rows_in_sql = False
+
+        def __init__(self, config=None):
+            super().__init__(config)
+            self.sql_banks = {"legacy-bank"}
+
+        def writes_memory_rows_in_sql_for(self, bank_id):
+            return bank_id in self.sql_banks
+
+        def owns_document_store_for(self, bank_id):
+            return bank_id not in self.sql_banks
+
+    store = PerBankStore({})
+    # A SQL-backed bank looks like Postgres (host does inline SQL, keeps documents in SQL)...
+    assert store.writes_memory_rows_in_sql_for("legacy-bank") is True
+    assert store.owns_document_store_for("legacy-bank") is False
+    # ...a store-backed bank owns its rows and its document store.
+    assert store.writes_memory_rows_in_sql_for("new-bank") is False
+    assert store.owns_document_store_for("new-bank") is True
+    # The process-level gate (cross-store recovery loop) still fires off the class attr.
+    assert store.writes_memory_rows_in_sql is False
+
+
 # ---------------------------------------------------------------------------
 # Interface conformance: the stub must stay a COMPLETE, signature-compatible
 # implementation of every MemoriesExtension method. This is the guard that keeps
@@ -33,6 +33,7 @@ async def _insert_memory(
    text: str,
    fact_type: str = "experience",
    document_id: str | None = None,
+    chunk_id: str | None = None,
 ) -> uuid.UUID:
    """Seed one memory through the store, bypassing the LLM retain pipeline.

@@ -47,7 +48,7 @@ async def _insert_memory(
        tags=[],
        context=None,
        document_id=document_id,
-        chunk_id=None,
+        chunk_id=chunk_id,
        metadata=None,
        observation_scopes=None,
        entities=[],
@@ -435,6 +436,187 @@ class TestDocumentUpsertObservationCleanup:
        await memory.delete_bank(bank_id, request_context=request_context)


+# ---------------------------------------------------------------------------
+# Tests: delta retain chunk delete (regression for orphan observations, #3294)
+# ---------------------------------------------------------------------------
+
+
+async def _seed_chunked_document(memory: MemoryEngine, conn, bank_id: str, chunk_texts: list[str]) -> tuple[str, list]:
+    """One document with ``len(chunk_texts)`` chunks, each owning one fact.
+
+    Returns the document id and, per chunk, the (chunk_id, fact_id) it owns.
+    """
+    doc_id = str(uuid.uuid4())
+    await conn.execute(
+        """
+        INSERT INTO documents (id, bank_id, original_text, content_hash, created_at, updated_at)
+        VALUES ($1, $2, $3, 'hash-old', NOW(), NOW())
+        """,
+        doc_id,
+        bank_id,
+        "\n\n".join(chunk_texts),
+    )
+    seeded = []
+    for idx, text in enumerate(chunk_texts):
+        chunk_id = f"{bank_id}_{doc_id}_{idx}"
+        await conn.execute(
+            """
+            INSERT INTO chunks (chunk_id, document_id, bank_id, chunk_index, chunk_text, content_hash)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            """,
+            chunk_id,
+            doc_id,
+            bank_id,
+            idx,
+            text,
+            f"chunk-hash-{idx}",
+        )
+        fact_id = await _insert_memory(memory, conn, bank_id, text, "experience", document_id=doc_id, chunk_id=chunk_id)
+        seeded.append((chunk_id, fact_id))
+    return doc_id, seeded
+
+
+class TestDeltaChunkDeleteObservationCleanup:
+    """Regression: delta retain drops facts by deleting their chunks, and that
+    cascade must invalidate the observations derived from them.
+
+    ``handle_document_tracking`` (full replace) sweeps observations before its
+    delete, but the delta path never calls it — it upserts the document row and
+    deletes the changed/removed chunks directly, cascading to memory_units. Every
+    delta re-ingest therefore used to leave the observations of the changed chunks
+    behind, valid and recallable, pointing at ids that no longer exist. Nothing
+    could reach them afterwards: consolidation batches are built from facts, so an
+    observation whose sources are all gone is never selected into a batch again.
+    """
+
+    @pytest.mark.asyncio
+    async def test_chunk_delete_removes_observations_from_outgoing_memories(
+        self, memory: MemoryEngine, request_context: RequestContext
+    ):
+        """The observation of a deleted chunk's fact goes with it; its surviving
+        co-source is requeued for re-consolidation."""
+        from hindsight_api.engine.retain.chunk_storage import delete_chunks_by_ids
+
+        bank_id = f"test-delta-obs-cleanup-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+        pool = await memory._get_pool()
+
+        async with pool.acquire() as conn:
+            _doc_id, seeded = await _seed_chunked_document(
+                memory, conn, bank_id, ["Alice works at Google.", "Bob works at Microsoft."]
+            )
+            (outgoing_chunk, outgoing_fact), (_kept_chunk, kept_fact) = seeded
+            standalone_fact = await _insert_memory(memory, conn, bank_id, "Carol works at Netflix.")
+            obs_id = await _insert_observation(
+                memory,
+                conn,
+                bank_id,
+                "The team is spread across Google, Microsoft and Netflix.",
+                [outgoing_fact, kept_fact, standalone_fact],
+            )
+
+        async with pool.acquire() as conn:
+            async with conn.transaction():
+                invalidated = await delete_chunks_by_ids(conn, [outgoing_chunk], bank_id, ops=memory._backend.ops)
+
+        assert invalidated == 1, "delete_chunks_by_ids should report the observation it invalidated"
+
+        async with pool.acquire() as conn:
+            assert str(obs_id) not in await _get_observation_ids(conn, bank_id), (
+                "Observation derived from the deleted chunk's fact should have been invalidated "
+                "(regression #3294: the delta path cascaded the fact away and left the orphan)"
+            )
+            assert await _count_surviving(conn, bank_id, [outgoing_fact]) == 0, "The chunk's fact is gone"
+            # Both surviving co-sources lost an observation, so both are due for re-consolidation.
+            for survivor in (kept_fact, standalone_fact):
+                assert await _get_consolidated_at(conn, survivor, bank_id) is None, (
+                    "Surviving co-source should be reset for re-consolidation"
+                )
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+    @pytest.mark.asyncio
+    async def test_chunk_delete_keeps_observations_of_unchanged_chunks(
+        self, memory: MemoryEngine, request_context: RequestContext
+    ):
+        """Precision: an observation sourced only from chunks that stay is untouched.
+
+        A sweep keyed on the document rather than on the deleted chunks would take
+        this one too — and needlessly requeue the whole document for consolidation
+        on every small edit, which is exactly the case delta retain exists for.
+        """
+        from hindsight_api.engine.retain.chunk_storage import delete_chunks_by_ids
+
+        bank_id = f"test-delta-obs-keep-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+        pool = await memory._get_pool()
+
+        async with pool.acquire() as conn:
+            _doc_id, seeded = await _seed_chunked_document(
+                memory, conn, bank_id, ["Alice works at Google.", "Bob works at Microsoft."]
+            )
+            (outgoing_chunk, _outgoing_fact), (_kept_chunk, kept_fact) = seeded
+            kept_obs = await _insert_observation(memory, conn, bank_id, "Bob is at Microsoft.", [kept_fact])
+
+        async with pool.acquire() as conn:
+            async with conn.transaction():
+                invalidated = await delete_chunks_by_ids(conn, [outgoing_chunk], bank_id, ops=memory._backend.ops)
+
+        assert invalidated == 0, "No observation of the surviving chunk should have been touched"
+
+        async with pool.acquire() as conn:
+            assert str(kept_obs) in await _get_observation_ids(conn, bank_id), (
+                "Observation of an unchanged chunk must survive the delta delete"
+            )
+            assert await _get_consolidated_at(conn, kept_fact, bank_id) is not None, (
+                "An untouched fact must not be requeued for consolidation"
+            )
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+    @pytest.mark.asyncio
+    async def test_chunk_delete_sweeps_every_deleted_chunk(self, memory: MemoryEngine, request_context: RequestContext):
+        """Multiple chunks in one call: each one's observations are swept.
+
+        The reporter's bank lost the observations of a whole document at once
+        (25 fully orphaned from a single replace), so the sweep must cover the
+        entire ``chunks_to_delete`` list, not just the first entry.
+        """
+        from hindsight_api.engine.retain.chunk_storage import delete_chunks_by_ids
+
+        bank_id = f"test-delta-obs-multi-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+        pool = await memory._get_pool()
+
+        async with pool.acquire() as conn:
+            _doc_id, seeded = await _seed_chunked_document(
+                memory,
+                conn,
+                bank_id,
+                ["Alice works at Google.", "Bob works at Microsoft.", "Dan works at Apple."],
+            )
+            observations = [
+                await _insert_observation(memory, conn, bank_id, f"Observation of chunk {i}.", [fact])
+                for i, (_chunk, fact) in enumerate(seeded)
+            ]
+
+        async with pool.acquire() as conn:
+            async with conn.transaction():
+                invalidated = await delete_chunks_by_ids(
+                    conn, [chunk for chunk, _fact in seeded], bank_id, ops=memory._backend.ops
+                )
+
+        assert invalidated == 3
+
+        async with pool.acquire() as conn:
+            remaining = await _get_observation_ids(conn, bank_id)
+            assert [o for o in observations if str(o) in remaining] == [], (
+                "Every deleted chunk's observations should be swept, not just the first chunk's"
+            )
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
 # ---------------------------------------------------------------------------
 # Tests: delete_bank with fact_type filter
 # ---------------------------------------------------------------------------
@@ -50,6 +50,9 @@ INTEGRATIONS: dict[str, IntegrationMeta] = {
    "nemoclaw": IntegrationMeta("@vectorize-io/hindsight-nemoclaw", "NemoClaw"),
    "strands": IntegrationMeta("hindsight-strands", "Strands"),
    "claude-code": IntegrationMeta("hindsight-memory", "Claude Code"),
+    # Git-distributed plugin bundle (Agent Plugins standard), not a registry
+    # package — its changelog links to the source tree (see _package_url).
+    "agent-plugin": IntegrationMeta("hindsight-agent-plugin", "Agent Plugins"),
    "zcode": IntegrationMeta("hindsight-zcode", "ZCode"),
    "claude-agent-sdk": IntegrationMeta("hindsight-claude-agent-sdk", "Claude Agent SDK"),
    "llamaindex": IntegrationMeta("hindsight-llamaindex", "LlamaIndex"),
@@ -657,8 +660,10 @@ def _get_package_name(integration: str) -> str:


 def _package_url(integration: str, package_name: str) -> str:
-    if integration == "claude-code":
-        return "https://github.com/vectorize-io/hindsight/tree/main/hindsight-integrations/claude-code"
+    # Git-distributed plugin bundles have no npm/pypi package — link to the
+    # source tree instead of a registry page.
+    if integration in ("claude-code", "agent-plugin"):
+        return f"https://github.com/vectorize-io/hindsight/tree/main/hindsight-integrations/{integration}"
    if package_name.startswith("@"):
        return f"https://www.npmjs.com/package/{package_name}"
    return f"https://pypi.org/project/{package_name}/"
@@ -0,0 +1,92 @@
+---
+sidebar_position: 2
+title: "Agent Plugins Persistent Memory with Hindsight | Integration Guide"
+description: "Give any Agent Plugins client — Codex, Cursor, GitHub Copilot, Kiro, VS Code — long-term memory with Hindsight. One portable, standards-based plugin bundles the Hindsight MCP server and a memory skill for recall, retain, and reflect."
+---
+
+# Agent Plugins
+
+Portable long-term memory for any [Agent Plugins](https://agent-plugins.org) client, powered by [Hindsight](https://vectorize.io/hindsight).
+
+[Agent Plugins](https://agent-plugins.org) is the vendor-neutral open standard (developed with Amazon, Cursor, Microsoft, OpenAI, and Vercel) for packaging **Agent Skills + MCP servers** into a single distributable plugin. Instead of a separate integration per tool, Hindsight ships **one** plugin that every compatible client can load — at launch: **ChatGPT / Codex, Cursor, GitHub Copilot, Kiro, and VS Code**.
+
+## Quick Start
+
+:::tip Recommended: Hindsight Cloud
+[Sign up free](https://ui.hindsight.vectorize.io/signup) for a Hindsight Cloud API key — no self-hosting, no local daemon to manage.
+:::
+
+1. Get your `hsk_...` API key from [ui.hindsight.vectorize.io/connect](https://ui.hindsight.vectorize.io/connect).
+2. Set the environment variables the plugin reads:
+
+   ```bash
+   export HINDSIGHT_API_KEY="hsk_your_token"
+   export HINDSIGHT_BANK_ID="my-project"   # optional; defaults to "default"
+   ```
+
+3. Install the plugin in your client (through its plugin/MCP UI, or by pointing it at the plugin directory — installation is client-specific per the standard).
+
+Once installed, ask the agent something that depends on past context, or tell it a durable preference — it calls `recall` and `retain` automatically, guided by the bundled skill.
+
+## What's in the plugin
+
+The plugin is a thin, transport-only wrapper — all memory logic stays server-side in Hindsight. It follows the Agent Plugins `1.0.0` layout:
+
+```
+agent-plugin/
+├── plugin.json                       # manifest ($schema + name + metadata)
+├── mcp.json                          # Hindsight MCP server (Streamable HTTP)
+└── skills/
+    └── hindsight-memory/
+        └── SKILL.md                  # teaches the agent when to recall / retain / reflect
+```
+
+- **`mcp.json`** connects the client to Hindsight's built-in [MCP server](/developer/mcp-server) over Streamable HTTP.
+- **`skills/hindsight-memory/SKILL.md`** is loaded into the agent's context so it knows *when* to reach for memory, not just that the tools exist.
+
+## Memory tools
+
+Via the MCP server, the agent gets Hindsight's full memory surface. The three it reaches for most:
+
+| Tool | When | What it does |
+|------|------|--------------|
+| `recall` | Before answering, when past context could help | Semantic + keyword + graph + temporal retrieval over the bank |
+| `retain` | After learning a durable, reusable fact | Stores the fact for future sessions |
+| `reflect` | When a lookup is too shallow and you need synthesized reasoning | Disposition-aware reasoning over everything remembered |
+
+Additional tools (knowledge pages, mental models, documents, tags) are exposed too — see the [MCP Server reference](/developer/mcp-server).
+
+## Configuration
+
+The plugin reads two environment variables, interpolated into `mcp.json`:
+
+| Setting | Env Var | Default | Description |
+|---------|---------|---------|-------------|
+| API key | `HINDSIGHT_API_KEY` | — | Your `hsk_...` key. Sent as `Authorization: Bearer`. Required for Hindsight Cloud. |
+| Memory bank | `HINDSIGHT_BANK_ID` | `default` | Bank to read from and write to (sent as `X-Bank-Id`). Use one bank per user, project, or team for isolation. |
+
+:::note Env-var syntax varies by client
+Most clients substitute `${VAR}`; some (VS Code, Cursor) use `${env:VAR}`. If your client doesn't interpolate, paste the literal key and bank id into `mcp.json`.
+:::
+
+**Self-hosting:** replace the host in `mcp.json` (`https://api.hindsight.vectorize.io`) with your deployment's URL. A local server with the MCP endpoint open needs no API key.
+
+## Explicit tools vs. automatic capture
+
+Agent Plugins `1.0.0` standardizes **Skills + MCP**, not session lifecycle hooks. This plugin therefore delivers **explicit, tool-driven** memory that works identically across every supported client.
+
+For the fully automatic experience — recall injected before every prompt and transcripts retained on session end — use the native, hook-based integration built for your specific tool, such as [Claude Code](/sdks/integrations/claude-code) or [Codex](/sdks/integrations/codex). Both share the same Hindsight banks, so memory captured by the hook-based integration is recalled through the Agent Plugin, and vice versa.
+
+## Troubleshooting
+
+**No memories recalled**: `recall` returns results only after something has been retained. Retain a fact first, or seed the bank via the [API](/developer/api/quickstart).
+
+**401 Unauthorized**: Check `HINDSIGHT_API_KEY` is set and your client is interpolating it into the `Authorization` header (see the env-var syntax note above).
+
+**Wrong or empty memory**: Confirm `HINDSIGHT_BANK_ID` points at the bank you expect. Different tools writing to different banks won't share memory.
+
+## Learn more
+
+- [Agent Plugins standard](https://agent-plugins.org)
+- [Hindsight MCP Server reference](/developer/mcp-server)
+- [Hindsight Cloud sign-up](https://ui.hindsight.vectorize.io/signup)
@@ -198,6 +198,7 @@ For non-English banks (especially CJK) and the language/extraction-language trad
 | `HINDSIGHT_API_LLM_STRICT_SCHEMA_REFLECT` | Override `HINDSIGHT_API_LLM_STRICT_SCHEMA` for reflect's structured-output extraction only. | Inherits global |
 | `HINDSIGHT_API_LLM_STRICT_SCHEMA_CONSOLIDATION` | Override `HINDSIGHT_API_LLM_STRICT_SCHEMA` for consolidation only (both the batch consolidation call and observation dedup). | Inherits global |
 | `HINDSIGHT_API_LLM_SUPPORTS_MAX_ITEMS` | Whether the LLM backend accepts JSON Schema `maxItems` in structured-output schemas. Set to `false` for backends such as Bedrock Converse that reject this keyword; consolidation still enforces observation caps after parsing. | `true` |
+| `HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL` | Request structured output from the LiteLLM-backed providers (`litellm`, `litellmrouter`, `bedrock`) with a single forced tool call — the response schema becomes the tool's parameters — instead of `response_format`. Set to `true` for backends that reject `response_format` outright. This is region-dependent on Bedrock Claude: `ap-southeast-2` (`au.*` inference profiles) refuses the translated Converse `outputConfig` with `Extra inputs are not permitted`, while the same model in `us-east-1` (`us.*`) accepts it and needs nothing here. Verified against both. If the model answers without calling the tool, the reply is parsed as text as before. Other providers ignore it. | `false` |
 | `HINDSIGHT_API_LLM_OLLAMA_NUM_CTX` | Optional native Ollama `num_ctx` override for structured-output calls. Leave unset to use the model/server default; set a positive integer only when you need a larger context window. | Unset |
 | `HINDSIGHT_API_LLM_GEMINI_SAFETY_SETTINGS` | JSON-encoded list of `{category, threshold}` dicts for Gemini/VertexAI content safety filtering | `null` |
 | `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED` | Reuse the fixed system prefix via the provider's explicit prompt cache, billed at the cached-input rate (Gemini/Vertex `CachedContent`). The cached prefix is shared across all banks and soft-fails to an uncached call. Set to `false` to disable. See [Models](./models#provider-capabilities). | `true` |
@@ -27,7 +27,9 @@ Used for fact extraction, entity resolution, mental model consolidation, and ans
 Also supports **any OpenAI-compatible API** (e.g., Azure OpenAI, Together AI, Fireworks) and **100+ providers via LiteLLM** (e.g., AWS Bedrock, Azure OpenAI, Together AI).

 :::tip OpenAI-Compatible Providers
-Hindsight works with any provider that exposes an OpenAI-compatible API (e.g., Azure OpenAI). Simply set `HINDSIGHT_API_LLM_PROVIDER=openai` and configure `HINDSIGHT_API_LLM_BASE_URL` to point to your provider's endpoint.
+Hindsight works with any provider that exposes an OpenAI-compatible API. Set `HINDSIGHT_API_LLM_PROVIDER=openai` and point `HINDSIGHT_API_LLM_BASE_URL` at the endpoint that serves `/chat/completions` — for most providers that is the URL ending in `/v1`, **not** the account or resource root.
+
+**Azure OpenAI does not serve the API at the resource root**, so `https://<resource>.openai.azure.com` on its own returns `404 Resource not found`. See [Azure OpenAI Setup](#azure-openai-setup) for the two URL shapes that work.

 The `openai` provider talks to the **Chat Completions API** (`/v1/chat/completions`). For the newer **Responses API** (`/v1/responses`), use `HINDSIGHT_API_LLM_PROVIDER=openai-responses` — see the tip below. Both accept a custom `HINDSIGHT_API_LLM_BASE_URL`, so an OpenAI-compatible endpoint that exposes `/v1/responses` works the same way as a Chat Completions one.

@@ -576,6 +578,56 @@ one, run one replica on this provider and give the others an API-key lane.

 ---

+### Azure OpenAI Setup
+
+Azure OpenAI is reached through the **`openai`** provider — there is no `azure`
+provider, and setting one fails at startup with
+`Invalid LLM provider: azure`.
+
+The one thing that trips people up is the base URL. Azure does not serve the
+OpenAI API at the resource root, so the endpoint shown in the Azure portal is
+not usable on its own:
+
+| `HINDSIGHT_API_LLM_BASE_URL` | Result |
+|---|---|
+| `https://<resource>.openai.azure.com` | `404 Resource not found` |
+| `https://<resource>.openai.azure.com/openai/deployments/<deployment>` | `404 Resource not found` (no `api-version`) |
+| `https://<resource>.openai.azure.com/openai/v1` | works |
+| `https://<resource>.openai.azure.com/openai/deployments/<deployment>?api-version=<version>` | works |
+
+**Recommended — the v1 surface:**
+
+```bash
+export HINDSIGHT_API_LLM_PROVIDER=openai
+export HINDSIGHT_API_LLM_API_KEY=<azure-openai-resource-key>
+export HINDSIGHT_API_LLM_MODEL=<deployment-name>
+export HINDSIGHT_API_LLM_BASE_URL=https://<resource>.openai.azure.com/openai/v1
+```
+
+**Or the deployment-scoped form.** Keep the `api-version` query string — Hindsight
+parses it out of the base URL and passes it to the SDK:
+
+```bash
+export HINDSIGHT_API_LLM_BASE_URL=https://<resource>.openai.azure.com/openai/deployments/<deployment>?api-version=2025-01-01-preview
+```
+
+**Important notes:**
+- `HINDSIGHT_API_LLM_MODEL` is your **deployment name**, not the model name. A
+  `gpt-4o` deployed as `my-gpt4o` is configured as `my-gpt4o`.
+- The key is the Azure OpenAI **resource** key (`az cognitiveservices account
+  keys list -n <resource> -g <group>`). An API Management subscription key is a
+  different credential: with APIM in front, the base URL must be the APIM route
+  and APIM has to forward the `api-key` header. Test against the Azure endpoint
+  directly first to isolate which layer is failing.
+- Gateways and proxies must preserve the same path shape (`/openai/v1` or
+  `/openai/deployments/...?api-version=`).
+- Azure accepts the `prompt_cache_key` field that
+  [`HINDSIGHT_API_LLM_CACHE_AFFINITY`](./configuration#llm-provider) sends under
+  `auto`, on every `api-version` from `2024-02-01` onward, so the default needs
+  no adjustment for Azure.
+
+---
+
 ### Vertex AI Setup (Google Cloud)

 Google Cloud's Vertex AI provides access to Gemini models via the native Google GenAI SDK.
@@ -1,5 +1,15 @@
 {
  "integrations": [
+    {
+      "id": "agent-plugin",
+      "name": "Agent Plugins",
+      "description": "Portable long-term memory for any Agent Plugins client (Codex, Cursor, GitHub Copilot, Kiro, VS Code). One standards-based plugin bundles Hindsight's MCP server and a memory skill — recall, retain, and reflect everywhere.",
+      "type": "official",
+      "by": "hindsight",
+      "category": "tool",
+      "link": "/sdks/integrations/agent-plugin",
+      "icon": "/img/icons/mcp.png"
+    },
    {
      "id": "coding-agents",
      "name": "Coding Agents",
@@ -48,6 +48,12 @@ HINDSIGHT_API_LLM_BASE_URL=https://api.openai.com/v1
 # HINDSIGHT_API_REFLECT_LLM_CACHE_AFFINITY=xai_conv_id
 # HINDSIGHT_API_CONSOLIDATION_LLM_CACHE_AFFINITY=none

+# Ask litellm/litellmrouter/bedrock for structured output via a forced tool call
+# instead of response_format. Enable it for backends that reject response_format
+# outright -- e.g. Bedrock Claude in ap-southeast-2 ("Extra inputs are not permitted");
+# the same model in us-east-1 accepts response_format and needs nothing here.
+# HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL=false
+
 # Diagnostic: on any LLM 4xx, log the exact assembled request ([LLM_4XX_DUMP]) --
 # serialized request config (message bodies stripped) + capped per-message previews.
 # For debugging otherwise-unreproducible rejected calls. Off by default.
@@ -0,0 +1,77 @@
+# Hindsight — Agent Plugin
+
+A portable [Agent Plugin](https://agent-plugins.org) (spec `1.0.0`) that gives any
+compatible agent client long-term memory via [Hindsight](https://hindsight.vectorize.io).
+
+Agent Plugins is the vendor-neutral standard (AWS, Cursor, GitHub/Microsoft, OpenAI,
+Vercel) for packaging **Agent Skills + MCP servers** into one distributable plugin. At
+launch it is supported by **ChatGPT/Codex, Cursor, GitHub Copilot, Kiro, and VS Code**.
+
+This is the *portable* front door to Hindsight: one artifact, every supported client. It
+carries the same `retain` / `recall` / `reflect` memory as our per-IDE integrations, but
+as a single standards-based bundle instead of N hand-rolled configs.
+
+## What's in the bundle
+
+```
+agent-plugin/
+├── plugin.json                       # manifest ($schema + name + metadata)
+├── mcp.json                          # Hindsight MCP server (Streamable HTTP)
+└── skills/
+    └── hindsight-memory/
+        └── SKILL.md                  # teaches the agent when to recall/retain/reflect
+```
+
+- **`mcp.json`** points the client at Hindsight's built-in MCP server (retain, recall,
+  reflect, knowledge pages, and more — see [MCP Server docs](https://hindsight.vectorize.io/developer/mcp-server)).
+  The plugin is transport-only; all memory logic stays server-side.
+- **`skills/hindsight-memory/SKILL.md`** is loaded into the agent's context so it knows
+  *when* to reach for memory, not just that the tools exist.
+
+## Configuration
+
+The plugin reads two environment variables (values are interpolated into `mcp.json`):
+
+| Variable | Required | Purpose |
+|----------|----------|---------|
+| `HINDSIGHT_API_KEY` | yes (Cloud) | Your `hsk_...` key from [ui.hindsight.vectorize.io/connect](https://ui.hindsight.vectorize.io/connect). Sent as `Authorization: Bearer`. |
+| `HINDSIGHT_BANK_ID` | no | Memory bank to scope to (sent as `X-Bank-Id`). Defaults to `default`. Use one bank per user/project/team. |
+
+**Self-hosting:** replace the host in `mcp.json` (`https://api.hindsight.vectorize.io`)
+with your deployment's URL. A local server with the MCP endpoint open needs no API key.
+
+> Env-var interpolation syntax varies by client. Most use `${VAR}`; some (VS Code,
+> Cursor) prefer `${env:VAR}`. If your client doesn't substitute, paste the literal key
+> and bank id into `mcp.json` instead.
+
+## Install
+
+Installation and distribution are intentionally left to each client by the spec. Common
+paths:
+
+- **VS Code / GitHub Copilot / Cursor / Kiro** — add this plugin directory through the
+  client's plugin/MCP UI, or drop it where the client discovers plugins, then set the
+  two environment variables above.
+- **Codex / ChatGPT** — register the plugin per the client's Agent Plugins support.
+
+Once installed, ask the agent something that depends on past context (or tell it a
+durable preference) and it will call `recall` / `retain` automatically, guided by the
+skill.
+
+## Want automatic capture (hooks)?
+
+Agent Plugins `1.0.0` standardizes **Skills + MCP**, not session lifecycle hooks. This
+plugin therefore delivers **explicit, tool-driven** memory that works identically
+everywhere. For the fully automatic experience — recall injected before every prompt and
+transcripts retained on session end — use the native, hook-based integration for your
+tool (e.g. [`hindsight-integrations/claude-code`](../claude-code)). The two share the
+same banks, so memory captured by one is recalled by the other.
+
+## Validate the manifests
+
+```bash
+python3 hindsight-integrations/agent-plugin/validate.py
+```
+
+Checks that `plugin.json` and `mcp.json` parse and satisfy the Agent Plugins `1.0.0`
+required-field contract.
@@ -0,0 +1,13 @@
+{
+  "$schema": "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json",
+  "mcpServers": {
+    "hindsight": {
+      "type": "streamable-http",
+      "url": "https://api.hindsight.vectorize.io/mcp",
+      "headers": {
+        "Authorization": "Bearer ${HINDSIGHT_API_KEY}",
+        "X-Bank-Id": "${HINDSIGHT_BANK_ID}"
+      }
+    }
+  }
+}
@@ -0,0 +1,29 @@
+{
+  "$schema": "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json",
+  "name": "hindsight",
+  "version": "1.0.0",
+  "description": "Long-term memory for AI agents. Recalls relevant past context before you answer and retains durable facts as you work, backed by Hindsight's retain/recall/reflect memory engine.",
+  "author": {
+    "name": "Hindsight (Vectorize)",
+    "url": "https://hindsight.vectorize.io"
+  },
+  "homepage": "https://hindsight.vectorize.io",
+  "repository": "https://github.com/vectorize-io/hindsight",
+  "license": "MIT",
+  "keywords": [
+    "memory",
+    "long-term-memory",
+    "recall",
+    "retain",
+    "reflect",
+    "mcp",
+    "agent-memory"
+  ],
+  "extensions": {
+    "io.vectorize.hindsight": {
+      "bankSelection": "Set HINDSIGHT_BANK_ID to scope memory to a user, project, or team. The mcp.json endpoint is per-bank, so tools never need a bank_id argument.",
+      "selfHosted": "Point HINDSIGHT_MCP_URL at your own deployment instead of Hindsight Cloud; the plugin is transport-only and stays otherwise identical.",
+      "deepIntegration": "For automatic capture/injection (session hooks that recall before each prompt and retain transcripts on exit), see the native Claude Code plugin at hindsight-integrations/claude-code. This portable plugin exposes the same memory via explicit MCP tools."
+    }
+  }
+}
@@ -0,0 +1,64 @@
+---
+name: hindsight-memory
+description: Long-term memory for the agent via Hindsight. Use to recall relevant past context before answering, retain durable facts as you learn them, and reflect over accumulated memory for the "why" behind a decision. Load whenever continuity across sessions matters — the user refers to earlier work, states a lasting preference, or asks a question that prior context could answer.
+---
+
+# Hindsight long-term memory
+
+This plugin connects the agent to **Hindsight**, a long-term memory engine. Memory
+persists across sessions in a **bank** (scoped by `HINDSIGHT_BANK_ID`), so what you
+retain now is available to recall in future conversations.
+
+The `hindsight` MCP server exposes the tools below. Prefer these over guessing from
+scratch when the answer might live in past context.
+
+## When to recall (read memory)
+
+Call **`recall`** at the start of a task, or whenever the user:
+
+- refers to earlier work, a past decision, or "the thing we set up",
+- states a preference or constraint that may already be recorded,
+- asks a question that accumulated project/user context could answer.
+
+```
+recall(query: "how do we deploy the API and which region")
+```
+
+`recall` runs semantic + keyword + graph + temporal retrieval and returns the most
+relevant memories. Ground your answer in what comes back, and say when nothing
+relevant was found rather than inventing continuity.
+
+## When to retain (write memory)
+
+Call **`retain`** when you learn something **durable and reusable** — worth having in
+a future session, not just this one:
+
+- stable user preferences ("prefers pnpm; deploys from `main` only"),
+- project facts and decisions ("staging DB is Postgres 16 on Neon"),
+- outcomes and gotchas ("the flaky test was a timezone bug, fixed in #482").
+
+```
+retain(content: "The user deploys the API to us-east-1 via GitHub Actions on push to main.")
+```
+
+Do **not** retain transient chatter, secrets, or anything the user asked you to keep
+out of memory. Retain the fact, not the whole transcript.
+
+## When to reflect (reason over memory)
+
+Call **`reflect`** when a single recall is too shallow and you need synthesized
+reasoning over everything remembered — the *why* behind a behavior, or a judgment that
+weighs many facts together:
+
+```
+reflect(query: "What has repeatedly caused our CI to flake, and what should we standardize?")
+```
+
+`reflect` is slower and disposition-aware; use it deliberately, not for lookups.
+
+## Bank scope
+
+All tools operate on the bank selected by the connection (`HINDSIGHT_BANK_ID`, default
+`default`). One bank = one memory store — keep a project's or a user's memory in its
+own bank so context stays isolated and relevant. You never pass a bank id to a tool;
+it is implicit from the endpoint.
@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""Validate this Agent Plugin against the Agent Plugins 1.0.0 required-field contract.
+
+Config-only plugin (no runtime code to unit-test), so the meaningful check is that the
+manifests parse and satisfy the spec's structural requirements. Runs as a script
+(`python3 validate.py`) and is imported by CI as a pytest.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+PLUGIN_DIR = Path(__file__).resolve().parent
+SPEC = "1.0.0"
+PLUGIN_SCHEMA = f"https://agent-plugins.org/schemas/{SPEC}/plugin.schema.json"
+MCP_SCHEMA = f"https://agent-plugins.org/schemas/{SPEC}/mcp.schema.json"
+
+# https://agent-plugins.org/schemas/1.0.0/plugin.schema.json — name pattern
+NAME_RE = re.compile(r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$")
+AUTHOR_KEYS = {"name", "email", "url"}
+SERVER_TYPES = {"stdio", "streamable-http", "sse"}
+
+
+def _load(name: str) -> dict:
+    return json.loads((PLUGIN_DIR / name).read_text())
+
+
+def validate_plugin_manifest() -> None:
+    m = _load("plugin.json")
+    assert m.get("$schema") == PLUGIN_SCHEMA, f"plugin.json $schema must be {PLUGIN_SCHEMA}"
+    name = m.get("name")
+    assert isinstance(name, str) and 1 <= len(name) <= 64, "name must be a 1-64 char string"
+    assert NAME_RE.match(name), f"name {name!r} violates the Agent Plugins name pattern"
+    author = m.get("author")
+    if author is not None:
+        assert isinstance(author, dict), "author must be an object"
+        assert set(author).issubset(AUTHOR_KEYS), f"author allows only {AUTHOR_KEYS}"
+    extensions = m.get("extensions")
+    if extensions is not None:
+        assert isinstance(extensions, dict), "extensions must be an object"
+        for key in extensions:
+            assert "." in key, f"extension namespace {key!r} must be reverse-domain"
+
+
+def validate_mcp_manifest() -> None:
+    m = _load("mcp.json")
+    assert m.get("$schema") == MCP_SCHEMA, f"mcp.json $schema must be {MCP_SCHEMA}"
+    servers = m.get("mcpServers")
+    assert isinstance(servers, dict) and servers, "mcp.json needs a non-empty mcpServers object"
+    for server_name, server in servers.items():
+        stype = server.get("type")
+        assert stype in SERVER_TYPES, f"{server_name}: type must be one of {SERVER_TYPES}"
+        if stype == "stdio":
+            assert server.get("command"), f"{server_name}: stdio server requires command"
+        else:  # streamable-http | sse
+            assert server.get("url"), f"{server_name}: {stype} server requires url"
+        headers = server.get("headers", {})
+        assert all(isinstance(v, str) for v in headers.values()), (
+            f"{server_name}: header values must be strings"
+        )
+
+
+def validate_skills() -> None:
+    skills_dir = PLUGIN_DIR / "skills"
+    assert skills_dir.is_dir(), "skills/ directory is missing"
+    skill_files = list(skills_dir.glob("*/SKILL.md"))
+    assert skill_files, "expected at least one skills/<name>/SKILL.md"
+    for skill in skill_files:
+        text = skill.read_text()
+        assert text.startswith("---"), f"{skill} must open with YAML frontmatter"
+        assert "name:" in text and "description:" in text, (
+            f"{skill} frontmatter needs name and description"
+        )
+
+
+def test_plugin_manifest() -> None:
+    validate_plugin_manifest()
+
+
+def test_mcp_manifest() -> None:
+    validate_mcp_manifest()
+
+
+def test_skills() -> None:
+    validate_skills()
+
+
+def main() -> int:
+    for check in (validate_plugin_manifest, validate_mcp_manifest, validate_skills):
+        check()
+        print(f"ok: {check.__name__}")
+    print("Agent Plugin is valid against Agent Plugins 1.0.0 required fields.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -13,7 +13,7 @@ print_info() { echo -e "${GREEN}[INFO]${NC} $1"; }
 print_warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
 print_error() { echo -e "${RED}[ERROR]${NC} $1"; }

-VALID_INTEGRATIONS=("ag2" "agent-framework" "agentcore" "agno" "aider" "ai-sdk" "autogen" "chat" "claude-agent-sdk" "claude-code" "cline" "cloudflare-oauth-proxy" "coding-agents" "codex" "composio" "continue" "copilot-cli" "crewai" "cursor" "cursor-cli" "devin-desktop" "dify" "eve" "flowise" "gemini-spark" "github-copilot" "google-adk" "haystack" "langgraph" "litellm" "llamaindex" "n8n" "nemoclaw" "obsidian" "omo" "openai-agents" "openclaw" "opencode" "openhands" "paperclip" "pipecat" "pydantic-ai" "roo-code" "smolagents" "strands" "superagent" "vapi" "zcode" "zed")
+VALID_INTEGRATIONS=("ag2" "agent-framework" "agent-plugin" "agentcore" "agno" "aider" "ai-sdk" "autogen" "chat" "claude-agent-sdk" "claude-code" "cline" "cloudflare-oauth-proxy" "coding-agents" "codex" "composio" "continue" "copilot-cli" "crewai" "cursor" "cursor-cli" "devin-desktop" "dify" "eve" "flowise" "gemini-spark" "github-copilot" "google-adk" "haystack" "langgraph" "litellm" "llamaindex" "n8n" "nemoclaw" "obsidian" "omo" "openai-agents" "openclaw" "opencode" "openhands" "paperclip" "pipecat" "pydantic-ai" "roo-code" "smolagents" "strands" "superagent" "vapi" "zcode" "zed")

 usage() {
    print_error "Usage: $0 <integration> <version>"
@@ -58,6 +58,8 @@ get_current_version() {
        grep '"version"' "$dir/package.json" | head -1 | sed 's/.*"version": "\(.*\)".*/\1/'
    elif [ -f "$dir/.claude-plugin/plugin.json" ]; then
        grep '"version"' "$dir/.claude-plugin/plugin.json" | head -1 | sed 's/.*"version": "\(.*\)".*/\1/'
+    elif [ -f "$dir/plugin.json" ]; then
+        grep '"version"' "$dir/plugin.json" | head -1 | sed 's/.*"version": "\(.*\)".*/\1/'
    elif [ -f "$dir/settings.json" ] && grep -q '"version"' "$dir/settings.json"; then
        grep '"version"' "$dir/settings.json" | head -1 | sed 's/.*"version": "\(.*\)".*/\1/'
    else
@@ -155,6 +157,10 @@ elif [ -f "$INTEGRATION_DIR/.claude-plugin/plugin.json" ]; then
    print_info "Updating version in $INTEGRATION_DIR/.claude-plugin/plugin.json"
    sed -i.bak "s/\"version\": \".*\"/\"version\": \"$VERSION\"/" "$INTEGRATION_DIR/.claude-plugin/plugin.json"
    rm "$INTEGRATION_DIR/.claude-plugin/plugin.json.bak"
+elif [ -f "$INTEGRATION_DIR/plugin.json" ]; then
+    print_info "Updating version in $INTEGRATION_DIR/plugin.json"
+    sed -i.bak "s/\"version\": \".*\"/\"version\": \"$VERSION\"/" "$INTEGRATION_DIR/plugin.json"
+    rm "$INTEGRATION_DIR/plugin.json.bak"
 elif [ -f "$INTEGRATION_DIR/settings.json" ] && grep -q '"version"' "$INTEGRATION_DIR/settings.json"; then
    print_info "Updating version in $INTEGRATION_DIR/settings.json"
    sed -i.bak "s/\"version\": \".*\"/\"version\": \"$VERSION\"/" "$INTEGRATION_DIR/settings.json"
@@ -198,6 +198,7 @@ For non-English banks (especially CJK) and the language/extraction-language trad
 | `HINDSIGHT_API_LLM_STRICT_SCHEMA_REFLECT` | Override `HINDSIGHT_API_LLM_STRICT_SCHEMA` for reflect's structured-output extraction only. | Inherits global |
 | `HINDSIGHT_API_LLM_STRICT_SCHEMA_CONSOLIDATION` | Override `HINDSIGHT_API_LLM_STRICT_SCHEMA` for consolidation only (both the batch consolidation call and observation dedup). | Inherits global |
 | `HINDSIGHT_API_LLM_SUPPORTS_MAX_ITEMS` | Whether the LLM backend accepts JSON Schema `maxItems` in structured-output schemas. Set to `false` for backends such as Bedrock Converse that reject this keyword; consolidation still enforces observation caps after parsing. | `true` |
+| `HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL` | Request structured output from the LiteLLM-backed providers (`litellm`, `litellmrouter`, `bedrock`) with a single forced tool call — the response schema becomes the tool's parameters — instead of `response_format`. Set to `true` for backends that reject `response_format` outright. This is region-dependent on Bedrock Claude: `ap-southeast-2` (`au.*` inference profiles) refuses the translated Converse `outputConfig` with `Extra inputs are not permitted`, while the same model in `us-east-1` (`us.*`) accepts it and needs nothing here. Verified against both. If the model answers without calling the tool, the reply is parsed as text as before. Other providers ignore it. | `false` |
 | `HINDSIGHT_API_LLM_OLLAMA_NUM_CTX` | Optional native Ollama `num_ctx` override for structured-output calls. Leave unset to use the model/server default; set a positive integer only when you need a larger context window. | Unset |
 | `HINDSIGHT_API_LLM_GEMINI_SAFETY_SETTINGS` | JSON-encoded list of `{category, threshold}` dicts for Gemini/VertexAI content safety filtering | `null` |
 | `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED` | Reuse the fixed system prefix via the provider's explicit prompt cache, billed at the cached-input rate (Gemini/Vertex `CachedContent`). The cached prefix is shared across all banks and soft-fails to an uncached call. Set to `false` to disable. See [Models](./models#provider-capabilities). | `true` |
@@ -50,7 +50,9 @@ Also supports **any OpenAI-compatible API** (e.g., Azure OpenAI, Together AI, Fi

 > **💡 OpenAI-Compatible Providers**
 >
-Hindsight works with any provider that exposes an OpenAI-compatible API (e.g., Azure OpenAI). Simply set `HINDSIGHT_API_LLM_PROVIDER=openai` and configure `HINDSIGHT_API_LLM_BASE_URL` to point to your provider's endpoint.
+Hindsight works with any provider that exposes an OpenAI-compatible API. Set `HINDSIGHT_API_LLM_PROVIDER=openai` and point `HINDSIGHT_API_LLM_BASE_URL` at the endpoint that serves `/chat/completions` — for most providers that is the URL ending in `/v1`, **not** the account or resource root.
+
+**Azure OpenAI does not serve the API at the resource root**, so `https://<resource>.openai.azure.com` on its own returns `404 Resource not found`. See [Azure OpenAI Setup](#azure-openai-setup) for the two URL shapes that work.

 The `openai` provider talks to the **Chat Completions API** (`/v1/chat/completions`). For the newer **Responses API** (`/v1/responses`), use `HINDSIGHT_API_LLM_PROVIDER=openai-responses` — see the tip below. Both accept a custom `HINDSIGHT_API_LLM_BASE_URL`, so an OpenAI-compatible endpoint that exposes `/v1/responses` works the same way as a Chat Completions one.

@@ -636,6 +638,56 @@ one, run one replica on this provider and give the others an API-key lane.

 ---

+### Azure OpenAI Setup
+
+Azure OpenAI is reached through the **`openai`** provider — there is no `azure`
+provider, and setting one fails at startup with
+`Invalid LLM provider: azure`.
+
+The one thing that trips people up is the base URL. Azure does not serve the
+OpenAI API at the resource root, so the endpoint shown in the Azure portal is
+not usable on its own:
+
+| `HINDSIGHT_API_LLM_BASE_URL` | Result |
+|---|---|
+| `https://<resource>.openai.azure.com` | `404 Resource not found` |
+| `https://<resource>.openai.azure.com/openai/deployments/<deployment>` | `404 Resource not found` (no `api-version`) |
+| `https://<resource>.openai.azure.com/openai/v1` | works |
+| `https://<resource>.openai.azure.com/openai/deployments/<deployment>?api-version=<version>` | works |
+
+**Recommended — the v1 surface:**
+
+```bash
+export HINDSIGHT_API_LLM_PROVIDER=openai
+export HINDSIGHT_API_LLM_API_KEY=<azure-openai-resource-key>
+export HINDSIGHT_API_LLM_MODEL=<deployment-name>
+export HINDSIGHT_API_LLM_BASE_URL=https://<resource>.openai.azure.com/openai/v1
+```
+
+**Or the deployment-scoped form.** Keep the `api-version` query string — Hindsight
+parses it out of the base URL and passes it to the SDK:
+
+```bash
+export HINDSIGHT_API_LLM_BASE_URL=https://<resource>.openai.azure.com/openai/deployments/<deployment>?api-version=2025-01-01-preview
+```
+
+**Important notes:**
+- `HINDSIGHT_API_LLM_MODEL` is your **deployment name**, not the model name. A
+  `gpt-4o` deployed as `my-gpt4o` is configured as `my-gpt4o`.
+- The key is the Azure OpenAI **resource** key (`az cognitiveservices account
+  keys list -n <resource> -g <group>`). An API Management subscription key is a
+  different credential: with APIM in front, the base URL must be the APIM route
+  and APIM has to forward the `api-key` header. Test against the Azure endpoint
+  directly first to isolate which layer is failing.
+- Gateways and proxies must preserve the same path shape (`/openai/v1` or
+  `/openai/deployments/...?api-version=`).
+- Azure accepts the `prompt_cache_key` field that
+  [`HINDSIGHT_API_LLM_CACHE_AFFINITY`](./configuration#llm-provider) sends under
+  `auto`, on every `api-version` from `2024-02-01` onward, so the default needs
+  no adjustment for Azure.
+
+---
+
 ### Vertex AI Setup (Google Cloud)

 Google Cloud's Vertex AI provides access to Gemini models via the native Google GenAI SDK.
Author	SHA1	Message	Date
Ben	82c6db52ad	feat(agent-plugin): add portable Hindsight plugin for the Agent Plugins standard Add a vendor-neutral Hindsight plugin conforming to Vercel's Agent Plugins 1.0.0 standard (plugin.json + mcp.json + skills/SKILL.md), so one artifact gives long-term memory to any compatible client (Codex, Cursor, GitHub Copilot, Kiro, VS Code) instead of a per-IDE integration. The plugin is a thin transport wrapper over Hindsight's existing MCP server (retain / recall / reflect); a bundled skill teaches the agent when to use it. Wiring: - CI: test-agent-plugin-integration job runs the manifest validator, gated on hindsight-integrations/agent-plugin/** changes. - Docs: integrations.json gallery entry + docs-integrations/agent-plugin.md. - Release: agent-plugin added to release-integration.sh and the changelog generator; both learn to read a root-level plugin.json and link the changelog to the source tree (git-distributed bundle, no registry package).	2026-08-11 11:21:54 -04:00
Nicolò Boschi	475895f0a2	fix(retain): fold shared document_id items on the sync path (#3363 ) (#3386 ) The sync retain batch endpoint rejected any batch whose items shared a document_id, contradicting the RetainRequest schema/example, the MemoryItem.document_id docs, and the SDK's batch-level documentId (which inlines one id into every item). The guard existed to avoid a race, but that race is only real on the queued path, where children fan out to parallel workers. The synchronous path processes sub-batches sequentially. retain_batch_async now folds items sharing an explicit document_id into one document, in request order, running each document in a single orchestrator pass. A single pass is required: splitting one document across sub-batches that carry different bodies trips the streaming pipeline's content-hash ownership check and silently drops later sub-batches. Batches with no shared document_id are unchanged. The queued path (submit_async_retain) keeps the guard, with a message that points clients at async=false for folding.	2026-08-11 16:17:56 +02:00
Nicolò Boschi	efc179f715	feat(memories): per-bank store capabilities on main (#3388 ) Re-applies the per-bank store-capability seam onto current main. The pluggable memories backend (#2917) is on main, but the per-bank capabilities landed later on feat/pluggable-memories-provider (#3350, plus fix #3381) while main advanced ~179 commits. A pluggable memories store may keep memory rows outside SQL and/or own the document store. The process-level flags writes_memory_rows_in_sql / owns_document_store gain per-bank forms — writes_memory_rows_in_sql_for(bank_id) and owns_document_store_for(bank_id), defaulting to the class attrs — and every bank-scoped call site in memory_engine, consolidation/consolidator, retain/* and reflect/tools consults the per-bank form. Process-level maintenance gates keep reading the class attr. Also two NameError fixes of the same class (a bare bank_id where the in-scope variable differs): list_banks (row["bank_id"], originally #3381) and get_chunk (chunk["bank_id"], newly surfaced by pyflakes while rebasing). Conflict resolution: only consolidation/consolidator.py conflicted — main added consolidation sites since the branch; all are bank_id-scoped, so all convert to the per-bank form. Validation: pyflakes on all changed engine files reports 0 undefined names; py_compile clean; per-bank + list_banks unit tests included.	2026-08-11 16:00:23 +02:00
Nicolò Boschi	7b35d2c6f2	feat(llm): opt-in forced-tool structured output for LiteLLM providers (#3300 ) (#3382 ) * feat(llm): opt-in forced-tool structured output for LiteLLM providers (#3300) Bedrock Claude rejects the structured-output route Hindsight uses. LiteLLM sends a well-formed Converse `outputConfig`; Bedrock's Anthropic layer rewrites it to snake_case internally and its own validator then refuses the key: BedrockException - {"message": "The model returned the following errors: output_config.format: Extra inputs are not permitted"} Every `LiteLLMLLM.call()` with a `response_format` fails on that provider, so retain returns 500 and consolidation degrades to "skipping batch" with no API error at all. Reflect is unaffected because it goes through `call_with_tools()`, which emits `toolConfig` — and Bedrock accepts that. The reporter's boto3 repro isolates it to the transport, not the schema: the same trivial schema fails via `outputConfig` and succeeds via `toolConfig`. This is a different failure from the two earlier Bedrock schema fixes (#1289 `minimum`/`maximum`, #2500 `maxItems`). Those leaked one unsupported keyword into an otherwise-accepted request and were fixed by not emitting it; here the whole `response_format` route is refused, so no amount of schema sanitizing helps. HINDSIGHT_API_LLM_STRUCTURED_OUTPUT_FORCED_TOOL (default false) makes the LiteLLM-backed providers — litellm, litellmrouter, bedrock — ask for structured output the way `anthropic_llm.py` already does natively: one tool whose parameters are the response schema, forced via tool_choice. The tool call's arguments are substituted for the message content before the existing parse block, so markdown-stripping, `parse_llm_json` repair, retries, usage accounting and tracing are untouched. If the model answers without calling the tool — a gateway that drops tool_choice — the text is parsed as before. Default false because every other LiteLLM backend handles `response_format` natively; this only pays off where the backend refuses it. * docs(llm): record the verified Bedrock behaviour behind the forced-tool flag Reproduced on a real AWS account. In ap-southeast-2 with the au.* inference profile, raw boto3 converse (no litellm at all) refuses the outputConfig structured-output route while accepting the identical schema via toolConfig; the same model in us-east-1 with us.* accepts both. So this is region/inference-profile dependent, not "Bedrock Claude is broken" — which is why the flag stays opt-in rather than being keyed off the provider. Also: the rejected key comes back as `model: Extra inputs are not permitted`, not the `output_config.format:` the issue quotes. Same validator and signature, but operators grepping for that exact string would not find it, so the docs now name the behaviour instead of the key.	2026-08-11 14:44:16 +02:00
Nicolò Boschi	d4ac97d643	docs: correct the Azure OpenAI base URL (#3385 ) Reported in #3377 and verified against a live Azure OpenAI resource. The OpenAI-compatible tip told users to point HINDSIGHT_API_LLM_BASE_URL at "your provider's endpoint", which for Azure reads as the resource root -- and Azure does not serve the API there, so it returns 404 Resource not found. Measured against a real resource (gpt-5-mini deployment): https://<res>.openai.azure.com 404 https://<res>.openai.azure.com/openai/deployments/<dep> 404 (no api-version) https://<res>.openai.azure.com/openai/v1 works .../openai/deployments/<dep>?api-version=2025-01-01-preview works Adds an Azure OpenAI Setup section with both working shapes and the three things that actually bite: the model is the deployment name, the key is the resource key (an APIM subscription key is a different setup), and gateways must preserve the path shape. Also records that Azure accepts the prompt_cache_key field sent under cache_affinity=auto (#3271) on every api-version from 2024-02-01 onward, so that default needs no Azure carve-out -- an explicitly untested risk when #3271 merged, now closed.	2026-08-11 13:45:51 +02:00
Nicolò Boschi	a2b018dce7	fix(retain): sweep observations when delta retain deletes chunks (#3384 ) Delta retain drops a document's outgoing facts by deleting their chunks and letting the FK cascade take the memory_units with them. Nothing swept the observations derived from those facts: the sweep lives in handle_document_tracking, which only the full-replace path calls. Every re-ingest that took the delta path — a small edit to an existing document, exactly what delta retain is for — therefore left the observations of the changed chunks behind, still valid and still recallable, pointing at source_memory_ids that no longer resolved. Those rows were unreachable afterwards: consolidation batches are built from facts, so an observation whose sources are all gone is never selected into a batch again, and no runtime path deletes it. Sweep in delete_chunks_by_ids, before the cascade, so the invariant holds at the choke point rather than at one call site. It returns the number it invalidated and the delta log line now carries that count unconditionally — "the sweep matched nothing" and "the sweep never ran" were indistinguishable from the outside, which is what made this hard to diagnose. The sweep is keyed on the deleted chunks, not the document, so an edit to one chunk leaves the other chunks' observations alone instead of requeueing the whole document for consolidation. Supersedes #3302. Reported and diagnosed by @fhiltscher. Fixes #3294.	2026-08-11 13:26:41 +02:00