feat(admin): add /admin config surface (API + control plane)

Server-level admin surface, gated by HINDSIGHT_API_ENABLE_ADMIN_API with an optional, independent HINDSIGHT_API_ADMIN_TOKEN: - API: GET /admin/config returns the resolved HindsightConfig with credentials redacted (set -> "***", unset -> null) via the credential denylist plus a name-suffix heuristic; features.admin_api exposed on /version. - Control plane: top-level /admin page reusing the shared (now data-driven) Sidebar with a single "Configuration" item + a dedicated AdminHeader; server proxy forwards an independent HINDSIGHT_CP_ADMIN_TOKEN to /admin/*. - Regenerated OpenAPI spec + Python/TS/Go clients; docs + .env.example; admin i18n across all locales. Groundwork for the visibility leg of #2034. The connectivity test + health surfacing land separately as a per-bank health endpoint.
fix(api): stop forcing vchordrq.probes session GUC on listless vchord indexes (#2076 )
2026-06-09 15:38:13 +02:00 · 2026-06-09 13:54:36 +02:00 · 2026-06-09 13:29:20 +02:00 · 2026-06-09 13:23:11 +02:00 · 2026-06-08 16:45:22 -04:00 · 2026-06-08 16:43:06 -04:00
912 changed files with 44813 additions and 23506 deletions
@@ -172,7 +172,8 @@ For each non-trivial change:
 If any files in `hindsight-integrations/` were added or changed, verify:
 - **Tests exist** — the integration must have tests that simulate/exercise the external framework (not just pure unit tests of helpers). Check for a `tests/` directory with meaningful test files.
 - **CI job exists** — check `.github/workflows/test.yml` for a corresponding `test-<name>-integration` job. If missing, flag it.
- **Release process** — check that the integration name is in the `VALID_INTEGRATIONS` array in `scripts/release-integration.sh`. If missing, flag it.
+- **Release process** — check that the integration name is in the `VALID_INTEGRATIONS` array in `scripts/release-integration.sh` AND in the `INTEGRATIONS` dict in `hindsight-dev/hindsight_dev/generate_changelog.py` (the changelog generator keeps its own list; a release fails at the changelog step if the name is missing there). If either is missing, flag it.
+- **Docs gallery + sidebar entry** — the integration must have an entry in `hindsight-docs/src/data/integrations.json`. This file is the **single source of truth** that drives both the integrations gallery and the docs sidebar (the sidebar category is injected from it at render time across all docs versions). The entry needs an internal `/sdks/integrations/<slug>` `link` and a matching page at `hindsight-docs/docs-integrations/<slug>.md(x)`. The `hindsight-docs/scripts/check-integrations.mjs` build step enforces both directions — forward: every internal JSON entry has a doc page; reverse: every released tag (`integrations/<name>/vX.Y.Z`) appears in the JSON (private infra like `cloudflare-oauth-proxy` is in the script's `EXCLUDED` set). Flag any integration that is released (or being released) but missing from `integrations.json`, and any JSON entry without a doc page. Do **not** hand-edit `versioned_sidebars/*.json` to add integration links — they are positional placeholders filled from the JSON.
 - **Code standards** — the integration code must follow all Python style rules (type hints, no raw dicts, no tuple returns, etc.).

 ### 10. Check MCP tool registration completeness
@@ -217,6 +218,7 @@ Present a clear summary organized by severity:
 - Direct DB access (raw SQL / `acquire_with_retry` / `fq_table`) in an `api/` handler instead of a `MemoryEngine` method
 - Tenant-scoped data accessed without authentication enforced in the engine (`_authenticate_tenant` / `get_bank_profile`)
 - New integration missing tests, CI job, or release-integration.sh entry
+- Released/added integration missing from `hindsight-docs/src/data/integrations.json`, or a JSON entry with no `docs-integrations/<slug>` page (fails the docs build via `check-integrations.mjs`)
 - New PostgreSQL table missing from `BACKUP_TABLES` in `admin/cli.py` (silent data loss on restore)

 **Should fix** — issues that hurt code quality:
@@ -159,3 +159,20 @@ HINDSIGHT_API_LOG_LEVEL=info
 # When set, visitors see a login page and must enter the key before
 # accessing the dashboard or any /api/* routes (except /api/health).
 # HINDSIGHT_CP_ACCESS_KEY=your-shared-secret-key
+
+# Optional: Token the CP forwards to the dataplane admin API (/admin/*).
+# Must match HINDSIGHT_API_ADMIN_TOKEN below. Leave unset for an open admin API.
+# HINDSIGHT_CP_ADMIN_TOKEN=your-admin-token
+
+# -----------------------------------------------------------------------------
+# Admin surface (Optional, server-level)
+# -----------------------------------------------------------------------------
+
+# Enable the admin API (GET /admin/config) and the Control Plane /admin page.
+# Off by default — the admin surface is invisible (404) until enabled.
+# HINDSIGHT_API_ENABLE_ADMIN_API=true
+
+# Optional: Require this bearer token for the admin API. When unset, the admin
+# API is open (once enabled). When set, callers must send
+# `Authorization: Bearer <token>`. Independent of the tenant API key.
+# HINDSIGHT_API_ADMIN_TOKEN=your-admin-token
@@ -22,6 +22,8 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v6
+        with:
+          fetch-depth: 0 # fetch tags so check-released-integrations can see them
      - uses: actions/setup-node@v6
        with:
          node-version: 20
@@ -23,6 +23,7 @@ on:
          - retain
          - recall
          - recall-with-observations
+          - recall-temporal
          - consolidation
          - graph-maintenance
        default: ""
@@ -10,6 +10,7 @@ jobs:
    runs-on: ubuntu-latest
    permissions:
      id-token: write  # for PyPI trusted publishing
+      contents: write  # for creating GitHub releases (Obsidian plugin assets)

    steps:
    - uses: actions/checkout@v6
@@ -112,6 +113,26 @@ jobs:
      working-directory: ./hindsight-integrations/${{ steps.info.outputs.integration }}
      run: npm run build

+    # ── Obsidian plugin — attach BRAT / community-store install assets ───────
+    # Obsidian plugins install from GitHub *release assets* (main.js,
+    # manifest.json, styles.css), not from npm — the npm publish below only
+    # gives us a versioned artifact. BRAT and the community store read these
+    # three files off the release for the tag.
+    - name: Attach Obsidian release assets
+      if: steps.type.outputs.type == 'typescript' && steps.info.outputs.integration == 'obsidian'
+      working-directory: ./hindsight-integrations/obsidian
+      env:
+        GH_TOKEN: ${{ github.token }}
+      run: |
+        TAG="${{ steps.info.outputs.tag }}"
+        if gh release view "$TAG" >/dev/null 2>&1; then
+          gh release upload "$TAG" main.js manifest.json styles.css --clobber
+        else
+          gh release create "$TAG" main.js manifest.json styles.css \
+            --title "Obsidian plugin v${{ steps.info.outputs.version }}" \
+            --notes "Hindsight Obsidian plugin v${{ steps.info.outputs.version }}. Install via BRAT (point it at this release) or copy main.js/manifest.json/styles.css into <vault>/.obsidian/plugins/hindsight/."
+        fi
+
    - name: Publish TypeScript package to npm
      if: steps.type.outputs.type == 'typescript'
      working-directory: ./hindsight-integrations/${{ steps.info.outputs.integration }}
@@ -34,27 +34,35 @@ jobs:
      integrations-ai-sdk: ${{ steps.filter.outputs.integrations-ai-sdk }}
      integrations-chat: ${{ steps.filter.outputs.integrations-chat }}
      integrations-claude-code: ${{ steps.filter.outputs.integrations-claude-code }}
+      integrations-cline: ${{ steps.filter.outputs.integrations-cline }}
      integrations-codex: ${{ steps.filter.outputs.integrations-codex }}
+      integrations-cursor-cli: ${{ steps.filter.outputs.integrations-cursor-cli }}
      integrations-crewai: ${{ steps.filter.outputs.integrations-crewai }}
      integrations-litellm: ${{ steps.filter.outputs.integrations-litellm }}
      integrations-pydantic-ai: ${{ steps.filter.outputs.integrations-pydantic-ai }}
      integrations-ag2: ${{ steps.filter.outputs.integrations-ag2 }}
      integrations-autogen: ${{ steps.filter.outputs.integrations-autogen }}
+      integrations-langgraph: ${{ steps.filter.outputs.integrations-langgraph }}
      integrations-llamaindex: ${{ steps.filter.outputs.integrations-llamaindex }}
      integrations-paperclip: ${{ steps.filter.outputs.integrations-paperclip }}
      integrations-opencode: ${{ steps.filter.outputs.integrations-opencode }}
      integrations-n8n: ${{ steps.filter.outputs.integrations-n8n }}
      integrations-cloudflare-oauth-proxy: ${{ steps.filter.outputs.integrations-cloudflare-oauth-proxy }}
+      integrations-superagent: ${{ steps.filter.outputs.integrations-superagent }}
      integrations-lockfiles: ${{ steps.filter.outputs.integrations-lockfiles }}
      integrations-openai-agents: ${{ steps.filter.outputs.integrations-openai-agents }}
      integrations-pipecat: ${{ steps.filter.outputs.integrations-pipecat }}
      integrations-agentcore: ${{ steps.filter.outputs.integrations-agentcore }}
      integrations-smolagents: ${{ steps.filter.outputs.integrations-smolagents }}
+      integrations-claude-agent-sdk: ${{ steps.filter.outputs.integrations-claude-agent-sdk }}
      integrations-dify: ${{ steps.filter.outputs.integrations-dify }}
      integrations-gemini-spark: ${{ steps.filter.outputs.integrations-gemini-spark }}
      integrations-vapi: ${{ steps.filter.outputs.integrations-vapi }}
      integrations-flowise: ${{ steps.filter.outputs.integrations-flowise }}
      integrations-google-adk: ${{ steps.filter.outputs.integrations-google-adk }}
+      integrations-obsidian: ${{ steps.filter.outputs.integrations-obsidian }}
+      integrations-omo: ${{ steps.filter.outputs.integrations-omo }}
+      integrations-haystack: ${{ steps.filter.outputs.integrations-haystack }}
      tools-agent-sdk: ${{ steps.filter.outputs.tools-agent-sdk }}
      integrations-roo-code: ${{ steps.filter.outputs.integrations-roo-code }}
      dev: ${{ steps.filter.outputs.dev }}
@@ -98,6 +106,9 @@ jobs:
          docs:
            - 'hindsight-docs/**'
            - '*.md'
+            # Integration changes can add/rename integrations, which the docs
+            # build's integrations check validates against integrations.json.
+            - 'hindsight-integrations/**'
          embed:
            - 'hindsight-embed/**'
          all-npm:
@@ -116,8 +127,12 @@ jobs:
            - 'hindsight-integrations/chat/**'
          integrations-claude-code:
            - 'hindsight-integrations/claude-code/**'
+          integrations-cline:
+            - 'hindsight-integrations/cline/**'
          integrations-codex:
            - 'hindsight-integrations/codex/**'
+          integrations-cursor-cli:
+            - 'hindsight-integrations/cursor-cli/**'
          integrations-crewai:
            - 'hindsight-integrations/crewai/**'
          integrations-litellm:
@@ -128,8 +143,12 @@ jobs:
            - 'hindsight-integrations/ag2/**'
          integrations-autogen:
            - 'hindsight-integrations/autogen/**'
+          integrations-langgraph:
+            - 'hindsight-integrations/langgraph/**'
          integrations-llamaindex:
            - 'hindsight-integrations/llamaindex/**'
+          integrations-haystack:
+            - 'hindsight-integrations/haystack/**'
          integrations-paperclip:
            - 'hindsight-integrations/paperclip/**'
          integrations-opencode:
@@ -138,6 +157,8 @@ jobs:
            - 'hindsight-integrations/n8n/**'
          integrations-cloudflare-oauth-proxy:
            - 'hindsight-integrations/cloudflare-oauth-proxy/**'
+          integrations-superagent:
+            - 'hindsight-integrations/superagent/**'
          integrations-lockfiles:
            - 'hindsight-integrations/*/package-lock.json'
            - 'hindsight-integrations/*/package.json'
@@ -150,6 +171,8 @@ jobs:
            - 'hindsight-integrations/agentcore/**'
          integrations-smolagents:
            - 'hindsight-integrations/smolagents/**'
+          integrations-claude-agent-sdk:
+            - 'hindsight-integrations/claude-agent-sdk/**'
          integrations-dify:
            - 'hindsight-integrations/dify/**'
          integrations-gemini-spark:
@@ -160,6 +183,10 @@ jobs:
            - 'hindsight-integrations/flowise/**'
          integrations-google-adk:
            - 'hindsight-integrations/google-adk/**'
+          integrations-obsidian:
+            - 'hindsight-integrations/obsidian/**'
+          integrations-omo:
+            - 'hindsight-integrations/omo/**'
          tools-agent-sdk:
            - 'hindsight-tools/hindsight-agent-sdk/**'
          integrations-roo-code:
@@ -420,6 +447,58 @@ jobs:
      working-directory: ./hindsight-integrations/claude-code
      run: python -m pytest tests/ -v

+  test-omo-integration:
+    needs: [detect-changes]
+    if: >-
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-omo == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    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: '3.11'
+
+    - name: Install pytest
+      run: pip install pytest
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/omo
+      run: python -m pytest tests/ -v
+
+  test-cline-integration:
+    needs: [detect-changes]
+    if: >-
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-cline == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    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: '3.11'
+
+    - name: Install pytest
+      run: pip install pytest
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/cline
+      run: python -m pytest tests/ -v
+
  test-codex-integration:
    needs: [detect-changes]
    if: >-
@@ -446,6 +525,32 @@ jobs:
      working-directory: ./hindsight-integrations/codex
      run: python -m pytest tests/ -v

+  test-cursor-cli-integration:
+    needs: [detect-changes]
+    if: >-
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-cursor-cli == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    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: '3.11'
+
+    - name: Install pytest
+      run: pip install pytest
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/cursor-cli
+      run: python -m pytest tests/ -v
+
  build-ai-sdk-integration:
    needs: [detect-changes]
    if: >-
@@ -819,17 +924,28 @@ jobs:
      with:
        ref: ${{ github.event.pull_request.head.sha || '' }}

+    - name: Install uv
+      uses: astral-sh/setup-uv@v7
+      with:
+        enable-cache: true
+        prune-cache: false
+
    - name: Set up Python
      uses: actions/setup-python@v6
      with:
-        python-version: '3.11'
+        python-version-file: ".python-version"

-    - name: Install pytest
-      run: pip install pytest
+    - name: Build roo-code integration
+      working-directory: ./hindsight-integrations/roo-code
+      run: uv build
+
+    - name: Install dependencies
+      working-directory: ./hindsight-integrations/roo-code
+      run: uv sync --frozen

    - name: Run tests
      working-directory: ./hindsight-integrations/roo-code
-      run: python -m pytest tests/ -v
+      run: uv run pytest tests -v

  build-control-plane:
    needs: [detect-changes]
@@ -911,6 +1027,7 @@ jobs:
    - uses: actions/checkout@v6
      with:
        ref: ${{ github.event.pull_request.head.sha || '' }}
+        fetch-depth: 0 # fetch tags so check-released-integrations can see them

    - name: Set up Node.js
      uses: actions/setup-node@v6
@@ -919,6 +1036,12 @@ jobs:
        cache: 'npm'
        cache-dependency-path: package-lock.json

+    # Fail fast before the (slow) build: every integrations.json entry must have a
+    # doc page, and every released integration tag must be in integrations.json.
+    # Needs no npm install (pure Node) and uses the tags fetched above.
+    - name: Check integrations (single source of truth)
+      run: node hindsight-docs/scripts/check-integrations.mjs
+
    - name: Install dependencies
      run: npm ci --workspace=hindsight-docs

@@ -2946,6 +3069,41 @@ jobs:
      working-directory: ./hindsight-integrations/flowise
      run: npm test

+  test-obsidian-integration:
+    needs: [detect-changes]
+    if: >-
+      github.event_name != 'pull_request_review' &&
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-obsidian == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v6
+      with:
+        ref: ${{ github.event.pull_request.head.sha || '' }}
+
+    - name: Set up Node.js
+      uses: actions/setup-node@v6
+      with:
+        node-version: '22'
+
+    - name: Install dependencies
+      working-directory: ./hindsight-integrations/obsidian
+      run: npm install --no-audit --no-fund
+
+    - name: Type check
+      working-directory: ./hindsight-integrations/obsidian
+      run: npx tsc --noEmit
+
+    - name: Build
+      working-directory: ./hindsight-integrations/obsidian
+      run: npm run build
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/obsidian
+      run: npm test
+
  test-crewai-integration:
    needs: [detect-changes]
    if: >-
@@ -3021,6 +3179,45 @@ jobs:
      working-directory: ./hindsight-integrations/vapi
      run: uv run pytest tests -v

+  test-superagent-integration:
+    needs: [detect-changes]
+    if: >-
+      github.event_name != 'pull_request_review' &&
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-superagent == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v6
+      with:
+        ref: ${{ github.event.pull_request.head.sha || '' }}
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v7
+      with:
+        enable-cache: true
+        prune-cache: false
+
+    - name: Set up Python
+      uses: actions/setup-python@v6
+      with:
+        python-version-file: ".python-version"
+
+    - name: Build superagent integration
+      working-directory: ./hindsight-integrations/superagent
+      run: uv build
+
+    - name: Install dependencies
+      working-directory: ./hindsight-integrations/superagent
+      run: uv sync --frozen
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/superagent
+      # PR CI runs only the deterministic bucket; the real-LLM E2E bucket
+      # (requires_real_llm) needs live Hindsight + provider keys and runs separately.
+      run: uv run pytest tests -v -m "not requires_real_llm"
+
  test-litellm-integration:
    needs: [detect-changes]
    if: >-
@@ -3056,7 +3253,9 @@ jobs:

    - name: Run tests
      working-directory: ./hindsight-integrations/litellm
-      run: uv run pytest tests -v
+      # PR CI runs only the deterministic bucket; the real-LLM E2E bucket
+      # (requires_real_llm) needs live Hindsight + provider keys and runs separately.
+      run: uv run pytest tests -v -m "not requires_real_llm"

  test-pydantic-ai-integration:
    needs: [detect-changes]
@@ -3095,6 +3294,45 @@ jobs:
      working-directory: ./hindsight-integrations/pydantic-ai
      run: uv run pytest tests -v

+  test-langgraph-integration:
+    needs: [detect-changes]
+    if: >-
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-langgraph == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    steps:
+    - uses: actions/checkout@v6
+      with:
+        ref: ${{ github.event.pull_request.head.sha || '' }}
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v7
+      with:
+        enable-cache: true
+        prune-cache: false
+
+    - name: Set up Python
+      uses: actions/setup-python@v6
+      with:
+        python-version-file: ".python-version"
+
+    - name: Build langgraph integration
+      working-directory: ./hindsight-integrations/langgraph
+      run: uv build
+
+    - name: Install dependencies
+      working-directory: ./hindsight-integrations/langgraph
+      run: uv sync --frozen
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/langgraph
+      # PR CI runs only the deterministic bucket; the real-LLM E2E bucket
+      # (requires_real_llm) needs a live Hindsight server and runs separately.
+      run: uv run pytest tests -v -m "not requires_real_llm"
+
  test-llamaindex-integration:
    needs: [detect-changes]
    if: >-
@@ -3134,6 +3372,45 @@ jobs:
      # (requires_real_llm) needs a live Hindsight server and runs separately.
      run: uv run pytest tests -v -m "not requires_real_llm"

+  test-haystack-integration:
+    needs: [detect-changes]
+    if: >-
+      github.event_name != 'pull_request_review' &&
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-haystack == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v6
+      with:
+        ref: ${{ github.event.pull_request.head.sha || '' }}
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v7
+      with:
+        enable-cache: true
+        prune-cache: false
+
+    - name: Set up Python
+      uses: actions/setup-python@v6
+      with:
+        python-version-file: ".python-version"
+
+    - name: Build haystack integration
+      working-directory: ./hindsight-integrations/haystack
+      run: uv build
+
+    - name: Install dependencies
+      working-directory: ./hindsight-integrations/haystack
+      run: uv sync --frozen
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/haystack
+      # PR CI runs only the deterministic bucket; the real-LLM E2E bucket
+      # (requires_real_llm) needs a live Hindsight server and runs separately.
+      run: uv run pytest tests -v -m "not requires_real_llm"
+
  test-openai-agents-integration:
    needs: [detect-changes]
    if: >-
@@ -3169,7 +3446,47 @@ jobs:

    - name: Run tests
      working-directory: ./hindsight-integrations/openai-agents
-      run: uv run pytest tests -v
+      # PR CI runs only the deterministic bucket; the real-LLM E2E bucket
+      # (requires_real_llm) needs a live Hindsight server and runs separately.
+      run: uv run pytest tests -v -m "not requires_real_llm"
+
+  test-claude-agent-sdk-integration:
+    needs: [detect-changes]
+    if: >-
+      (github.event_name == 'workflow_dispatch' ||
+      needs.detect-changes.outputs.integrations-claude-agent-sdk == 'true' ||
+      needs.detect-changes.outputs.ci == 'true')
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v6
+      with:
+        ref: ${{ github.event.pull_request.head.sha || '' }}
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v7
+      with:
+        enable-cache: true
+        prune-cache: false
+
+    - name: Set up Python
+      uses: actions/setup-python@v6
+      with:
+        python-version-file: ".python-version"
+
+    - name: Build claude-agent-sdk integration
+      working-directory: ./hindsight-integrations/claude-agent-sdk
+      run: uv build
+
+    - name: Install dependencies
+      working-directory: ./hindsight-integrations/claude-agent-sdk
+      run: uv sync --frozen
+
+    - name: Run tests
+      working-directory: ./hindsight-integrations/claude-agent-sdk
+      # PR CI runs only the deterministic bucket; the real-LLM E2E bucket
+      # (requires_real_llm) needs a live Hindsight server and runs separately.
+      run: uv run pytest tests -v -m "not requires_real_llm"

  test-agentcore-integration:
    needs: [detect-changes]
@@ -4034,10 +4351,13 @@ jobs:
      - build-openclaw-integration
      - smoke-openclaw-install
      - test-claude-code-integration
+      - test-cline-integration
      - test-codex-integration
+      - test-cursor-cli-integration
      - build-ai-sdk-integration
      - test-ai-sdk-integration-deno
      - test-opencode-integration
+      - test-omo-integration
      - test-cloudflare-oauth-proxy-integration
      - build-chat-integration
      - test-paperclip-integration
@@ -4069,16 +4389,22 @@ jobs:
      - test-smolagents-integration
      - test-dify-integration
      - test-flowise-integration
+      - test-obsidian-integration
      - test-crewai-integration
+      - test-langgraph-integration
+      - test-superagent-integration
      - test-litellm-integration
      - test-pydantic-ai-integration
      - test-llamaindex-integration
+      - test-openai-agents-integration
      - test-agentcore-integration
+      - test-haystack-integration
      - test-pip-slim
      - test-embed
      - test-embed-windows
      - test-hindsight-all
      - test-hindsight-agent-sdk
+      - test-claude-agent-sdk-integration
      - test-doc-examples
      - test-upgrade
      - verify-generated-files
@@ -59,4 +59,5 @@ hindsight-integrations/_drafts/
 # Changelog is now tracked in hindsight-docs/src/pages/changelog.md
 # CHANGELOG.md

-blog-post*
+blog-post*
+.worktrees/
@@ -2,8 +2,8 @@ apiVersion: v2
 name: hindsight
 description: Hindsight helm chart
 type: application
-version: 0.7.2
-appVersion: "0.7.2"
+version: 0.8.0
+appVersion: "0.8.0"
 keywords:
  - ai
  - memory
@@ -1,6 +1,6 @@
 {
  "name": "@vectorize-io/hindsight-all",
-  "version": "0.7.2",
+  "version": "0.8.0",
  "description": "Node.js programmatic lifecycle manager for Hindsight — embeds a local hindsight daemon in a Node application. Pair with @vectorize-io/hindsight-client for memory operations.",
  "main": "dist/index.js",
  "types": "dist/index.d.ts",
@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"

 [project]
 name = "hindsight-all-slim"
-version = "0.7.2"
+version = "0.8.0"
 description = "Hindsight: Agent Memory That Works Like Human Memory - Slim All-in-One Bundle"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    "hindsight-api-slim==0.7.2",
+    "hindsight-api-slim==0.8.0",
    "hindsight-client>=0.0.7",
    "hindsight-embed>=0.1.0",
 ]
@@ -4,12 +4,12 @@ build-backend = "hatchling.build"

 [project]
 name = "hindsight-all"
-version = "0.7.2"
+version = "0.8.0"
 description = "Hindsight: Agent Memory That Works Like Human Memory - All-in-One Bundle"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    "hindsight-api-slim[all]==0.7.2",
+    "hindsight-api-slim[all]==0.8.0",
    "hindsight-client>=0.0.7",
    "hindsight-embed>=0.1.0",
 ]
@@ -21,7 +21,7 @@ hindsight-embed = { workspace = true }

 [project.optional-dependencies]
 local-llm = [
-    "hindsight-api-slim[local-llm]==0.7.2",
+    "hindsight-api-slim[local-llm]==0.8.0",
 ]
 test = [
    "pytest>=7.0.0",
@@ -53,4 +53,4 @@ __all__ = [
    "RemoteTEICrossEncoder",
    "LLMConfig",
 ]
-__version__ = "0.7.2"
+__version__ = "0.8.0"
@@ -54,23 +54,20 @@ _INDEX_TYPE_KEYWORDS = {
 #   pre-dispatcher code (internal benchmarks tuned around our embedding count
 #   and recall floor; see the link_utils / pool init call sites for the
 #   latency-vs-recall framing).
-# - vchord exposes vchordrq.probes (no default; see VectorChord issue #392)
-#   and vchordrq.epsilon (default 1.9). probes = 10 / 30 are starting
-#   defaults pending a workload-specific sweep — vchordrq's recall curve
-#   shape differs from HNSW's, so the pgvector numbers don't translate
-#   directly. Revisit with a per-cluster benchmark once we have production
-#   recall data; until then these are deliberately conservative on the
-#   high-recall path. We leave epsilon at its default; tightening it is a
-#   separate trade-off.
+# - vchord exposes vchordrq.probes, but its shape must match the index's
+#   build.internal.lists hierarchy. VectorChord 1.1 added per-index fallback
+#   parameters for this reason: a session GUC overrides every vchordrq index,
+#   and a single value can be invalid for listless or mixed-layout indexes.
+#   Hindsight's built-in vchord clause does not set lists, so the safe default
+#   is no session-level probe override; deployments that partition vchordrq
+#   indexes should attach probes to the index storage parameters instead.
 # - pgvectorscale / pg_diskann / scann do not expose an equivalent per-statement
 #   knob in the engine today, so the dispatcher returns no statements for them.
 _ANN_TUNING_LOW_LATENCY: dict[str, tuple[tuple[str, str], ...]] = {
    "pgvector": (("hnsw.ef_search", "60"),),
-    "vchord": (("vchordrq.probes", "10"),),
 }
 _ANN_TUNING_HIGH_RECALL: dict[str, tuple[tuple[str, str], ...]] = {
    "pgvector": (("hnsw.ef_search", "200"),),
-    "vchord": (("vchordrq.probes", "30"),),
 }

 _EXTENSION_INSTALL_SQL = {
@@ -463,7 +463,8 @@ def import_bank_command(
    typer.echo(
        f"Imported bank '{result.bank_id}': {result.documents_imported} doc(s), "
        f"{result.facts_imported} fact(s), {result.observations_imported} observation(s), "
-        f"{result.mental_models_imported} mental model(s), {result.directives_imported} directive(s), "
+        f"{result.mental_models_imported} mental model(s), "
+        f"{result.mental_model_history_imported} mm-history row(s), {result.directives_imported} directive(s), "
        f"{result.webhooks_imported} webhook(s), {result.history_rows_imported} history row(s)"
    )

@@ -0,0 +1,156 @@
+"""Repair: install maintenance routines on the ``public`` / base-schema run.
+
+The original maintenance-routines migration (``e5f6a7b8c9d0``) only created the
+shared ``public.banks_needing_consolidation()`` and
+``public.schemas_with_expired_rows(...)`` routines when the run had *no*
+``target_schema`` at all. But the single-tenant runtime always migrates an
+explicit schema — which defaults to ``public`` — so on every default
+PostgreSQL deployment the migration was stamped as applied while the functions
+were never created. Background maintenance then logs::
+
+    Retention sweep failed for llm_requests: function public.schemas_with_expired_rows(...) does not exist
+    Consolidation reconcile discovery failed: function public.banks_needing_consolidation() does not exist
+
+See https://github.com/vectorize-io/hindsight/issues/2056.
+
+Because ``e5f6a7b8c9d0`` is already stamped on affected ``0.8.0`` databases,
+editing it would not re-run it there. This forward migration re-installs the
+functions idempotently (``CREATE OR REPLACE``) on the run that targets the
+shared ``public`` schema (base run with no ``target_schema``, or an explicit
+``target_schema=public``), self-healing already-upgraded deployments and
+covering fresh upgrades from earlier versions.
+
+Per-tenant runs against a non-``public`` schema still skip it: re-issuing
+``CREATE OR REPLACE FUNCTION public....`` from each concurrent tenant migration
+aborts with ``tuple concurrently updated`` on the ``pg_proc`` catalog row, and
+the base/public run has already created the functions for every tenant to use.
+Runs that target ``public`` are serialized by the per-schema migration advisory
+lock, so only one wins the create.
+
+PostgreSQL only — the worker poller and these tables are not wired for Oracle,
+so the Oracle slot is intentionally absent (mirrors ``e5f6a7b8c9d0``).
+
+Revision ID: b2d4f6a8c1e3
+Revises: e5f6a7b8c9d0
+Create Date: 2026-06-08
+"""
+
+from collections.abc import Sequence
+
+from alembic import context, op
+
+from hindsight_api.alembic._dialect import run_for_dialect
+
+revision: str = "b2d4f6a8c1e3"
+down_revision: str | Sequence[str] | None = "e5f6a7b8c9d0"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _should_install_public_routines(target_schema: str | None) -> bool:
+    """True for the run that must (re)create the shared ``public.*`` routines.
+
+    The routines physically live in ``public`` (hard-coded ``public.`` qualifier
+    in the SQL below), so they must be installed exactly once — on the base run
+    (no ``target_schema``) or on the run that explicitly targets ``public``. A
+    run against any other tenant schema skips it to avoid concurrent
+    ``CREATE OR REPLACE`` on the same ``pg_proc`` row.
+    """
+    return not target_schema or target_schema == "public"
+
+
+def _pg_upgrade() -> None:
+    if not _should_install_public_routines(context.config.get_main_option("target_schema")):
+        return
+    # Banks with eligible-but-unscheduled facts and no in-flight consolidation.
+    # Auto-consolidation is filtered here only at the bank level (cheap prune);
+    # the full hierarchical resolution (global -> tenant -> bank, plus
+    # enable_observations) is done by the caller for the small returned set.
+    op.execute(
+        """
+        CREATE OR REPLACE FUNCTION public.banks_needing_consolidation()
+        RETURNS TABLE(schema_name text, bank_id text)
+        LANGUAGE plpgsql STABLE
+        AS $fn$
+        DECLARE
+            sch text;
+        BEGIN
+            FOR sch IN
+                SELECT n.nspname
+                FROM pg_class c
+                JOIN pg_namespace n ON n.oid = c.relnamespace
+                WHERE c.relname = 'memory_units' AND c.relkind = 'r'
+            LOOP
+                RETURN QUERY EXECUTE format($q$
+                    SELECT %1$L::text, m.bank_id
+                    FROM %1$I.memory_units m
+                    JOIN %1$I.banks b ON b.bank_id = m.bank_id
+                    WHERE m.consolidated_at IS NULL
+                      AND m.consolidation_failed_at IS NULL
+                      AND m.fact_type IN ('experience', 'world')
+                      AND COALESCE(b.config -> 'enable_auto_consolidation', 'true'::jsonb) <> 'false'::jsonb
+                      AND NOT EXISTS (
+                          SELECT 1 FROM %1$I.async_operations o
+                          WHERE o.bank_id = m.bank_id
+                            AND o.operation_type = 'consolidation'
+                            AND o.status IN ('pending', 'processing')
+                      )
+                    GROUP BY m.bank_id
+                $q$, sch);
+            END LOOP;
+        END;
+        $fn$;
+        """
+    )
+
+    # Schemas holding at least one row of p_table older than p_days. p_ts_col is
+    # the timestamp column to compare. Returns nothing when p_days <= 0
+    # (retention disabled).
+    op.execute(
+        """
+        CREATE OR REPLACE FUNCTION public.schemas_with_expired_rows(
+            p_table text, p_ts_col text, p_days int
+        )
+        RETURNS SETOF text
+        LANGUAGE plpgsql STABLE
+        AS $fn$
+        DECLARE
+            sch text;
+            has_expired boolean;
+        BEGIN
+            IF p_days IS NULL OR p_days <= 0 THEN
+                RETURN;
+            END IF;
+            FOR sch IN
+                SELECT n.nspname
+                FROM pg_class c
+                JOIN pg_namespace n ON n.oid = c.relnamespace
+                WHERE c.relname = p_table AND c.relkind = 'r'
+            LOOP
+                EXECUTE format(
+                    'SELECT EXISTS (SELECT 1 FROM %I.%I WHERE %I < NOW() - make_interval(days => $1))',
+                    sch, p_table, p_ts_col
+                ) INTO has_expired USING p_days;
+                IF has_expired THEN
+                    RETURN NEXT sch;
+                END IF;
+            END LOOP;
+        END;
+        $fn$;
+        """
+    )
+
+
+def _pg_downgrade() -> None:
+    # No-op: ``e5f6a7b8c9d0`` owns the lifecycle of these functions and drops
+    # them on its own downgrade. This migration only ever (re)creates them, so
+    # there is nothing to undo without racing that migration's DROP.
+    pass
+
+
+def upgrade() -> None:
+    run_for_dialect(pg=_pg_upgrade)
+
+
+def downgrade() -> None:
+    run_for_dialect(pg=_pg_downgrade)
@@ -0,0 +1,153 @@
+"""Add server-side routines for background maintenance sweeps.
+
+Installs two PL/pgSQL discovery routines in the ``public`` schema. Both loop
+over every schema that actually holds the relevant table (via ``pg_class``), so
+a single function call covers all tenants in one round-trip instead of the
+per-tenant query storm that a client-side loop would create at thousands of
+tenants.
+
+- ``public.banks_needing_consolidation()`` -> (schema_name, bank_id) for banks
+  that have eligible-but-unscheduled facts (``consolidated_at IS NULL AND
+  consolidation_failed_at IS NULL`` for consolidatable fact types), have
+  auto-consolidation not explicitly disabled at the bank level, and have no
+  consolidation operation already pending/processing. Drives the periodic
+  reconcile that re-schedules consolidation after a terminal failure left facts
+  stranded (see HINDSIGHT_API_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS).
+
+- ``public.schemas_with_expired_rows(p_table, p_ts_col, p_days)`` -> schema
+  names that hold at least one ``p_table`` row older than ``p_days``. Drives the
+  cross-tenant retention sweeps for ``audit_log`` and ``llm_requests``; the loop
+  then issues a DELETE only against the returned schemas.
+
+These are read-only (STABLE) discovery routines — the caller performs the
+enqueue/DELETE — so installing them never mutates data.
+
+PostgreSQL only — the worker poller and these tables are not wired for Oracle,
+so the Oracle slot is intentionally absent (mirrors the audit_log / llm_requests
+table migrations). The routines live in ``public`` and are CREATE OR REPLACE, so
+running this migration once per tenant schema is idempotent.
+
+Revision ID: e5f6a7b8c9d0
+Revises: a7b8c9d0e1f2
+Create Date: 2026-06-05
+"""
+
+from collections.abc import Sequence
+
+from alembic import context, op
+
+from hindsight_api.alembic._dialect import run_for_dialect
+
+revision: str = "e5f6a7b8c9d0"
+down_revision: str | Sequence[str] | None = "a7b8c9d0e1f2"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _is_base_schema_run() -> bool:
+    """True only for the base-schema migration (no per-tenant target_schema).
+
+    These routines live in the shared ``public`` schema, so they must be created
+    exactly once. Running ``CREATE OR REPLACE FUNCTION public....`` again from each
+    concurrent per-tenant migration aborts with ``tuple concurrently updated`` on
+    the ``pg_proc`` catalog row, so tenant runs skip it (the base run already
+    created the function for every tenant to use).
+    """
+    return not context.config.get_main_option("target_schema")
+
+
+def _pg_upgrade() -> None:
+    if not _is_base_schema_run():
+        return
+    # Banks with eligible-but-unscheduled facts and no in-flight consolidation.
+    # Auto-consolidation is filtered here only at the bank level (cheap prune);
+    # the full hierarchical resolution (global -> tenant -> bank, plus
+    # enable_observations) is done by the caller for the small returned set.
+    op.execute(
+        """
+        CREATE OR REPLACE FUNCTION public.banks_needing_consolidation()
+        RETURNS TABLE(schema_name text, bank_id text)
+        LANGUAGE plpgsql STABLE
+        AS $fn$
+        DECLARE
+            sch text;
+        BEGIN
+            FOR sch IN
+                SELECT n.nspname
+                FROM pg_class c
+                JOIN pg_namespace n ON n.oid = c.relnamespace
+                WHERE c.relname = 'memory_units' AND c.relkind = 'r'
+            LOOP
+                RETURN QUERY EXECUTE format($q$
+                    SELECT %1$L::text, m.bank_id
+                    FROM %1$I.memory_units m
+                    JOIN %1$I.banks b ON b.bank_id = m.bank_id
+                    WHERE m.consolidated_at IS NULL
+                      AND m.consolidation_failed_at IS NULL
+                      AND m.fact_type IN ('experience', 'world')
+                      AND COALESCE(b.config -> 'enable_auto_consolidation', 'true'::jsonb) <> 'false'::jsonb
+                      AND NOT EXISTS (
+                          SELECT 1 FROM %1$I.async_operations o
+                          WHERE o.bank_id = m.bank_id
+                            AND o.operation_type = 'consolidation'
+                            AND o.status IN ('pending', 'processing')
+                      )
+                    GROUP BY m.bank_id
+                $q$, sch);
+            END LOOP;
+        END;
+        $fn$;
+        """
+    )
+
+    # Schemas holding at least one row of p_table older than p_days. p_ts_col is
+    # the timestamp column to compare. Returns nothing when p_days <= 0
+    # (retention disabled).
+    op.execute(
+        """
+        CREATE OR REPLACE FUNCTION public.schemas_with_expired_rows(
+            p_table text, p_ts_col text, p_days int
+        )
+        RETURNS SETOF text
+        LANGUAGE plpgsql STABLE
+        AS $fn$
+        DECLARE
+            sch text;
+            has_expired boolean;
+        BEGIN
+            IF p_days IS NULL OR p_days <= 0 THEN
+                RETURN;
+            END IF;
+            FOR sch IN
+                SELECT n.nspname
+                FROM pg_class c
+                JOIN pg_namespace n ON n.oid = c.relnamespace
+                WHERE c.relname = p_table AND c.relkind = 'r'
+            LOOP
+                EXECUTE format(
+                    'SELECT EXISTS (SELECT 1 FROM %I.%I WHERE %I < NOW() - make_interval(days => $1))',
+                    sch, p_table, p_ts_col
+                ) INTO has_expired USING p_days;
+                IF has_expired THEN
+                    RETURN NEXT sch;
+                END IF;
+            END LOOP;
+        END;
+        $fn$;
+        """
+    )
+
+
+def _pg_downgrade() -> None:
+    if not _is_base_schema_run():
+        return
+    op.execute("DROP FUNCTION IF EXISTS public.banks_needing_consolidation()")
+    op.execute("DROP FUNCTION IF EXISTS public.schemas_with_expired_rows(text, text, int)")
+
+
+def upgrade() -> None:
+    run_for_dialect(pg=_pg_upgrade)
+
+
+def downgrade() -> None:
+    run_for_dialect(pg=_pg_downgrade)
@@ -6,6 +6,8 @@ the FastAPI application with all API endpoints.
 """

 import asyncio
+import dataclasses
+import hmac
 import json
 import logging
 import re
@@ -79,7 +81,7 @@ def FieldWithDefault(default_factory: Callable, **kwargs) -> Any:
    return Field(default_factory=default_factory, json_schema_extra=json_extra, **kwargs)


-from hindsight_api.config import get_config
+from hindsight_api.config import _get_raw_config, get_config
 from hindsight_api.engine.memory_engine import Budget, _current_schema, _get_tiktoken_encoding, fq_table
 from hindsight_api.engine.providers.none_llm import LLMNotAvailableError
 from hindsight_api.engine.response_models import VALID_RECALL_FACT_TYPES, MemoryFact, TokenUsage
@@ -1206,6 +1208,33 @@ class BankConfigResponse(BaseModel):
    overrides: dict[str, Any] = Field(description="Bank-specific configuration overrides only (Python field names)")


+class AdminConfigResponse(BaseModel):
+    """Response model for the server-level (admin) configuration view.
+
+    Returns the resolved ``HindsightConfig`` as a flat dict keyed by Python field
+    name. Credential fields (API keys, tokens, service-account keys, base URLs) are
+    masked: present as ``"***"`` when set and ``None`` when unset, so an operator can
+    see which credentials are configured without ever seeing their values.
+    """
+
+    config: dict[str, Any] = Field(
+        description="Resolved server-level configuration (Python field names); credentials are redacted"
+    )
+
+
+# Name suffixes that mark a config field as secret-bearing. Used in addition to
+# HindsightConfig._CREDENTIAL_FIELDS so the admin config view never leaks a provider
+# credential even if the (denylist) credential set misses a field. Suffixes are
+# singular on purpose — "_token" must not also match value-bearing "_tokens" fields
+# like recall_max_tokens.
+_SENSITIVE_FIELD_SUFFIXES = ("_api_key", "_token", "_secret", "_access_key", "_account_key", "_password")
+
+
+def _is_sensitive_config_field(field_name: str, credential_fields: set[str]) -> bool:
+    """Whether a config field must be redacted in the admin view."""
+    return field_name in credential_fields or field_name.endswith(_SENSITIVE_FIELD_SUFFIXES)
+
+
 class GraphDataResponse(BaseModel):
    """Response model for graph data endpoint."""

@@ -2413,6 +2442,7 @@ class FeaturesInfo(BaseModel):
    mcp: bool = Field(description="Whether MCP (Model Context Protocol) server is enabled")
    worker: bool = Field(description="Whether the background worker is enabled")
    bank_config_api: bool = Field(description="Whether per-bank configuration API is enabled")
+    admin_api: bool = Field(description="Whether the admin API (/admin) is enabled")
    file_upload_api: bool = Field(description="Whether file upload/conversion API is enabled")
    document_export_api: bool = Field(description="Whether the document export endpoint is enabled")
    document_import_api: bool = Field(description="Whether the document import endpoint is enabled")
@@ -2971,6 +3001,31 @@ def _register_routes(app: FastAPI):
                api_key = authorization.strip()
        return RequestContext(api_key=api_key)

+    def require_admin(authorization: str | None = Header(default=None)) -> None:
+        """Guard for the admin surface.
+
+        - 404 when the admin API is disabled (so the surface is invisible by default).
+        - When ``HINDSIGHT_API_ADMIN_TOKEN`` is set, require it as a bearer token
+          (or a bare token) and reject with 401 otherwise. When unset, the admin API
+          is open (auth is optional) — consistent with the rest of the deployment.
+        """
+        config = _get_raw_config()
+        if not config.enable_admin_api:
+            raise HTTPException(
+                status_code=404,
+                detail="Admin API is disabled. Set HINDSIGHT_API_ENABLE_ADMIN_API=true to enable.",
+            )
+        expected = config.admin_api_token
+        if expected:
+            token = None
+            if authorization:
+                if authorization.lower().startswith("bearer "):
+                    token = authorization[7:].strip()
+                else:
+                    token = authorization.strip()
+            if not token or not hmac.compare_digest(token, expected):
+                raise HTTPException(status_code=401, detail="Invalid or missing admin token")
+
    def precheck_for(operation: str):
        """
        Build a FastAPI dependency that runs ``OperationValidator.precheck``.
@@ -3080,6 +3135,7 @@ def _register_routes(app: FastAPI):
                mcp=config.mcp_enabled,
                worker=config.worker_enabled,
                bank_config_api=config.enable_bank_config_api,
+                admin_api=config.enable_admin_api,
                file_upload_api=config.enable_file_upload_api,
                document_export_api=config.enable_document_export_api,
                document_import_api=config.enable_document_import_api,
@@ -3088,6 +3144,33 @@ def _register_routes(app: FastAPI):
            ),
        )

+    @app.get(
+        "/admin/config",
+        response_model=AdminConfigResponse,
+        summary="Get resolved server-level configuration",
+        description="Returns the resolved server-level configuration with credentials redacted. "
+        "Gated by HINDSIGHT_API_ENABLE_ADMIN_API and, when set, HINDSIGHT_API_ADMIN_TOKEN.",
+        tags=["Admin"],
+        operation_id="get_admin_config",
+        dependencies=[Depends(require_admin)],
+    )
+    async def admin_config_endpoint() -> AdminConfigResponse:
+        """Expose the resolved ``HindsightConfig`` for operator inspection.
+
+        Sensitive fields (the credential denylist plus any field whose name ends in a
+        secret-bearing suffix — see ``_is_sensitive_config_field``) are masked so values
+        never leave the server: ``"***"`` when set, ``None`` when unset. All other fields
+        are returned as-is.
+        """
+        config = _get_raw_config()
+        credential_fields = type(config).get_credential_fields()
+        raw = dataclasses.asdict(config)
+        redacted = {
+            key: ("***" if value is not None else None) if _is_sensitive_config_field(key, credential_fields) else value
+            for key, value in raw.items()
+        }
+        return AdminConfigResponse(config=redacted)
+
    @app.get(
        "/metrics",
        summary="Prometheus metrics endpoint",
@@ -309,6 +309,7 @@ ENV_RERANKER_LITELLM_TIMEOUT = "HINDSIGHT_API_RERANKER_LITELLM_TIMEOUT"
 ENV_RERANKER_LITELLM_SDK_TIMEOUT = "HINDSIGHT_API_RERANKER_LITELLM_SDK_TIMEOUT"
 ENV_RERANKER_GOOGLE_TIMEOUT = "HINDSIGHT_API_RERANKER_GOOGLE_TIMEOUT"
 ENV_RERANKER_MAX_CANDIDATES = "HINDSIGHT_API_RERANKER_MAX_CANDIDATES"
+ENV_SEMANTIC_MIN_SIMILARITY = "HINDSIGHT_API_SEMANTIC_MIN_SIMILARITY"
 ENV_RERANKER_FLASHRANK_MODEL = "HINDSIGHT_API_RERANKER_FLASHRANK_MODEL"
 ENV_RERANKER_FLASHRANK_CACHE_DIR = "HINDSIGHT_API_RERANKER_FLASHRANK_CACHE_DIR"
 ENV_RERANKER_FLASHRANK_CPU_MEM_ARENA = "HINDSIGHT_API_RERANKER_FLASHRANK_CPU_MEM_ARENA"
@@ -350,6 +351,8 @@ ENV_MCP_ENABLED = "HINDSIGHT_API_MCP_ENABLED"
 ENV_MCP_ENABLED_TOOLS = "HINDSIGHT_API_MCP_ENABLED_TOOLS"
 ENV_MCP_STATELESS = "HINDSIGHT_API_MCP_STATELESS"
 ENV_ENABLE_BANK_CONFIG_API = "HINDSIGHT_API_ENABLE_BANK_CONFIG_API"
+ENV_ENABLE_ADMIN_API = "HINDSIGHT_API_ENABLE_ADMIN_API"
+ENV_ADMIN_API_TOKEN = "HINDSIGHT_API_ADMIN_TOKEN"
 ENV_DEFAULT_BANK_TEMPLATE = "HINDSIGHT_API_DEFAULT_BANK_TEMPLATE"
 ENV_GRAPH_RETRIEVER = "HINDSIGHT_API_GRAPH_RETRIEVER"
 ENV_RECALL_MAX_CONCURRENT = "HINDSIGHT_API_RECALL_MAX_CONCURRENT"
@@ -437,6 +440,7 @@ ENV_CONSOLIDATION_LLM_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_LLM_BATCH_SIZE"
 ENV_CONSOLIDATION_DEDUP_THRESHOLD = "HINDSIGHT_API_CONSOLIDATION_DEDUP_THRESHOLD"
 ENV_CONSOLIDATION_LLM_PARALLELISM = "HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM"
 ENV_CONSOLIDATION_MAX_TOKENS = "HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS"
+ENV_CONSOLIDATION_MAX_COMPLETION_TOKENS = "HINDSIGHT_API_CONSOLIDATION_MAX_COMPLETION_TOKENS"
 ENV_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS = "HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS"
 ENV_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS_PER_OBSERVATION = (
    "HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS_PER_OBSERVATION"
@@ -552,6 +556,9 @@ ENV_LLM_TRACE_SCOPES = "HINDSIGHT_API_LLM_TRACE_SCOPES"
 ENV_LLM_TRACE_RETENTION_DAYS = "HINDSIGHT_API_LLM_TRACE_RETENTION_DAYS"
 ENV_LLM_TRACE_MAX_CHARS = "HINDSIGHT_API_LLM_TRACE_MAX_CHARS"

+# Background maintenance settings
+ENV_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS = "HINDSIGHT_API_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS"
+
 # Disposition settings
 ENV_DISPOSITION_SKEPTICISM = "HINDSIGHT_API_DISPOSITION_SKEPTICISM"
 ENV_DISPOSITION_LITERALISM = "HINDSIGHT_API_DISPOSITION_LITERALISM"
@@ -661,6 +668,7 @@ DEFAULT_RERANKER_LITELLM_TIMEOUT = 60.0
 DEFAULT_RERANKER_LITELLM_SDK_TIMEOUT = 60.0
 DEFAULT_RERANKER_GOOGLE_TIMEOUT = 60.0
 DEFAULT_RERANKER_MAX_CANDIDATES = 300
+DEFAULT_SEMANTIC_MIN_SIMILARITY = 0.3
 # Minimum BM25 score a row must exceed to enter fusion. 0.0 gates out
 # zero-score (non-matching) rows on backends — notably VectorChord — whose
 # operator ranks every document rather than pre-filtering to term matches.
@@ -786,6 +794,8 @@ DEFAULT_MCP_ENABLED = True
 DEFAULT_MCP_ENABLED_TOOLS: list[str] | None = None  # None = all tools enabled
 DEFAULT_MCP_STATELESS = False  # False = stateful (supports SSE/GET); True = stateless (POST-only)
 DEFAULT_ENABLE_BANK_CONFIG_API = True
+DEFAULT_ENABLE_ADMIN_API = False  # Admin surface (server config view) is off unless explicitly enabled
+DEFAULT_ADMIN_API_TOKEN: str | None = None  # None = admin API open (when enabled); set = required bearer token
 DEFAULT_DEFAULT_BANK_TEMPLATE: dict | None = None  # BankTemplateManifest dict applied to newly-created banks
 DEFAULT_GRAPH_RETRIEVER = "link_expansion"
 DEFAULT_RECALL_MAX_CONCURRENT = 32  # Max concurrent recall operations per worker
@@ -861,6 +871,10 @@ DEFAULT_CONSOLIDATION_LLM_PARALLELISM = (
    # scopes degrade to sequential automatically; matches retain_max_concurrent.
 )
 DEFAULT_CONSOLIDATION_MAX_TOKENS = 512  # Max tokens for recall when finding related observations
+# Unset by default: the key is omitted from the LLM call so every provider keeps its current implicit output
+# budget — 100% backwards compatible. Operators on providers with a low hidden default (notably Bedrock imported
+# models, which cap at 4096 and truncate structured consolidation JSON) set this explicitly to fix #1939.
+DEFAULT_CONSOLIDATION_MAX_COMPLETION_TOKENS = None
 DEFAULT_CONSOLIDATION_RECALL_BUDGET = "low"  # Budget level for consolidation recall (low/mid/high)
 DEFAULT_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS = (
    4096  # Total token budget for source facts in consolidation recall (-1 = unlimited)
@@ -939,6 +953,12 @@ DEFAULT_LLM_TRACE_SCOPES = ""  # Empty = trace all call scopes
 DEFAULT_LLM_TRACE_RETENTION_DAYS = 1  # Retain trace rows for 1 day by default
 DEFAULT_LLM_TRACE_MAX_CHARS = 50000  # Truncate stored input/output beyond this many chars

+# Background maintenance defaults
+# Periodic reconcile that re-schedules consolidation for banks with eligible-but-unscheduled
+# facts (e.g. after a consolidation operation failed terminally and left them unscheduled).
+# 0 disables the reconcile sweep.
+DEFAULT_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS = 300
+
 # Default MCP tool descriptions (can be customized via env vars)
 DEFAULT_MCP_RETAIN_DESCRIPTION = """Store important information to long-term memory.

@@ -1324,6 +1344,7 @@ class HindsightConfig:
    reranker_tei_max_concurrent: int
    reranker_tei_http_timeout: float
    reranker_max_candidates: int
+    semantic_min_similarity: float
    bm25_min_score: float
    recall_max_candidates_per_source: int
    recall_strategy_boosts: dict[str, str]
@@ -1370,6 +1391,10 @@ class HindsightConfig:
    mcp_enabled_tools: list[str] | None  # None = all tools; explicit list = allowlist
    mcp_stateless: bool  # True = stateless HTTP (POST-only); False = stateful (supports GET/SSE)
    enable_bank_config_api: bool
+    # Admin surface (static, server-level only). enable_admin_api gates the /admin API +
+    # control-plane page; admin_api_token (when set) is the required bearer token.
+    enable_admin_api: bool
+    admin_api_token: str | None
    # Default bank template (static, server-level only). When set, the manifest is applied
    # to every newly-created bank, overriding the env/config defaults for any fields it sets.
    default_bank_template: dict | None
@@ -1438,6 +1463,7 @@ class HindsightConfig:
    consolidation_llm_batch_size: int
    consolidation_llm_parallelism: int
    consolidation_max_tokens: int
+    consolidation_max_completion_tokens: int | None
    consolidation_recall_budget: str
    consolidation_source_facts_max_tokens: int
    consolidation_source_facts_max_tokens_per_observation: int
@@ -1531,6 +1557,11 @@ class HindsightConfig:
    llm_trace_retention_days: int  # -1 = keep forever, >0 = delete after N days
    llm_trace_max_chars: int  # Truncate stored input/output beyond this many chars

+    # Background maintenance configuration (static - server-level only)
+    # Interval for the periodic sweep that re-schedules consolidation for banks with
+    # eligible-but-unscheduled facts. 0 = disabled.
+    consolidation_reconcile_interval_seconds: int
+
    # Webhook configuration (static - server-level only, not per-bank)
    webhook_url: str | None  # Global webhook URL (None = disabled)
    webhook_secret: str | None  # HMAC signing secret (None = unsigned)
@@ -1589,6 +1620,8 @@ class HindsightConfig:
        # File parser credentials
        "file_parser_iris_token",
        "file_parser_llama_parse_api_key",
+        # Admin surface token (never exposed via the admin config view itself)
+        "admin_api_token",
    }

    # CONFIGURABLE_FIELDS: Safe behavioral settings that can be customized per-tenant/bank
@@ -1731,6 +1764,11 @@ class HindsightConfig:
            self.text_search_extension_pg_search_tokenizer
        )

+        if not 0.0 <= self.semantic_min_similarity <= 1.0:
+            raise ValueError(
+                f"Invalid semantic_min_similarity: {self.semantic_min_similarity}. Must be between 0.0 and 1.0"
+            )
+
        # When LLM provider is "none", force chunks-only mode and disable LLM-dependent features
        if self.llm_provider == "none":
            self.retain_extraction_mode = "chunks"
@@ -2108,6 +2146,7 @@ class HindsightConfig:
                os.getenv(ENV_RERANKER_TEI_HTTP_TIMEOUT, str(DEFAULT_RERANKER_TEI_HTTP_TIMEOUT))
            ),
            reranker_max_candidates=int(os.getenv(ENV_RERANKER_MAX_CANDIDATES, str(DEFAULT_RERANKER_MAX_CANDIDATES))),
+            semantic_min_similarity=float(os.getenv(ENV_SEMANTIC_MIN_SIMILARITY, str(DEFAULT_SEMANTIC_MIN_SIMILARITY))),
            bm25_min_score=float(os.getenv(ENV_BM25_MIN_SCORE, str(DEFAULT_BM25_MIN_SCORE))),
            recall_max_candidates_per_source=int(
                os.getenv(ENV_RECALL_MAX_CANDIDATES_PER_SOURCE, str(DEFAULT_RECALL_MAX_CANDIDATES_PER_SOURCE))
@@ -2189,6 +2228,8 @@ class HindsightConfig:
            mcp_stateless=os.getenv(ENV_MCP_STATELESS, str(DEFAULT_MCP_STATELESS)).lower() == "true",
            enable_bank_config_api=os.getenv(ENV_ENABLE_BANK_CONFIG_API, str(DEFAULT_ENABLE_BANK_CONFIG_API)).lower()
            == "true",
+            enable_admin_api=os.getenv(ENV_ENABLE_ADMIN_API, str(DEFAULT_ENABLE_ADMIN_API)).lower() == "true",
+            admin_api_token=os.getenv(ENV_ADMIN_API_TOKEN) or DEFAULT_ADMIN_API_TOKEN,
            default_bank_template=_parse_default_bank_template(os.getenv(ENV_DEFAULT_BANK_TEMPLATE)),
            # Recall
            graph_retriever=os.getenv(ENV_GRAPH_RETRIEVER, DEFAULT_GRAPH_RETRIEVER),
@@ -2334,6 +2375,11 @@ class HindsightConfig:
            consolidation_max_tokens=int(
                os.getenv(ENV_CONSOLIDATION_MAX_TOKENS, str(DEFAULT_CONSOLIDATION_MAX_TOKENS))
            ),
+            consolidation_max_completion_tokens=(
+                int(os.getenv(ENV_CONSOLIDATION_MAX_COMPLETION_TOKENS))
+                if os.getenv(ENV_CONSOLIDATION_MAX_COMPLETION_TOKENS)
+                else DEFAULT_CONSOLIDATION_MAX_COMPLETION_TOKENS
+            ),
            consolidation_recall_budget=os.getenv(ENV_CONSOLIDATION_RECALL_BUDGET, DEFAULT_CONSOLIDATION_RECALL_BUDGET),
            consolidation_source_facts_max_tokens=int(
                os.getenv(ENV_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS, str(DEFAULT_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS))
@@ -2455,6 +2501,13 @@ class HindsightConfig:
                os.getenv(ENV_LLM_TRACE_RETENTION_DAYS, str(DEFAULT_LLM_TRACE_RETENTION_DAYS))
            ),
            llm_trace_max_chars=int(os.getenv(ENV_LLM_TRACE_MAX_CHARS, str(DEFAULT_LLM_TRACE_MAX_CHARS))),
+            # Background maintenance configuration (static, server-level only)
+            consolidation_reconcile_interval_seconds=int(
+                os.getenv(
+                    ENV_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS,
+                    str(DEFAULT_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS),
+                )
+            ),
            # Webhook configuration (static, server-level only)
            webhook_url=os.getenv(ENV_WEBHOOK_URL) or DEFAULT_WEBHOOK_URL,
            webhook_secret=os.getenv(ENV_WEBHOOK_SECRET) or DEFAULT_WEBHOOK_SECRET,
@@ -107,11 +107,11 @@ def _safe_json(data: Any) -> str | None:
        return None


-_SWEEP_INTERVAL_SECONDS = 3600  # Run retention sweep every hour
-
-
 class AuditLogger:
-    """Fire-and-forget audit log writer with optional retention sweep."""
+    """Fire-and-forget audit log writer.
+
+    Retention of old rows is handled by the background :class:`MaintenanceLoop`.
+    """

    def __init__(
        self,
@@ -119,14 +119,11 @@ class AuditLogger:
        schema_getter: Callable[[], str],
        enabled: bool,
        allowed_actions: list[str],
-        retention_days: int = -1,
    ) -> None:
        self._pool_getter = pool_getter
        self._schema_getter = schema_getter
        self._enabled = enabled
        self._allowed_actions: frozenset[str] | None = frozenset(allowed_actions) if allowed_actions else None
-        self._retention_days = retention_days
-        self._sweep_task: asyncio.Task | None = None

    def is_enabled(self, action: str) -> bool:
        """Check if audit logging is enabled for this action."""
@@ -176,48 +173,6 @@ class AuditLogger:
        except Exception as e:
            logger.warning(f"Audit log write failed for action={entry.action}: {e}")

-    def start_retention_sweep(self) -> None:
-        """Start the periodic retention sweep if retention is configured."""
-        if self._retention_days <= 0 or not self._enabled:
-            return
-        try:
-            self._sweep_task = asyncio.create_task(self._sweep_loop())
-        except RuntimeError:
-            logger.debug("Cannot start retention sweep: no running event loop")
-
-    async def stop_retention_sweep(self) -> None:
-        """Stop the periodic retention sweep."""
-        if self._sweep_task and not self._sweep_task.done():
-            self._sweep_task.cancel()
-            try:
-                await self._sweep_task
-            except asyncio.CancelledError:
-                pass
-            self._sweep_task = None
-
-    async def _sweep_loop(self) -> None:
-        """Periodically delete audit log entries older than retention_days."""
-        while True:
-            await self._run_sweep()
-            await asyncio.sleep(_SWEEP_INTERVAL_SECONDS)
-
-    async def _run_sweep(self) -> None:
-        """Delete expired audit log entries. Concurrent-safe via row-level deletes."""
-        pool = self._pool_getter()
-        if pool is None:
-            return
-        try:
-            schema = self._schema_getter()
-            table = f"{schema}.audit_log"
-            async with acquire_with_retry(pool, max_retries=1) as conn:
-                result = await conn.execute(
-                    f"DELETE FROM {table} WHERE started_at < NOW() - INTERVAL '{self._retention_days} days'"
-                )
-                if result and result != "DELETE 0":
-                    logger.info(f"Audit log retention sweep: {result}")
-        except Exception as e:
-            logger.warning(f"Audit log retention sweep failed: {e}")
-

@asynccontextmanager
 async def audit_context(
@@ -2116,6 +2116,12 @@ async def _consolidate_batch_with_llm(
                "response_format": response_model,
                "scope": "consolidation",
            }
+            # Only request an explicit output budget when configured. Left unset by default the key is
+            # omitted, so each provider keeps its implicit default (backwards compatible). Operators on
+            # providers with a low hidden cap (notably Bedrock imported models, which truncate structured
+            # consolidation JSON) set HINDSIGHT_API_CONSOLIDATION_MAX_COMPLETION_TOKENS to fix it.
+            if config.consolidation_max_completion_tokens is not None:
+                call_kwargs["max_completion_tokens"] = config.consolidation_max_completion_tokens
            if inner_max_retries is not None:
                call_kwargs["max_retries"] = inner_max_retries
            if cached_prefix_name is not None:
@@ -46,7 +46,6 @@ from ..config import (
    ENV_RERANKER_FLASHRANK_CPU_MEM_ARENA,
    ENV_RERANKER_FLASHRANK_MODEL,
    ENV_RERANKER_GOOGLE_PROJECT_ID,
-    ENV_RERANKER_LITELLM_SDK_API_KEY,
    ENV_RERANKER_LOCAL_FORCE_CPU,
    ENV_RERANKER_LOCAL_MAX_CONCURRENT,
    ENV_RERANKER_LOCAL_MODEL,
@@ -1199,7 +1198,7 @@ class LiteLLMSDKCrossEncoder(CrossEncoderModel):

    def __init__(
        self,
-        api_key: str,
+        api_key: str | None = None,
        model: str = DEFAULT_RERANKER_LITELLM_SDK_MODEL,
        api_base: str | None = None,
        timeout: float = 60.0,
@@ -1209,7 +1208,8 @@ class LiteLLMSDKCrossEncoder(CrossEncoderModel):
        Initialize LiteLLM SDK cross-encoder client.

        Args:
-            api_key: API key for the reranking provider
+            api_key: API key for the reranking provider (optional — omit for
+                     providers that use ambient credentials, e.g. AWS Bedrock with IAM)
            model: Model name with provider prefix (e.g., "deepinfra/Qwen3-reranker-8B")
            api_base: Custom base URL for API (optional)
            timeout: Request timeout in seconds (default: 60.0)
@@ -1284,8 +1284,9 @@ class LiteLLMSDKCrossEncoder(CrossEncoderModel):
                "model": self.model,
                "query": query,
                "documents": texts,
-                "api_key": self.api_key,
            }
+            if self.api_key:
+                rerank_kwargs["api_key"] = self.api_key
            if self.api_base:
                rerank_kwargs["api_base"] = self.api_base

@@ -1697,13 +1698,8 @@ def create_cross_encoder_from_env() -> CrossEncoderModel:
            timeout=config.reranker_litellm_timeout,
        )
    elif provider == "litellm-sdk":
-        api_key = config.reranker_litellm_sdk_api_key
-        if not api_key:
-            raise ValueError(
-                f"{ENV_RERANKER_LITELLM_SDK_API_KEY} is required when {ENV_RERANKER_PROVIDER} is 'litellm-sdk'"
-            )
        return LiteLLMSDKCrossEncoder(
-            api_key=api_key,
+            api_key=config.reranker_litellm_sdk_api_key or None,
            model=config.reranker_litellm_sdk_model,
            api_base=config.reranker_litellm_sdk_api_base,
            max_tokens_per_doc=config.reranker_litellm_max_tokens_per_doc,
@@ -35,8 +35,6 @@ from .db_utils import acquire_with_retry

 logger = logging.getLogger(__name__)

-_SWEEP_INTERVAL_SECONDS = 3600  # Run retention sweep every hour
-

 # ── bank/operation attribution (carried across the async call chain) ──────────

@@ -310,8 +308,8 @@ class LLMTraceRecorder:

    Implements ``record_llm_call`` so it can be registered with
    :func:`hindsight_api.tracing.register_span_recorder`. Writes are
-    fire-and-forget and never surface errors into the calling path; an optional
-    retention sweep deletes rows older than ``retention_days``.
+    fire-and-forget and never surface errors into the calling path. Retention of
+    old rows is handled by the background :class:`MaintenanceLoop`.
    """

    def __init__(
@@ -320,16 +318,13 @@ class LLMTraceRecorder:
        schema_getter: Callable[[], str],
        enabled: bool,
        allowed_scopes: list[str],
-        retention_days: int = -1,
        max_chars: int = 50000,
    ) -> None:
        self._pool_getter = pool_getter
        self._schema_getter = schema_getter
        self._enabled = enabled
        self._allowed_scopes: frozenset[str] | None = frozenset(allowed_scopes) if allowed_scopes else None
-        self._retention_days = retention_days
        self._max_chars = max_chars
-        self._sweep_task: asyncio.Task | None = None
        # In-flight fire-and-forget write tasks, bucketed by trace_id so
        # attach_memory_ids can await only *its own* operation's writes before the
        # post-operation UPDATE (otherwise the UPDATE could race ahead of the
@@ -543,46 +538,3 @@ class LLMTraceRecorder:
                )
        except Exception as e:
            logger.warning(f"LLM trace memory_id attach failed for trace={trace_id}: {e}")
-
-    # ── retention sweep ───────────────────────────────────────────────────────
-
-    def start_retention_sweep(self) -> None:
-        """Start the periodic retention sweep if retention is configured."""
-        if self._retention_days <= 0 or not self._enabled:
-            return
-        try:
-            self._sweep_task = asyncio.create_task(self._sweep_loop())
-        except RuntimeError:
-            logger.debug("Cannot start llm trace retention sweep: no running event loop")
-
-    async def stop_retention_sweep(self) -> None:
-        """Stop the periodic retention sweep."""
-        if self._sweep_task and not self._sweep_task.done():
-            self._sweep_task.cancel()
-            try:
-                await self._sweep_task
-            except asyncio.CancelledError:
-                pass
-            self._sweep_task = None
-
-    async def _sweep_loop(self) -> None:
-        while True:
-            await self._run_sweep()
-            await asyncio.sleep(_SWEEP_INTERVAL_SECONDS)
-
-    async def _run_sweep(self) -> None:
-        """Delete trace rows older than retention_days. Concurrent-safe."""
-        pool = self._pool_getter()
-        if pool is None:
-            return
-        try:
-            schema = self._schema_getter()
-            table = f"{schema}.llm_requests"
-            async with acquire_with_retry(pool, max_retries=1) as conn:
-                result = await conn.execute(
-                    f"DELETE FROM {table} WHERE started_at < NOW() - INTERVAL '{self._retention_days} days'"
-                )
-                if result and result != "DELETE 0":
-                    logger.info(f"LLM trace retention sweep: {result}")
-        except Exception as e:
-            logger.warning(f"LLM trace retention sweep failed: {e}")
@@ -0,0 +1,214 @@
+"""Background maintenance loop.
+
+A single periodic loop that drives all of Hindsight's recurring housekeeping
+from one place, so we don't spawn a separate ``asyncio`` task per concern:
+
+- **Retention sweeps** (hourly): delete ``audit_log`` and ``llm_requests`` rows
+  older than their configured retention, across *all* tenant schemas.
+- **Consolidation reconcile** (configurable, default 5 min): re-schedule
+  consolidation for banks that have eligible-but-unscheduled facts and no
+  in-flight consolidation. This recovers facts that were stranded when a
+  consolidation operation failed terminally and left them with
+  ``consolidated_at IS NULL AND consolidation_failed_at IS NULL`` and nothing to
+  re-trigger them.
+
+The loop wakes on a short fixed tick and runs each job when its own
+``last_run + interval`` is due (run-at-start, then on interval), so adding jobs
+with different cadences doesn't burst CPU. Cross-tenant discovery goes through
+server-side PL/pgSQL routines (``public.schemas_with_expired_rows`` and
+``public.banks_needing_consolidation``) — one round-trip each — instead of a
+per-schema query storm, which matters at thousands of tenants.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import TYPE_CHECKING
+
+from ..config import HindsightConfig, get_config
+from ..models import RequestContext
+from .db_utils import acquire_with_retry
+from .schema import _is_oracle
+
+if TYPE_CHECKING:
+    from .memory_engine import MemoryEngine
+
+logger = logging.getLogger(__name__)
+
+# Short tick so jobs with different cadences share one loop without per-job tasks.
+_TICK_SECONDS = 60
+# Retention sweeps are not time-sensitive; hourly matches the previous per-sweep cadence.
+_RETENTION_INTERVAL_SECONDS = 3600
+
+
+class MaintenanceLoop:
+    """Owns the single periodic maintenance task for a :class:`MemoryEngine`."""
+
+    def __init__(self, engine: "MemoryEngine") -> None:
+        self._engine = engine
+        self._task: asyncio.Task | None = None
+        self._stop = asyncio.Event()
+        # Monotonic timestamps of the last run per job, keyed by job name.
+        self._last_run: dict[str, float] = {}
+
+    # ── lifecycle ──────────────────────────────────────────────────────────
+
+    def start(self) -> None:
+        """Start the loop if any maintenance job is enabled. Idempotent."""
+        if self._task and not self._task.done():
+            return
+        # PostgreSQL-only: the retention sweeps target PG-only tables (audit_log,
+        # llm_requests) and the reconcile relies on PG-only PL/pgSQL routines
+        # installed by the maintenance-routines migration. Oracle support is
+        # intentionally absent (mirrors that PG-only migration).
+        if _is_oracle():
+            logger.debug("Maintenance loop not started: PostgreSQL-only")
+            return
+        if not self._any_job_enabled():
+            logger.debug("Maintenance loop not started: no jobs enabled")
+            return
+        self._stop.clear()
+        try:
+            self._task = asyncio.create_task(self._run())
+        except RuntimeError:
+            logger.debug("Cannot start maintenance loop: no running event loop")
+
+    async def stop(self) -> None:
+        """Stop the loop and wait for the current tick to finish."""
+        self._stop.set()
+        if self._task and not self._task.done():
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+        self._task = None
+
+    @staticmethod
+    def _any_job_enabled() -> bool:
+        cfg = get_config()
+        reconcile_on = cfg.consolidation_reconcile_interval_seconds > 0
+        audit_on = cfg.audit_log_enabled and cfg.audit_log_retention_days > 0
+        llm_on = cfg.llm_trace_enabled and cfg.llm_trace_retention_days > 0
+        return reconcile_on or audit_on or llm_on
+
+    # ── loop ───────────────────────────────────────────────────────────────
+
+    async def _run(self) -> None:
+        while not self._stop.is_set():
+            try:
+                await self._tick()
+            except Exception:
+                logger.exception("Maintenance tick failed")
+            try:
+                await asyncio.wait_for(self._stop.wait(), timeout=_TICK_SECONDS)
+            except asyncio.TimeoutError:
+                pass
+
+    def _is_due(self, job: str, interval_seconds: int) -> bool:
+        """True if ``job`` has never run or its interval has elapsed; marks it run now."""
+        now = time.monotonic()
+        last = self._last_run.get(job)
+        if last is not None and (now - last) < interval_seconds:
+            return False
+        self._last_run[job] = now
+        return True
+
+    async def _tick(self) -> None:
+        cfg = get_config()
+        if self._is_due("retention", _RETENTION_INTERVAL_SECONDS):
+            await self._run_retention(cfg)
+        interval = cfg.consolidation_reconcile_interval_seconds
+        if interval > 0 and self._is_due("reconcile", interval):
+            await self._run_reconcile()
+
+    # ── retention ──────────────────────────────────────────────────────────
+
+    async def _run_retention(self, cfg: HindsightConfig) -> None:
+        # Retention days are static server-level config, so one global cutoff
+        # applies to every tenant schema (the routine sweeps them all).
+        if cfg.audit_log_enabled and cfg.audit_log_retention_days > 0:
+            await self._purge_expired("audit_log", "started_at", cfg.audit_log_retention_days)
+        if cfg.llm_trace_enabled and cfg.llm_trace_retention_days > 0:
+            await self._purge_expired("llm_requests", "started_at", cfg.llm_trace_retention_days)
+
+    async def _purge_expired(self, table: str, ts_col: str, days: int) -> None:
+        """Delete rows older than ``days`` from ``table`` across every tenant schema."""
+        backend = self._engine._backend
+        try:
+            async with acquire_with_retry(backend, max_retries=1) as conn:
+                rows = await conn.fetch(
+                    "SELECT * FROM public.schemas_with_expired_rows($1, $2, $3)", table, ts_col, days
+                )
+                for row in rows:
+                    schema = row[0]
+                    # schema names come from pg_class; quote defensively all the same.
+                    qschema = '"' + schema.replace('"', '""') + '"'
+                    result = await conn.execute(
+                        f"DELETE FROM {qschema}.{table} WHERE {ts_col} < NOW() - make_interval(days => $1)",
+                        days,
+                    )
+                    if result and result != "DELETE 0":
+                        logger.info(f"Retention sweep {schema}.{table}: {result}")
+        except Exception as e:
+            logger.warning(f"Retention sweep failed for {table}: {e}")
+
+    # ── consolidation reconcile ──────────────────────────────────────────────
+
+    async def _run_reconcile(self) -> None:
+        """Re-schedule consolidation for banks with eligible-but-unscheduled facts."""
+        engine = self._engine
+        try:
+            async with acquire_with_retry(engine._backend, max_retries=1) as conn:
+                rows = await conn.fetch("SELECT schema_name, bank_id FROM public.banks_needing_consolidation()")
+        except Exception as e:
+            logger.warning(f"Consolidation reconcile discovery failed: {e}")
+            return
+        if not rows:
+            return
+
+        # Only enqueue into schemas the worker actually polls (tenant discovery),
+        # otherwise the op would never be claimed and would block future reconciles
+        # for that bank. The tenant_id (when the extension provides one) lets
+        # config resolution honor tenant-level overrides.
+        try:
+            tenants = await engine._tenant_extension.list_tenants()
+        except Exception as e:
+            logger.warning(f"Consolidation reconcile tenant discovery failed: {e}")
+            return
+        tenant_by_schema = {t.schema: t for t in tenants}
+        default_schema = get_config().database_schema
+
+        from .memory_engine import _current_schema
+
+        submitted = 0
+        skipped_unknown = 0
+        for row in rows:
+            schema = row["schema_name"]
+            bank_id = row["bank_id"]
+            tenant = tenant_by_schema.get(schema)
+            if tenant is None and schema != default_schema:
+                skipped_unknown += 1
+                continue
+            tenant_id = tenant.tenant_id if tenant else None
+            token = _current_schema.set(schema)
+            try:
+                context = RequestContext(internal=True, tenant_id=tenant_id)
+                resolved = await engine._config_resolver.resolve_full_config(bank_id, context)
+                # Mirror the retain-time auto-consolidation gate (memory_engine): both
+                # observations and auto-consolidation must be enabled for this bank.
+                if not (resolved.enable_observations and resolved.enable_auto_consolidation):
+                    continue
+                await engine.submit_async_consolidation(bank_id=bank_id, request_context=context)
+                submitted += 1
+            except Exception as e:
+                logger.warning(f"Consolidation reconcile failed for bank {bank_id} in {schema}: {e}")
+            finally:
+                _current_schema.reset(token)
+
+        if submitted or skipped_unknown:
+            logger.info(
+                f"Consolidation reconcile: scheduled {submitted} bank(s)"
+                + (f", skipped {skipped_unknown} in unrecognized schema(s)" if skipped_unknown else "")
+            )
@@ -57,7 +57,10 @@ from .operation_metadata import (
    BatchRetainParentMetadata,
    ConsolidationMetadata,
    RefreshMentalModelMetadata,
+    RetainExtractionErrors,
    RetainMetadata,
+    RetainOutcomeAggregate,
+    RetainOutcomeMetadata,
 )
 from .sql import SQLDialect, create_sql_dialect

@@ -928,7 +931,6 @@ class MemoryEngine(MemoryEngineInterface):
            schema_getter=get_current_schema,
            enabled=config.audit_log_enabled,
            allowed_actions=config.audit_log_actions,
-            retention_days=config.audit_log_retention_days,
        )

        # Per-bank LLM request tracer (disabled by default). Registered as a
@@ -939,13 +941,18 @@ class MemoryEngine(MemoryEngineInterface):
            schema_getter=get_current_schema,
            enabled=config.llm_trace_enabled,
            allowed_scopes=config.llm_trace_scopes,
-            retention_days=config.llm_trace_retention_days,
            max_chars=config.llm_trace_max_chars,
        )
        from ..tracing import register_span_recorder

        register_span_recorder(self._llm_recorder)

+        # Background maintenance loop (retention sweeps + consolidation reconcile),
+        # created in initialize() once the pool/backend is ready.
+        from .maintenance import MaintenanceLoop
+
+        self._maintenance_loop: MaintenanceLoop | None = None
+
        # Backpressure mechanism: limit concurrent searches to prevent overwhelming the database
        # Configurable via HINDSIGHT_API_RECALL_MAX_CONCURRENT (default: 50)
        self._search_semaphore = asyncio.Semaphore(get_config().recall_max_concurrent)
@@ -2036,6 +2043,47 @@ class MemoryEngine(MemoryEngineInterface):
        except Exception as e:
            logger.error(f"Failed to mark operation as completed {operation_id}: {e}")

+    async def _write_retain_outcome_metadata(self, operation_id: str | None, unit_ids: list[list[str]]) -> None:
+        """Persist completed retain outcome fields before the operation is marked completed."""
+        if not operation_id:
+            return
+
+        unit_ids_count = sum(len(group) for group in unit_ids)
+        try:
+            backend = await self._get_backend()
+            async with acquire_with_retry(backend) as conn:
+                row = await conn.fetchrow(
+                    f"SELECT result_metadata FROM {fq_table('async_operations')} WHERE operation_id = $1",
+                    uuid.UUID(operation_id),
+                )
+                if not row:
+                    return
+
+                metadata = conn.parse_json(row["result_metadata"]) or {}
+                extraction_errors = RetainExtractionErrors()
+                extraction_errors.merge_metadata(metadata)
+                outcome = RetainOutcomeMetadata(
+                    unit_ids_count=unit_ids_count,
+                    extraction_errors_count=extraction_errors.count,
+                    extraction_errors_sample=extraction_errors.sample,
+                )
+
+                await conn.execute(
+                    f"""
+                    UPDATE {fq_table("async_operations")}
+                    SET result_metadata = COALESCE(result_metadata, '{{}}'::jsonb) || $2::jsonb,
+                        updated_at = now()
+                    WHERE operation_id = $1
+                    """,
+                    uuid.UUID(operation_id),
+                    json.dumps(outcome.to_dict()),
+                )
+        except Exception as e:
+            # Best-effort, but log loudly: the whole point of this metadata is to
+            # give clients a reliable success/silent-failure signal, so a missing
+            # write silently regresses them to the ambiguous pre-fix behaviour.
+            logger.warning(f"Failed to write retain outcome metadata for {operation_id}: {e}")
+
    async def _mark_operation_completed_and_fire_webhook(
        self,
        operation_id: str,
@@ -2147,14 +2195,16 @@ class MemoryEngine(MemoryEngineInterface):

            # Get all sibling operations (including this one).
            # This query runs in the same transaction, so it sees the current
-            # child's updated status. Pull error_message too so a parent that
+            # child's updated status. Pull result_metadata for completed
+            # children so the parent exposes the same outcome counters as the
+            # individual retain operations. Pull error_message too so a parent that
            # fails can inherit a representative child reason -- otherwise
            # downstream consumers (dashboards, alert filters) lose the actual
            # cause once a batch has children. See the worker poller's
            # _summarise_child_error_messages for the propagation rationale.
            siblings = await conn.fetch(
                f"""
-                SELECT status, error_message
+                SELECT status, error_message, result_metadata
                FROM {fq_table("async_operations")}
                WHERE bank_id = $1
                AND result_metadata::jsonb @> $2::jsonb
@@ -2196,14 +2246,22 @@ class MemoryEngine(MemoryEngineInterface):
                )
            elif all_completed:
                new_status = "completed"
+                outcome_aggregate = RetainOutcomeAggregate()
+                for sibling in siblings:
+                    sibling_metadata = conn.parse_json(sibling["result_metadata"]) or {}
+                    outcome_aggregate.add_metadata(sibling_metadata)
                await conn.execute(
                    f"""
                    UPDATE {fq_table("async_operations")}
-                    SET status = $2, updated_at = NOW(), completed_at = NOW()
+                    SET status = $2,
+                        result_metadata = COALESCE(result_metadata, '{{}}'::jsonb) || $3::jsonb,
+                        updated_at = NOW(),
+                        completed_at = NOW()
                    WHERE operation_id = $1
                    """,
                    uuid.UUID(parent_operation_id),
                    new_status,
+                    json.dumps(outcome_aggregate.to_outcome_metadata().to_dict()),
                )

            logger.info(f"Updated parent operation {parent_operation_id} to status '{new_status}' (all children done)")
@@ -2455,11 +2513,9 @@ class MemoryEngine(MemoryEngineInterface):
                await conn.execute('SET search_path TO "$user", public, bm25_catalog, tokenizer_catalog')

            # SET (not SET LOCAL) so per-backend ANN tuning persists for the
-            # connection lifetime. Each backend exposes its own GUC: pgvector
-            # uses hnsw.ef_search, vchord uses vchordrq.probes. The dispatcher
-            # returns the right one for the configured extension, tuned for
-            # the higher recall the per-fact_type semantic queries in
-            # retrieve_semantic_bm25_combined() need.
+            # connection lifetime. The dispatcher returns only safe, portable
+            # knobs for the configured extension; VectorChord probe tuning is
+            # index-shaped and should be stored on vchordrq indexes instead.
            for guc, value in ann_search_tuning_settings(configured_vector_extension(), kind="high_recall"):
                try:
                    await conn.execute(f"SET {guc} = {value}")
@@ -2585,11 +2641,13 @@ class MemoryEngine(MemoryEngineInterface):
        self._task_backend.set_executor(self.execute_task)
        await self._task_backend.initialize()

-        # Start audit log retention sweep (if configured)
-        self._audit_logger.start_retention_sweep()
+        # Start the background maintenance loop: cross-tenant retention sweeps
+        # (audit_log, llm_requests) plus the consolidation reconcile that
+        # re-schedules banks with eligible-but-unscheduled facts.
+        from .maintenance import MaintenanceLoop

-        # Start LLM trace retention sweep (if configured)
-        self._llm_recorder.start_retention_sweep()
+        self._maintenance_loop = MaintenanceLoop(self)
+        self._maintenance_loop.start()

        self._initialized = True
        logger.info("Memory system initialized (pool and task backend started)")
@@ -2654,11 +2712,11 @@ class MemoryEngine(MemoryEngineInterface):
        """Close the connection pool and shutdown background workers."""
        logger.info("close() started")

-        # Stop audit log retention sweep
-        await self._audit_logger.stop_retention_sweep()
+        # Stop the background maintenance loop (retention sweeps + reconcile)
+        if self._maintenance_loop is not None:
+            await self._maintenance_loop.stop()

-        # Stop LLM trace retention sweep and unregister the recorder
-        await self._llm_recorder.stop_retention_sweep()
+        # Unregister the LLM trace recorder span hook
        from ..tracing import unregister_span_recorder

        unregister_span_recorder(self._llm_recorder)
@@ -3121,6 +3179,8 @@ 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
@@ -5,8 +5,10 @@ These dataclasses define the structure of result_metadata for different operatio
 The metadata is exposed in the API for debugging purposes and may change without notice.
 """

-from dataclasses import asdict, dataclass
-from typing import Any
+from dataclasses import asdict, dataclass, field
+from typing import Any, Mapping
+
+MAX_EXTRACTION_ERROR_SAMPLES = 5


@dataclass
@@ -48,6 +50,79 @@ class RetainMetadata:
        return asdict(self)


+@dataclass
+class RetainExtractionErrors:
+    """Non-fatal fact extraction failures observed inside one retain operation."""
+
+    count: int = 0
+    sample: list[str] = field(default_factory=list)
+
+    def add(self, message: str) -> None:
+        """Record one extraction error while keeping the stored sample bounded."""
+        self.count += 1
+        if len(self.sample) < MAX_EXTRACTION_ERROR_SAMPLES:
+            self.sample.append(message[:500])
+
+    def merge_metadata(self, metadata: Mapping[str, Any]) -> None:
+        """Merge errors already present on an operation result_metadata object."""
+        self.count += int(metadata.get("extraction_errors_count") or 0)
+
+        sample = metadata.get("extraction_errors_sample") or []
+        if isinstance(sample, str):
+            sample = [sample]
+        if isinstance(sample, list):
+            for entry in sample:
+                if isinstance(entry, str) and len(self.sample) < MAX_EXTRACTION_ERROR_SAMPLES:
+                    self.sample.append(entry[:500])
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to the public result_metadata field shape."""
+        data: dict[str, Any] = {"extraction_errors_count": self.count}
+        if self.sample:
+            data["extraction_errors_sample"] = self.sample
+        return data
+
+
+@dataclass
+class RetainOutcomeMetadata:
+    """Machine-readable outcome metadata for a completed retain operation."""
+
+    unit_ids_count: int
+    extraction_errors_count: int = 0
+    extraction_errors_sample: list[str] = field(default_factory=list)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to dict for JSON serialization, omitting empty optional samples."""
+        data: dict[str, Any] = {
+            "unit_ids_count": self.unit_ids_count,
+            "extraction_errors_count": self.extraction_errors_count,
+        }
+        if self.extraction_errors_sample:
+            data["extraction_errors_sample"] = self.extraction_errors_sample[:MAX_EXTRACTION_ERROR_SAMPLES]
+        return data
+
+
+@dataclass
+class RetainOutcomeAggregate:
+    """Aggregate retain outcome metadata from child retain operations."""
+
+    unit_ids_count: int = 0
+    extraction_errors: RetainExtractionErrors = field(default_factory=RetainExtractionErrors)
+
+    def add_metadata(self, metadata: Mapping[str, Any]) -> None:
+        """Fold one child operation's result_metadata into the aggregate."""
+        self.unit_ids_count += int(metadata.get("unit_ids_count") or 0)
+        self.extraction_errors.merge_metadata(metadata)
+
+    def to_outcome_metadata(self) -> RetainOutcomeMetadata:
+        """Return the aggregate in the public result_metadata field shape."""
+        return RetainOutcomeMetadata(
+            unit_ids_count=self.unit_ids_count,
+            extraction_errors_count=self.extraction_errors.count,
+            extraction_errors_sample=self.extraction_errors.sample,
+        )
+
+
@dataclass
 class ConsolidationMetadata:
    """Metadata for consolidation operations."""
@@ -16,6 +16,7 @@ from pydantic import BaseModel, ConfigDict, Field, create_model, field_validator

 from ...config import get_config
 from ..llm_wrapper import LLMConfig, OutputTooLongError, sanitize_llm_output
+from ..operation_metadata import RetainExtractionErrors
 from ..response_models import TokenUsage
 from .entity_labels import (
    EntityLabelsConfig,
@@ -510,11 +511,11 @@ LANGUAGE: MANDATORY — Detect the language of the input text and produce ALL ou
 FACT FORMAT - BE CONCISE
 ══════════════════════════════════════════════════════════════════════════

-1. **what**: Core fact - concise but complete (1-2 sentences max)
-2. **when**: Temporal info if mentioned. "N/A" if none. Use day name when known.
-3. **where**: Location if relevant. "N/A" if none.
-4. **who**: People involved with relationships. "N/A" if just general info.
-5. **why**: Context/significance ONLY if important. "N/A" if obvious.
+1. "what": Core fact - concise but complete (1-2 sentences max)
+2. "when": Temporal info if mentioned. "N/A" if none. Use day name when known.
+3. "where": Location if relevant. "N/A" if none.
+4. "who": People involved with relationships. "N/A" if just general info.
+5. "why": Context/significance ONLY if important. "N/A" if obvious.

 CONCISENESS: Capture the essence, not every word. One good sentence beats three mediocre ones.

@@ -1710,6 +1711,39 @@ logger = logging.getLogger(__name__)
 SECONDS_PER_FACT = 0.01


+async def _write_batch_extraction_errors(
+    pool: Any,
+    operation_id: str | None,
+    schema: str | None,
+    errors: RetainExtractionErrors,
+) -> None:
+    """Persist non-fatal Batch API extraction errors into operation result_metadata."""
+    if not pool or not operation_id or errors.count == 0:
+        return
+
+    from ..db_utils import acquire_with_retry
+    from ..task_backend import fq_table
+
+    # `errors` is the complete set for this extraction run, so overwrite the
+    # extraction_errors_* keys rather than folding in what's already stored. On
+    # batch crash recovery the resumed batch reprocesses every result and
+    # recomputes `errors` from scratch; reading + merging the prior run's
+    # counters here would double-count them. The SQL `||` merge still preserves
+    # unrelated keys (e.g. batch_id) already on result_metadata.
+    table = fq_table("async_operations", schema)
+    async with acquire_with_retry(pool) as conn:
+        await conn.execute(
+            f"""
+            UPDATE {table}
+            SET result_metadata = COALESCE(result_metadata, '{{}}'::jsonb) || $2::jsonb,
+                updated_at = now()
+            WHERE operation_id = $1
+            """,
+            operation_id,
+            json.dumps(errors.to_dict()),
+        )
+
+
 async def extract_facts_from_contents_batch_api(
    contents: list[RetainContent],
    llm_config,
@@ -1887,6 +1921,7 @@ async def extract_facts_from_contents_batch_api(
    all_facts_from_llm = []
    chunks_metadata = []
    total_usage = TokenUsage()
+    extraction_errors = RetainExtractionErrors()

    for chunk_idx, (chunk_content, content_index, chunk_index_in_content, event_date, context) in enumerate(
        all_chunks_info
@@ -1895,7 +1930,9 @@ async def extract_facts_from_contents_batch_api(
        result = results_by_id.get(custom_id)

        if not result:
-            logger.warning(f"Missing result for {custom_id}, skipping")
+            message = f"{custom_id}: missing batch result"
+            logger.warning(message)
+            extraction_errors.add(message)
            chunks_metadata.append(
                ChunkMetadata(
                    chunk_text=chunk_content, fact_count=0, content_index=content_index, chunk_index=chunk_idx
@@ -1905,7 +1942,9 @@ async def extract_facts_from_contents_batch_api(

        # Check for errors
        if result.get("error"):
-            logger.error(f"Error in {custom_id}: {result['error']}")
+            message = f"{custom_id}: {result['error']}"
+            logger.error(f"Error in {message}")
+            extraction_errors.add(message)
            chunks_metadata.append(
                ChunkMetadata(
                    chunk_text=chunk_content, fact_count=0, content_index=content_index, chunk_index=chunk_idx
@@ -1918,7 +1957,9 @@ async def extract_facts_from_contents_batch_api(
        choices = response_body.get("choices", [])

        if not choices:
-            logger.warning(f"No choices in response for {custom_id}")
+            message = f"{custom_id}: no choices in response"
+            logger.warning(message)
+            extraction_errors.add(message)
            chunks_metadata.append(
                ChunkMetadata(
                    chunk_text=chunk_content, fact_count=0, content_index=content_index, chunk_index=chunk_idx
@@ -1933,7 +1974,9 @@ async def extract_facts_from_contents_batch_api(
        try:
            extraction_response_json = json.loads(content_str)
        except json.JSONDecodeError as e:
-            logger.error(f"Failed to parse JSON for {custom_id}: {e}")
+            message = f"{custom_id}: failed to parse JSON: {e}"
+            logger.error(message)
+            extraction_errors.add(message)
            chunks_metadata.append(
                ChunkMetadata(
                    chunk_text=chunk_content, fact_count=0, content_index=content_index, chunk_index=chunk_idx
@@ -2106,7 +2149,9 @@ async def extract_facts_from_contents_batch_api(
                fact = Fact(fact=combined_text, fact_type=fact_type, **fact_data)
                chunk_facts.append(fact)
            except Exception as e:
-                logger.error(f"Failed to create Fact model for fact {i}: {e}")
+                message = f"{custom_id}: failed to create Fact model for fact {i}: {e}"
+                logger.error(message)
+                extraction_errors.add(message)
                continue

        all_facts_from_llm.extend(chunk_facts)
@@ -2171,6 +2216,8 @@ async def extract_facts_from_contents_batch_api(
    # Step 8: Auto-tag facts from label groups with tag=True
    _inject_label_tags(extracted_facts, config)

+    await _write_batch_extraction_errors(pool, operation_id, schema, extraction_errors)
+
    logger.info(f"Batch API extracted {len(extracted_facts)} facts from {len(all_chunks_info)} chunks")

    return extracted_facts, chunks_metadata, total_usage
@@ -574,12 +574,10 @@ async def compute_semantic_links_ann(
    # the transaction end handles both.
    rows: list = []
    async with conn.transaction():
-        # Transaction-local ANN tuning. Each supported backend exposes its own
-        # GUC (hnsw.ef_search on pgvector, vchordrq.probes on vchord); the
-        # dispatcher returns the right knob for the configured backend with a
-        # value tuned for top-50 semantic link creation (lower recall but much
-        # lower latency than the recall-side default). SET LOCAL auto-reverts
-        # at commit, so we don't pollute the pool for subsequent queries.
+        # Transaction-local ANN tuning. The dispatcher only returns GUCs that
+        # are safe to apply at session/transaction scope for the configured
+        # backend. VectorChord probe values are index-shaped, so vchordrq uses
+        # index storage fallback parameters instead of a blanket SET LOCAL.
        for guc, value in ann_search_tuning_settings(configured_vector_extension(), kind="low_latency"):
            await conn.execute(f"SET LOCAL {guc} = {value}")

@@ -636,7 +634,7 @@ async def compute_semantic_links_ann(
            logger.debug(f"[ANN] fact_type={fact_type}: {len(ft_rows)} rows in {time_mod.time() - t_query:.3f}s")
            rows.extend(ft_rows)
    # Transaction commits here. _ann_seeds is dropped (ON COMMIT DROP).
-    # hnsw.ef_search reverts (SET LOCAL).
+    # Transaction-local ANN tuning reverts (SET LOCAL).

    for row in rows:
        sim = float(min(1.0, max(0.0, row["similarity"])))
@@ -137,6 +137,7 @@ async def retrieve_semantic_bm25_combined(
    """
    result_dict: dict[str, tuple[list[RetrievalResult], list[RetrievalResult]]] = {ft: ([], []) for ft in fact_types}

+    config = get_config()
    tokens = tokenize_query(query_text)

    # Over-fetch for HNSW approximation; semantic results trimmed to limit in Python.
@@ -148,8 +149,6 @@ async def retrieve_semantic_bm25_combined(
    )
    table = fq_table("memory_units")

-    config = get_config()
-
    # Use the SQL dialect to build backend-specific query arms, avoiding
    # inline if/else branches for each database.
    # Use getattr for backward compat: raw asyncpg connections (used in some
@@ -201,6 +200,7 @@ async def retrieve_semantic_bm25_combined(
            embedding_param="$1",
            bank_id_param="$2",
            fetch_limit=hnsw_fetch,
+            min_similarity=config.semantic_min_similarity,
            tags_clause=tags_clause,
            groups_clause=groups_clause,
            extra_where=created_range_clause,
@@ -274,6 +274,7 @@ async def retrieve_semantic_bm25_combined(
                    embedding_param="$1",
                    bank_id_param="$2",
                    fetch_limit=hnsw_fetch,
+                    min_similarity=config.semantic_min_similarity,
                    tags_clause=fb_tags_clause,
                    groups_clause=fb_groups_clause,
                    extra_where=fb_created_clause,
@@ -358,12 +358,15 @@ class SearchTracer:
        """
        self.rrf_merged = []
        for rank, (doc_id, data, rrf_meta) in enumerate(merged_results, start=1):
+            source_ranks = rrf_meta.get("source_ranks")
+            if source_ranks is None:
+                source_ranks = {key: value for key, value in rrf_meta.items() if key.endswith("_rank")}
            self.rrf_merged.append(
                RRFMergeResult(
                    node_id=doc_id,
                    text=data.get("text", ""),
                    rrf_score=rrf_meta.get("rrf_score", 0.0),
-                    source_ranks=rrf_meta.get("source_ranks", {}),
+                    source_ranks=source_ranks,
                    final_rrf_rank=rank,
                )
            )
@@ -371,6 +371,7 @@ class SQLDialect(ABC):
        embedding_param: str,
        bank_id_param: str,
        fetch_limit: int,
+        min_similarity: float,
        tags_clause: str = "",
        groups_clause: str = "",
        extra_where: str = "",
@@ -387,6 +388,7 @@ class SQLDialect(ABC):
            embedding_param: Parameter placeholder for query embedding.
            bank_id_param: Parameter placeholder for bank_id.
            fetch_limit: Max rows to fetch (over-fetched for HNSW approximation).
+            min_similarity: Minimum cosine similarity to include.
            tags_clause: Optional WHERE clause fragment for tag filtering.
            groups_clause: Optional WHERE clause fragment for tag group filtering.
            extra_where: Optional additional WHERE clause fragment (e.g. time range filter).
@@ -234,6 +234,7 @@ class OracleDialect(SQLDialect):
        embedding_param: str,
        bank_id_param: str,
        fetch_limit: int,
+        min_similarity: float,
        tags_clause: str = "",
        groups_clause: str = "",
        extra_where: str = "",
@@ -249,7 +250,7 @@ class OracleDialect(SQLDialect):
            f" WHERE bank_id = {bank_id_param}"
            f"   AND fact_type = '{fact_type}'"
            f"   AND embedding IS NOT NULL"
-            f"   AND (1 - VECTOR_DISTANCE(embedding, {embedding_param}, COSINE)) >= 0.3"
+            f"   AND (1 - VECTOR_DISTANCE(embedding, {embedding_param}, COSINE)) >= {min_similarity}"
            f"   {tags_clause}"
            f"   {groups_clause}"
            f"   {extra_where}"
@@ -148,6 +148,7 @@ class PostgreSQLDialect(SQLDialect):
        embedding_param: str,
        bank_id_param: str,
        fetch_limit: int,
+        min_similarity: float,
        tags_clause: str = "",
        groups_clause: str = "",
        extra_where: str = "",
@@ -161,7 +162,7 @@ class PostgreSQLDialect(SQLDialect):
            f" WHERE bank_id = {bank_id_param}"
            f"   AND fact_type = '{fact_type}'"
            f"   AND embedding IS NOT NULL"
-            f"   AND (1 - (embedding <=> {embedding_param}::vector)) >= 0.3"
+            f"   AND (1 - (embedding <=> {embedding_param}::vector)) >= {min_similarity}"
            f"   {tags_clause}"
            f"   {groups_clause}"
            f"   {extra_where}"
@@ -40,6 +40,11 @@ class Tenant:
    """

    schema: str
+    # Optional tenant identifier. When provided, background maintenance (e.g. the
+    # consolidation reconcile sweep) can build a RequestContext carrying this id so
+    # tenant-level config overrides are honored. Leave as None for single-tenant
+    # setups or extensions that do not key config by tenant id.
+    tenant_id: str | None = None


 class TenantExtension(Extension, ABC):
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "hindsight-api-slim"
-version = "0.7.2"
+version = "0.8.0"
 description = "Hindsight: Agent Memory That Works Like Human Memory"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -81,6 +81,10 @@ local-ml = [
    # Local ML models for embeddings/reranking
    "sentence-transformers>=3.3.0",
    "transformers>=4.53.0", # Security fixes for ReDoS vulnerabilities
+    # transformers (incl. latest 5.x) hard-requires tokenizers<=0.23.0 via a
+    # runtime check; without this cap an in-place upgrade can pull tokenizers
+    # 0.23.1 and break local embeddings/reranker startup. See issue #2055.
+    "tokenizers>=0.22.0,<=0.23.0",
    "torch>=2.6.0", # CVE fix for remote code execution
    "einops>=0.8.2",
    "flashrank>=0.2.0",
@@ -100,6 +104,7 @@ local-onnx = [
    # In-process ONNX Runtime embeddings without an Ollama/TEI sidecar
    "onnxruntime>=1.17.0",
    "transformers>=4.53.0",
+    "tokenizers>=0.22.0,<=0.23.0", # See issue #2055 (transformers caps tokenizers<=0.23.0)
    "huggingface-hub>=0.20.0",
    "numpy>=1.26.0",
 ]
@@ -182,12 +187,14 @@ dev = [
 [tool.ruff]
 line-length = 120
 target-version = "py311"
-exclude = [
-    "tests/",
-    "**/tests/",
-]

 [tool.ruff.lint]
+# Tests are formatted (via `ruff format`) but excluded from lint rules, which
+# are too noisy for test code (unused imports/vars, import ordering).
+exclude = [
+    "tests/**",
+    "**/tests/**",
+]
 select = [
    "E",      # pycodestyle errors
    "W",      # pycodestyle warnings
@@ -1,6 +1,7 @@
 """
 Pytest configuration and shared fixtures.
 """
+
 import asyncio
 import os
 from pathlib import Path
@@ -20,6 +21,16 @@ from hindsight_api.pg0 import EmbeddedPostgres
 DEFAULT_PG0_INSTANCE_NAME = "hindsight-test"
 DEFAULT_PG0_PORT = int(os.environ.get("HINDSIGHT_TEST_PG_PORT", "5556"))

+# Keep the background MaintenanceLoop from auto-starting during tests. In
+# production it sweeps retention and re-schedules consolidation, but its timers
+# would race shared-pg0 test data (e.g. delete llm_requests/audit_log rows a test
+# just inserted). Disabling the reconcile interval and llm-trace retention — with
+# audit retention already off by default — leaves no job enabled, so the loop
+# never starts. Tests that exercise it call MaintenanceLoop methods
+# (_run_reconcile / _purge_expired) directly.
+os.environ.setdefault("HINDSIGHT_API_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS", "0")
+os.environ.setdefault("HINDSIGHT_API_LLM_TRACE_RETENTION_DAYS", "-1")
+

 # Load environment variables from .env at the start of test session
 def pytest_configure(config):
@@ -66,6 +77,7 @@ def pg0_db_url(db_url, tmp_path_factory, worker_id):
    if db_url and not _parse_pg0_url(db_url)[0]:
        # Plain postgresql:// URL - use it directly but still run migrations
        from hindsight_api.migrations import run_migrations
+
        run_migrations(db_url)
        return db_url

@@ -117,6 +129,7 @@ def pg0_db_url(db_url, tmp_path_factory, worker_id):
    # Run migrations - uses PostgreSQL advisory lock internally,
    # so safe to call from multiple workers (only one will actually run migrations)
    from hindsight_api.migrations import run_migrations
+
    run_migrations(url)

    # Clean up stale test data from previous sessions. Per-bank vector indexes
@@ -147,8 +160,7 @@ def _cleanup_stale_test_data(db_url: str) -> None:
        conn = await asyncpg.connect(db_url)
        try:
            idx_rows = await conn.fetch(
-                "SELECT indexname FROM pg_indexes "
-                "WHERE schemaname = 'public' AND indexname LIKE 'idx_mu_emb_%'"
+                "SELECT indexname FROM pg_indexes WHERE schemaname = 'public' AND indexname LIKE 'idx_mu_emb_%'"
            )
            if idx_rows:
                for row in idx_rows:
@@ -156,10 +168,20 @@ def _cleanup_stale_test_data(db_url: str) -> None:

            # Truncate test data in dependency order
            for table in [
-                "entity_cooccurrences", "unit_entities", "memory_links",
-                "entities", "memory_units", "chunks", "documents",
-                "mental_models", "directives", "async_operations",
-                "audit_log", "webhooks", "file_storage", "banks",
+                "entity_cooccurrences",
+                "unit_entities",
+                "memory_links",
+                "entities",
+                "memory_units",
+                "chunks",
+                "documents",
+                "mental_models",
+                "directives",
+                "async_operations",
+                "audit_log",
+                "webhooks",
+                "file_storage",
+                "banks",
            ]:
                try:
                    await conn.execute(f"TRUNCATE {table} CASCADE")
@@ -242,8 +264,7 @@ def oracle_db_url(_oracle_admin_dsn):
        # Create test user (idempotent — skip if already exists)
        try:
            cursor.execute(
-                f'CREATE USER {test_user} IDENTIFIED BY "{test_pass}" '
-                f"DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS"
+                f'CREATE USER {test_user} IDENTIFIED BY "{test_pass}" DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS'
            )
        except oracledb.DatabaseError as e:
            if hasattr(e.args[0], "code") and e.args[0].code == 1920:
@@ -410,13 +431,12 @@ def cross_encoder(tmp_path_factory, worker_id):

    return ce

+
@pytest.fixture(scope="session")
 def query_analyzer():
    return DateparserQueryAnalyzer()


-
-
@pytest_asyncio.fixture(scope="function")
 async def memory(pg0_db_url, embeddings, cross_encoder, query_analyzer):
    """
@@ -23,6 +23,7 @@ from urllib.parse import urlparse
 # Helpers
 # ---------------------------------------------------------------------------

+
 def _log(step: int, total: int, msg: str) -> None:
    print(f"  [{step}/{total}] {msg}")

@@ -64,8 +65,7 @@ def _bootstrap_test_user(admin_dsn: dict[str, str]) -> str:
        # Create user (skip if already exists - ORA-01920)
        try:
            cursor.execute(
-                f'CREATE USER {test_user} IDENTIFIED BY "{test_pass}" '
-                f"DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS"
+                f'CREATE USER {test_user} IDENTIFIED BY "{test_pass}" DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS'
            )
        except oracledb.DatabaseError as e:
            if hasattr(e.args[0], "code") and e.args[0].code == 1920:
@@ -100,6 +100,7 @@ def _bootstrap_test_user(admin_dsn: dict[str, str]) -> str:
 # Main
 # ---------------------------------------------------------------------------

+
 async def _run() -> None:
    total_steps = 8

@@ -290,6 +291,7 @@ def main() -> int:
    except Exception as exc:
        print(f"\nFAILED: {exc}", file=sys.stderr)
        import traceback
+
        traceback.print_exc()
        return 1

@@ -12,6 +12,7 @@ Usage in tests:
    )
 """

+import asyncio
 import json
 import logging
 import os
@@ -36,6 +37,19 @@ _JUDGE_API_KEY = os.getenv(
 )
 _JUDGE_BASE_URL = os.getenv("HINDSIGHT_TEST_JUDGE_BASE_URL", "")

+# Flakiness hardening. A single temperature-0 judge call still occasionally flips
+# its verdict on borderline phrasing — the dominant source of hs_llm_core
+# flakiness. When the primary verdict is "not met", we ask for a few independent
+# second opinions (at a higher temperature so the samples genuinely differ) and
+# uphold the failure only if the majority agrees. Verdicts that pass on the first
+# call are returned immediately, so passing tests are unaffected in cost or
+# behaviour, and genuine failures (where every judge agrees) still fail.
+_JUDGE_CONFIRMATIONS = int(os.getenv("HINDSIGHT_TEST_JUDGE_CONFIRMATIONS", "2"))
+_JUDGE_CONFIRM_TEMPERATURE = float(os.getenv("HINDSIGHT_TEST_JUDGE_CONFIRM_TEMPERATURE", "0.5"))
+# Retry transient judge-call errors (rate limits, 5xx) so judge infrastructure
+# hiccups never fail the test under evaluation.
+_JUDGE_CALL_ATTEMPTS = int(os.getenv("HINDSIGHT_TEST_JUDGE_CALL_ATTEMPTS", "3"))
+

 class JudgeVerdict(BaseModel):
    meets_criteria: bool
@@ -58,6 +72,58 @@ def _get_judge():
    return _judge_instance


+async def _judge_once(
+    response: str,
+    criteria: str,
+    context: str | None,
+    temperature: float,
+) -> JudgeVerdict:
+    """Run a single judge verdict, retrying transient call errors."""
+    judge = _get_judge()
+    context_block = f"\n\nContext provided to the system:\n{context}" if context else ""
+    messages = [
+        {
+            "role": "system",
+            "content": (
+                "You are a test evaluation judge. Given a response and evaluation criteria, "
+                "determine whether the response meets the criteria. "
+                'Respond with JSON: {"meets_criteria": true/false, "reasoning": "brief explanation"}'
+            ),
+        },
+        {
+            "role": "user",
+            "content": (
+                f"## Response to evaluate\n{response}\n"
+                f"{context_block}\n"
+                f"## Criteria\n{criteria}\n\n"
+                "Does the response meet the criteria?"
+            ),
+        },
+    ]
+
+    last_error: Exception | None = None
+    for attempt in range(max(1, _JUDGE_CALL_ATTEMPTS)):
+        try:
+            result = await judge.call(
+                messages=messages,
+                response_format=JudgeVerdict,
+                max_completion_tokens=256,
+                temperature=temperature,
+                scope="test_judge",
+            )
+            if isinstance(result, JudgeVerdict):
+                return result
+            if isinstance(result, dict):
+                return JudgeVerdict(**result)
+            return JudgeVerdict(**json.loads(str(result)))
+        except Exception as e:  # transient provider error — retry before giving up
+            last_error = e
+            logger.warning(f"Judge call failed (attempt {attempt + 1}/{_JUDGE_CALL_ATTEMPTS}): {e}")
+            await asyncio.sleep(1.0 * (attempt + 1))
+
+    raise RuntimeError(f"Judge call failed after {_JUDGE_CALL_ATTEMPTS} attempts: {last_error}") from last_error
+
+
 async def evaluate(
    response: str,
    criteria: str,
@@ -65,6 +131,12 @@ async def evaluate(
 ) -> JudgeVerdict:
    """Ask the judge LLM whether a response meets the given criteria.

+    The primary verdict is deterministic (temperature 0). If it says the criteria
+    are NOT met, we collect a few independent higher-temperature second opinions
+    and overrule the failure only when the majority disagrees — smoothing out the
+    single-call noise that makes these tests flaky. See the module-level
+    ``_JUDGE_CONFIRMATIONS`` notes.
+
    Args:
        response: The LLM-generated text to evaluate.
        criteria: Plain-English description of what the response should contain/satisfy.
@@ -73,43 +145,33 @@ async def evaluate(
    Returns:
        JudgeVerdict with meets_criteria bool and reasoning string.
    """
-    judge = _get_judge()
+    primary = await _judge_once(response, criteria, context, temperature=0.0)
+    if primary.meets_criteria or _JUDGE_CONFIRMATIONS <= 0:
+        return primary

-    context_block = f"\n\nContext provided to the system:\n{context}" if context else ""
-
-    result = await judge.call(
-        messages=[
-            {
-                "role": "system",
-                "content": (
-                    "You are a test evaluation judge. Given a response and evaluation criteria, "
-                    "determine whether the response meets the criteria. "
-                    "Respond with JSON: {\"meets_criteria\": true/false, \"reasoning\": \"brief explanation\"}"
-                ),
-            },
-            {
-                "role": "user",
-                "content": (
-                    f"## Response to evaluate\n{response}\n"
-                    f"{context_block}\n"
-                    f"## Criteria\n{criteria}\n\n"
-                    "Does the response meet the criteria?"
-                ),
-            },
-        ],
-        response_format=JudgeVerdict,
-        max_completion_tokens=256,
-        temperature=0.0,
-        scope="test_judge",
+    # Primary says "not met": get independent second opinions before trusting it.
+    confirmations = await asyncio.gather(
+        *(
+            _judge_once(response, criteria, context, temperature=_JUDGE_CONFIRM_TEMPERATURE)
+            for _ in range(_JUDGE_CONFIRMATIONS)
+        ),
+        return_exceptions=True,
    )
+    verdicts = [primary] + [c for c in confirmations if isinstance(c, JudgeVerdict)]
+    met = sum(1 for v in verdicts if v.meets_criteria)
+    not_met = len(verdicts) - met

-    if isinstance(result, JudgeVerdict):
-        return result
-
-    # Fallback: parse raw dict/string
-    if isinstance(result, dict):
-        return JudgeVerdict(**result)
-    return JudgeVerdict(**json.loads(str(result)))
+    if met > not_met:
+        agreeing = next(v for v in verdicts if v.meets_criteria)
+        logger.info(f"Judge: primary 'not met' overruled by majority ({met}/{len(verdicts)} met). Criteria: {criteria}")
+        return JudgeVerdict(
+            meets_criteria=True,
+            reasoning=f"Majority of {len(verdicts)} judges met criteria (primary verdict overruled as noise). {agreeing.reasoning}",
+        )
+    return JudgeVerdict(
+        meets_criteria=False,
+        reasoning=f"{not_met}/{len(verdicts)} judges agree criteria not met. {primary.reasoning}",
+    )


 async def assert_meets_criteria(
@@ -124,7 +186,7 @@ async def assert_meets_criteria(
    """
    verdict = await evaluate(response=response, criteria=criteria, context=context)
    if not verdict.meets_criteria:
-        fail_msg = msg or f"LLM judge: criteria not met"
+        fail_msg = msg or "LLM judge: criteria not met"
        raise AssertionError(
            f"{fail_msg}\n"
            f"  Criteria: {criteria}\n"
@@ -0,0 +1,128 @@
+"""Tests for the admin surface: GET /admin/config + the admin_api feature flag.
+
+These are deterministic (no LLM): the endpoint only reads server-level config. We
+toggle env vars + clear the config cache to exercise the enable flag, the optional
+admin token, and credential redaction.
+"""
+
+import httpx
+import pytest
+import pytest_asyncio
+
+from hindsight_api.api import create_app
+from hindsight_api.config import clear_config_cache
+
+
+@pytest_asyncio.fixture
+async def admin_client(memory):
+    """Async test client for the FastAPI app (mock LLM)."""
+    app = create_app(memory, initialize_memory=False)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        yield client
+
+
+def _set_env(monkeypatch, **values: str | None) -> None:
+    """Set/unset env vars and reset the cached config so the next read reflects them."""
+    for key, value in values.items():
+        if value is None:
+            monkeypatch.delenv(key, raising=False)
+        else:
+            monkeypatch.setenv(key, value)
+    clear_config_cache()
+
+
+@pytest.fixture(autouse=True)
+def _restore_config_cache():
+    """Ensure the global config cache is reset after each test."""
+    yield
+    clear_config_cache()
+
+
+@pytest.mark.asyncio
+async def test_admin_config_disabled_by_default(admin_client, monkeypatch):
+    """When the admin API is disabled (default), the endpoint is invisible (404)."""
+    _set_env(monkeypatch, HINDSIGHT_API_ENABLE_ADMIN_API=None, HINDSIGHT_API_ADMIN_TOKEN=None)
+
+    response = await admin_client.get("/admin/config")
+
+    assert response.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_admin_config_enabled_no_token(admin_client, monkeypatch):
+    """When enabled without a token, the endpoint is open and returns config."""
+    _set_env(monkeypatch, HINDSIGHT_API_ENABLE_ADMIN_API="true", HINDSIGHT_API_ADMIN_TOKEN=None)
+
+    response = await admin_client.get("/admin/config")
+
+    assert response.status_code == 200
+    config = response.json()["config"]
+    # A representative spread of non-credential fields should be present.
+    assert "llm_provider" in config
+    assert "enable_admin_api" in config
+    assert config["enable_admin_api"] is True
+
+
+@pytest.mark.asyncio
+async def test_admin_config_redacts_credentials(admin_client, monkeypatch):
+    """Credential fields are masked, never returned in cleartext."""
+    _set_env(
+        monkeypatch,
+        HINDSIGHT_API_ENABLE_ADMIN_API="true",
+        HINDSIGHT_API_ADMIN_TOKEN="s3cret-token",
+        HINDSIGHT_API_LLM_API_KEY="super-secret-key",
+    )
+
+    response = await admin_client.get("/admin/config", headers={"Authorization": "Bearer s3cret-token"})
+
+    assert response.status_code == 200
+    config = response.json()["config"]
+    # The configured LLM key is present but masked.
+    assert config["llm_api_key"] == "***"
+    assert "super-secret-key" not in response.text
+    # Provider keys that fall back to the LLM key (and aren't in the credential
+    # denylist) must also be masked — the view redacts by name, not just the set.
+    assert config["embeddings_openrouter_api_key"] == "***"
+    assert config["reranker_openrouter_api_key"] == "***"
+    # The admin token must never leak through its own config view.
+    assert config["admin_api_token"] == "***"
+    assert "s3cret-token" not in response.text
+    # Value-bearing fields that merely contain "token" in their name (plural) are
+    # NOT redacted — they carry useful config, not secrets.
+    assert config["recall_max_tokens"] != "***"
+
+
+@pytest.mark.asyncio
+async def test_admin_config_requires_token_when_set(admin_client, monkeypatch):
+    """With a token configured, missing/wrong tokens are rejected; the right one passes."""
+    _set_env(
+        monkeypatch,
+        HINDSIGHT_API_ENABLE_ADMIN_API="true",
+        HINDSIGHT_API_ADMIN_TOKEN="right-token",
+    )
+
+    missing = await admin_client.get("/admin/config")
+    assert missing.status_code == 401
+
+    wrong = await admin_client.get("/admin/config", headers={"Authorization": "Bearer wrong-token"})
+    assert wrong.status_code == 401
+
+    bearer = await admin_client.get("/admin/config", headers={"Authorization": "Bearer right-token"})
+    assert bearer.status_code == 200
+
+    # A bare token (no "Bearer " prefix) is also accepted.
+    bare = await admin_client.get("/admin/config", headers={"Authorization": "right-token"})
+    assert bare.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_version_reports_admin_api_flag(admin_client, monkeypatch):
+    """The /version feature flags track the admin enable flag."""
+    _set_env(monkeypatch, HINDSIGHT_API_ENABLE_ADMIN_API="true")
+    enabled = await admin_client.get("/version")
+    assert enabled.json()["features"]["admin_api"] is True
+
+    _set_env(monkeypatch, HINDSIGHT_API_ENABLE_ADMIN_API="false")
+    disabled = await admin_client.get("/version")
+    assert disabled.json()["features"]["admin_api"] is False
@@ -1,6 +1,7 @@
 """
 Tests for agent management API (profile, disposition).
 """
+
 import pytest
 import uuid
 from hindsight_api import MemoryEngine, RequestContext
@@ -17,9 +18,7 @@ class TestAgentProfile:
    """Tests for agent profile management."""

    @pytest.mark.asyncio
-    async def test_get_bank_profile_no_auto_create_returns_none(
-        self, memory: MemoryEngine, request_context
-    ):
+    async def test_get_bank_profile_no_auto_create_returns_none(self, memory: MemoryEngine, request_context):
        """When create_if_missing=False is passed, a missing bank returns None
        rather than being silently auto-created. This is what read-only
        endpoints (HTTP GET, polling, etc.) must use to avoid creating banks
@@ -27,28 +26,20 @@ class TestAgentProfile:
        bank_id = unique_agent_id("test_no_auto_create")

        # First call with create_if_missing=False on a non-existent bank
-        result = await memory.get_bank_profile(
-            bank_id, request_context=request_context, create_if_missing=False
-        )
+        result = await memory.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
        assert result is None, "Expected None for missing bank with create_if_missing=False"

        # Verify the bank was NOT created as a side effect
-        result_again = await memory.get_bank_profile(
-            bank_id, request_context=request_context, create_if_missing=False
-        )
+        result_again = await memory.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
        assert result_again is None, "Bank must not exist after read-only call"

        # And explicit auto-create still works
-        created = await memory.get_bank_profile(
-            bank_id, request_context=request_context, create_if_missing=True
-        )
+        created = await memory.get_bank_profile(bank_id, request_context=request_context, create_if_missing=True)
        assert created is not None
        assert created["disposition"]["skepticism"] == 3

        # Now read-only call sees it
-        seen = await memory.get_bank_profile(
-            bank_id, request_context=request_context, create_if_missing=False
-        )
+        seen = await memory.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
        assert seen is not None
        assert seen["disposition"]["skepticism"] == 3

@@ -122,11 +113,7 @@ class TestAgentEndpoint:
        bank_id = unique_agent_id("test_put_create")

        request = CreateBankRequest(
-            disposition=DispositionTraits(
-                skepticism=4,
-                literalism=5,
-                empathy=2
-            ),
+            disposition=DispositionTraits(skepticism=4, literalism=5, empathy=2),
        )

        profile = await memory.get_bank_profile(bank_id, request_context=request_context)
@@ -155,7 +142,7 @@ class TestAgentDispositionIntegration:
        disposition = {
            "skepticism": 5,  # Very skeptical
            "literalism": 4,  # High literalism
-            "empathy": 2,     # Low empathy
+            "empathy": 2,  # Low empathy
        }
        await memory.update_bank_disposition(bank_id, disposition, request_context=request_context)

@@ -163,7 +150,7 @@ class TestAgentDispositionIntegration:
            bank_id=bank_id,
            contents=[
                {"content": "Traditional painting techniques have been used for centuries"},
-                {"content": "Modern digital art is changing the art world"}
+                {"content": "Modern digital art is changing the art world"},
            ],
            request_context=request_context,
        )
@@ -103,6 +103,11 @@ async def test_small_async_batch_no_splitting(memory, request_context):
    assert status["result_metadata"]["num_sub_batches"] == 1  # Single sub-batch
    assert len(status["child_operations"]) == 1
    assert status["child_operations"][0]["status"] == "completed"
+    child_meta = await _child_metadata(memory, bank_id, operation_id, request_context)
+    assert child_meta["unit_ids_count"] > 0
+    assert child_meta["extraction_errors_count"] == 0
+    assert status["result_metadata"]["unit_ids_count"] == child_meta["unit_ids_count"]
+    assert status["result_metadata"]["extraction_errors_count"] == 0


@pytest.mark.asyncio
@@ -166,6 +171,19 @@ async def test_large_async_batch_auto_splits(memory, request_context):

    # Parent status should be aggregated as "completed"
    assert parent_status["status"] == "completed"
+    child_unit_counts = []
+    for child in child_ops:
+        child_status = await memory.get_operation_status(
+            bank_id=bank_id,
+            operation_id=child["operation_id"],
+            request_context=request_context,
+        )
+        child_meta = child_status["result_metadata"]
+        assert child_meta["unit_ids_count"] > 0
+        assert child_meta["extraction_errors_count"] == 0
+        child_unit_counts.append(child_meta["unit_ids_count"])
+    assert parent_status["result_metadata"]["unit_ids_count"] == sum(child_unit_counts)
+    assert parent_status["result_metadata"]["extraction_errors_count"] == 0


@pytest.mark.asyncio
@@ -461,6 +479,42 @@ async def _child_metadata(memory, bank_id: str, parent_operation_id: str, reques
    return child["result_metadata"]


+@pytest.mark.asyncio
+async def test_retain_outcome_metadata_records_zero_counts(memory, request_context, monkeypatch):
+    """Completed retain operations expose explicit zero outcome counters."""
+    from hindsight_api.engine.response_models import TokenUsage
+    from hindsight_api.engine.retain import fact_extraction
+
+    async def empty_extract_facts_from_contents(
+        *args: object, **kwargs: object
+    ) -> tuple[list[object], list[object], TokenUsage]:
+        return [], [], TokenUsage()
+
+    monkeypatch.setattr(fact_extraction, "extract_facts_from_contents", empty_extract_facts_from_contents)
+
+    bank_id = "test_retain_outcome_zero_counts"
+    result = await memory.submit_async_retain(
+        bank_id=bank_id,
+        contents=[{"content": "No extracted facts for this item."}],
+        request_context=request_context,
+    )
+    await asyncio.sleep(0.2)
+
+    parent = await memory.get_operation_status(
+        bank_id=bank_id,
+        operation_id=result["operation_id"],
+        request_context=request_context,
+    )
+    child_meta = await _child_metadata(memory, bank_id, result["operation_id"], request_context)
+
+    assert child_meta["unit_ids_count"] == 0
+    assert child_meta["extraction_errors_count"] == 0
+    assert "extraction_errors_sample" not in child_meta
+    assert parent["result_metadata"]["unit_ids_count"] == 0
+    assert parent["result_metadata"]["extraction_errors_count"] == 0
+    assert "extraction_errors_sample" not in parent["result_metadata"]
+
+
@pytest.mark.asyncio
 async def test_retain_records_user_provided_document_ids(memory, request_context):
    """User-supplied document_ids land in child op result_metadata.document_ids."""
@@ -643,9 +643,7 @@ class TestDefaultBankTemplateEnvVar:
        yield default_template

    @pytest.mark.asyncio
-    async def test_default_template_applied_on_new_bank(
-        self, api_client, bank_id, _patched_default_template
-    ):
+    async def test_default_template_applied_on_new_bank(self, api_client, bank_id, _patched_default_template):
        """Creating a new bank applies the default template (config + mental models + directives)."""
        # Trigger bank auto-creation via GET profile
        resp = await api_client.put(f"/v1/default/banks/{bank_id}", json={})
@@ -730,9 +728,7 @@ class TestDefaultBankTemplateEnvVar:
        assert config_resp.json()["overrides"] == {}

    @pytest.mark.asyncio
-    async def test_default_template_malformed_is_swallowed(
-        self, api_client, bank_id, monkeypatch
-    ):
+    async def test_default_template_malformed_is_swallowed(self, api_client, bank_id, monkeypatch):
        """A malformed default template is logged and ignored — bank creation still succeeds."""
        from hindsight_api.config import _get_raw_config

@@ -4,6 +4,7 @@ Integration test for API base path support.
 Tests that the API works correctly when deployed with a base path (e.g., /hindsight)
 for reverse proxy deployments.
 """
+
 import os
 import pytest
 import pytest_asyncio
@@ -27,10 +28,7 @@ async def api_client_with_base_path(memory):

    # Use base_url with base path
    transport = httpx.ASGITransport(app=app)
-    async with httpx.AsyncClient(
-        transport=transport,
-        base_url=f"http://test{base_path}"
-    ) as client:
+    async with httpx.AsyncClient(transport=transport, base_url=f"http://test{base_path}") as client:
        yield client

    # Cleanup: unset base path
@@ -122,10 +120,10 @@ async def test_base_path_full_workflow(api_client_with_base_path):
            "items": [
                {
                    "content": "The API supports base path deployment for reverse proxy use cases.",
-                    "context": "testing base path feature"
+                    "context": "testing base path feature",
                }
            ]
-        }
+        },
    )
    assert response.status_code == 200
    result = response.json()
@@ -133,10 +131,7 @@ async def test_base_path_full_workflow(api_client_with_base_path):

    # 3. Recall the memory
    response = await api_client_with_base_path.post(
-        f"/v1/default/banks/{bank_id}/memories/recall",
-        json={
-            "query": "base path support"
-        }
+        f"/v1/default/banks/{bank_id}/memories/recall", json={"query": "base path support"}
    )
    assert response.status_code == 200
    recall_result = response.json()
@@ -7,21 +7,24 @@ Tests cover:
 - Hard error when provider doesn't support the batch API (no silent fallback)
 - Worker recovery on restart
 """
-import pytest
+
 import asyncio
-import logging
 import json
+import logging
 import uuid
 from datetime import datetime, timezone
 from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
 from hindsight_api import RequestContext
-from hindsight_api.engine.retain.fact_extraction import (
-    extract_facts_from_contents_batch_api,
-    extract_facts_from_contents,
-    RetainContent,
-)
 from hindsight_api.config import HindsightConfig
 from hindsight_api.engine.llm_wrapper import create_llm_provider
+from hindsight_api.engine.retain.fact_extraction import (
+    RetainContent,
+    extract_facts_from_contents,
+    extract_facts_from_contents_batch_api,
+)
 from hindsight_api.worker.poller import WorkerPoller

 logger = logging.getLogger(__name__)
@@ -103,19 +106,21 @@ async def test_batch_api_normal_flow(mock_llm_config, test_contents, hindsight_c
                        "choices": [
                            {
                                "message": {
-                                    "content": json.dumps({
-                                        "facts": [
-                                            {
-                                                "what": "Alice is a senior software engineer at TechCorp",
-                                                "when": "present",
-                                                "where": "TechCorp",
-                                                "who": "Alice",
-                                                "why": "Professional background information",
-                                                "fact_type": "world",
-                                                "fact_kind": "conversation",
-                                            }
-                                        ]
-                                    })
+                                    "content": json.dumps(
+                                        {
+                                            "facts": [
+                                                {
+                                                    "what": "Alice is a senior software engineer at TechCorp",
+                                                    "when": "present",
+                                                    "where": "TechCorp",
+                                                    "who": "Alice",
+                                                    "why": "Professional background information",
+                                                    "fact_type": "world",
+                                                    "fact_kind": "conversation",
+                                                }
+                                            ]
+                                        }
+                                    )
                                }
                            }
                        ],
@@ -130,19 +135,21 @@ async def test_batch_api_normal_flow(mock_llm_config, test_contents, hindsight_c
                        "choices": [
                            {
                                "message": {
-                                    "content": json.dumps({
-                                        "facts": [
-                                            {
-                                                "what": "Bob joined the team last month as a junior developer",
-                                                "when": "last month",
-                                                "where": "team",
-                                                "who": "Bob",
-                                                "why": "New team member information",
-                                                "fact_type": "world",
-                                                "fact_kind": "conversation",
-                                            }
-                                        ]
-                                    })
+                                    "content": json.dumps(
+                                        {
+                                            "facts": [
+                                                {
+                                                    "what": "Bob joined the team last month as a junior developer",
+                                                    "when": "last month",
+                                                    "where": "team",
+                                                    "who": "Bob",
+                                                    "why": "New team member information",
+                                                    "fact_type": "world",
+                                                    "fact_kind": "conversation",
+                                                }
+                                            ]
+                                        }
+                                    )
                                }
                            }
                        ],
@@ -211,6 +218,7 @@ async def test_batch_api_crash_recovery(mock_llm_config, test_contents, hindsigh
        schema = request_context.tenant_id

        from hindsight_api.engine.task_backend import fq_table
+
        table = fq_table("async_operations", schema)

        # Create operation with batch_id already stored
@@ -221,11 +229,13 @@ async def test_batch_api_crash_recovery(mock_llm_config, test_contents, hindsigh
            """,
            operation_id,
            bank_id,
-            json.dumps({
-                "batch_id": batch_id,
-                "batch_provider": "openai",
-                "chunk_count": 2,
-            }),
+            json.dumps(
+                {
+                    "batch_id": batch_id,
+                    "batch_provider": "openai",
+                    "chunk_count": 2,
+                }
+            ),
        )

        # Mock batch API responses for resume scenario
@@ -248,19 +258,21 @@ async def test_batch_api_crash_recovery(mock_llm_config, test_contents, hindsigh
                        "choices": [
                            {
                                "message": {
-                                    "content": json.dumps({
-                                        "facts": [
-                                            {
-                                                "what": "Alice is a senior software engineer",
-                                                "when": "present",
-                                                "where": "TechCorp",
-                                                "who": "Alice",
-                                                "why": "Background",
-                                                "fact_type": "world",
-                                                "fact_kind": "conversation",
-                                            }
-                                        ]
-                                    })
+                                    "content": json.dumps(
+                                        {
+                                            "facts": [
+                                                {
+                                                    "what": "Alice is a senior software engineer",
+                                                    "when": "present",
+                                                    "where": "TechCorp",
+                                                    "who": "Alice",
+                                                    "why": "Background",
+                                                    "fact_type": "world",
+                                                    "fact_kind": "conversation",
+                                                }
+                                            ]
+                                        }
+                                    )
                                }
                            }
                        ],
@@ -275,19 +287,21 @@ async def test_batch_api_crash_recovery(mock_llm_config, test_contents, hindsigh
                        "choices": [
                            {
                                "message": {
-                                    "content": json.dumps({
-                                        "facts": [
-                                            {
-                                                "what": "Bob is a junior developer",
-                                                "when": "last month",
-                                                "where": "team",
-                                                "who": "Bob",
-                                                "why": "New member",
-                                                "fact_type": "world",
-                                                "fact_kind": "conversation",
-                                            }
-                                        ]
-                                    })
+                                    "content": json.dumps(
+                                        {
+                                            "facts": [
+                                                {
+                                                    "what": "Bob is a junior developer",
+                                                    "when": "last month",
+                                                    "where": "team",
+                                                    "who": "Bob",
+                                                    "why": "New member",
+                                                    "fact_type": "world",
+                                                    "fact_kind": "conversation",
+                                                }
+                                            ]
+                                        }
+                                    )
                                }
                            }
                        ],
@@ -331,6 +345,105 @@ async def test_batch_api_crash_recovery(mock_llm_config, test_contents, hindsigh
            pass


+@pytest.mark.asyncio
+async def test_batch_api_records_non_fatal_extraction_errors(
+    mock_llm_config, test_contents, hindsight_config, memory, request_context
+):
+    """Batch API skipped chunks are surfaced in operation result_metadata."""
+    bank_id = f"test_batch_errors_{datetime.now(timezone.utc).timestamp()}"
+    operation_id = str(uuid.uuid4())
+
+    try:
+        await memory.get_bank_profile(bank_id, request_context=request_context)
+        pool = memory._pool
+        schema = request_context.tenant_id
+
+        from hindsight_api.engine.task_backend import fq_table
+
+        table = fq_table("async_operations", schema)
+        await pool.execute(
+            f"""
+            INSERT INTO {table} (operation_id, operation_type, bank_id, status, result_metadata)
+            VALUES ($1, 'retain', $2, 'processing', $3::jsonb)
+            """,
+            operation_id,
+            bank_id,
+            json.dumps({}),
+        )
+
+        batch_id = "batch_partial_errors"
+        mock_llm_config._provider_impl.supports_batch_api = AsyncMock(return_value=True)
+        mock_llm_config._provider_impl.submit_batch = AsyncMock(return_value={"batch_id": batch_id})
+        mock_llm_config._provider_impl.get_batch_status = AsyncMock(
+            return_value={
+                "status": "completed",
+                "request_counts": {"total": 2, "completed": 2, "failed": 0},
+            }
+        )
+        mock_llm_config._provider_impl.retrieve_batch_results = AsyncMock(
+            return_value=[
+                {
+                    "custom_id": "chunk_0",
+                    "response": {
+                        "body": {
+                            "choices": [
+                                {
+                                    "message": {
+                                        "content": json.dumps(
+                                            {
+                                                "facts": [
+                                                    {
+                                                        "what": "Alice is a senior software engineer",
+                                                        "when": "present",
+                                                        "where": "TechCorp",
+                                                        "who": "Alice",
+                                                        "why": "Background",
+                                                        "fact_type": "world",
+                                                        "fact_kind": "conversation",
+                                                    }
+                                                ]
+                                            }
+                                        )
+                                    }
+                                }
+                            ],
+                            "usage": {"prompt_tokens": 100, "completion_tokens": 50, "total_tokens": 150},
+                        }
+                    },
+                }
+            ]
+        )
+
+        facts, chunks, usage = await extract_facts_from_contents_batch_api(
+            contents=test_contents,
+            llm_config=mock_llm_config,
+            agent_name="test_agent",
+            config=hindsight_config,
+            pool=pool,
+            operation_id=operation_id,
+            schema=schema,
+        )
+
+        assert len(facts) == 1
+        assert len(chunks) == 2
+        assert chunks[1].fact_count == 0
+        assert usage.total_tokens == 150
+
+        row = await pool.fetchrow(f"SELECT result_metadata FROM {table} WHERE operation_id = $1", operation_id)
+        metadata = (
+            json.loads(row["result_metadata"]) if isinstance(row["result_metadata"], str) else row["result_metadata"]
+        )
+        assert metadata["batch_id"] == batch_id
+        assert metadata["extraction_errors_count"] == 1
+        assert metadata["extraction_errors_sample"] == ["chunk_1: missing batch result"]
+
+    finally:
+        try:
+            await memory.delete_bank(bank_id, request_context=request_context)
+        except Exception:
+            pass
+
+
@pytest.mark.asyncio
 async def test_batch_api_raises_for_unsupported_provider(mock_llm_config, test_contents, hindsight_config):
    """Batch extraction must surface a hard error (not silently fall back) when
@@ -372,6 +485,7 @@ async def test_worker_batch_recovery(memory, request_context):
        schema = request_context.tenant_id

        from hindsight_api.engine.task_backend import fq_table
+
        table = fq_table("async_operations", schema)

        # Create orphaned batch operation (simulates worker crash during polling)
@@ -389,16 +503,19 @@ async def test_worker_batch_recovery(memory, request_context):
            """,
            operation_id,
            bank_id,
-            json.dumps({
-                "batch_id": batch_id,
-                "batch_provider": "openai",
-                "chunk_count": 1,
-            }),
+            json.dumps(
+                {
+                    "batch_id": batch_id,
+                    "batch_provider": "openai",
+                    "chunk_count": 1,
+                }
+            ),
            json.dumps(task_payload),
        )

        # Create WorkerPoller
        from hindsight_api.extensions.builtin.tenant import DefaultTenantExtension
+
        tenant_extension = DefaultTenantExtension(config={"schema": schema} if schema else {})

        poller = WorkerPoller(
@@ -462,13 +579,7 @@ async def test_batch_api_via_extract_facts_from_contents(
                    "custom_id": "chunk_0",
                    "response": {
                        "body": {
-                            "choices": [
-                                {
-                                    "message": {
-                                        "content": json.dumps({"facts": []})
-                                    }
-                                }
-                            ],
+                            "choices": [{"message": {"content": json.dumps({"facts": []})}}],
                            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
                        }
                    },
@@ -10,6 +10,7 @@ To run:
 To skip in CI:
    Add @pytest.mark.skip at the test level
 """
+
 import pytest
 import os
 import asyncio
@@ -115,7 +116,9 @@ def integration_config():
    return config


-@pytest.mark.skip(reason="Real API test - takes minutes and costs money. Run manually with: pytest tests/test_batch_api_integration.py::test_real_openai_batch_api -v -s")
+@pytest.mark.skip(
+    reason="Real API test - takes minutes and costs money. Run manually with: pytest tests/test_batch_api_integration.py::test_real_openai_batch_api -v -s"
+)
@pytest.mark.integration  # Mark as integration test
@pytest.mark.slow  # Mark as slow test
@pytest.mark.asyncio
@@ -174,17 +177,21 @@ async def test_real_openai_batch_api(real_llm_config, test_contents_real, integr
        logger.info("\n" + "=" * 80)
        logger.info("✅ BATCH COMPLETED SUCCESSFULLY")
        logger.info("=" * 80)
-        logger.info(f"Total duration: {total_duration:.1f} seconds ({total_duration/60:.1f} minutes)")
+        logger.info(f"Total duration: {total_duration:.1f} seconds ({total_duration / 60:.1f} minutes)")
        logger.info(f"Facts extracted: {len(facts)}")
        logger.info(f"Chunks processed: {len(chunks)}")
-        logger.info(f"Token usage: {usage.input_tokens} input + {usage.output_tokens} output = {usage.total_tokens} total")
-        logger.info(f"Estimated cost: ${(usage.input_tokens * 0.00015 / 1000 + usage.output_tokens * 0.0006 / 1000):.4f}")
+        logger.info(
+            f"Token usage: {usage.input_tokens} input + {usage.output_tokens} output = {usage.total_tokens} total"
+        )
+        logger.info(
+            f"Estimated cost: ${(usage.input_tokens * 0.00015 / 1000 + usage.output_tokens * 0.0006 / 1000):.4f}"
+        )
        logger.info("=" * 80)

        # Log sample facts
        logger.info("\n📋 Sample extracted facts:")
        for i, fact in enumerate(facts[:5]):  # Show first 5 facts
-            logger.info(f"\nFact {i+1}:")
+            logger.info(f"\nFact {i + 1}:")
            logger.info(f"  Type: {fact.fact_type}")
            logger.info(f"  Text: {fact.fact_text[:100]}...")
            logger.info(f"  Entities: {fact.entities}")
@@ -212,11 +219,13 @@ async def test_real_openai_batch_api(real_llm_config, test_contents_real, integr
            f.write(f"Contents: {len(test_contents_real)} items\n")
            f.write(f"Poll Interval: {integration_config.retain_batch_poll_interval_seconds}s\n\n")
            f.write(f"Results:\n")
-            f.write(f"  Total Duration: {total_duration:.1f}s ({total_duration/60:.1f} min)\n")
+            f.write(f"  Total Duration: {total_duration:.1f}s ({total_duration / 60:.1f} min)\n")
            f.write(f"  Facts Extracted: {len(facts)}\n")
            f.write(f"  Chunks Processed: {len(chunks)}\n")
            f.write(f"  Token Usage: {usage.total_tokens} ({usage.input_tokens} in + {usage.output_tokens} out)\n")
-            f.write(f"  Estimated Cost: ${(usage.input_tokens * 0.00015 / 1000 + usage.output_tokens * 0.0006 / 1000):.4f}\n")
+            f.write(
+                f"  Estimated Cost: ${(usage.input_tokens * 0.00015 / 1000 + usage.output_tokens * 0.0006 / 1000):.4f}\n"
+            )

        logger.info(f"\n📄 Timing report written to: {report_path}")

@@ -174,9 +174,7 @@ def test_async_children_packs_small_items_by_budget():
    num_items = max(4, (tokens_per_batch // max(item_tokens, 1)) * 3)
    contents = [{"content": item_text, "document_id": f"doc-{i}"} for i in range(num_items)]
    total = sum(count_tokens(c["content"]) for c in contents)
-    assert total > tokens_per_batch, (
-        f"Test setup error: {total} tokens does not exceed budget {tokens_per_batch}"
-    )
+    assert total > tokens_per_batch, f"Test setup error: {total} tokens does not exceed budget {tokens_per_batch}"

    children = _split_contents_into_async_children(contents, tokens_per_batch)

@@ -104,8 +104,7 @@ class TestCausalRelationsValidation:
            for rel in facts[0].causal_relations:
                # This should never happen due to validation
                assert False, (
-                    f"First fact should not have causal relations, "
-                    f"but found: target_index={rel.target_fact_index}"
+                    f"First fact should not have causal relations, but found: target_index={rel.target_fact_index}"
                )

    @pytest.mark.asyncio
@@ -139,11 +138,13 @@ class TestCausalRelationsValidation:
        for i, fact in enumerate(facts):
            if fact.causal_relations:
                for rel in fact.causal_relations:
-                    all_relations.append({
-                        "from_fact": i,
-                        "to_fact": rel.target_fact_index,
-                        "type": rel.relation_type,
-                    })
+                    all_relations.append(
+                        {
+                            "from_fact": i,
+                            "to_fact": rel.target_fact_index,
+                            "type": rel.relation_type,
+                        }
+                    )

        # If causal relations were extracted, verify they form a valid chain
        if all_relations:
@@ -226,6 +227,5 @@ class TestCausalRelationsValidation:
            if fact.causal_relations:
                for rel in fact.causal_relations:
                    assert rel.relation_type in valid_types, (
-                        f"Invalid relation_type '{rel.relation_type}'. "
-                        f"Must be one of: {valid_types}"
+                        f"Invalid relation_type '{rel.relation_type}'. Must be one of: {valid_types}"
                    )
@@ -40,7 +40,11 @@ After searching for weeks, I finally found a cheaper apartment in Brooklyn.
        llm_config = LLMConfig.from_env()

        facts, _, _ = await extract_facts_from_text(
-            text=text, event_date=datetime(2024, 3, 15), context=context, llm_config=llm_config, agent_name="TestUser",
+            text=text,
+            event_date=datetime(2024, 3, 15),
+            context=context,
+            llm_config=llm_config,
+            agent_name="TestUser",
            config=_get_raw_config(),
        )

@@ -109,7 +113,11 @@ The renovation took three months and cost $15,000.
        llm_config = LLMConfig.from_env()

        facts, _, _ = await extract_facts_from_text(
-            text=text, event_date=datetime(2024, 6, 1), context=context, llm_config=llm_config, agent_name="TestUser",
+            text=text,
+            event_date=datetime(2024, 6, 1),
+            context=context,
+            llm_config=llm_config,
+            agent_name="TestUser",
            config=_get_raw_config(),
        )

@@ -140,7 +148,11 @@ Machine learning fascinated me so much that I changed my career to data science.
        llm_config = LLMConfig.from_env()

        facts, _, _ = await extract_facts_from_text(
-            text=text, event_date=datetime(2024, 1, 1), context=context, llm_config=llm_config, agent_name="TestUser",
+            text=text,
+            event_date=datetime(2024, 1, 1),
+            context=context,
+            llm_config=llm_config,
+            agent_name="TestUser",
            config=_get_raw_config(),
        )

@@ -168,7 +180,11 @@ The new role enabled me to lead a team of engineers.
        llm_config = LLMConfig.from_env()

        facts, _, _ = await extract_facts_from_text(
-            text=text, event_date=datetime(2024, 2, 15), context=context, llm_config=llm_config, agent_name="TestUser",
+            text=text,
+            event_date=datetime(2024, 2, 15),
+            context=context,
+            llm_config=llm_config,
+            agent_name="TestUser",
            config=_get_raw_config(),
        )

@@ -180,4 +196,3 @@ The new role enabled me to lead a team of engineers.
                        f"Invalid target_fact_index {rel.target_fact_index} in fact {i}. "
                        f"Must reference previous facts only (valid range: 0 to {i - 1})"
                    )
-
@@ -130,8 +130,7 @@ async def test_store_chunks_batch_second_call_with_identical_payload(memory):
            await _seed_bank_and_document(conn, bank_id, document_id)

            chunks = [
-                ChunkMetadata(chunk_text=f"chunk-{i}", fact_count=1, content_index=0, chunk_index=i)
-                for i in range(5)
+                ChunkMetadata(chunk_text=f"chunk-{i}", fact_count=1, content_index=0, chunk_index=i) for i in range(5)
            ]

            await chunk_storage.store_chunks_batch(conn, bank_id, document_id, chunks, ops=ops)
@@ -1,6 +1,7 @@
 """
 Test chunking functionality for large documents.
 """
+
 import pytest
 from hindsight_api.engine.retain.fact_extraction import chunk_text

@@ -53,4 +54,3 @@ def test_chunk_text_64k():
    # Verify we didn't lose content
    combined_length = sum(len(chunk) for chunk in chunks)
    assert combined_length >= len(text) * 0.95, "Lost too much content during chunking"
-
@@ -344,4 +344,7 @@ class TestFactoryFunction:
                assert isinstance(encoder, CohereCrossEncoder)
                assert encoder.api_key == "test_key"
                assert encoder.model == "cohere-rerank-v3-english"
-                assert encoder.base_url == "https://my-endpoint.inference.ai.azure.com/models/cohere-rerank-v3-english/invoke"
+                assert (
+                    encoder.base_url
+                    == "https://my-endpoint.inference.ai.azure.com/models/cohere-rerank-v3-english/invoke"
+                )
@@ -18,10 +18,12 @@ def setup_test_env():
    # Save original environment values
    env_vars_to_save = [
        "HINDSIGHT_API_RETAIN_MAX_COMPLETION_TOKENS",
+        "HINDSIGHT_API_CONSOLIDATION_MAX_COMPLETION_TOKENS",
        "HINDSIGHT_API_RETAIN_CHUNK_SIZE",
        "HINDSIGHT_API_LLM_PROVIDER",
        "HINDSIGHT_API_LLM_MODEL",
        "HINDSIGHT_API_LLM_REASONING_EFFORT",
+        "HINDSIGHT_API_SEMANTIC_MIN_SIMILARITY",
        "HINDSIGHT_API_DATABASE_URL",
        "HINDSIGHT_API_MIGRATION_DATABASE_URL",
    ]
@@ -103,6 +105,48 @@ def test_valid_retain_config_succeeds():
    assert config.retain_chunk_size == 3000


+def test_semantic_min_similarity_reads_from_env():
+    """Semantic retrieval min similarity can be configured at the server level."""
+    from hindsight_api.config import HindsightConfig
+
+    os.environ["HINDSIGHT_API_SEMANTIC_MIN_SIMILARITY"] = "0.58"
+
+    config = HindsightConfig.from_env()
+    assert config.semantic_min_similarity == 0.58
+
+
+def test_semantic_min_similarity_must_be_between_zero_and_one():
+    """Invalid semantic min similarity fails fast during configuration loading."""
+    from hindsight_api.config import HindsightConfig
+
+    os.environ["HINDSIGHT_API_SEMANTIC_MIN_SIMILARITY"] = "1.5"
+
+    with pytest.raises(ValueError, match="semantic_min_similarity"):
+        HindsightConfig.from_env()
+
+
+def test_consolidation_max_completion_tokens_defaults_to_unset():
+    """By default consolidation sends no explicit output budget (backwards compatible)."""
+    from hindsight_api.config import HindsightConfig
+
+    os.environ["HINDSIGHT_API_LLM_PROVIDER"] = "mock"
+    os.environ.pop("HINDSIGHT_API_CONSOLIDATION_MAX_COMPLETION_TOKENS", None)
+
+    config = HindsightConfig.from_env()
+    assert config.consolidation_max_completion_tokens is None
+
+
+def test_consolidation_max_completion_tokens_env_override():
+    """HINDSIGHT_API_CONSOLIDATION_MAX_COMPLETION_TOKENS controls consolidation LLM output budget."""
+    from hindsight_api.config import HindsightConfig
+
+    os.environ["HINDSIGHT_API_LLM_PROVIDER"] = "mock"
+    os.environ["HINDSIGHT_API_CONSOLIDATION_MAX_COMPLETION_TOKENS"] = "8192"
+
+    config = HindsightConfig.from_env()
+    assert config.consolidation_max_completion_tokens == 8192
+
+
 def test_log_config_masks_database_urls(caplog):
    """Config startup logs must not expose database credentials."""
    from hindsight_api.config import HindsightConfig
@@ -371,9 +371,7 @@ class TestRecoverConsolidation:
                    mem_id,
                )

-        result = await memory_no_llm_verify.retry_failed_consolidation(
-            bank_id, request_context=request_context
-        )
+        result = await memory_no_llm_verify.retry_failed_consolidation(bank_id, request_context=request_context)

        assert result["retried_count"] == 2

@@ -394,9 +392,7 @@ class TestRecoverConsolidation:
        bank_id = f"test-recover-zero-{uuid.uuid4().hex[:8]}"
        await memory_no_llm_verify.get_bank_profile(bank_id=bank_id, request_context=request_context)

-        result = await memory_no_llm_verify.retry_failed_consolidation(
-            bank_id, request_context=request_context
-        )
+        result = await memory_no_llm_verify.retry_failed_consolidation(bank_id, request_context=request_context)

        assert result["retried_count"] == 0

@@ -410,14 +406,10 @@ class TestRecoverConsolidation:

        async with memory_no_llm_verify._pool.acquire() as conn:
            (mem_id,) = await _insert_memories(conn, bank_id, ["Grace is an expert rock climber."])
-            await conn.execute(
-                "UPDATE memory_units SET consolidation_failed_at = NOW() WHERE id = $1", mem_id
-            )
+            await conn.execute("UPDATE memory_units SET consolidation_failed_at = NOW() WHERE id = $1", mem_id)

        # Recover
-        recover_result = await memory_no_llm_verify.retry_failed_consolidation(
-            bank_id, request_context=request_context
-        )
+        recover_result = await memory_no_llm_verify.retry_failed_consolidation(bank_id, request_context=request_context)
        assert recover_result["retried_count"] == 1

        # Now consolidate with a healthy LLM
@@ -460,9 +452,7 @@ class TestRecoverConsolidation:
                ["Henry is a professional chef.", "Henry trained at Le Cordon Bleu."],
            )
            for mem_id in ids:
-                await conn.execute(
-                    "UPDATE memory_units SET consolidation_failed_at = NOW() WHERE id = $1", mem_id
-                )
+                await conn.execute("UPDATE memory_units SET consolidation_failed_at = NOW() WHERE id = $1", mem_id)

        app = create_app(memory_no_llm_verify, initialize_memory=False)
        transport = httpx.ASGITransport(app=app)
@@ -80,9 +80,7 @@ async def _pending_consolidation_ops(memory, bank_id: str) -> list[str]:


@pytest.mark.asyncio
-async def test_round_limited_consolidation_leaves_followup_pending_op(
-    memory: MemoryEngine, request_context
-):
+async def test_round_limited_consolidation_leaves_followup_pending_op(memory: MemoryEngine, request_context):
    """A round-limited consolidation must leave a new ``pending`` consolidation
    op in ``async_operations`` for the same bank so the worker poller can
    drain the backlog without external intervention."""
@@ -147,9 +145,7 @@ async def test_round_limited_consolidation_leaves_followup_pending_op(
            op_id,
        )
    assert row is not None
-    assert row["status"] == "completed", (
-        f"first consolidation op should be marked completed, got {row['status']}"
-    )
+    assert row["status"] == "completed", f"first consolidation op should be marked completed, got {row['status']}"

    # 4. Backlog must remain (round limit kept one round under the total)
    unconsolidated_after = await _count_unconsolidated(memory, bank_id)
@@ -169,8 +165,6 @@ async def test_round_limited_consolidation_leaves_followup_pending_op(
        f"backlog. Found {len(pending_ops)} pending ops; backlog still has "
        f"{unconsolidated_after} unconsolidated memory_units."
    )
-    assert pending_ops[0] != str(op_id), (
-        "The pending op must be a NEW row, not the original op we just executed."
-    )
+    assert pending_ops[0] != str(op_id), "The pending op must be a NEW row, not the original op we just executed."

    await memory.delete_bank(bank_id, request_context=request_context)
@@ -1,9 +1,9 @@
 """Tests for consolidation retry budget configurability (issue #1042)."""

-import pytest
-
 from unittest.mock import AsyncMock, MagicMock

+import pytest
+
 from hindsight_api.engine.consolidation.consolidator import _consolidate_batch_with_llm


@@ -24,6 +24,7 @@ def mock_config():
    config.observations_mission = None
    config.consolidation_max_attempts = 3
    config.consolidation_llm_max_retries = None
+    config.consolidation_max_completion_tokens = None
    return config


@@ -68,6 +69,32 @@ class TestConsolidationRetryBudget:
        )
        assert mock_llm_config.call.call_args.kwargs.get("max_retries") == 3

+    @pytest.mark.asyncio
+    async def test_max_completion_tokens_threaded_to_call(self, mock_llm_config, mock_config):
+        """consolidation_max_completion_tokens is passed to llm_config.call()."""
+        mock_config.consolidation_max_completion_tokens = 8192
+        await _consolidate_batch_with_llm(
+            llm_config=mock_llm_config,
+            memories=[{"id": "m1", "text": "test"}],
+            union_observations=[],
+            union_source_facts={},
+            config=mock_config,
+        )
+        assert mock_llm_config.call.call_args.kwargs.get("max_completion_tokens") == 8192
+
+    @pytest.mark.asyncio
+    async def test_max_completion_tokens_not_passed_when_none(self, mock_llm_config, mock_config):
+        """When consolidation_max_completion_tokens is None, max_completion_tokens is omitted (no regression)."""
+        mock_config.consolidation_max_completion_tokens = None
+        await _consolidate_batch_with_llm(
+            llm_config=mock_llm_config,
+            memories=[{"id": "m1", "text": "test"}],
+            union_observations=[],
+            union_source_facts={},
+            config=mock_config,
+        )
+        assert "max_completion_tokens" not in mock_llm_config.call.call_args.kwargs
+
    @pytest.mark.asyncio
    async def test_max_retries_not_passed_when_none(self, mock_llm_config, mock_config):
        """When consolidation_llm_max_retries is None, max_retries is not passed."""
@@ -230,8 +230,7 @@ async def test_backoff_matches_schedule_by_retry_count(memory, retry_count):

    delta = (excinfo.value.retry_at - before).total_seconds()
    assert expected_backoff <= delta <= expected_backoff + 10, (
-        f"retry_count={retry_count}: expected backoff ~{expected_backoff}s, "
-        f"got delta={delta:.2f}s"
+        f"retry_count={retry_count}: expected backoff ~{expected_backoff}s, got delta={delta:.2f}s"
    )

    await _cleanup(pool, bank_id, op_id)
@@ -266,8 +265,6 @@ async def test_retry_is_indefinite(memory):

    delta = (excinfo.value.retry_at - before).total_seconds()
    cap = _CONSOLIDATION_RETRY_BACKOFF_MAX_SECONDS
-    assert cap <= delta <= cap + 10, (
-        f"At retry_count=100 expected backoff at cap (~{cap}s), got {delta:.2f}s"
-    )
+    assert cap <= delta <= cap + 10, f"At retry_count=100 expected backoff at cap (~{cap}s), got {delta:.2f}s"

    await _cleanup(pool, bank_id, op_id)
@@ -77,9 +77,7 @@ async def test_round_limit_caps_processed_memories(memory: MemoryEngine, request
    assert result["memories_processed"] <= round_limit

    # Must have re-queued consolidation for remaining work
-    mock_requeue.assert_called_once_with(
-        bank_id=bank_id, request_context=request_context, observation_scopes=None
-    )
+    mock_requeue.assert_called_once_with(bank_id=bank_id, request_context=request_context, observation_scopes=None)

    # Mental model refresh should be skipped on intermediate round
    assert result.get("mental_models_refreshed", 0) == 0
@@ -113,10 +113,7 @@ def _mock_llm_one_obs_per_fact():
        # example UUIDs in its OUTPUT samples — read user only.
        prompt = "\n".join(m.get("content", "") for m in messages if m.get("role") == "user")
        fact_ids = re.findall(r"\[([0-9a-f-]{36})\]", prompt)
-        creates = [
-            _CreateAction(text=f"Observation about fact {fid[:8]}", source_fact_ids=[fid])
-            for fid in fact_ids
-        ]
+        creates = [_CreateAction(text=f"Observation about fact {fid[:8]}", source_fact_ids=[fid]) for fid in fact_ids]
        return _ConsolidationBatchResponse(creates=creates)

    mock_llm.set_response_callback(callback)
@@ -170,11 +167,13 @@ async def test_combined_mode_parallel_writes_to_memory_tag_set(memory: MemoryEng

        assert result["status"] == "completed"
        tag_sets = _ag_sorted(await _fetch_observation_tag_sets(memory, bank_id))
-        assert tag_sets == _ag_sorted([
-            frozenset({"user:alice"}),
-            frozenset({"user:bob"}),
-            frozenset({"user:carol"}),
-        ])
+        assert tag_sets == _ag_sorted(
+            [
+                frozenset({"user:alice"}),
+                frozenset({"user:bob"}),
+                frozenset({"user:carol"}),
+            ]
+        )
    finally:
        await memory.delete_bank(bank_id, request_context=request_context)

@@ -373,8 +372,7 @@ async def test_overlapping_scopes_serialise_under_parallelism(memory: MemoryEngi
        # The whole point: lock invariant per scope.
        for scope, peak in max_concurrent.items():
            assert peak <= 1, (
-                f"scope {set(scope) or '<untagged>'} had {peak} concurrent in-flight recalls; "
-                "lock invariant violated"
+                f"scope {set(scope) or '<untagged>'} had {peak} concurrent in-flight recalls; lock invariant violated"
            )
        # Sanity: we DID see recalls for the shared scope, so the test wasn't trivial.
        assert frozenset({"a"}) in max_concurrent
@@ -421,9 +419,7 @@ async def test_per_batch_log_line_attributes_only_own_work(memory: MemoryEngine,
                patch.object(memory, "submit_async_consolidation"),
                caplog.at_level(logging.INFO, logger="hindsight_api.engine.consolidation.consolidator"),
            ):
-                await run_consolidation_job(
-                    memory_engine=memory, bank_id=bank_id, request_context=request_context
-                )
+                await run_consolidation_job(memory_engine=memory, bank_id=bank_id, request_context=request_context)
        finally:
            memory._consolidation_llm_config = original_llm

@@ -452,9 +448,7 @@ async def test_per_batch_log_line_attributes_only_own_work(memory: MemoryEngine,
        assert processed_values == sorted(processed_values), (
            f"processed counter must be monotonic, got {processed_values}"
        )
-        assert max(processed_values) == 3, (
-            f"final cumulative processed should be 3, got {max(processed_values)}"
-        )
+        assert max(processed_values) == 3, f"final cumulative processed should be 3, got {max(processed_values)}"
        assert set(processed_values) == {1, 2, 3}, (
            f"each batch should bump the counter by exactly 1, got {processed_values}"
        )
@@ -466,9 +460,7 @@ async def test_per_batch_log_line_attributes_only_own_work(memory: MemoryEngine,
            assert m_llm_time, f"expected llm=Xs timing, got: {line}"
            # Sanity: a single mock-LLM call is fast — under a second easily.
            # If snapshot leaked, this would catch concurrent batches' LLM time too.
-            assert float(m_llm_time.group(1)) < 5.0, (
-                f"llm timing implausibly large for a single mock-LLM call: {line}"
-            )
+            assert float(m_llm_time.group(1)) < 5.0, f"llm timing implausibly large for a single mock-LLM call: {line}"
    finally:
        await memory.delete_bank(bank_id, request_context=request_context)

@@ -71,9 +71,7 @@ async def _count_unconsolidated(memory, bank_id: str) -> int:


@pytest.mark.asyncio
-async def test_requeue_failure_propagates_to_worker_retry(
-    memory: MemoryEngine, request_context
-):
+async def test_requeue_failure_propagates_to_worker_retry(memory: MemoryEngine, request_context):
    """When the in-task ``submit_async_consolidation`` call raises, the op
    must NOT be silently completed. The consolidator's work for this round
    is durably committed (memories marked consolidated_at in their own
@@ -183,8 +181,6 @@ async def test_requeue_failure_propagates_to_worker_retry(
        f"unconsolidated_remaining={unconsolidated_after}"
    )

-    assert call_count["n"] == 1, (
-        f"only one in-task submit_async_consolidation call expected, got {call_count['n']}"
-    )
+    assert call_count["n"] == 1, f"only one in-task submit_async_consolidation call expected, got {call_count['n']}"

    await memory.delete_bank(bank_id, request_context=request_context)
@@ -63,10 +63,7 @@ async def test_concurrent_submits_leave_one_pending(memory, request_context, no_
    await _ensure_bank(pool, bank_id)
    try:
        results = await asyncio.gather(
-            *(
-                memory.submit_async_consolidation(bank_id=bank_id, request_context=request_context)
-                for _ in range(5)
-            )
+            *(memory.submit_async_consolidation(bank_id=bank_id, request_context=request_context) for _ in range(5))
        )
        assert await _count_pending(pool, bank_id) == 1
        op_ids = {r["operation_id"] for r in results}
@@ -287,7 +287,9 @@ class TestEmbeddingDimension:
        # Try to change dimension - should raise RuntimeError.
        # Retry on transient OID errors from concurrent xdist schema drops.
        _assert_raises_runtime_error_with_retry(
-            db_url, 768, schema,
+            db_url,
+            768,
+            schema,
            expected_messages=["Cannot change embedding dimension", "1 rows with embeddings"],
        )

@@ -332,7 +334,9 @@ class TestEmbeddingDimension:
        # Try to change dimension - should raise RuntimeError.
        # Retry on transient OID errors from concurrent xdist schema drops.
        _assert_raises_runtime_error_with_retry(
-            db_url, 768, schema,
+            db_url,
+            768,
+            schema,
            expected_messages=["Cannot change embedding dimension", "mental_models"],
        )

@@ -193,18 +193,28 @@ class TestPostgreSQLDialect:

    def test_build_semantic_arm(self, d):
        arm = d.build_semantic_arm(
-            table="schema.memory_units", cols="id, text", fact_type="world",
-            embedding_param="$1", bank_id_param="$2", fetch_limit=100,
+            table="schema.memory_units",
+            cols="id, text",
+            fact_type="world",
+            embedding_param="$1",
+            bank_id_param="$2",
+            fetch_limit=100,
+            min_similarity=0.58,
        )
        assert "1 - (embedding <=> $1::vector)" in arm
+        assert ">= 0.58" in arm
        assert "fact_type = 'world'" in arm
        assert "LIMIT 100" in arm
        assert "'semantic' AS source" in arm

    def test_build_bm25_arm_native(self, d):
        arm = d.build_bm25_arm(
-            table="schema.memory_units", cols="id, text", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="schema.memory_units",
+            cols="id, text",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
        )
        assert "ts_rank_cd" in arm
        assert "to_tsquery" in arm
@@ -215,8 +225,12 @@ class TestPostgreSQLDialect:

    def test_build_bm25_arm_native_uses_configured_language(self, d):
        arm = d.build_bm25_arm(
-            table="schema.memory_units", cols="id, text", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="schema.memory_units",
+            cols="id, text",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
            bm25_language="french",
        )
        # Both the score and the WHERE filter must use the configured dictionary
@@ -225,8 +239,12 @@ class TestPostgreSQLDialect:

    def test_build_bm25_arm_vchord(self, d):
        arm = d.build_bm25_arm(
-            table="t", cols="id", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="t",
+            cols="id",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
            text_search_extension="vchord",
        )
        assert "to_bm25query" in arm
@@ -239,16 +257,26 @@ class TestPostgreSQLDialect:
        rows with a genuine query-term match, mirroring native tsvector's `@@`.
        """
        arm = d.build_bm25_arm(
-            table="t", cols="id", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="t",
+            cols="id",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
            text_search_extension="vchord",
        )
-        assert "-(search_vector <&> to_bm25query('idx_memory_units_text_search', tokenize($4, 'llmlingua2'))) > 0" in arm
+        assert (
+            "-(search_vector <&> to_bm25query('idx_memory_units_text_search', tokenize($4, 'llmlingua2'))) > 0" in arm
+        )

    def test_build_bm25_arm_vchord_honors_custom_min_score(self, d):
        arm = d.build_bm25_arm(
-            table="t", cols="id", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="t",
+            cols="id",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
            text_search_extension="vchord",
            bm25_min_score=2.5,
        )
@@ -256,8 +284,12 @@ class TestPostgreSQLDialect:

    def test_build_bm25_arm_pgroonga(self, d):
        arm = d.build_bm25_arm(
-            table="schema.memory_units", cols="id, text", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="schema.memory_units",
+            cols="id, text",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
            text_search_extension="pgroonga",
        )
        # pgroonga uses the &@~ operator + pgroonga_score for ranking. Escape
@@ -271,8 +303,12 @@ class TestPostgreSQLDialect:
    def test_build_bm25_arm_pgroonga_ignores_bm25_language(self, d):
        """pgroonga's tokenizer is fixed at index creation; bm25_language must not leak in."""
        arm = d.build_bm25_arm(
-            table="t", cols="id", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="t",
+            cols="id",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
            text_search_extension="pgroonga",
            bm25_language="french",
        )
@@ -280,8 +316,12 @@ class TestPostgreSQLDialect:

    def test_build_bm25_arm_pg_search(self, d):
        arm = d.build_bm25_arm(
-            table="schema.memory_units", cols="id, text", fact_type="world",
-            bank_id_param="$2", limit_param="$3", text_param="$4",
+            table="schema.memory_units",
+            cols="id, text",
+            fact_type="world",
+            bank_id_param="$2",
+            limit_param="$3",
+            text_param="$4",
            text_search_extension="pg_search",
        )
        assert "paradedb.score(id)" in arm
@@ -359,18 +399,28 @@ class TestOracleDialect:

    def test_build_semantic_arm(self, d):
        arm = d.build_semantic_arm(
-            table="memory_units", cols="id, text", fact_type="world",
-            embedding_param=":1", bank_id_param=":2", fetch_limit=100,
+            table="memory_units",
+            cols="id, text",
+            fact_type="world",
+            embedding_param=":1",
+            bank_id_param=":2",
+            fetch_limit=100,
+            min_similarity=0.58,
        )
        assert "VECTOR_DISTANCE" in arm
+        assert ">= 0.58" in arm
        assert "fact_type = 'world'" in arm
        assert "FETCH FIRST 100 ROWS ONLY" in arm
        assert "'semantic' AS source" in arm

    def test_build_bm25_arm(self, d):
        arm = d.build_bm25_arm(
-            table="memory_units", cols="id, text", fact_type="world",
-            bank_id_param=":2", limit_param=":3", text_param=":4",
+            table="memory_units",
+            cols="id, text",
+            fact_type="world",
+            bank_id_param=":2",
+            limit_param=":3",
+            text_param=":4",
            arm_index=0,
        )
        assert "CONTAINS" in arm
@@ -381,12 +431,22 @@ class TestOracleDialect:
    def test_build_bm25_arm_unique_labels(self, d):
        """Each arm_index produces a unique SCORE label to avoid conflicts in UNION ALL."""
        arm0 = d.build_bm25_arm(
-            table="t", cols="id", fact_type="world",
-            bank_id_param=":2", limit_param=":3", text_param=":4", arm_index=0,
+            table="t",
+            cols="id",
+            fact_type="world",
+            bank_id_param=":2",
+            limit_param=":3",
+            text_param=":4",
+            arm_index=0,
        )
        arm1 = d.build_bm25_arm(
-            table="t", cols="id", fact_type="experience",
-            bank_id_param=":2", limit_param=":3", text_param=":4", arm_index=1,
+            table="t",
+            cols="id",
+            fact_type="experience",
+            bank_id_param=":2",
+            limit_param=":3",
+            text_param=":4",
+            arm_index=1,
        )
        assert "SCORE(10)" in arm0
        assert "SCORE(11)" in arm1
@@ -472,9 +532,7 @@ class TestOracleQueryRewriter:
        """Verify JSONB ->> boolean comparison is rewritten to JSON_VALUE."""
        from hindsight_api.engine.db.oracle import _rewrite_pg_to_oracle

-        query, _, _ = _rewrite_pg_to_oracle(
-            "WHERE (trigger->>'refresh_after_consolidation')::boolean = true"
-        )
+        query, _, _ = _rewrite_pg_to_oracle("WHERE (trigger->>'refresh_after_consolidation')::boolean = true")
        assert "JSON_VALUE" in query
        assert "'true'" in query
        assert "->>" not in query
@@ -491,9 +549,7 @@ class TestOracleQueryRewriter:
        """Verify ->> works with quoted column names."""
        from hindsight_api.engine.db.oracle import _rewrite_pg_to_oracle

-        query, _, _ = _rewrite_pg_to_oracle(
-            "ORDER BY (result_metadata->>'sub_batch_index')::int"
-        )
+        query, _, _ = _rewrite_pg_to_oracle("ORDER BY (result_metadata->>'sub_batch_index')::int")
        assert "JSON_VALUE" in query
        assert "->>" not in query

@@ -679,9 +735,7 @@ class TestOracleOpsInsertFactsBatch:
    @pytest.mark.asyncio
    async def test_tags_json_decoded_to_list(self, ops, mock_conn):
        """Tags JSON strings must be decoded to Python lists, not passed as strings."""
-        await ops.insert_facts_batch(
-            conn=mock_conn, **{**self._make_batch(1), "tags_list": ['["tag1", "tag2"]']}
-        )
+        await ops.insert_facts_batch(conn=mock_conn, **{**self._make_batch(1), "tags_list": ['["tag1", "tag2"]']})
        _, rows_data = mock_conn.executemany.call_args.args
        assert rows_data[0][13] == ["tag1", "tag2"]
        assert isinstance(rows_data[0][13], list)
@@ -689,9 +743,7 @@ class TestOracleOpsInsertFactsBatch:
    @pytest.mark.asyncio
    async def test_empty_tags_becomes_empty_list(self, ops, mock_conn):
        """Empty/falsy tags string must become [], not crash or pass empty string."""
-        await ops.insert_facts_batch(
-            conn=mock_conn, **{**self._make_batch(1), "tags_list": [""]}
-        )
+        await ops.insert_facts_batch(conn=mock_conn, **{**self._make_batch(1), "tags_list": [""]})
        _, rows_data = mock_conn.executemany.call_args.args
        assert rows_data[0][13] == []

@@ -36,16 +36,10 @@ class TestPassthrough:

 class TestSchemeNormalization:
    def test_asyncpg_scheme_stripped(self) -> None:
-        assert (
-            to_libpq_url("postgresql+asyncpg://user:pass@host:5432/db")
-            == "postgresql://user:pass@host:5432/db"
-        )
+        assert to_libpq_url("postgresql+asyncpg://user:pass@host:5432/db") == "postgresql://user:pass@host:5432/db"

    def test_postgres_asyncpg_scheme_normalized(self) -> None:
-        assert (
-            to_libpq_url("postgres+asyncpg://user:pass@host/db")
-            == "postgresql://user:pass@host/db"
-        )
+        assert to_libpq_url("postgres+asyncpg://user:pass@host/db") == "postgresql://user:pass@host/db"

    def test_bare_postgres_scheme_normalized_to_postgresql(self) -> None:
        assert to_libpq_url("postgres://user:pass@host/db") == "postgresql://user:pass@host/db"
@@ -68,9 +62,7 @@ class TestSslParamRename:
        assert to_libpq_url("postgresql://h/d?ssl=require") == "postgresql://h/d?sslmode=require"

    def test_ssl_param_preserved_among_other_params(self) -> None:
-        result = to_libpq_url(
-            "postgresql+asyncpg://h/d?ssl=require&application_name=hindsight&connect_timeout=10"
-        )
+        result = to_libpq_url("postgresql+asyncpg://h/d?ssl=require&application_name=hindsight&connect_timeout=10")
        assert result.startswith("postgresql://h/d?")
        # Query order should be preserved; ssl renamed, others untouched.
        assert "sslmode=require" in result
@@ -80,10 +72,7 @@ class TestSslParamRename:

    def test_sslmode_not_double_renamed(self) -> None:
        """An already-correct sslmode= param must not be altered."""
-        assert (
-            to_libpq_url("postgresql+asyncpg://h/d?sslmode=require")
-            == "postgresql://h/d?sslmode=require"
-        )
+        assert to_libpq_url("postgresql+asyncpg://h/d?sslmode=require") == "postgresql://h/d?sslmode=require"


 class TestProductionConfigs:
@@ -132,10 +121,7 @@ class TestEdgeCases:
        assert result == "postgresql://user:my%2Basyncpgpass@host/db"

    def test_url_without_query_string(self) -> None:
-        assert (
-            to_libpq_url("postgresql+asyncpg://user:pass@host/db")
-            == "postgresql://user:pass@host/db"
-        )
+        assert to_libpq_url("postgresql+asyncpg://user:pass@host/db") == "postgresql://user:pass@host/db"

    def test_url_with_port_and_path_only(self) -> None:
        assert to_libpq_url("postgresql+asyncpg://host:5432/db") == "postgresql://host:5432/db"
@@ -126,22 +126,30 @@ class TestDeltaEditorialFusion:

            # Phase 1: Ingest SEO best practices
            await memory.retain_async(
-                bank_id=bank_id, content=SEO_BEST_PRACTICES,
-                document_id="seo-best-practices", request_context=request_context,
+                bank_id=bank_id,
+                content=SEO_BEST_PRACTICES,
+                document_id="seo-best-practices",
+                request_context=request_context,
            )
            mm_after_seo = await memory.refresh_mental_model(
-                bank_id=bank_id, mental_model_id=mm_id, request_context=request_context,
+                bank_id=bank_id,
+                mental_model_id=mm_id,
+                request_context=request_context,
            )
            seo_content = mm_after_seo["content"]
            assert len(seo_content) > 100, f"First refresh produced too little content: {len(seo_content)} chars"

            # Phase 2: Ingest brand voice -> delta refresh
            await memory.retain_async(
-                bank_id=bank_id, content=BRAND_VOICE,
-                document_id="brand-voice", request_context=request_context,
+                bank_id=bank_id,
+                content=BRAND_VOICE,
+                document_id="brand-voice",
+                request_context=request_context,
            )
            mm_after_brand = await memory.refresh_mental_model(
-                bank_id=bank_id, mental_model_id=mm_id, request_context=request_context,
+                bank_id=bank_id,
+                mental_model_id=mm_id,
+                request_context=request_context,
            )
            fused = mm_after_brand["content"]
            rr = mm_after_brand.get("reflect_response") or {}
@@ -156,8 +164,7 @@ class TestDeltaEditorialFusion:
                "vocabulary rules": ["jargon", "leverage", "empower", "forbidden"],
            }.items():
                assert any(s in fused_lower for s in signals), (
-                    f"Brand voice concept '{concept}' missing (looked for {signals}).\n"
-                    f"Fused content:\n{fused[:500]}"
+                    f"Brand voice concept '{concept}' missing (looked for {signals}).\nFused content:\n{fused[:500]}"
                )

            # SEO concepts still present (not wiped by delta)
@@ -167,8 +174,7 @@ class TestDeltaEditorialFusion:
                "seo": ["meta", "e-e-a-t", "seo", "search"],
            }.items():
                assert any(s in fused_lower for s in signals), (
-                    f"SEO concept '{concept}' missing (looked for {signals}).\n"
-                    f"Fused content:\n{fused[:500]}"
+                    f"SEO concept '{concept}' missing (looked for {signals}).\nFused content:\n{fused[:500]}"
                )

            # Brand voice overrides generic tone
@@ -177,15 +183,9 @@ class TestDeltaEditorialFusion:
            )

            # No duplicate paragraphs
-            lines = [
-                ln.strip() for ln in fused.split("\n")
-                if ln.strip() and not ln.strip().startswith("#")
-            ]
+            lines = [ln.strip() for ln in fused.split("\n") if ln.strip() and not ln.strip().startswith("#")]
            dupes = {line: cnt for line, cnt in Counter(lines).items() if cnt > 1}
-            assert not dupes, (
-                "Duplicate paragraphs:\n" +
-                "\n".join(f"  [{c}x] {t[:80]}" for t, c in dupes.items())
-            )
+            assert not dupes, "Duplicate paragraphs:\n" + "\n".join(f"  [{c}x] {t[:80]}" for t, c in dupes.items())

            # based_on accumulates from both docs
            obs_count = len(rr.get("based_on", {}).get("observation", []))
@@ -146,7 +146,10 @@ async def test_delta_retain_appended_content(memory, request_context):

        # Second version — original content + new content appended
        # This should preserve facts from the first chunk and add new ones
-        v2_content = v1_content + "\n\nBob joined Google as a product manager in 2024. He previously worked at Meta on AR/VR products."
+        v2_content = (
+            v1_content
+            + "\n\nBob joined Google as a product manager in 2024. He previously worked at Meta on AR/VR products."
+        )

        v2_units = await memory.retain_async(
            bank_id=bank_id,
@@ -387,9 +390,7 @@ async def test_delta_retain_links_preserved_for_unchanged_chunks(memory, request
                document_id,
            )

-        assert v2_link_count == v1_link_count, (
-            f"Links should be preserved: v1={v1_link_count}, v2={v2_link_count}"
-        )
+        assert v2_link_count == v1_link_count, f"Links should be preserved: v1={v1_link_count}, v2={v2_link_count}"

    finally:
        await memory.delete_bank(bank_id, request_context=request_context)
@@ -456,11 +457,13 @@ async def test_delta_retain_tags_propagated_to_existing_units(memory, request_co
        # v1 with tag "team-a"
        await memory.retain_batch_async(
            bank_id=bank_id,
-            contents=[{
-                "content": content,
-                "document_id": document_id,
-                "tags": ["team-a"],
-            }],
+            contents=[
+                {
+                    "content": content,
+                    "document_id": document_id,
+                    "tags": ["team-a"],
+                }
+            ],
            request_context=request_context,
        )

@@ -476,11 +479,13 @@ async def test_delta_retain_tags_propagated_to_existing_units(memory, request_co
        # v2 with same content but different tags
        await memory.retain_batch_async(
            bank_id=bank_id,
-            contents=[{
-                "content": content,
-                "document_id": document_id,
-                "tags": ["team-b", "important"],
-            }],
+            contents=[
+                {
+                    "content": content,
+                    "document_id": document_id,
+                    "tags": ["team-b", "important"],
+                }
+            ],
            request_context=request_context,
        )

@@ -692,7 +697,9 @@ async def test_delta_retain_empty_to_content(memory, request_context):

        doc_v2 = await memory.get_document(document_id, bank_id, request_context=request_context)
        assert doc_v2 is not None
-        assert doc_v2["memory_unit_count"] > 0 or len(v2_units) > 0, "Should have facts after updating with real content"
+        assert doc_v2["memory_unit_count"] > 0 or len(v2_units) > 0, (
+            "Should have facts after updating with real content"
+        )

    finally:
        await memory.delete_bank(bank_id, request_context=request_context)
@@ -775,11 +782,13 @@ async def test_delta_retain_with_user_entities(memory, request_context):
        # v1 with user entities
        await memory.retain_batch_async(
            bank_id=bank_id,
-            contents=[{
-                "content": content,
-                "document_id": document_id,
-                "entities": [{"text": "Project Alpha", "type": "PROJECT"}],
-            }],
+            contents=[
+                {
+                    "content": content,
+                    "document_id": document_id,
+                    "entities": [{"text": "Project Alpha", "type": "PROJECT"}],
+                }
+            ],
            request_context=request_context,
        )

@@ -797,14 +806,16 @@ async def test_delta_retain_with_user_entities(memory, request_context):
        v2_content = content + "\n\nThe timeline is on track for Q2 delivery."
        await memory.retain_batch_async(
            bank_id=bank_id,
-            contents=[{
-                "content": v2_content,
-                "document_id": document_id,
-                "entities": [
-                    {"text": "Project Alpha", "type": "PROJECT"},
-                    {"text": "Q2 Deadline", "type": "MILESTONE"},
-                ],
-            }],
+            contents=[
+                {
+                    "content": v2_content,
+                    "document_id": document_id,
+                    "entities": [
+                        {"text": "Project Alpha", "type": "PROJECT"},
+                        {"text": "Q2 Deadline", "type": "MILESTONE"},
+                    ],
+                }
+            ],
            request_context=request_context,
        )

@@ -867,9 +878,7 @@ async def test_delta_retain_recall_with_chunks(memory, request_context):
        facts_with_chunks = [r for r in result.results if r.chunk_id]
        if facts_with_chunks and result.chunks:
            for fact in facts_with_chunks:
-                assert fact.chunk_id in result.chunks, (
-                    f"Chunk {fact.chunk_id} should be in returned chunks"
-                )
+                assert fact.chunk_id in result.chunks, f"Chunk {fact.chunk_id} should be in returned chunks"

    finally:
        await memory.delete_bank(bank_id, request_context=request_context)
@@ -1024,8 +1033,7 @@ async def test_processed_content_tokens_appended_reports_delta(memory, request_c
            return
        assert second > 0, "Partial-delta retain should report a positive token count"
        assert second < submitted_tokens, (
-            "Partial-delta retain should report fewer processed tokens "
-            "than the full submitted payload"
+            "Partial-delta retain should report fewer processed tokens than the full submitted payload"
        )
    finally:
        memory._operation_validator = None
@@ -143,9 +143,7 @@ async def test_delta_detects_unchanged_after_first_retain(memory, request_contex
                bank_id,
                document_id,
            )
-        assert v2_count == v1_count, (
-            f"Memory unit count changed on same-content upsert: {v1_count} -> {v2_count}"
-        )
+        assert v2_count == v1_count, f"Memory unit count changed on same-content upsert: {v1_count} -> {v2_count}"

        # Third retain — verify stability
        v3_units = await memory.retain_async(
@@ -163,9 +161,7 @@ async def test_delta_detects_unchanged_after_first_retain(memory, request_contex
                bank_id,
                document_id,
            )
-        assert v3_count == v1_count, (
-            f"Memory unit count changed on third upsert: {v1_count} -> {v3_count}"
-        )
+        assert v3_count == v1_count, f"Memory unit count changed on third upsert: {v1_count} -> {v3_count}"

    finally:
        await memory.delete_bank(bank_id, request_context=request_context)
@@ -371,9 +367,7 @@ async def test_concurrent_upserts_no_duplicates(memory_no_llm, request_context):
        # splitter may cut mid-text, so later chunks might not start with the prefix.
        winning_person = f"Person_{winning_version}"
        wrong_version_units = [
-            (r["text"], r["chunk_id"], r["unit_id"])
-            for r in units
-            if winning_person not in r["text"]
+            (r["text"], r["chunk_id"], r["unit_id"]) for r in units if winning_person not in r["text"]
        ]
        assert not wrong_version_units, (
            f"Found {len(wrong_version_units)} memory units NOT from winning version "
@@ -397,8 +391,7 @@ async def test_concurrent_upserts_no_duplicates(memory_no_llm, request_context):
        )

        logger.info(
-            f"Concurrent test passed: version {winning_version} won with "
-            f"{len(unit_texts)} memory units, no duplicates"
+            f"Concurrent test passed: version {winning_version} won with {len(unit_texts)} memory units, no duplicates"
        )

    finally:
@@ -1,6 +1,7 @@
 """
 Tests for document chunks API, reprocess, nodes_by_fact_type, and graph document/chunk filtering.
 """
+
 from datetime import datetime, timezone

 import httpx
@@ -223,9 +224,7 @@ async def test_graph_chunk_id_filter(api_client, bank_id):
    await _retain(api_client, bank_id, "doc-chunk-test", "Alice works at Google. " * 20)

    # First get chunks to find a valid chunk_id
-    response = await api_client.get(
-        f"/v1/default/banks/{bank_id}/documents/doc-chunk-test/chunks"
-    )
+    response = await api_client.get(f"/v1/default/banks/{bank_id}/documents/doc-chunk-test/chunks")
    assert response.status_code == 200
    chunks_data = response.json()
    if chunks_data["total"] == 0:
@@ -251,11 +250,14 @@ async def test_graph_chunk_id_filter(api_client, bank_id):
@pytest.mark.asyncio
 async def test_http_list_document_chunks(api_client, bank_id):
    """HTTP GET .../documents/{id}/chunks returns chunks."""
-    await _retain(api_client, bank_id, "doc-http-chunks", "Alice works at Google on AI research. Bob works at Meta on VR systems. " * 20)
-
-    response = await api_client.get(
-        f"/v1/default/banks/{bank_id}/documents/doc-http-chunks/chunks"
+    await _retain(
+        api_client,
+        bank_id,
+        "doc-http-chunks",
+        "Alice works at Google on AI research. Bob works at Meta on VR systems. " * 20,
    )
+
+    response = await api_client.get(f"/v1/default/banks/{bank_id}/documents/doc-http-chunks/chunks")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
@@ -266,9 +268,7 @@ async def test_http_list_document_chunks(api_client, bank_id):
@pytest.mark.asyncio
 async def test_http_list_document_chunks_not_found(api_client, bank_id):
    """HTTP GET .../documents/{id}/chunks returns 404 for non-existent document."""
-    response = await api_client.get(
-        f"/v1/default/banks/{bank_id}/documents/nonexistent/chunks"
-    )
+    response = await api_client.get(f"/v1/default/banks/{bank_id}/documents/nonexistent/chunks")
    assert response.status_code == 404


@@ -277,9 +277,7 @@ async def test_http_reprocess_document(api_client, bank_id):
    """HTTP POST .../documents/{id}/reprocess returns success with operation_id."""
    await _retain(api_client, bank_id, "doc-http-reprocess", "Alice works at Google.")

-    response = await api_client.post(
-        f"/v1/default/banks/{bank_id}/documents/doc-http-reprocess/reprocess"
-    )
+    response = await api_client.post(f"/v1/default/banks/{bank_id}/documents/doc-http-reprocess/reprocess")
    assert response.status_code == 200
    data = response.json()
    assert data["success"] is True
@@ -289,9 +287,7 @@ async def test_http_reprocess_document(api_client, bank_id):
@pytest.mark.asyncio
 async def test_http_reprocess_document_not_found(api_client, bank_id):
    """HTTP POST .../documents/{id}/reprocess returns 404 for non-existent document."""
-    response = await api_client.post(
-        f"/v1/default/banks/{bank_id}/documents/nonexistent/reprocess"
-    )
+    response = await api_client.post(f"/v1/default/banks/{bank_id}/documents/nonexistent/reprocess")
    assert response.status_code == 404


@@ -300,9 +296,7 @@ async def test_http_get_document_includes_nodes_by_fact_type(api_client, bank_id
    """HTTP GET .../documents/{id} includes nodes_by_fact_type."""
    await _retain(api_client, bank_id, "doc-http-comp", "Alice works at Google on AI research.")

-    response = await api_client.get(
-        f"/v1/default/banks/{bank_id}/documents/doc-http-comp"
-    )
+    response = await api_client.get(f"/v1/default/banks/{bank_id}/documents/doc-http-comp")
    assert response.status_code == 200
    data = response.json()
    assert "nodes_by_fact_type" in data
@@ -1,6 +1,7 @@
 """
 Tests for document tracking and upsert functionality.
 """
+
 import logging
 from datetime import datetime, timezone
 from unittest.mock import patch
@@ -357,9 +358,7 @@ async def test_document_persisted_with_zero_facts_async_submit(memory_real_llm,
            elapsed += wait_interval

            # Check if document exists
-            doc = await memory.get_document(
-                "doc-async-zero-facts", bank_id, request_context=request_context
-            )
+            doc = await memory.get_document("doc-async-zero-facts", bank_id, request_context=request_context)
            if doc is not None:
                break

@@ -86,9 +86,7 @@ async def _import(memory, bank_id, archive, request_context, on_conflict="skip")
    inline and is already completed when submit returns.
    """
    submission = await memory.import_documents_async(bank_id, archive, request_context, on_conflict)
-    status = await memory.get_operation_status(
-        bank_id, submission["operation_id"], request_context=request_context
-    )
+    status = await memory.get_operation_status(bank_id, submission["operation_id"], request_context=request_context)
    assert status["status"] == "completed", status
    return status["result_metadata"]

@@ -340,12 +338,8 @@ async def test_bank_roundtrip_carries_mental_model_history(memory, request_conte
            mental_model_id="mm-1",
            request_context=request_context,
        )
-        await memory.update_mental_model(
-            bank, mental_model_id="mm-1", content="v2", request_context=request_context
-        )
-        await memory.update_mental_model(
-            bank, mental_model_id="mm-1", content="v3", request_context=request_context
-        )
+        await memory.update_mental_model(bank, mental_model_id="mm-1", content="v2", request_context=request_context)
+        await memory.update_mental_model(bank, mental_model_id="mm-1", content="v3", request_context=request_context)
        # Two refreshes → two snapshots (previous content v1 then v2), newest-first.
        before = await memory.get_mental_model_history(bank, "mm-1", request_context=request_context)
        assert [h["previous_content"] for h in before] == ["v2", "v1"]
@@ -475,13 +469,11 @@ async def _bank_snapshot(memory, bank_id):
    backend = await memory._get_backend()
    async with acquire_with_retry(backend) as conn:
        docs = await conn.fetch(
-            f"SELECT id, COALESCE(length(original_text), 0) AS len FROM {fq_table('documents')} "
-            f"WHERE bank_id = $1",
+            f"SELECT id, COALESCE(length(original_text), 0) AS len FROM {fq_table('documents')} WHERE bank_id = $1",
            bank_id,
        )
        chunks = await conn.fetch(
-            f"SELECT document_id, chunk_index, length(chunk_text) AS len FROM {fq_table('chunks')} "
-            f"WHERE bank_id = $1",
+            f"SELECT document_id, chunk_index, length(chunk_text) AS len FROM {fq_table('chunks')} WHERE bank_id = $1",
            bank_id,
        )
        ftypes = await conn.fetch(
@@ -591,9 +583,7 @@ async def test_export_import_observations(memory, request_context):
        backend = await memory._get_backend()
        async with acquire_with_retry(backend) as conn:
            async with conn.transaction():
-                await _create_observation_directly(
-                    conn, memory, src, source_ids, "Alice and Bob are colleagues."
-                )
+                await _create_observation_directly(conn, memory, src, source_ids, "Alice and Bob are colleagues.")

        # Export WITHOUT observations -> none in the archive (the bank may also
        # contain auto-consolidation observations; the flag is what gates them).
@@ -762,9 +752,7 @@ async def test_include_observations_requires_whole_bank_export(memory, request_c
        await _retain(memory, src, "Alice works at Google.", request_context, "doc-1")
        # Subset export (document_ids set) + observations must be rejected.
        with pytest.raises(ValueError, match="whole bank"):
-            await memory.export_documents_async(
-                src, request_context, ["doc-1"], include_observations=True
-            )
+            await memory.export_documents_async(src, request_context, ["doc-1"], include_observations=True)
        # Whole-bank export with observations is fine; subset without observations is fine.
        await memory.export_documents_async(src, request_context, include_observations=True)
        await memory.export_documents_async(src, request_context, ["doc-1"])
@@ -80,11 +80,7 @@ def test_parse_entity_labels_dict_format():

 def test_parse_entity_labels_dict_format_defaults():
    """Dict format parses attributes correctly."""
-    raw = {
-        "attributes": [
-            {"key": "topic", "values": [{"value": "math", "description": "Mathematics"}]}
-        ]
-    }
+    raw = {"attributes": [{"key": "topic", "values": [{"value": "math", "description": "Mathematics"}]}]}
    result = parse_entity_labels(raw)
    assert result is not None
    assert len(result.attributes) == 1
@@ -178,9 +174,7 @@ def test_build_labels_model_free_values_optional():
    """type='text', optional=True → str | None field."""
    from hindsight_api.engine.retain.entity_labels import build_labels_model

-    labels_cfg = EntityLabelsConfig(
-        attributes=[LabelGroup(key="topic", type="text", optional=True, values=[])]
-    )
+    labels_cfg = EntityLabelsConfig(attributes=[LabelGroup(key="topic", type="text", optional=True, values=[])])
    Model = build_labels_model(labels_cfg)
    assert Model is not None
    schema = Model.model_json_schema()
@@ -194,9 +188,7 @@ def test_build_labels_model_free_values_always_optional():
    """type='text' with optional=False is still treated as str | None — always optional."""
    from hindsight_api.engine.retain.entity_labels import build_labels_model

-    labels_cfg = EntityLabelsConfig(
-        attributes=[LabelGroup(key="topic", type="text", optional=False, values=[])]
-    )
+    labels_cfg = EntityLabelsConfig(attributes=[LabelGroup(key="topic", type="text", optional=False, values=[])])
    Model = build_labels_model(labels_cfg)
    assert Model is not None
    schema = Model.model_json_schema()
@@ -210,9 +202,7 @@ def test_build_labels_model_free_values_multi_still_optional():
    """type='text' is always str | None — multi-values only applies to enum types."""
    from hindsight_api.engine.retain.entity_labels import build_labels_model

-    labels_cfg = EntityLabelsConfig(
-        attributes=[LabelGroup(key="tags", type="text", values=[])]
-    )
+    labels_cfg = EntityLabelsConfig(attributes=[LabelGroup(key="tags", type="text", values=[])])
    Model = build_labels_model(labels_cfg)
    assert Model is not None
    schema = Model.model_json_schema()
@@ -226,9 +216,7 @@ def test_build_labels_model_free_values_no_values_still_creates_field():
    """type='text' group with no values still creates a field (description holds examples)."""
    from hindsight_api.engine.retain.entity_labels import build_labels_model

-    labels_cfg = EntityLabelsConfig(
-        attributes=[LabelGroup(key="mood", type="text", values=[])]
-    )
+    labels_cfg = EntityLabelsConfig(attributes=[LabelGroup(key="mood", type="text", values=[])])
    Model = build_labels_model(labels_cfg)
    assert Model is not None
    assert "mood" in Model.model_json_schema()["properties"]
@@ -549,9 +537,7 @@ def test_label_entity_post_processing_invalid_value_ignored():
    from hindsight_api.engine.retain.entity_labels import build_labels_lookup, parse_entity_labels
    from hindsight_api.engine.retain.fact_extraction import Entity

-    labels_cfg = parse_entity_labels(
-        [{"key": "pedagogy", "values": [{"value": "scaffolding", "description": ""}]}]
-    )
+    labels_cfg = parse_entity_labels([{"key": "pedagogy", "values": [{"value": "scaffolding", "description": ""}]}])
    labels_lookup = build_labels_lookup(labels_cfg)

    labels_data = {"pedagogy": "unknown_value"}
@@ -665,9 +651,7 @@ def test_free_values_label_is_single_value():
    """type='text' groups are always single-value (str | None)."""
    from hindsight_api.engine.retain.entity_labels import build_labels_model, parse_entity_labels

-    labels_cfg = parse_entity_labels(
-        [{"key": "topic", "type": "text", "values": []}]
-    )
+    labels_cfg = parse_entity_labels([{"key": "topic", "type": "text", "values": []}])
    Model = build_labels_model(labels_cfg)
    assert Model is not None
    schema = Model.model_json_schema()
@@ -681,9 +665,7 @@ def test_free_values_label_not_in_lookup():
    """type='text' group values do NOT appear in the lookup set (no fixed vocabulary)."""
    from hindsight_api.engine.retain.entity_labels import build_labels_lookup, parse_entity_labels

-    labels_cfg = parse_entity_labels(
-        [{"key": "topic", "type": "text", "values": [{"value": "algebra"}]}]
-    )
+    labels_cfg = parse_entity_labels([{"key": "topic", "type": "text", "values": [{"value": "algebra"}]}])
    lookup = build_labels_lookup(labels_cfg)
    assert "topic:algebra" not in lookup  # example hints not added to lookup
    assert len(lookup) == 0
@@ -725,9 +707,7 @@ def test_optional_label_string_none_produces_no_entity():

    # LLM returned the string "None" instead of JSON null — must not be stored
    entity_texts = _run_label_post_processing(labels_cfg, {"engagement": "None"})
-    assert entity_texts == set(), (
-        f"String 'None' must not produce engagement:None entity, got: {entity_texts}"
-    )
+    assert entity_texts == set(), f"String 'None' must not produce engagement:None entity, got: {entity_texts}"


 def test_optional_label_null_does_not_affect_other_labels():
@@ -744,9 +724,7 @@ def test_optional_label_null_does_not_affect_other_labels():
    # engagement is null, but topic is set
    entity_texts = _run_label_post_processing(labels_cfg, {"engagement": None, "topic": "math"})
    assert "topic:math" in entity_texts, f"Expected topic:math entity, got: {entity_texts}"
-    assert not any("engagement" in t for t in entity_texts), (
-        f"engagement should not appear, got: {entity_texts}"
-    )
+    assert not any("engagement" in t for t in entity_texts), f"engagement should not appear, got: {entity_texts}"


 def test_free_form_entities_false_clears_entities():
@@ -982,9 +960,7 @@ async def test_retain_extracts_single_value_label(memory_real_llm, request_conte
            )

        entity_names = {r["canonical_name"].lower() for r in rows}
-        assert "engagement:active" in entity_names, (
-            f"Expected 'engagement:active' label entity. Got: {entity_names}"
-        )
+        assert "engagement:active" in entity_names, f"Expected 'engagement:active' label entity. Got: {entity_names}"
        # In labels-only mode, free-form entities like 'Maria' should be absent
        assert not any("maria" in n for n in entity_names), (
            f"Free-form entity 'Maria' should not appear in labels-only mode. Got: {entity_names}"
@@ -1054,9 +1030,7 @@ async def test_retain_extracts_multi_value_label(memory_real_llm, request_contex
        entity_names = {r["canonical_name"].lower() for r in rows}
        # At least one pedagogy label should be assigned
        pedagogy_labels = {n for n in entity_names if n.startswith("pedagogy:")}
-        assert len(pedagogy_labels) > 0, (
-            f"Expected at least one pedagogy:* label entity. Got: {entity_names}"
-        )
+        assert len(pedagogy_labels) > 0, f"Expected at least one pedagogy:* label entity. Got: {entity_names}"
    finally:
        await memory.delete_bank(bank_id, request_context=request_context)

@@ -1118,9 +1092,7 @@ async def test_retain_extracts_free_values_label(memory_real_llm, request_contex
        entity_names = {r["canonical_name"].lower() for r in rows}
        # A topic:* entity must exist — value is free-form so we only check the prefix
        topic_entities = {n for n in entity_names if n.startswith("topic:")}
-        assert len(topic_entities) > 0, (
-            f"Expected at least one topic:* free-value entity. Got: {entity_names}"
-        )
+        assert len(topic_entities) > 0, f"Expected at least one topic:* free-value entity. Got: {entity_names}"
        # The value must not be the literal string "none" or "null"
        assert not any(n in ("topic:none", "topic:null", "topic:n/a") for n in topic_entities), (
            f"topic entity should not be a null sentinel. Got: {topic_entities}"
@@ -1188,18 +1160,14 @@ async def test_retain_extracts_map_type_entities(memory_real_llm, request_contex
        entity_names = {r["canonical_name"].lower() for r in rows}
        # Should have person:name:* entity
        name_entities = {n for n in entity_names if n.startswith("person:name:")}
-        assert len(name_entities) > 0, (
-            f"Expected at least one person:name:* entity. Got: {entity_names}"
-        )
+        assert len(name_entities) > 0, f"Expected at least one person:name:* entity. Got: {entity_names}"
        # Name should contain "alice" somewhere
        assert any("alice" in n for n in name_entities), (
            f"Expected person:name entity containing 'alice'. Got: {name_entities}"
        )
        # Should have person:organization:* entity mentioning google
        org_entities = {n for n in entity_names if n.startswith("person:organization:")}
-        assert len(org_entities) > 0, (
-            f"Expected at least one person:organization:* entity. Got: {entity_names}"
-        )
+        assert len(org_entities) > 0, f"Expected at least one person:organization:* entity. Got: {entity_names}"
        assert any("google" in n for n in org_entities), (
            f"Expected person:organization entity containing 'google'. Got: {org_entities}"
        )
@@ -2036,9 +2004,7 @@ async def test_retain_multivalue_tag_entities_all_stored(memory_real_llm, reques

        # The core assertion from GH-1558: tags and entities should match
        # Tags show both but entities only show a subset → BUG
-        assert len(use_tags) >= 2, (
-            f"Expected at least 2 use:* tags. Got: {use_tags}"
-        )
+        assert len(use_tags) >= 2, f"Expected at least 2 use:* tags. Got: {use_tags}"
        assert len(use_entities) >= 2, (
            f"GH-1558 BUG: Expected at least 2 use:* entities in unit_entities, "
            f"but only got {len(use_entities)}: {use_entities}. "
@@ -2097,8 +2063,7 @@ async def test_retain_multivalue_tag_entities_second_retain(memory_real_llm, req
        await memory_real_llm.retain_async(
            bank_id=bank_id,
            content=(
-                "## Authentication Flow (use-001)\n\n"
-                "The authentication flow use-001 handles user login via OAuth2."
+                "## Authentication Flow (use-001)\n\nThe authentication flow use-001 handles user login via OAuth2."
            ),
            request_context=request_context,
        )
@@ -2145,9 +2110,7 @@ async def test_retain_multivalue_tag_entities_second_retain(memory_real_llm, req
        use_entities = {n for n in entity_names if n.startswith("use:")}
        use_tags = {t for t in all_tags if t.startswith("use:")}

-        assert len(use_tags) >= 2, (
-            f"Expected at least 2 use:* tags on second retain. Got: {use_tags}"
-        )
+        assert len(use_tags) >= 2, f"Expected at least 2 use:* tags on second retain. Got: {use_tags}"
        assert len(use_entities) >= 2, (
            f"GH-1558 BUG: On second retain, expected at least 2 use:* entities "
            f"but only got {len(use_entities)}: {use_entities}. "
@@ -2155,9 +2118,7 @@ async def test_retain_multivalue_tag_entities_second_retain(memory_real_llm, req
            f"Entity resolution may be merging similar names."
        )
        missing = use_tags - use_entities
-        assert len(missing) == 0, (
-            f"GH-1558 BUG: Tags present but entities missing after second retain: {missing}"
-        )
+        assert len(missing) == 0, f"GH-1558 BUG: Tags present but entities missing after second retain: {missing}"
    finally:
        await memory_real_llm.delete_bank(bank_id, request_context=request_context)

@@ -2243,9 +2204,7 @@ async def test_entity_resolution_does_not_merge_distinct_label_values(memory, re
            )

        # We should get 2 DISTINCT entity IDs, not the same ID twice
-        assert len(resolved_entity_ids) == 2, (
-            f"Expected 2 resolved entity IDs, got {len(resolved_entity_ids)}"
-        )
+        assert len(resolved_entity_ids) == 2, f"Expected 2 resolved entity IDs, got {len(resolved_entity_ids)}"
        unique_ids = set(resolved_entity_ids)
        assert len(unique_ids) == 2, (
            f"GH-1558 BUG: Entity resolution merged 'use:use-001' and 'use:use-002' "
@@ -2254,3 +2213,210 @@ async def test_entity_resolution_does_not_merge_distinct_label_values(memory, re
        )
    finally:
        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+# ─── User report: paired id/name map-entity extraction from [[...]] tags ───────
+#
+# Forum report (related to GH-1558): a user wants consistent extraction of a
+# structured `application` entity with BOTH an `id` and a `name` field for every
+# tagged element in their documents. They mark up source text with their own
+# `[[Matched Text (name, id)]]` notation, e.g.
+#     [[SystemA (SystemA, SYS001)]]   [[System-A (SystemA, SYS001)]]
+# and configure an entity label group like:
+#     application (tag)
+#       - id   (multi-values): SYS001, SYS002, SYS003, ...
+#       - name (multi-values): SystemA, SystemB, SystemC, ...
+#
+# Symptom: extraction is inconsistent. For a given tagged element they often get
+# only PART of the pair (e.g. application:name:SystemA but no application:id:SYS001),
+# and sometimes the element is missed entirely. It is noticeably worse when more
+# than one tagged element appears in the same chunk.
+#
+# These tests reproduce that scenario. The deterministic tests pin the mechanics
+# (map post-processing emits the full pair when the LLM returns both fields, and
+# faithfully drops half when it doesn't — there is no backfill, so the pairing
+# must come from the model). The hs_llm_core test exercises the real model
+# end-to-end and asserts that EVERY tagged element yields a COMPLETE {name, id}
+# pair — the assertion that surfaces the reported flakiness.
+
+
+# Known applications: canonical name → canonical id (the configured vocabulary).
+_KNOWN_APPLICATIONS = {
+    "SystemA": "SYS001",
+    "SystemB": "SYS002",
+    "SystemC": "SYS003",
+}
+
+
+def _build_application_label_config() -> dict:
+    """The user's reported entity_labels config: application map with id + name."""
+    return {
+        "entity_labels": [
+            {
+                "key": "application",
+                "type": "map",
+                "tag": True,
+                "description": "A known software system referenced in the text",
+                "fields": {
+                    "name": {
+                        "type": "multi-values",
+                        "description": "The human-readable application name",
+                        "values": [{"value": n} for n in _KNOWN_APPLICATIONS],
+                    },
+                    "id": {
+                        "type": "multi-values",
+                        "description": "The application identifier code",
+                        "values": [{"value": i} for i in _KNOWN_APPLICATIONS.values()],
+                    },
+                },
+            }
+        ],
+        "entities_allow_free_form": False,
+        "retain_extraction_mode": "verbose",
+    }
+
+
+def test_map_entity_emits_complete_id_name_pair():
+    """
+    Deterministic mechanics: when the LLM returns a map entity object with BOTH
+    fields populated, post-processing emits the full pair of label entities.
+
+    This isolates the post-processing step from LLM non-determinism — it proves
+    the pipeline is capable of producing the complete pair, so any missing half
+    seen end-to-end comes from the model's structured output, not from a bug here.
+    """
+    from hindsight_api.engine.retain.fact_extraction import Entity, _extract_map_entities
+
+    cfg = parse_entity_labels(_build_application_label_config()["entity_labels"])
+    assert cfg is not None
+    group = cfg.attributes[0]
+
+    validated: list[Entity] = []
+    existing_lower: set[str] = set()
+    # Simulated LLM output for one tagged element: [[SystemA (SystemA, SYS001)]]
+    _extract_map_entities(
+        entity_obj={"name": ["SystemA"], "id": ["SYS001"]},
+        fields=group.fields,
+        prefix="application:",
+        validated_entities=validated,
+        existing_texts_lower=existing_lower,
+    )
+
+    texts = {e.text for e in validated}
+    assert texts == {"application:name:SystemA", "application:id:SYS001"}, (
+        f"Expected the complete id/name pair, got: {texts}"
+    )
+
+
+def test_map_entity_partial_object_drops_half_the_pair():
+    """
+    Deterministic: documents the failure shape the user sees. If the LLM returns
+    only one field of the map object, post-processing faithfully emits only that
+    half — there is no inference of the missing member. This shows the pairing
+    must be guaranteed upstream (by the model), and post-processing won't backfill.
+    """
+    from hindsight_api.engine.retain.fact_extraction import Entity, _extract_map_entities
+
+    cfg = parse_entity_labels(_build_application_label_config()["entity_labels"])
+    group = cfg.attributes[0]
+
+    validated: list[Entity] = []
+    # LLM returned the name but omitted the id — the reported "part only" case.
+    _extract_map_entities(
+        entity_obj={"name": ["SystemA"]},
+        fields=group.fields,
+        prefix="application:",
+        validated_entities=validated,
+        existing_texts_lower=set(),
+    )
+
+    texts = {e.text for e in validated}
+    assert texts == {"application:name:SystemA"}, texts
+    assert "application:id:SYS001" not in texts
+
+
+@pytest.mark.asyncio
+@pytest.mark.hs_llm_core
+async def test_retain_application_tags_extract_complete_pairs(memory_real_llm, request_context):
+    """
+    User report reproducer (integration): retain a document whose source text is
+    marked up with `[[Matched Text (name, id)]]` tags referencing several known
+    applications, and assert that EVERY tagged element yields a COMPLETE
+    {application:name:*, application:id:*} pair.
+
+    The reported symptom is that some elements come back with only the name OR
+    only the id (and occasionally neither), especially with several tags in one
+    chunk. This test fails when any expected pair is incomplete, surfacing that
+    inconsistency.
+    """
+    from hindsight_api.engine.memory_engine import fq_table
+
+    bank_id = f"test-app-pairs-{uuid.uuid4().hex[:8]}"
+    # Three tagged elements in ONE chunk, with surface forms that differ from the
+    # canonical values (hyphenation, casing) so the model has to map each tag back
+    # onto the configured vocabulary — the "more than one item in the chunk"
+    # condition from the report.
+    elements = ["SystemA", "SystemB", "SystemC"]
+    expected_pairs = {
+        name: (
+            f"application:name:{name.lower()}",
+            f"application:id:{_KNOWN_APPLICATIONS[name].lower()}",
+        )
+        for name in elements
+    }
+    try:
+        await memory_real_llm.get_bank_profile(bank_id=bank_id, request_context=request_context)
+        await memory_real_llm._config_resolver.update_bank_config(
+            bank_id=bank_id,
+            updates=_build_application_label_config(),
+            context=request_context,
+        )
+
+        # Multiple tagged elements in a single document, mirroring the user's
+        # `[[Matched Text (name, id)]]` notation and varied surface forms.
+        unit_ids = await memory_real_llm.retain_async(
+            bank_id=bank_id,
+            content=(
+                "## Integration Architecture\n\n"
+                "The order pipeline routes events from [[SystemA (SystemA, SYS001)]] "
+                "into [[System-B (SystemB, SYS002)]] for enrichment. "
+                "Reconciliation is handled downstream by [[system c (SystemC, SYS003)]]. "
+                "Note that [[System-A (SystemA, SYS001)]] also emits audit records "
+                "consumed by [[SystemC (SystemC, SYS003)]]."
+            ),
+            request_context=request_context,
+        )
+
+        assert len(unit_ids) > 0, "Should have extracted at least one fact"
+
+        async with memory_real_llm._pool.acquire() as conn:
+            entity_rows = await conn.fetch(
+                f"""
+                SELECT e.canonical_name
+                FROM {fq_table("unit_entities")} ue
+                JOIN {fq_table("entities")} e ON e.id = ue.entity_id
+                WHERE ue.unit_id = ANY($1::uuid[])
+                """,
+                [u for u in unit_ids],
+            )
+        entity_names = {r["canonical_name"].lower() for r in entity_rows}
+        app_entities = {n for n in entity_names if n.startswith("application:")}
+
+        # Build a per-element completeness report so a failure is diagnostic.
+        report: list[str] = []
+        incomplete: list[str] = []
+        for element, (name_ent, id_ent) in expected_pairs.items():
+            has_name = name_ent in app_entities
+            has_id = id_ent in app_entities
+            if not (has_name and has_id):
+                incomplete.append(element)
+            report.append(f"  {element}: name={'OK' if has_name else 'MISSING'} id={'OK' if has_id else 'MISSING'}")
+
+        assert not incomplete, (
+            "User report reproduced: not every tagged element produced a complete "
+            f"id/name pair. Incomplete: {incomplete}\n"
+            "Per-element extraction:\n" + "\n".join(report) + "\n"
+            f"All application:* entities: {sorted(app_entities)}"
+        )
+    finally:
+        await memory_real_llm.delete_bank(bank_id, request_context=request_context)
@@ -305,9 +305,7 @@ class TestOracleFuzzyEntityResolution:
        conn = AsyncMock()
        conn.backend_type = "oracle"
        conn.fetch = AsyncMock(return_value=[])
-        entities_data = [
-            {"text": f"Entity {idx}", "nearby_entities": [], "event_date": None} for idx in range(5)
-        ]
+        entities_data = [{"text": f"Entity {idx}", "nearby_entities": [], "event_date": None} for idx in range(5)]

        with patch.object(resolver, "_resolve_from_candidates", new_callable=AsyncMock, return_value=[]):
            await resolver._resolve_entities_batch_oracle_fuzzy(
@@ -101,25 +101,19 @@ class RateLimitingValidator(OperationValidatorExtension):
    async def validate_retain(self, ctx: RetainContext) -> ValidationResult:
        self.retain_counts[ctx.bank_id] += 1
        if self.retain_counts[ctx.bank_id] > self.max_attempts:
-            return ValidationResult.reject(
-                f"Retain limit exceeded for bank {ctx.bank_id}"
-            )
+            return ValidationResult.reject(f"Retain limit exceeded for bank {ctx.bank_id}")
        return ValidationResult.accept()

    async def validate_recall(self, ctx: RecallContext) -> ValidationResult:
        self.recall_counts[ctx.bank_id] += 1
        if self.recall_counts[ctx.bank_id] > self.max_attempts:
-            return ValidationResult.reject(
-                f"Recall limit exceeded for bank {ctx.bank_id}"
-            )
+            return ValidationResult.reject(f"Recall limit exceeded for bank {ctx.bank_id}")
        return ValidationResult.accept()

    async def validate_reflect(self, ctx: ReflectContext) -> ValidationResult:
        self.reflect_counts[ctx.bank_id] += 1
        if self.reflect_counts[ctx.bank_id] > self.max_attempts:
-            return ValidationResult.reject(
-                f"Reflect limit exceeded for bank {ctx.bank_id}"
-            )
+            return ValidationResult.reject(f"Reflect limit exceeded for bank {ctx.bank_id}")
        return ValidationResult.accept()


@@ -579,9 +573,7 @@ class TestMemoryEngineTenantAuth:
    """Tests for tenant authentication in MemoryEngine."""

    @pytest.mark.asyncio
-    async def test_retain_requires_tenant_request_when_extension_configured(
-        self, memory_with_tenant
-    ):
+    async def test_retain_requires_tenant_request_when_extension_configured(self, memory_with_tenant):
        """Retain fails without RequestContext when tenant extension is configured."""
        memory = memory_with_tenant

@@ -621,9 +613,7 @@ class TestMemoryEngineTenantAuth:
        assert "Invalid API key" in str(exc_info.value)

    @pytest.mark.asyncio
-    async def test_recall_requires_tenant_request_when_extension_configured(
-        self, memory_with_tenant
-    ):
+    async def test_recall_requires_tenant_request_when_extension_configured(self, memory_with_tenant):
        """Recall fails without RequestContext when tenant extension is configured."""
        memory = memory_with_tenant

@@ -861,8 +851,7 @@ class RecordingPrecheckValidator(OperationValidatorExtension):
    instantiable; the tests here only exercise precheck.
    """

-    def __init__(self, *, reject: bool = False, status_code: int = 402,
-                 reason: str = "rejected by precheck") -> None:
+    def __init__(self, *, reject: bool = False, status_code: int = 402, reason: str = "rejected by precheck") -> None:
        super().__init__(config={})
        self.reject = reject
        self.status_code = status_code
@@ -1027,9 +1016,7 @@ class TestPrecheckHttpWiring:
        assert body_parses == ["retain"]

    def test_precheck_rejection_returns_status_and_reason(self):
-        validator = RecordingPrecheckValidator(
-            reject=True, status_code=402, reason="Insufficient credits"
-        )
+        validator = RecordingPrecheckValidator(reject=True, status_code=402, reason="Insufficient credits")
        app, _ = self._build_app(validator)
        client = TestClient(app)

@@ -1045,9 +1032,7 @@ class TestPrecheckHttpWiring:
        deserialises the body. We send an oversized body and verify the
        body-parse counter never incremented.
        """
-        validator = RecordingPrecheckValidator(
-            reject=True, status_code=402, reason="rejected by precheck"
-        )
+        validator = RecordingPrecheckValidator(reject=True, status_code=402, reason="rejected by precheck")
        app, body_parses = self._build_app(validator)
        client = TestClient(app)

@@ -1063,9 +1048,7 @@ class TestPrecheckHttpWiring:
        )

    def test_precheck_rejection_skips_body_parse_for_recall(self):
-        validator = RecordingPrecheckValidator(
-            reject=True, status_code=402, reason="rejected"
-        )
+        validator = RecordingPrecheckValidator(reject=True, status_code=402, reason="rejected")
        app, body_parses = self._build_app(validator)
        client = TestClient(app)

@@ -1078,9 +1061,7 @@ class TestPrecheckHttpWiring:
        assert body_parses == []

    def test_precheck_rejection_skips_body_parse_for_reflect(self):
-        validator = RecordingPrecheckValidator(
-            reject=True, status_code=402, reason="rejected"
-        )
+        validator = RecordingPrecheckValidator(reject=True, status_code=402, reason="rejected")
        app, body_parses = self._build_app(validator)
        client = TestClient(app)

@@ -1,6 +1,7 @@
 """
 Test to analyze fact extraction token usage and identify optimization opportunities.
 """
+
 import asyncio
 import logging
 import time
@@ -63,9 +64,9 @@ async def test_fact_extraction_basic_analysis(llm_config):

    duration = time.time() - start_time

-    logger.info(f"\n{'='*60}")
+    logger.info(f"\n{'=' * 60}")
    logger.info(f"EXTRACTION RESULTS")
-    logger.info(f"{'='*60}")
+    logger.info(f"{'=' * 60}")
    logger.info(f"Duration: {duration:.2f}s")
    logger.info(f"Chunks: {len(chunks)}")
    logger.info(f"Facts extracted: {len(facts)}")
@@ -86,13 +87,13 @@ async def test_fact_extraction_basic_analysis(llm_config):
    # Show sample facts
    logger.info(f"\nSample facts (first 10):")
    for i, fact in enumerate(facts[:10]):
-        logger.info(f"\n  [{i+1}] {fact.fact_type}: {fact.fact[:150]}...")
+        logger.info(f"\n  [{i + 1}] {fact.fact_type}: {fact.fact[:150]}...")

    # Show facts containing key terms
    key_terms = ["kubernetes", "k8s", "CKA", "certification", "Alice"]
-    logger.info(f"\n{'='*60}")
+    logger.info(f"\n{'=' * 60}")
    logger.info(f"FACTS CONTAINING KEY TERMS")
-    logger.info(f"{'='*60}")
+    logger.info(f"{'=' * 60}")

    for term in key_terms:
        matching = [f for f in facts if term.lower() in f.fact.lower()]
@@ -1,6 +1,7 @@
 """
 Unit tests for metadata inclusion in fact extraction LLM prompt.
 """
+
 from datetime import datetime

 from hindsight_api.engine.retain.fact_extraction import _build_user_message
@@ -109,8 +109,7 @@ User: Perfect, I'll make a reservation for Saturday at 7pm.

        # Output should not be more than 5x the input
        assert ratio < 5.0, (
-            f"Output/input ratio {ratio:.2f} is too high! "
-            f"Input: {input_length} chars, Output: {output_length} chars"
+            f"Output/input ratio {ratio:.2f} is too high! Input: {input_length} chars, Output: {output_length} chars"
        )

    @pytest.mark.asyncio
@@ -168,16 +167,12 @@ I edited about 20 photos from my recent trip to the mountains.
        # Output should not be more than 4x the input for longer texts
        # (ratio should decrease as input grows)
        assert ratio < 4.0, (
-            f"Output/input ratio {ratio:.2f} is too high! "
-            f"Input: {input_length} chars, Output: {output_length} chars"
+            f"Output/input ratio {ratio:.2f} is too high! Input: {input_length} chars, Output: {output_length} chars"
        )

        # Also check that individual facts aren't excessively long
        max_fact_length = max(len(f.fact) for f in facts) if facts else 0
-        assert max_fact_length < 1000, (
-            f"Individual fact too long: {max_fact_length} chars. "
-            f"Facts should be concise."
-        )
+        assert max_fact_length < 1000, f"Individual fact too long: {max_fact_length} chars. Facts should be concise."

    @pytest.mark.asyncio
    async def test_token_ratio_with_locomo_conversation(self):
@@ -190,11 +185,7 @@ I edited about 20 photos from my recent trip to the mountains.
        import os

        # Load locomo conversation
-        fixture_path = os.path.join(
-            os.path.dirname(__file__),
-            "fixtures",
-            "locomo_conversation_sample.json"
-        )
+        fixture_path = os.path.join(os.path.dirname(__file__), "fixtures", "locomo_conversation_sample.json")
        with open(fixture_path, "r") as f:
            data = json.load(f)

@@ -246,8 +237,7 @@ I edited about 20 photos from my recent trip to the mountains.
        max_expected_facts = num_turns * 2  # At most 2 facts per conversation turn

        assert len(facts) <= max_expected_facts, (
-            f"Too many facts: {len(facts)} for {num_turns} conversation turns. "
-            f"Expected at most {max_expected_facts}."
+            f"Too many facts: {len(facts)} for {num_turns} conversation turns. Expected at most {max_expected_facts}."
        )

    @pytest.mark.asyncio
@@ -279,7 +269,7 @@ I'm planning to visit Japan next year.
        )

        # Count approximate number of statements (sentences)
-        num_statements = len([s for s in text.split('.') if s.strip()])
+        num_statements = len([s for s in text.split(".") if s.strip()])

        print(f"\nNumber of facts test:")
        print(f"  Input statements: ~{num_statements}")
@@ -5,6 +5,7 @@ This ensures that when multiple facts are extracted from a long conversation,
 their relative order is preserved via time offsets, allowing retrieval to
 distinguish between things said earlier vs later.
 """
+
 import pytest
 from datetime import datetime, timezone
 from hindsight_api import MemoryEngine, RequestContext
@@ -20,11 +21,9 @@ async def test_fact_ordering_within_conversation(memory, request_context):
    await memory.get_bank_profile(bank_id, request_context=request_context)

    # Update disposition to match Marcus
-    await memory.update_bank_disposition(bank_id, {
-        "skepticism": 3,
-        "literalism": 3,
-        "empathy": 3
-    }, request_context=request_context)
+    await memory.update_bank_disposition(
+        bank_id, {"skepticism": 3, "literalism": 3, "empathy": 3}, request_context=request_context
+    )

    # A conversation where Marcus changes his position
    conversation = """
@@ -51,7 +50,7 @@ Marcus: Yeah, I realized I was being too optimistic about their defense.
    results = await memory.recall_async(
        bank_id=bank_id,
        query="Marcus prediction Rams",
-        fact_type=['experience', 'world'],
+        fact_type=["experience", "world"],
        budget=Budget.LOW,
        max_tokens=8192,
        request_context=request_context,
@@ -59,37 +58,41 @@ Marcus: Yeah, I realized I was being too optimistic about their defense.

    print(f"\n=== Retrieved {len(results.results)} facts ===")
    for i, result in enumerate(results.results):
-        print(f"{i+1}. [{result.mentioned_at}] {result.text[:100]}")
+        print(f"{i + 1}. [{result.mentioned_at}] {result.text[:100]}")

    # Get all facts (Marcus's predictions/statements)
    agent_facts = results.results

    print(f"\n=== Agent facts (Marcus's statements) ===")
    for i, fact in enumerate(agent_facts):
-        print(f"{i+1}. [{fact.mentioned_at}] {fact.text}")
+        print(f"{i + 1}. [{fact.mentioned_at}] {fact.text}")

    # Check that agent facts have different timestamps
    if len(agent_facts) >= 2:
        # Parse timestamps
-        timestamps = [datetime.fromisoformat(f.mentioned_at.replace('Z', '+00:00')) for f in agent_facts]
+        timestamps = [datetime.fromisoformat(f.mentioned_at.replace("Z", "+00:00")) for f in agent_facts]

        # Verify timestamps are different (have time offsets)
        unique_timestamps = set(timestamps)
-        assert len(unique_timestamps) == len(timestamps), \
+        assert len(unique_timestamps) == len(timestamps), (
            f"Expected unique timestamps for each fact, but got duplicates: {timestamps}"
+        )

        # Sort facts by timestamp for ordering check
        # Note: recall returns by relevance, not time order
-        sorted_facts = sorted(agent_facts, key=lambda f: datetime.fromisoformat(f.mentioned_at.replace('Z', '+00:00')))
-        sorted_timestamps = [datetime.fromisoformat(f.mentioned_at.replace('Z', '+00:00')) for f in sorted_facts]
+        sorted_facts = sorted(agent_facts, key=lambda f: datetime.fromisoformat(f.mentioned_at.replace("Z", "+00:00")))
+        sorted_timestamps = [datetime.fromisoformat(f.mentioned_at.replace("Z", "+00:00")) for f in sorted_facts]

        # Verify sorted timestamps are in ascending order
        for i in range(len(sorted_timestamps) - 1):
-            assert sorted_timestamps[i] < sorted_timestamps[i + 1], \
-                f"Facts should have sequential timestamps. Fact {i} ({sorted_timestamps[i]}) >= Fact {i+1} ({sorted_timestamps[i+1]})"
+            assert sorted_timestamps[i] < sorted_timestamps[i + 1], (
+                f"Facts should have sequential timestamps. Fact {i} ({sorted_timestamps[i]}) >= Fact {i + 1} ({sorted_timestamps[i + 1]})"
+            )

        # Verify facts have distinct timestamps (ordering is preserved)
-        time_diffs = [(sorted_timestamps[i+1] - sorted_timestamps[i]).total_seconds() for i in range(len(sorted_timestamps) - 1)]
+        time_diffs = [
+            (sorted_timestamps[i + 1] - sorted_timestamps[i]).total_seconds() for i in range(len(sorted_timestamps) - 1)
+        ]
        print(f"\n=== Time differences between facts: {time_diffs} seconds ===")

        # Each fact should have a positive time difference (uniqueness already checked above)
@@ -108,7 +111,7 @@ Marcus: Yeah, I realized I was being too optimistic about their defense.
    all_text = " ".join(agent_texts)

    # Look for evidence of the predictions being captured (may be merged or separate)
-    has_prediction_info = '27' in all_text or 'rams' in all_text or 'prediction' in all_text
+    has_prediction_info = "27" in all_text or "rams" in all_text or "prediction" in all_text

    assert has_prediction_info, "Facts should contain information about Marcus's predictions"
    print(f"\n✅ Facts capture prediction information")
@@ -121,7 +124,6 @@ Marcus: Yeah, I realized I was being too optimistic about their defense.

@pytest.mark.asyncio
 async def test_multiple_documents_ordering(memory, request_context):
-
    bank_id = "test_multi_doc_agent"

    await memory.get_bank_profile(bank_id, request_context=request_context)  # Auto-creates with defaults
@@ -149,7 +151,7 @@ Alice: I reconsidered the team's experience level.
        bank_id=bank_id,
        contents=[
            {"content": conv1, "context": "project discussion 1", "event_date": time1},
-            {"content": conv2, "context": "project discussion 2", "event_date": time2}
+            {"content": conv2, "context": "project discussion 2", "event_date": time2},
        ],
        request_context=request_context,
    )
@@ -168,18 +170,21 @@ Alice: I reconsidered the team's experience level.
    agent_facts = results.results

    for i, fact in enumerate(agent_facts):
-        print(f"{i+1}. [{fact.mentioned_at}] {fact.text[:80]}")
+        print(f"{i + 1}. [{fact.mentioned_at}] {fact.text[:80]}")

    # Each conversation's facts should have different timestamps.
    # Filter out observations — they inherit their source fact's timestamp,
    # which can collapse the unique set. Also skip facts without timestamps.
-    source_facts = [f for f in agent_facts if f.mentioned_at is not None and getattr(f, "fact_type", "") != "observation"]
+    source_facts = [
+        f for f in agent_facts if f.mentioned_at is not None and getattr(f, "fact_type", "") != "observation"
+    ]
    if len(source_facts) >= 2:
-        timestamps = [datetime.fromisoformat(f.mentioned_at.replace('Z', '+00:00')) for f in source_facts]
+        timestamps = [datetime.fromisoformat(f.mentioned_at.replace("Z", "+00:00")) for f in source_facts]
        unique_timestamps = set(timestamps)

-        assert len(unique_timestamps) >= 2, \
+        assert len(unique_timestamps) >= 2, (
            f"Expected multiple unique timestamps across conversations, got: {len(unique_timestamps)}"
+        )

        print(f"\n✅ Facts from {len(source_facts)} statements have {len(unique_timestamps)} unique timestamps")

@@ -216,7 +216,9 @@ async def test_file_retain_validation_errors(memory_no_llm_verify):

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        # Create bank
-        bank_response = await client.put("/v1/default/banks/test-validation-bank", json={"name": "Test Validation Bank"})
+        bank_response = await client.put(
+            "/v1/default/banks/test-validation-bank", json={"name": "Test Validation Bank"}
+        )
        assert bank_response.status_code in (200, 201)

        # Test: metadata count mismatch
@@ -98,9 +98,7 @@ def seaweedfs_container():
        DockerContainer(image="chrislusf/seaweedfs:latest")
        .with_exposed_ports(SEAWEEDFS_S3_PORT)
        .with_volume_mapping(s3_config_file.name, "/etc/seaweedfs/s3.json", "ro")
-        .with_command(
-            f"server -s3 -s3.port={SEAWEEDFS_S3_PORT} -s3.config=/etc/seaweedfs/s3.json -ip.bind=0.0.0.0"
-        )
+        .with_command(f"server -s3 -s3.port={SEAWEEDFS_S3_PORT} -s3.config=/etc/seaweedfs/s3.json -ip.bind=0.0.0.0")
    )

    container.start()
@@ -309,9 +309,7 @@ async def test_api_errors_surface_the_response_body():
    llm = _make_fireworks(http_client=client)

    with pytest.raises(httpx.HTTPStatusError, match="invalid field 'userUploaded'"):
-        await llm.submit_batch(
-            [{"custom_id": "c0", "method": "POST", "url": "/v1/chat/completions", "body": {}}]
-        )
+        await llm.submit_batch([{"custom_id": "c0", "method": "POST", "url": "/v1/chat/completions", "body": {}}])

    await client.aclose()

@@ -33,9 +33,7 @@ def _make_client(create_side_effect=None):
    elif callable(create_side_effect):
        create_mock.side_effect = create_side_effect
    else:
-        create_mock.return_value = SimpleNamespace(
-            name="cachedContents/test-cache-name-001"
-        )
+        create_mock.return_value = SimpleNamespace(name="cachedContents/test-cache-name-001")

    client = MagicMock()
    client.aio = MagicMock()
@@ -127,18 +125,12 @@ async def test_first_call_creates_subsequent_reuses():
@pytest.mark.asyncio
 async def test_different_prefixes_create_separately():
    client, create_mock = _make_client(
-        create_side_effect=lambda *a, **kw: SimpleNamespace(
-            name=f"cachedContents/created-{create_mock.call_count}"
-        )
+        create_side_effect=lambda *a, **kw: SimpleNamespace(name=f"cachedContents/created-{create_mock.call_count}")
    )
    mgr = GeminiCacheManager(client)

-    name_a = await mgr.get_or_create(
-        model="m", system_instruction="A", response_schema=None
-    )
-    name_b = await mgr.get_or_create(
-        model="m", system_instruction="B", response_schema=None
-    )
+    name_a = await mgr.get_or_create(model="m", system_instruction="A", response_schema=None)
+    name_b = await mgr.get_or_create(model="m", system_instruction="B", response_schema=None)
    assert name_a != name_b
    assert create_mock.call_count == 2

@@ -155,9 +147,7 @@ async def test_minimum_token_count_error_returns_none():
    client, _ = _make_client(create_side_effect=err)
    mgr = GeminiCacheManager(client)

-    result = await mgr.get_or_create(
-        model="m", system_instruction="tiny", response_schema=None
-    )
+    result = await mgr.get_or_create(model="m", system_instruction="tiny", response_schema=None)
    assert result is None


@@ -169,9 +159,7 @@ async def test_other_sdk_errors_also_return_none():
    client, _ = _make_client(create_side_effect=err)
    mgr = GeminiCacheManager(client)

-    result = await mgr.get_or_create(
-        model="m", system_instruction="ok-sized prefix", response_schema=None
-    )
+    result = await mgr.get_or_create(model="m", system_instruction="ok-sized prefix", response_schema=None)
    assert result is None


@@ -194,12 +182,8 @@ async def test_failed_create_does_not_poison_cache():

    mgr = GeminiCacheManager(client)

-    first = await mgr.get_or_create(
-        model="m", system_instruction="prefix", response_schema=None
-    )
-    second = await mgr.get_or_create(
-        model="m", system_instruction="prefix", response_schema=None
-    )
+    first = await mgr.get_or_create(model="m", system_instruction="prefix", response_schema=None)
+    second = await mgr.get_or_create(model="m", system_instruction="prefix", response_schema=None)

    assert first is None
    assert second == "cachedContents/recovered"
@@ -214,9 +198,7 @@ async def test_refreshes_after_ttl_margin(monkeypatch):
    """An entry created at t=0 with ttl=10 and margin=2 should be
    treated as stale at t>=8 and trigger a recreate."""
    client, create_mock = _make_client(
-        create_side_effect=lambda *a, **kw: SimpleNamespace(
-            name=f"cachedContents/v{create_mock.call_count}"
-        )
+        create_side_effect=lambda *a, **kw: SimpleNamespace(name=f"cachedContents/v{create_mock.call_count}")
    )
    mgr = GeminiCacheManager(client, ttl_seconds=10, refresh_margin_seconds=2)

@@ -226,24 +208,18 @@ async def test_refreshes_after_ttl_margin(monkeypatch):
        lambda: fake_now["t"],
    )

-    first = await mgr.get_or_create(
-        model="m", system_instruction="p", response_schema=None
-    )
+    first = await mgr.get_or_create(model="m", system_instruction="p", response_schema=None)
    assert first == "cachedContents/v1"

    # Advance to just before the refresh boundary — should reuse.
    fake_now["t"] = 1000.0 + 7.0
-    again = await mgr.get_or_create(
-        model="m", system_instruction="p", response_schema=None
-    )
+    again = await mgr.get_or_create(model="m", system_instruction="p", response_schema=None)
    assert again == "cachedContents/v1"
    assert create_mock.call_count == 1

    # Advance past the refresh boundary — should recreate.
    fake_now["t"] = 1000.0 + 9.0
-    refreshed = await mgr.get_or_create(
-        model="m", system_instruction="p", response_schema=None
-    )
+    refreshed = await mgr.get_or_create(model="m", system_instruction="p", response_schema=None)
    assert refreshed == "cachedContents/v2"
    assert create_mock.call_count == 2

@@ -295,9 +271,7 @@ async def test_gemini_llm_uses_cache_when_enabled(monkeypatch):
    # Replace the SDK-shaped client with a fake whose caches.create returns
    # a predictable name. The lazy import inside get_or_create_cached_prefix
    # picks up the patched module-level GeminiCacheManager naturally.
-    fake_create = AsyncMock(
-        return_value=SimpleNamespace(name="cachedContents/from-llm-test")
-    )
+    fake_create = AsyncMock(return_value=SimpleNamespace(name="cachedContents/from-llm-test"))
    llm._client = MagicMock()
    llm._client.aio = MagicMock()
    llm._client.aio.caches = MagicMock()
@@ -333,7 +307,9 @@ async def test_call_falls_back_to_uncached_when_cache_400s():
    from hindsight_api.engine.providers.gemini_cache import GeminiCacheManager, _CacheEntry
    from hindsight_api.engine.providers.gemini_llm import GeminiLLM

-    llm = GeminiLLM(provider="gemini", api_key="not-real-key", base_url="", model="gemini-test", prompt_cache_enabled=True)
+    llm = GeminiLLM(
+        provider="gemini", api_key="not-real-key", base_url="", model="gemini-test", prompt_cache_enabled=True
+    )

    # Seed a cache manager entry that maps to the (now invalid) cache name.
    mgr = GeminiCacheManager(client=MagicMock())
@@ -413,9 +389,7 @@ def test_fingerprint_changes_with_tools():
    """Two prefixes that differ ONLY in tools must hash differently —
    otherwise a loop that adds a tool would silently reuse a stale
    cache that doesn't know about it."""
-    tools_a = [
-        {"type": "function", "function": {"name": "search", "description": "search", "parameters": {}}}
-    ]
+    tools_a = [{"type": "function", "function": {"name": "search", "description": "search", "parameters": {}}}]
    tools_b = [
        {"type": "function", "function": {"name": "search", "description": "search", "parameters": {}}},
        {"type": "function", "function": {"name": "fetch", "description": "fetch", "parameters": {}}},
@@ -455,7 +429,10 @@ async def test_get_or_create_passes_tools_to_create():

    mgr = GeminiCacheManager(client)
    tools = [
-        {"type": "function", "function": {"name": "search", "description": "do a search", "parameters": {"type": "object"}}}
+        {
+            "type": "function",
+            "function": {"name": "search", "description": "do a search", "parameters": {"type": "object"}},
+        }
    ]
    name = await mgr.get_or_create(
        model="gemini-3.1-flash-lite",
@@ -34,9 +34,7 @@ from hindsight_api.engine.consolidation.consolidator import run_consolidation_jo
 from hindsight_api.engine.llm_trace import LLMRequestEntry
 from hindsight_api.engine.llm_wrapper import LLMConfig

-_GEMINI_API_KEY = (
-    os.getenv("HINDSIGHT_GEMINI_API_KEY") or os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")
-)
+_GEMINI_API_KEY = os.getenv("HINDSIGHT_GEMINI_API_KEY") or os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")
 _RUN = os.getenv("HINDSIGHT_RUN_GEMINI_EVALS") == "1" and bool(_GEMINI_API_KEY)

 pytestmark = pytest.mark.skipif(
@@ -133,13 +133,17 @@ async def test_call_applies_safety_settings():
    assert hasattr(config_arg, "safety_settings"), "Config should have safety_settings"
    assert config_arg.safety_settings is not None

-    categories = [s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings]
+    categories = [
+        s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings
+    ]
    assert "HARM_CATEGORY_HARASSMENT" in categories
    assert "HARM_CATEGORY_HATE_SPEECH" in categories
    assert "HARM_CATEGORY_SEXUALLY_EXPLICIT" in categories
    assert "HARM_CATEGORY_DANGEROUS_CONTENT" in categories

-    thresholds = [s.threshold.value if hasattr(s.threshold, "value") else str(s.threshold) for s in config_arg.safety_settings]
+    thresholds = [
+        s.threshold.value if hasattr(s.threshold, "value") else str(s.threshold) for s in config_arg.safety_settings
+    ]
    assert all(t == "BLOCK_NONE" for t in thresholds)


@@ -212,7 +216,9 @@ async def test_call_with_tools_applies_safety_settings():

    assert config_arg is not None
    assert config_arg.safety_settings is not None
-    categories = [s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings]
+    categories = [
+        s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings
+    ]
    assert "HARM_CATEGORY_HARASSMENT" in categories


@@ -266,7 +272,9 @@ async def test_with_config_overrides_instance_settings():

    config_arg = provider._provider_impl._client.aio.models.generate_content.call_args.kwargs.get("config")
    assert config_arg is not None
-    categories = [s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings]
+    categories = [
+        s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings
+    ]
    # Should use override_settings (HATE_SPEECH), not instance_settings (HARASSMENT)
    assert "HARM_CATEGORY_HATE_SPEECH" in categories
    assert "HARM_CATEGORY_HARASSMENT" not in categories
@@ -285,7 +293,9 @@ async def test_with_config_none_falls_back_to_instance():

    config_arg = provider._provider_impl._client.aio.models.generate_content.call_args.kwargs.get("config")
    assert config_arg is not None
-    categories = [s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings]
+    categories = [
+        s.category.value if hasattr(s.category, "value") else str(s.category) for s in config_arg.safety_settings
+    ]
    assert "HARM_CATEGORY_HARASSMENT" in categories


@@ -97,19 +97,23 @@ class TestGoogleCrossEncoder:
    async def test_predict_single_query(self):
        """Test prediction with a single query and multiple documents."""
        mock_creds = _make_mock_credentials()
-        mock_client = _make_mock_httpx_client([
-            _make_rank_response([("1", 0.95), ("0", 0.30)]),
-        ])
+        mock_client = _make_mock_httpx_client(
+            [
+                _make_rank_response([("1", 0.95), ("0", 0.30)]),
+            ]
+        )

        encoder = GoogleCrossEncoder(project_id="test-project")
        with patch("google.auth.default", return_value=(mock_creds, "p")):
            await encoder.initialize()
        encoder._client = mock_client

-        scores = await encoder.predict([
-            ("What is AI?", "AI is artificial intelligence"),
-            ("What is AI?", "The sky is blue"),
-        ])
+        scores = await encoder.predict(
+            [
+                ("What is AI?", "AI is artificial intelligence"),
+                ("What is AI?", "The sky is blue"),
+            ]
+        )

        assert len(scores) == 2
        assert scores[0] == 0.30  # id="0" -> index 0
@@ -119,21 +123,25 @@ class TestGoogleCrossEncoder:
    async def test_predict_multiple_queries(self):
        """Test prediction with multiple distinct queries."""
        mock_creds = _make_mock_credentials()
-        mock_client = _make_mock_httpx_client([
-            _make_rank_response([("0", 0.9), ("1", 0.1)]),
-            _make_rank_response([("0", 0.8)]),
-        ])
+        mock_client = _make_mock_httpx_client(
+            [
+                _make_rank_response([("0", 0.9), ("1", 0.1)]),
+                _make_rank_response([("0", 0.8)]),
+            ]
+        )

        encoder = GoogleCrossEncoder(project_id="test-project")
        with patch("google.auth.default", return_value=(mock_creds, "p")):
            await encoder.initialize()
        encoder._client = mock_client

-        scores = await encoder.predict([
-            ("Query A", "Doc A1"),
-            ("Query A", "Doc A2"),
-            ("Query B", "Doc B1"),
-        ])
+        scores = await encoder.predict(
+            [
+                ("Query A", "Doc A1"),
+                ("Query A", "Doc A2"),
+                ("Query B", "Doc B1"),
+            ]
+        )

        assert len(scores) == 3
        assert scores[0] == 0.9
@@ -161,10 +169,12 @@ class TestGoogleCrossEncoder:
    async def test_predict_batching(self):
        """Test that >200 records are split into batches."""
        mock_creds = _make_mock_credentials()
-        mock_client = _make_mock_httpx_client([
-            _make_rank_response([(str(i), 0.5) for i in range(200)]),
-            _make_rank_response([(str(i), 0.3) for i in range(50)]),
-        ])
+        mock_client = _make_mock_httpx_client(
+            [
+                _make_rank_response([(str(i), 0.5) for i in range(200)]),
+                _make_rank_response([(str(i), 0.3) for i in range(50)]),
+            ]
+        )

        encoder = GoogleCrossEncoder(project_id="test-project")
        with patch("google.auth.default", return_value=(mock_creds, "p")):
@@ -181,9 +191,11 @@ class TestGoogleCrossEncoder:
        """Test that Authorization header is sent with requests."""
        mock_creds = _make_mock_credentials()
        mock_creds.token = "test-bearer-token"
-        mock_client = _make_mock_httpx_client([
-            _make_rank_response([("0", 0.9)]),
-        ])
+        mock_client = _make_mock_httpx_client(
+            [
+                _make_rank_response([("0", 0.9)]),
+            ]
+        )

        encoder = GoogleCrossEncoder(project_id="test-project")
        with patch("google.auth.default", return_value=(mock_creds, "p")):
@@ -167,9 +167,7 @@ class TestEnqueueRelinkVictims:
            assert await _queue_unit_ids(conn, bank_id) == [str(survivor)]

    @pytest.mark.asyncio
-    async def test_excludes_deleted_units_themselves(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_excludes_deleted_units_themselves(self, memory: MemoryEngine, request_context: RequestContext):
        """A unit being deleted that linked TO another deleted unit must not enqueue itself."""
        bank_id = f"test-gm-self-{uuid.uuid4().hex[:8]}"
        await _ensure_bank(memory, bank_id, request_context)
@@ -270,9 +268,7 @@ class TestDeleteDocumentEnqueue:

 class TestRelinkPass:
    @pytest.mark.asyncio
-    async def test_drains_empty_queue_cleanly(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_drains_empty_queue_cleanly(self, memory: MemoryEngine, request_context: RequestContext):
        bank_id = f"test-gm-empty-{uuid.uuid4().hex[:8]}"
        await _ensure_bank(memory, bank_id, request_context)

@@ -285,9 +281,7 @@ class TestRelinkPass:
        }

    @pytest.mark.asyncio
-    async def test_skips_missing_unit_silently(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_skips_missing_unit_silently(self, memory: MemoryEngine, request_context: RequestContext):
        """Unit deleted between enqueue and drain: worker dequeues and no-ops."""
        bank_id = f"test-gm-miss-{uuid.uuid4().hex[:8]}"
        await _ensure_bank(memory, bank_id, request_context)
@@ -309,9 +303,7 @@ class TestRelinkPass:
            assert await _queue_unit_ids(conn, bank_id) == []

    @pytest.mark.asyncio
-    async def test_tops_up_temporal_when_under_cap(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_tops_up_temporal_when_under_cap(self, memory: MemoryEngine, request_context: RequestContext):
        """A victim under the temporal cap gets new outgoing links to neighbours
        that were never linked at retain time."""
        bank_id = f"test-gm-topup-{uuid.uuid4().hex[:8]}"
@@ -365,9 +357,7 @@ class TestRelinkPass:
            assert await _queue_unit_ids(conn, bank_id) == []

    @pytest.mark.asyncio
-    async def test_no_topup_when_victim_at_cap(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_no_topup_when_victim_at_cap(self, memory: MemoryEngine, request_context: RequestContext):
        """If the victim already has cap links, probing is skipped."""
        bank_id = f"test-gm-atcap-{uuid.uuid4().hex[:8]}"
        await _ensure_bank(memory, bank_id, request_context)
@@ -414,9 +404,7 @@ class TestRelinkPass:

 class TestOrphanEntityPrune:
    @pytest.mark.asyncio
-    async def test_prunes_entities_with_no_unit_references(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_prunes_entities_with_no_unit_references(self, memory: MemoryEngine, request_context: RequestContext):
        """An entity with zero unit_entities rows is an orphan and should be
        deleted by the sweep."""
        bank_id = f"test-gm-orphan-{uuid.uuid4().hex[:8]}"
@@ -435,9 +423,7 @@ class TestOrphanEntityPrune:
        assert result["orphan_entities_pruned"] == 2

        async with pool.acquire() as conn:
-            survivors = await conn.fetch(
-                "SELECT id FROM entities WHERE bank_id = $1 ORDER BY id", bank_id
-            )
+            survivors = await conn.fetch("SELECT id FROM entities WHERE bank_id = $1 ORDER BY id", bank_id)
            survivor_ids = {str(r["id"]) for r in survivors}
            assert survivor_ids == {str(referenced)}
            # Confirm orphans are gone.
@@ -445,9 +431,7 @@ class TestOrphanEntityPrune:
                assert orphan not in survivor_ids

    @pytest.mark.asyncio
-    async def test_does_not_touch_other_banks(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_does_not_touch_other_banks(self, memory: MemoryEngine, request_context: RequestContext):
        """The sweep is scoped by bank — orphan entities in OTHER banks
        must not be touched."""
        bank_a = f"test-gm-scopea-{uuid.uuid4().hex[:8]}"
@@ -478,9 +462,7 @@ class TestOrphanEntityPrune:

 class TestStaleCooccurrencePrune:
    @pytest.mark.asyncio
-    async def test_prunes_cooccurrence_with_no_shared_unit(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_prunes_cooccurrence_with_no_shared_unit(self, memory: MemoryEngine, request_context: RequestContext):
        """Both entities still exist but no unit references both of them — the
        cooccurrence row is stale and should be pruned."""
        bank_id = f"test-gm-cocc-{uuid.uuid4().hex[:8]}"
@@ -515,9 +497,7 @@ class TestStaleCooccurrencePrune:
            assert remaining == 0

    @pytest.mark.asyncio
-    async def test_keeps_cooccurrence_with_shared_unit(
-        self, memory: MemoryEngine, request_context: RequestContext
-    ):
+    async def test_keeps_cooccurrence_with_shared_unit(self, memory: MemoryEngine, request_context: RequestContext):
        """If at least one unit still references both entities, the cooccurrence
        row stays."""
        bank_id = f"test-gm-keep-{uuid.uuid4().hex[:8]}"
@@ -7,6 +7,7 @@ Covers:
 - Per-bank vector indexes dropped on bank deletion
 - retrieve_semantic_bm25_combined groups results correctly by fact_type and source
 """
+
 import uuid
 from datetime import datetime, timezone

@@ -152,10 +153,7 @@ async def test_retrieve_semantic_bm25_grouped_by_fact_type(memory, request_conte
    try:
        await memory.retain_async(
            bank_id=bank_id,
-            content=(
-                "Alice is a software engineer at TechCorp. "
-                "She visited Paris in 2023 for a conference."
-            ),
+            content=("Alice is a software engineer at TechCorp. She visited Paris in 2023 for a conference."),
            context="background",
            event_date=datetime(2023, 6, 1, tzinfo=timezone.utc),
            request_context=request_context,
@@ -87,9 +87,7 @@ async def test_unique_violation_marks_failed_without_retry(memory):
        try:
            await memory.execute_task(task_dict)
        except RetryTaskAt as exc:
-            pytest.fail(
-                f"IntegrityConstraintViolationError must not be retried, but execute_task raised {exc!r}"
-            )
+            pytest.fail(f"IntegrityConstraintViolationError must not be retried, but execute_task raised {exc!r}")

    # The operation must be marked 'failed' (not left pending / retrying).
    row = await pool.fetchrow(
@@ -97,9 +95,7 @@ async def test_unique_violation_marks_failed_without_retry(memory):
        operation_id,
    )
    assert row is not None, "Operation row disappeared"
-    assert row["status"] == "failed", (
-        f"Expected status='failed' after integrity violation, got {row['status']!r}"
-    )
+    assert row["status"] == "failed", f"Expected status='failed' after integrity violation, got {row['status']!r}"
    assert row["error_message"] is not None
    assert "pk_chunks" in row["error_message"]

@@ -123,7 +119,7 @@ async def test_foreign_key_violation_also_not_retried(memory):
    await _create_pending_operation(pool, bank_id, operation_id)

    fk_violation = asyncpg.exceptions.ForeignKeyViolationError(
-        "insert or update on table \"memory_units\" violates foreign key constraint \"fk_bank\""
+        'insert or update on table "memory_units" violates foreign key constraint "fk_bank"'
    )

    task_dict = {
@@ -137,9 +133,7 @@ async def test_foreign_key_violation_also_not_retried(memory):
        try:
            await memory.execute_task(task_dict)
        except RetryTaskAt as exc:
-            pytest.fail(
-                f"ForeignKeyViolationError must not be retried, but execute_task raised {exc!r}"
-            )
+            pytest.fail(f"ForeignKeyViolationError must not be retried, but execute_task raised {exc!r}")

    row = await pool.fetchrow(
        "SELECT status FROM async_operations WHERE operation_id = $1",
@@ -68,5 +68,3 @@ async def test_iris_parser_converts_pdf(iris_parser: IrisParser):
 async def test_iris_parser_name(iris_parser: IrisParser):
    """IrisParser.name() should return 'iris'."""
    assert iris_parser.name() == "iris"
-
-
@@ -48,9 +48,7 @@ def _make_replacement_body() -> str:
    than one sub-batch.
    """
    lines = [
-        f"[role: user] turn {i}: alpha bravo charlie delta echo "
-        f"foxtrot golf hotel india juliet"
-        for i in range(20)
+        f"[role: user] turn {i}: alpha bravo charlie delta echo foxtrot golf hotel india juliet" for i in range(20)
    ]
    return "\n".join(lines)

@@ -79,9 +77,7 @@ async def test_large_same_id_replacement_preserves_full_body(memory, request_con
            request_context=request_context,
        )

-        doc_initial = await memory.get_document(
-            document_id, bank_id, request_context=request_context
-        )
+        doc_initial = await memory.get_document(document_id, bank_id, request_context=request_context)
        assert doc_initial is not None
        assert doc_initial["original_text"] == initial_body

@@ -95,9 +91,7 @@ async def test_large_same_id_replacement_preserves_full_body(memory, request_con
            request_context=request_context,
        )

-        doc_replaced = await memory.get_document(
-            document_id, bank_id, request_context=request_context
-        )
+        doc_replaced = await memory.get_document(document_id, bank_id, request_context=request_context)
        assert doc_replaced is not None

        stored = doc_replaced["original_text"]
@@ -105,10 +99,7 @@ async def test_large_same_id_replacement_preserves_full_body(memory, request_con
            f"stored body length {len(stored)} != submitted length "
            f"{len(replacement_body)} — partial replacement persisted"
        )
-        assert stored == replacement_body, (
-            "stored original_text does not exactly match the submitted "
-            "replacement body"
-        )
+        assert stored == replacement_body, "stored original_text does not exactly match the submitted replacement body"

    finally:
        await memory.delete_bank(bank_id, request_context=request_context)
@@ -143,9 +134,7 @@ async def test_repeated_large_same_id_replacement_is_idempotent(memory, request_
                request_context=request_context,
            )

-            doc = await memory.get_document(
-                document_id, bank_id, request_context=request_context
-            )
+            doc = await memory.get_document(document_id, bank_id, request_context=request_context)
            assert doc is not None, f"attempt {attempt}: document missing after retain"
            assert doc["original_text"] == replacement_body, (
                f"attempt {attempt}: stored body diverged from submitted body "
@@ -133,7 +133,9 @@ async def test_link_expansion_observation_graph_retrieval(memory_real_llm, reque

        assert obs_result is not None and obs_result.results is not None, "Should have observations after consolidation"
        # We should have observations from consolidation
-        assert len(obs_result.results) >= 1, f"Should have at least 1 observation about Python, got {len(obs_result.results)}"
+        assert len(obs_result.results) >= 1, (
+            f"Should have at least 1 observation about Python, got {len(obs_result.results)}"
+        )

        # Now test graph retrieval specifically
        # Query for Alice - should find Bob via shared "Python" entity
@@ -175,9 +177,7 @@ async def test_link_expansion_observation_graph_retrieval(memory_real_llm, reque

        assert world_result.trace is not None, "Should have trace data for world facts"
        world_retrieval_results = world_result.trace.get("retrieval_results", [])
-        world_graph_results = [
-            r for r in world_retrieval_results if r.get("method_name") == "graph"
-        ]
+        world_graph_results = [r for r in world_retrieval_results if r.get("method_name") == "graph"]

        if world_graph_results:
            world_graph_result = [r for r in world_graph_results if r.get("fact_type") == "world"][0]
@@ -192,7 +192,9 @@ async def test_link_expansion_observation_graph_retrieval(memory_real_llm, reque
                    print("  Found Bob's world fact via shared 'Python' entity!")

        print("\n✓ Link expansion observation test passed!")
-        print("  Entity traversal path verified (observations -> sources -> entities -> connected sources -> observations)")
+        print(
+            "  Entity traversal path verified (observations -> sources -> entities -> connected sources -> observations)"
+        )

    finally:
        await memory.delete_bank(bank_id, request_context=request_context)
@@ -259,15 +261,11 @@ async def test_link_expansion_world_fact_graph_retrieval(memory, request_context
        # Verify graph retrieval ran (it may or may not find new results depending
        # on whether semantic search already found everything)
        retrieval_results = result.trace.get("retrieval_results", [])
-        graph_results = [
-            r for r in retrieval_results if r.get("method_name") == "graph"
-        ]
+        graph_results = [r for r in retrieval_results if r.get("method_name") == "graph"]
        assert len(graph_results) > 0, "Should have graph retrieval results in trace"

        # The important thing is that recall works and returns relevant results
-        assert result.results is not None and len(result.results) > 0, (
-            "Should return results for 'Alice' query"
-        )
+        assert result.results is not None and len(result.results) > 0, "Should return results for 'Alice' query"

        # Alice's result should be at or near the top
        result_texts = [r.text for r in result.results]
@@ -1,4 +1,5 @@
 """Tests for link_utils datetime handling, temporal link computation, and semantic link splitting."""
+
 import numpy as np
 import pytest
 from datetime import datetime, timezone, timedelta
@@ -374,6 +375,7 @@ class TestComputeSemanticLinksWithinBatch:
        links = compute_semantic_links_within_batch(unit_ids, embs, top_k=3, threshold=0.5)
        # Each unit should have at most 3 outgoing links
        from collections import Counter
+
        from_counts = Counter(lnk[0] for lnk in links)
        for count in from_counts.values():
            assert count <= 3
@@ -509,16 +511,13 @@ class TestComputeSemanticLinksAnnPgBouncerSafety:
        )

    @pytest.mark.asyncio
-    @pytest.mark.parametrize(
-        ("ext", "guc"),
-        [("pgvector", "hnsw.ef_search"), ("vchord", "vchordrq.probes")],
-    )
-    async def test_uses_set_local_for_ann_tuning(self, mock_conn, monkeypatch, ext, guc):
+    async def test_uses_set_local_for_pgvector_ann_tuning(self, mock_conn, monkeypatch):
        """The per-backend ANN tuning GUC must be set with SET LOCAL so the
        change is scoped to the transaction. Without SET LOCAL, the setting
        would leak onto the pooled backend and affect subsequent recall
        queries that land on the same backend."""
-        monkeypatch.setenv("HINDSIGHT_API_VECTOR_EXTENSION", ext)
+        monkeypatch.setenv("HINDSIGHT_API_VECTOR_EXTENSION", "pgvector")
+        guc = "hnsw.ef_search"
        emb = [0.1] * 384
        await compute_semantic_links_ann(
            conn=mock_conn,
@@ -530,10 +529,29 @@ class TestComputeSemanticLinksAnnPgBouncerSafety:

        executed_sql = [call.args[0] for call in mock_conn.execute.call_args_list]
        tuning_statements = [s for s in executed_sql if guc in s]
-        assert tuning_statements, f"{guc} must be tuned for retain ANN under ext={ext}"
+        assert tuning_statements, f"{guc} must be tuned for retain ANN under pgvector"
        for stmt in tuning_statements:
-            assert stmt.strip().startswith("SET LOCAL"), (
-                f"{guc} must use SET LOCAL, got: {stmt}"
-            )
+            assert stmt.strip().startswith("SET LOCAL"), f"{guc} must use SET LOCAL, got: {stmt}"
        # And there must not be a RESET — SET LOCAL handles it at commit.
        assert not any(f"RESET {guc}" in s for s in executed_sql)
+
+    @pytest.mark.asyncio
+    async def test_vchord_ann_does_not_set_fixed_probe_count(self, mock_conn, monkeypatch):
+        """VectorChord probe counts must come from index/default config.
+
+        VectorChord requires vchordrq.probes to match the index's
+        build.internal.lists shape. Hindsight must not apply one fixed session
+        GUC across listless and partitioned vchordrq indexes.
+        """
+        monkeypatch.setenv("HINDSIGHT_API_VECTOR_EXTENSION", "vchord")
+        emb = [0.1] * 384
+        await compute_semantic_links_ann(
+            conn=mock_conn,
+            bank_id="bank-1",
+            unit_ids=["u1"],
+            embeddings=[emb],
+            fact_types=["world"],
+        )
+
+        executed_sql = [call.args[0] for call in mock_conn.execute.call_args_list]
+        assert not any("vchordrq.probes" in s for s in executed_sql)
@@ -1,6 +1,7 @@
 """
 Tests for list_documents pagination and tags filtering.
 """
+
 from datetime import datetime, timezone

 import pytest
@@ -28,25 +29,19 @@ async def test_list_documents_offset_pagination(memory, request_context):
            await _retain_doc(memory, bank_id, f"doc-{i:02d}", [], request_context)

        # All documents, ordered by created_at DESC → doc-03, doc-02, doc-01, doc-00
-        all_docs = await memory.list_documents(
-            bank_id=bank_id, limit=10, offset=0, request_context=request_context
-        )
+        all_docs = await memory.list_documents(bank_id=bank_id, limit=10, offset=0, request_context=request_context)
        assert all_docs["total"] == 4
        assert len(all_docs["items"]) == 4
        all_ids = [d["id"] for d in all_docs["items"]]

        # offset=2 should skip the first two and return the remaining two
-        page2 = await memory.list_documents(
-            bank_id=bank_id, limit=10, offset=2, request_context=request_context
-        )
+        page2 = await memory.list_documents(bank_id=bank_id, limit=10, offset=2, request_context=request_context)
        assert page2["total"] == 4  # total is always the full count
        assert len(page2["items"]) == 2
        assert [d["id"] for d in page2["items"]] == all_ids[2:]

        # offset beyond total returns empty items but correct total
-        beyond = await memory.list_documents(
-            bank_id=bank_id, limit=10, offset=10, request_context=request_context
-        )
+        beyond = await memory.list_documents(bank_id=bank_id, limit=10, offset=10, request_context=request_context)
        assert beyond["total"] == 4
        assert beyond["items"] == []

@@ -104,6 +104,35 @@ class TestLiteLLMSDKCrossEncoder:
            assert len(call_args.kwargs["documents"]) == 3
            assert call_args.kwargs["api_key"] == "test_key"

+    def test_constructor_without_api_key(self):
+        """api_key is optional (e.g. AWS Bedrock reranker with ambient IAM creds)."""
+        encoder = LiteLLMSDKCrossEncoder(model="bedrock/cohere.rerank-v3-5:0")
+        assert encoder.api_key is None
+
+    @pytest.mark.asyncio
+    async def test_predict_omits_api_key_for_ambient_credentials(self):
+        """When no api_key is set, it must not be injected into the rerank call.
+
+        litellm maps an explicit ``api_key`` to ``aws_access_key_id`` for Bedrock,
+        which overrides ambient IAM/task-role credentials; omitting it lets litellm
+        resolve credentials from the environment (regression test for IAM auth).
+        """
+        encoder = LiteLLMSDKCrossEncoder(model="bedrock/cohere.rerank-v3-5:0")
+
+        mock_response = MagicMock()
+        mock_response.results = [{"index": 0, "relevance_score": 0.9}]
+
+        mock_litellm = MagicMock()
+        mock_litellm.arerank = AsyncMock(return_value=mock_response)
+
+        with patch.dict("sys.modules", {"litellm": mock_litellm}):
+            await encoder.initialize()
+            await encoder.predict([("query", "document")])
+
+            mock_litellm.arerank.assert_called_once()
+            call_kwargs = mock_litellm.arerank.call_args.kwargs
+            assert "api_key" not in call_kwargs
+
    @pytest.mark.asyncio
    async def test_predict_multiple_queries(self):
        """Test prediction with multiple different queries (grouped efficiently)."""
@@ -278,11 +307,11 @@ class TestFactoryFunction:
                assert encoder.model == "deepinfra/Qwen3-reranker-8B"

    @pytest.mark.asyncio
-    async def test_create_litellm_sdk_missing_api_key(self):
-        """Test that factory raises error when API key is missing."""
+    async def test_create_litellm_sdk_without_api_key(self):
+        """Test that litellm-sdk works without an API key (e.g. AWS Bedrock with IAM)."""
        env_vars = {
            "HINDSIGHT_API_RERANKER_PROVIDER": "litellm-sdk",
-            "HINDSIGHT_API_RERANKER_LITELLM_SDK_MODEL": "deepinfra/Qwen3-reranker-8B",
+            "HINDSIGHT_API_RERANKER_LITELLM_SDK_MODEL": "bedrock/cohere.rerank-v3-5:0",
        }

        with patch.dict(os.environ, env_vars, clear=False):
@@ -295,8 +324,11 @@ class TestFactoryFunction:
            config = HindsightConfig.from_env()

            with patch("hindsight_api.config.get_config", return_value=config):
-                with pytest.raises(ValueError, match="HINDSIGHT_API_RERANKER_LITELLM_SDK_API_KEY is required"):
-                    create_cross_encoder_from_env()
+                encoder = create_cross_encoder_from_env()
+
+                assert isinstance(encoder, LiteLLMSDKCrossEncoder)
+                assert encoder.api_key is None
+                assert encoder.model == "bedrock/cohere.rerank-v3-5:0"

    @pytest.mark.asyncio
    async def test_create_litellm_sdk_with_custom_api_base(self):
@@ -57,7 +57,10 @@ class TestLiteLLMSDKEmbeddings:

    async def test_initialization_success(self, mock_litellm):
        """Test successful initialization."""
-        with patch("builtins.__import__", side_effect=lambda name, *args: mock_litellm if name == "litellm" else __import__(name, *args)):
+        with patch(
+            "builtins.__import__",
+            side_effect=lambda name, *args: mock_litellm if name == "litellm" else __import__(name, *args),
+        ):
            emb = LiteLLMSDKEmbeddings(
                api_key="test_key",
                model="cohere/embed-english-v3.0",
@@ -84,7 +87,10 @@ class TestLiteLLMSDKEmbeddings:

    async def test_initialization_without_api_key(self, mock_litellm):
        """Test initialization without api_key (e.g. AWS Bedrock with IAM auth)."""
-        with patch("builtins.__import__", side_effect=lambda name, *args: mock_litellm if name == "litellm" else __import__(name, *args)):
+        with patch(
+            "builtins.__import__",
+            side_effect=lambda name, *args: mock_litellm if name == "litellm" else __import__(name, *args),
+        ):
            emb = LiteLLMSDKEmbeddings(
                model="bedrock/amazon.titan-embed-text-v2:0",
                batch_size=100,
@@ -119,6 +125,7 @@ class TestLiteLLMSDKEmbeddings:

    async def test_initialization_missing_package(self):
        """Test initialization fails gracefully when litellm is not installed."""
+
        def mock_import(name, *args):
            if name == "litellm":
                raise ImportError("No module named 'litellm'")
@@ -208,9 +215,7 @@ class TestLiteLLMSDKEmbeddings:
        # Mock responses for each batch
        def mock_embedding_side_effect(model, input, **kwargs):
            mock_response = MagicMock()
-            mock_response.data = [
-                {"embedding": [float(i)] * 768, "index": i} for i in range(len(input))
-            ]
+            mock_response.data = [{"embedding": [float(i)] * 768, "index": i} for i in range(len(input))]
            return mock_response

        mock_litellm.embedding.side_effect = mock_embedding_side_effect
@@ -279,7 +284,10 @@ class TestLiteLLMSDKEmbeddings:

    async def test_custom_api_base(self, mock_litellm):
        """Test custom API base URL is passed to embedding calls."""
-        with patch("builtins.__import__", side_effect=lambda name, *args: mock_litellm if name == "litellm" else __import__(name, *args)):
+        with patch(
+            "builtins.__import__",
+            side_effect=lambda name, *args: mock_litellm if name == "litellm" else __import__(name, *args),
+        ):
            emb = LiteLLMSDKEmbeddings(
                api_key="test_key",
                model="cohere/embed-english-v3.0",
@@ -230,9 +230,7 @@ async def test_litellm_explicit_param_wins_over_extra_body():
    provider._acompletion = AsyncMock(return_value=_fake_litellm_response())

    with patch("hindsight_api.engine.providers.litellm_llm.get_metrics_collector"):
-        await provider.call(
-            messages=[{"role": "user", "content": "hi"}], temperature=0.9, scope="test", max_retries=0
-        )
+        await provider.call(messages=[{"role": "user", "content": "hi"}], temperature=0.9, scope="test", max_retries=0)

    assert provider._acompletion.call_args.kwargs.get("temperature") == 0.9

@@ -5,6 +5,7 @@ per-operation semaphores when `HINDSIGHT_API_{RETAIN,REFLECT,CONSOLIDATION}_LLM_
 is set. They patch the module-level semaphore registry so they can run without
 needing to re-import the module with custom env vars.
 """
+
 import asyncio
 from contextlib import AsyncExitStack
 from unittest.mock import patch
@@ -79,9 +80,7 @@ class TestSemaphoresForScope:
                "consolidation": consolidation_sem,
            },
        ):
-            assert _semaphores_for_scope("mental_model_delta_ops") == [
-                llm_wrapper._global_llm_semaphore
-            ]
+            assert _semaphores_for_scope("mental_model_delta_ops") == [llm_wrapper._global_llm_semaphore]
            assert _semaphores_for_scope("memory_think") == [llm_wrapper._global_llm_semaphore]
            assert _semaphores_for_scope("verification") == [llm_wrapper._global_llm_semaphore]

@@ -26,6 +26,7 @@ pytestmark = pytest.mark.hs_llm_mat
 _PROVIDER = os.environ.get("HINDSIGHT_API_LLM_PROVIDER", "")
 _MODEL = os.environ.get("HINDSIGHT_API_LLM_MODEL", "")

+
 def _get_api_key() -> str:
    """Get API key from HINDSIGHT_API_LLM_API_KEY (CI) or provider-specific env var."""
    key = os.environ.get("HINDSIGHT_API_LLM_API_KEY", "")
@@ -1,6 +1,7 @@
 """
 Test that LLM calls record token metrics via the metrics collector.
 """
+
 import os
 from unittest.mock import MagicMock, patch
 import pytest
@@ -31,7 +32,9 @@ async def test_llm_metrics_recorded_for_groq():
    mock_collector = MagicMock(spec=MetricsCollector)

    # Patch the provider module where get_metrics_collector is actually called
-    with patch("hindsight_api.engine.providers.openai_compatible_llm.get_metrics_collector", return_value=mock_collector):
+    with patch(
+        "hindsight_api.engine.providers.openai_compatible_llm.get_metrics_collector", return_value=mock_collector
+    ):
        llm = LLMProvider(
            provider="groq",
            api_key=api_key,
@@ -43,7 +46,7 @@ async def test_llm_metrics_recorded_for_groq():
        response = await llm.call(
            messages=[
                {"role": "system", "content": "You are a helpful assistant. Always respond."},
-                {"role": "user", "content": "What is 2+2? Reply with just the number."}
+                {"role": "user", "content": "What is 2+2? Reply with just the number."},
            ],
            max_completion_tokens=50,
            scope="test_metrics",
@@ -92,7 +95,9 @@ async def test_llm_metrics_recorded_for_structured_output():
    mock_collector = MagicMock(spec=MetricsCollector)

    # Patch the provider module where get_metrics_collector is actually called
-    with patch("hindsight_api.engine.providers.openai_compatible_llm.get_metrics_collector", return_value=mock_collector):
+    with patch(
+        "hindsight_api.engine.providers.openai_compatible_llm.get_metrics_collector", return_value=mock_collector
+    ):
        llm = LLMProvider(
            provider="groq",
            api_key=api_key,
@@ -180,7 +185,7 @@ async def test_return_usage_returns_tuple():
    result, usage = await llm.call(
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
-            {"role": "user", "content": "What is 2+2? Reply with just the number."}
+            {"role": "user", "content": "What is 2+2? Reply with just the number."},
        ],
        max_completion_tokens=50,
        return_usage=True,
@@ -51,9 +51,11 @@ class TestMockToolCalling:
        llm = LLMProvider(provider="mock", api_key="", base_url="", model="mock")

        # Set mock response to return tool calls
-        llm.set_mock_response([
-            {"name": "get_weather", "arguments": {"location": "Paris", "unit": "celsius"}},
-        ])
+        llm.set_mock_response(
+            [
+                {"name": "get_weather", "arguments": {"location": "Paris", "unit": "celsius"}},
+            ]
+        )

        result = await llm.call_with_tools(
            messages=[{"role": "user", "content": "What's the weather in Paris?"}],
@@ -105,10 +107,12 @@ class TestMockToolCalling:
        """Test handling multiple tool calls in one response."""
        llm = LLMProvider(provider="mock", api_key="", base_url="", model="mock")

-        llm.set_mock_response([
-            {"name": "get_weather", "arguments": {"location": "Paris"}},
-            {"name": "search", "arguments": {"query": "weather forecast"}},
-        ])
+        llm.set_mock_response(
+            [
+                {"name": "get_weather", "arguments": {"location": "Paris"}},
+                {"name": "search", "arguments": {"query": "weather forecast"}},
+            ]
+        )

        result = await llm.call_with_tools(
            messages=[{"role": "user", "content": "Weather in Paris and search for forecasts"}],
@@ -307,9 +307,7 @@ async def test_retain_creates_trace_rows_with_tokens(trace_api_client, bank_id):

    # Filtering by a trace_id returns only that operation run's calls.
    a_trace = entry["trace_id"]
-    resp = await trace_api_client.get(
-        f"/v1/default/banks/{bank_id}/llm-requests", params={"trace_id": a_trace}
-    )
+    resp = await trace_api_client.get(f"/v1/default/banks/{bank_id}/llm-requests", params={"trace_id": a_trace})
    assert resp.status_code == 200
    filtered = resp.json()
    assert filtered["total"] >= 1
@@ -402,9 +400,7 @@ async def test_memory_ids_mapped_to_retain_and_consolidation(trace_api_client, b
    # the retain that produced it (memory_ids) and any consolidation that consumed
    # it as a source (source_memory_ids).
    by_mem = (
-        await trace_api_client.get(
-            f"/v1/default/banks/{bank_id}/llm-requests", params={"memory_id": created[0]}
-        )
+        await trace_api_client.get(f"/v1/default/banks/{bank_id}/llm-requests", params={"memory_id": created[0]})
    ).json()
    assert by_mem["total"] >= 1
    for it in by_mem["items"]:
@@ -433,9 +429,7 @@ async def test_filter_by_status_and_operation(trace_api_client, bank_id):
        assert item["status"] == "success"
        assert item["operation"] == "retain"

-    response = await trace_api_client.get(
-        f"/v1/default/banks/{bank_id}/llm-requests", params={"status": "error"}
-    )
+    response = await trace_api_client.get(f"/v1/default/banks/{bank_id}/llm-requests", params={"status": "error"})
    assert response.json()["total"] == 0


@@ -448,9 +442,7 @@ async def test_stats_endpoint_includes_tokens(trace_api_client, bank_id):
    )
    await asyncio.sleep(1.0)

-    response = await trace_api_client.get(
-        f"/v1/default/banks/{bank_id}/llm-requests/stats", params={"period": "1d"}
-    )
+    response = await trace_api_client.get(f"/v1/default/banks/{bank_id}/llm-requests/stats", params={"period": "1d"})
    assert response.status_code == 200
    data = response.json()
    assert data["trunc"] == "day"
@@ -72,21 +72,23 @@ def create_mock_facts_from_content(content: str, ratio: float = 1.5, max_facts:
    If content has N sentences, return approximately N * ratio facts (capped at max_facts).
    """
    # Estimate sentences by splitting on periods
-    sentences = [s.strip() for s in content.split('.') if s.strip()]
+    sentences = [s.strip() for s in content.split(".") if s.strip()]
    num_facts = min(max(1, int(len(sentences) * ratio)), max_facts)

    facts = []
    for i in range(num_facts):
-        facts.append({
-            "what": f"Mock fact {i}: Something happened based on the content",
-            "when": "2024-06-15",
-            "where": "San Francisco",
-            "who": "John, Sarah",
-            "why": "Business reasons",
-            "fact_type": "world",
-            "entities": [{"text": "John", "type": "PERSON"}],
-            "causal_relations": [],
-        })
+        facts.append(
+            {
+                "what": f"Mock fact {i}: Something happened based on the content",
+                "when": "2024-06-15",
+                "where": "San Francisco",
+                "who": "John, Sarah",
+                "why": "Business reasons",
+                "fact_type": "world",
+                "entities": [{"text": "John", "type": "PERSON"}],
+                "causal_relations": [],
+            }
+        )

    return facts

@@ -122,6 +124,7 @@ class TestLargeBatchRetain:
    @pytest.fixture
    def disable_observations(self):
        from hindsight_api.config import _get_raw_config
+
        config = _get_raw_config()
        original = config.enable_observations
        config.enable_observations = False
@@ -147,11 +150,13 @@ class TestLargeBatchRetain:
        contents = []
        for i in range(num_items):
            content_text = generate_content(chars_per_item)
-            contents.append({
-                "content": content_text,
-                "context": f"Test content item {i + 1} of {num_items}",
-                "event_date": datetime.now(UTC),
-            })
+            contents.append(
+                {
+                    "content": content_text,
+                    "context": f"Test content item {i + 1} of {num_items}",
+                    "event_date": datetime.now(UTC),
+                }
+            )

        actual_total_chars = sum(len(c["content"]) for c in contents)
        logger.info(f"Created {num_items} content items with {actual_total_chars:,} total chars")
@@ -191,7 +196,7 @@ class TestLargeBatchRetain:
            return response_dict

        # Patch LLMProvider.call at the class level
-        with patch('hindsight_api.engine.llm_wrapper.LLMProvider.call', new=mock_llm_call):
+        with patch("hindsight_api.engine.llm_wrapper.LLMProvider.call", new=mock_llm_call):
            start_time = time.time()

            try:
@@ -247,11 +252,13 @@ class TestLargeBatchRetain:

        contents = []
        for i in range(num_items):
-            contents.append({
-                "content": generate_content(chars_per_item),
-                "context": f"Chunk test item {i + 1}",
-                "event_date": datetime.now(UTC),
-            })
+            contents.append(
+                {
+                    "content": generate_content(chars_per_item),
+                    "context": f"Chunk test item {i + 1}",
+                    "event_date": datetime.now(UTC),
+                }
+            )

        actual_total_chars = sum(len(c["content"]) for c in contents)
        logger.info(f"Created {num_items} items with {actual_total_chars:,} chars (should trigger chunking)")
@@ -275,7 +282,7 @@ class TestLargeBatchRetain:
                return response_dict, TokenUsage(input_tokens=100, output_tokens=50)
            return response_dict

-        with patch('hindsight_api.engine.llm_wrapper.LLMProvider.call', new=mock_llm_call):
+        with patch("hindsight_api.engine.llm_wrapper.LLMProvider.call", new=mock_llm_call):
            start_time = time.time()

            result = await memory.retain_batch_async(
@@ -305,9 +312,18 @@ class TestLargeBatchRetain:
        async def mock_llm_call(*args, **kwargs):
            # Small delay to simulate real LLM latency
            await asyncio.sleep(0.01)
-            mock_facts = [{"what": "Test fact", "when": "now", "where": "here",
-                          "who": "someone", "why": "testing", "fact_type": "world",
-                          "entities": [], "causal_relations": []}]
+            mock_facts = [
+                {
+                    "what": "Test fact",
+                    "when": "now",
+                    "where": "here",
+                    "who": "someone",
+                    "why": "testing",
+                    "fact_type": "world",
+                    "entities": [],
+                    "causal_relations": [],
+                }
+            ]
            response_dict = {"facts": mock_facts}

            return_usage = kwargs.get("return_usage", False)
@@ -315,16 +331,18 @@ class TestLargeBatchRetain:
                return response_dict, TokenUsage(input_tokens=10, output_tokens=10)
            return response_dict

-        with patch('hindsight_api.engine.llm_wrapper.LLMProvider.call', new=mock_llm_call):
+        with patch("hindsight_api.engine.llm_wrapper.LLMProvider.call", new=mock_llm_call):
            # Run 10 concurrent retain operations
            tasks = []
            for i in range(10):
                bank_id = f"pool-test-{uuid.uuid4().hex[:8]}"
-                contents = [{
-                    "content": f"Test content for concurrent operation {i}. " * 50,
-                    "context": f"Pool test {i}",
-                    "event_date": datetime.now(UTC),
-                }]
+                contents = [
+                    {
+                        "content": f"Test content for concurrent operation {i}. " * 50,
+                        "context": f"Pool test {i}",
+                        "event_date": datetime.now(UTC),
+                    }
+                ]
                tasks.append(
                    memory.retain_batch_async(bank_id=bank_id, contents=contents, request_context=request_context)
                )
@@ -43,14 +43,15 @@ class TestMainModuleExtensionLoading:
            loaded_extensions[name] = result
            return result

-        with patch("hindsight_api.main.MemoryEngine") as mock_engine, \
-             patch("hindsight_api.main.create_app") as mock_create_app, \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.load_extension", side_effect=tracking_load_extension), \
-             patch("hindsight_api.main.DefaultExtensionContext"), \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run"):  # Don't actually start uvicorn
-
+        with (
+            patch("hindsight_api.main.MemoryEngine") as mock_engine,
+            patch("hindsight_api.main.create_app") as mock_create_app,
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.load_extension", side_effect=tracking_load_extension),
+            patch("hindsight_api.main.DefaultExtensionContext"),
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run"),
+        ):  # Don't actually start uvicorn
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -63,17 +64,21 @@ class TestMainModuleExtensionLoading:
            mock_create_app.return_value = MagicMock()

            # Mock sys.argv to simulate CLI invocation
-            with patch.object(sys, 'argv', ['hindsight-api']):
+            with patch.object(sys, "argv", ["hindsight-api"]):
                from hindsight_api.main import main
+
                main()

        # Verify TENANT extension was loaded
-        assert "TENANT" in loaded_extensions, \
+        assert "TENANT" in loaded_extensions, (
            "main.py did not call load_extension('TENANT', ...) - extensions not loaded!"
-        assert loaded_extensions["TENANT"] is not None, \
+        )
+        assert loaded_extensions["TENANT"] is not None, (
            "load_extension('TENANT', ...) returned None despite env var being set"
-        assert isinstance(loaded_extensions["TENANT"], MockTenantExtension), \
+        )
+        assert isinstance(loaded_extensions["TENANT"], MockTenantExtension), (
            f"Expected MockTenantExtension, got {type(loaded_extensions['TENANT'])}"
+        )

    def test_main_loads_operation_validator_when_configured(self, monkeypatch):
        """
@@ -94,14 +99,15 @@ class TestMainModuleExtensionLoading:
            loaded_extensions[name] = result
            return result

-        with patch("hindsight_api.main.MemoryEngine") as mock_engine, \
-             patch("hindsight_api.main.create_app") as mock_create_app, \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.load_extension", side_effect=tracking_load_extension), \
-             patch("hindsight_api.main.DefaultExtensionContext"), \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run"):
-
+        with (
+            patch("hindsight_api.main.MemoryEngine") as mock_engine,
+            patch("hindsight_api.main.create_app") as mock_create_app,
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.load_extension", side_effect=tracking_load_extension),
+            patch("hindsight_api.main.DefaultExtensionContext"),
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run"),
+        ):
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -113,12 +119,14 @@ class TestMainModuleExtensionLoading:
            mock_engine.return_value = MagicMock()
            mock_create_app.return_value = MagicMock()

-            with patch.object(sys, 'argv', ['hindsight-api']):
+            with patch.object(sys, "argv", ["hindsight-api"]):
                from hindsight_api.main import main
+
                main()

-        assert "OPERATION_VALIDATOR" in loaded_extensions, \
+        assert "OPERATION_VALIDATOR" in loaded_extensions, (
            "main.py did not call load_extension('OPERATION_VALIDATOR', ...)"
+        )
        assert loaded_extensions["OPERATION_VALIDATOR"] is not None
        assert isinstance(loaded_extensions["OPERATION_VALIDATOR"], MockOperationValidator)

@@ -141,13 +149,14 @@ class TestMainModuleExtensionLoading:
            memory_engine_calls.append({"args": args, "kwargs": kwargs})
            return MagicMock()

-        with patch("hindsight_api.main.MemoryEngine", side_effect=capture_memory_engine), \
-             patch("hindsight_api.main.create_app") as mock_create_app, \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.DefaultExtensionContext"), \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run"):
-
+        with (
+            patch("hindsight_api.main.MemoryEngine", side_effect=capture_memory_engine),
+            patch("hindsight_api.main.create_app") as mock_create_app,
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.DefaultExtensionContext"),
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run"),
+        ):
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -158,8 +167,9 @@ class TestMainModuleExtensionLoading:
            mock_get_config.return_value = mock_config
            mock_create_app.return_value = MagicMock()

-            with patch.object(sys, 'argv', ['hindsight-api']):
+            with patch.object(sys, "argv", ["hindsight-api"]):
                from hindsight_api.main import main
+
                main()

        # Verify MemoryEngine was called
@@ -168,10 +178,10 @@ class TestMainModuleExtensionLoading:
        call_kwargs = memory_engine_calls[0]["kwargs"]

        # THE CRITICAL ASSERTION: tenant_extension must be passed and not None
-        assert "tenant_extension" in call_kwargs, \
-            "MemoryEngine was not called with tenant_extension parameter!"
-        assert call_kwargs["tenant_extension"] is not None, \
+        assert "tenant_extension" in call_kwargs, "MemoryEngine was not called with tenant_extension parameter!"
+        assert call_kwargs["tenant_extension"] is not None, (
            "tenant_extension was None - main.py did not pass loaded extension to MemoryEngine!"
+        )

    def test_main_sets_extension_context_on_tenant_extension(self, monkeypatch):
        """
@@ -198,13 +208,14 @@ class TestMainModuleExtensionLoading:
            context_created.append(ctx)
            return ctx

-        with patch("hindsight_api.main.MemoryEngine", side_effect=capture_memory_engine), \
-             patch("hindsight_api.main.create_app") as mock_create_app, \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.DefaultExtensionContext", side_effect=capture_context), \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run"):
-
+        with (
+            patch("hindsight_api.main.MemoryEngine", side_effect=capture_memory_engine),
+            patch("hindsight_api.main.create_app") as mock_create_app,
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.DefaultExtensionContext", side_effect=capture_context),
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run"),
+        ):
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -215,15 +226,15 @@ class TestMainModuleExtensionLoading:
            mock_get_config.return_value = mock_config
            mock_create_app.return_value = MagicMock()

-            with patch.object(sys, 'argv', ['hindsight-api']):
+            with patch.object(sys, "argv", ["hindsight-api"]):
                from hindsight_api.main import main
+
                main()

        # Verify context was created and set
        assert len(context_created) == 1, "DefaultExtensionContext should be created"
        assert captured_tenant_ext[0] is not None, "Tenant extension should be captured"
-        assert captured_tenant_ext[0]._context_set, \
-            "set_context was not called on tenant extension"
+        assert captured_tenant_ext[0]._context_set, "set_context was not called on tenant extension"

    def test_main_works_without_extensions(self, monkeypatch):
        """
@@ -240,12 +251,13 @@ class TestMainModuleExtensionLoading:
            memory_engine_calls.append({"args": args, "kwargs": kwargs})
            return MagicMock()

-        with patch("hindsight_api.main.MemoryEngine", side_effect=capture_memory_engine), \
-             patch("hindsight_api.main.create_app") as mock_create_app, \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run"):
-
+        with (
+            patch("hindsight_api.main.MemoryEngine", side_effect=capture_memory_engine),
+            patch("hindsight_api.main.create_app") as mock_create_app,
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run"),
+        ):
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -256,8 +268,9 @@ class TestMainModuleExtensionLoading:
            mock_get_config.return_value = mock_config
            mock_create_app.return_value = MagicMock()

-            with patch.object(sys, 'argv', ['hindsight-api']):
+            with patch.object(sys, "argv", ["hindsight-api"]):
                from hindsight_api.main import main
+
                main()

        # Should work without extensions
@@ -285,12 +298,13 @@ class TestMainModuleExtensionLoading:

        mock_app = MagicMock()

-        with patch("hindsight_api.main.MemoryEngine") as mock_engine, \
-             patch("hindsight_api.main.create_app", return_value=mock_app), \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run", side_effect=capture_uvicorn_run):
-
+        with (
+            patch("hindsight_api.main.MemoryEngine") as mock_engine,
+            patch("hindsight_api.main.create_app", return_value=mock_app),
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run", side_effect=capture_uvicorn_run),
+        ):
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -301,14 +315,14 @@ class TestMainModuleExtensionLoading:
            mock_get_config.return_value = mock_config
            mock_engine.return_value = MagicMock()

-            with patch.object(sys, 'argv', ['hindsight-api', '--workers', '1']):
+            with patch.object(sys, "argv", ["hindsight-api", "--workers", "1"]):
                from hindsight_api.main import main
+
                main()

        assert len(uvicorn_calls) == 1
        # With workers=1, should pass app object, not import string
-        assert uvicorn_calls[0]["app"] is mock_app, \
-            "main.py should pass app object (not import string) when workers=1"
+        assert uvicorn_calls[0]["app"] is mock_app, "main.py should pass app object (not import string) when workers=1"

    def test_main_uses_import_string_for_multiple_workers(self, monkeypatch):
        """
@@ -325,12 +339,13 @@ class TestMainModuleExtensionLoading:
        def capture_uvicorn_run(**kwargs):
            uvicorn_calls.append(kwargs)

-        with patch("hindsight_api.main.MemoryEngine") as mock_engine, \
-             patch("hindsight_api.main.create_app") as mock_create_app, \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run", side_effect=capture_uvicorn_run):
-
+        with (
+            patch("hindsight_api.main.MemoryEngine") as mock_engine,
+            patch("hindsight_api.main.create_app") as mock_create_app,
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run", side_effect=capture_uvicorn_run),
+        ):
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -342,14 +357,16 @@ class TestMainModuleExtensionLoading:
            mock_engine.return_value = MagicMock()
            mock_create_app.return_value = MagicMock()

-            with patch.object(sys, 'argv', ['hindsight-api', '--workers', '2']):
+            with patch.object(sys, "argv", ["hindsight-api", "--workers", "2"]):
                from hindsight_api.main import main
+
                main()

        assert len(uvicorn_calls) == 1
        # With workers > 1, should use import string
-        assert uvicorn_calls[0]["app"] == "hindsight_api.server:app", \
+        assert uvicorn_calls[0]["app"] == "hindsight_api.server:app", (
            "main.py should use import string when workers > 1"
+        )
        assert uvicorn_calls[0]["workers"] == 2

    def test_main_sets_keepalive_timeout(self, monkeypatch):
@@ -366,12 +383,13 @@ class TestMainModuleExtensionLoading:
        def capture_uvicorn_run(**kwargs):
            uvicorn_calls.append(kwargs)

-        with patch("hindsight_api.main.MemoryEngine") as mock_engine, \
-             patch("hindsight_api.main.create_app") as mock_create_app, \
-             patch("hindsight_api.main._get_raw_config") as mock_get_config, \
-             patch("hindsight_api.main.print_banner"), \
-             patch("uvicorn.run", side_effect=capture_uvicorn_run):
-
+        with (
+            patch("hindsight_api.main.MemoryEngine") as mock_engine,
+            patch("hindsight_api.main.create_app") as mock_create_app,
+            patch("hindsight_api.main._get_raw_config") as mock_get_config,
+            patch("hindsight_api.main.print_banner"),
+            patch("uvicorn.run", side_effect=capture_uvicorn_run),
+        ):
            mock_config = MagicMock()
            mock_config.host = "0.0.0.0"
            mock_config.port = 8888
@@ -383,15 +401,16 @@ class TestMainModuleExtensionLoading:
            mock_engine.return_value = MagicMock()
            mock_create_app.return_value = MagicMock()

-            with patch.object(sys, 'argv', ['hindsight-api']):
+            with patch.object(sys, "argv", ["hindsight-api"]):
                from hindsight_api.main import main
+
                main()

        assert len(uvicorn_calls) == 1
-        assert "timeout_keep_alive" in uvicorn_calls[0], \
-            "uvicorn config must set timeout_keep_alive"
-        assert uvicorn_calls[0]["timeout_keep_alive"] > 15, \
+        assert "timeout_keep_alive" in uvicorn_calls[0], "uvicorn config must set timeout_keep_alive"
+        assert uvicorn_calls[0]["timeout_keep_alive"] > 15, (
            "timeout_keep_alive must exceed aiohttp's 15s client default"
+        )


 # Mock extensions for testing
--- a/Show More
+++ b/Show More