Compare commits

..
Author SHA1 Message Date
Nicolò Boschi 4a9917bfd6 doc: add faq page 2026-02-16 16:55:23 +01:00
1503 changed files with 41630 additions and 229717 deletions
-15
View File
@@ -1,15 +0,0 @@
{
"$schema": "https://anthropic.com/claude-code/marketplace.schema.json",
"name": "hindsight",
"description": "Official Hindsight integrations for Claude Code",
"owner": {
"name": "vectorize-io"
},
"plugins": [
{
"name": "hindsight-memory",
"description": "Automatic long-term memory for Claude Code via Hindsight",
"source": "./hindsight-integrations/claude-code"
}
]
}
-205
View File
@@ -1,205 +0,0 @@
---
name: code-review
description: Review changed code against project standards. Checks for missing tests, dead code, type safety, lint issues, and coding conventions. Run after completing any implementation work.
user_invocable: true
---
# Code Review
Review all changed code against the project's quality standards and coding conventions.
## Code Standards
Read and internalize these standards before writing code. The review steps below verify compliance.
### Python Style
- Python 3.11+, type hints required
- Async throughout (asyncpg, async FastAPI)
- Pydantic models for request/response
- Ruff for linting (line-length 120)
- No Python files at project root - maintain clean directory structure
- **Never use multi-item tuple return values** — not even for internal/private functions. Always use a dataclass or Pydantic model. No exceptions, no "it's just two values" shortcuts. If a function returns more than one value, define a named type for it.
### Type Safety with Pydantic Models
**NEVER use raw `dict` types for structured data** — this applies to all code, including internal helpers and private functions. If the dict has known keys, it must be a dataclass or Pydantic model:
- Use Pydantic `BaseModel` for all data structures passed between functions
- Use `@dataclass` for lightweight internal data containers when Pydantic validation isn't needed
- Add `@field_validator` for type coercion (e.g., ensuring datetimes are timezone-aware)
- Avoid `dict.get()` patterns - use typed model attributes instead
- Parse external data (JSON, API responses) into Pydantic models at the boundary
- This catches type errors at parse time, not deep in business logic
- The only acceptable `dict` usage is for truly dynamic/unknown keys (e.g., arbitrary metadata, JSON blobs with no fixed schema)
```python
# BAD - error-prone dict access
def process(data: dict) -> str:
return data.get("name", "") # No validation, silent failures
# GOOD - typed and validated
class UserData(BaseModel):
name: str
created_at: datetime
@field_validator("created_at", mode="before")
@classmethod
def ensure_tz_aware(cls, v):
if isinstance(v, str):
v = datetime.fromisoformat(v.replace("Z", "+00:00"))
if v.tzinfo is None:
return v.replace(tzinfo=timezone.utc)
return v
def process(data: UserData) -> str:
return data.name # Type-safe, validated at construction
```
### TypeScript Style
- Next.js App Router for control plane
- Tailwind CSS with shadcn/ui components
### Code Comments
- **Always comment non-trivial technical decisions** with the reasoning behind the choice. If someone would ask "why is it done this way?", there should be a comment.
- **Keep comments up to date with history** — when changing an approach, update the comment to explain what was tried before and why it was changed. Comments serve as a tracker of previous implementations that likely had problems.
- Don't comment obvious code — only where the "why" isn't self-evident from the code itself.
```python
# BAD - no context for future readers
results = await asyncio.gather(*tasks, return_exceptions=True)
# GOOD - explains the non-obvious choice
# Use return_exceptions=True to avoid cancelling sibling tasks on failure.
# Previously we used TaskGroup but it cancelled all tasks when one failed,
# causing partial writes that left orphaned entity links (see #412).
results = await asyncio.gather(*tasks, return_exceptions=True)
```
### Branch Hygiene
- **Always start new feature branches from `origin/main`** — rebase to ensure a clean base.
- **Only include commits relevant to the PR/branch/feature** — no unrelated changes. If the branch contains commits that don't belong, they must be removed before merging.
### General Principles
- Don't add features, refactor code, or make "improvements" beyond what was asked
- Don't add unnecessary error handling for impossible scenarios
- Don't create helpers or abstractions for one-time operations
- No backwards-compatibility hacks (unused vars, re-exports, "removed" comments)
- Three similar lines of code is better than a premature abstraction
## Review Steps
### 1. Check branch hygiene
- Run `git log --oneline main..HEAD` to list all commits on the branch.
- Verify every commit is relevant to the feature/PR. Flag any unrelated commits.
- Check the branch is based on a recent `origin/main` (no stale base).
### 2. Identify changed files
Run `git diff --name-only HEAD` (unstaged) and `git diff --cached --name-only` (staged) to get all changed files. If there are no local changes, diff against the base branch using `git diff main...HEAD --name-only` and `git diff main...HEAD` to review all commits on the current branch.
### 3. Run linters
```bash
./scripts/hooks/lint.sh
```
Report any failures. Do NOT fix them yourself — just report.
### 4. Check for dead code
For each changed Python file, check for:
- Unused imports (Ruff should catch these, but verify)
- Functions/methods/classes that were added but are never called from anywhere
- Variables assigned but never read
- Commented-out code blocks that should be removed
For each changed TypeScript file, check for:
- Unused imports
- Unused variables or functions
- Commented-out code
### 5. Check type safety (Python)
For each changed Python file, check for violations:
- **No raw `dict` for structured data** — must use Pydantic model or dataclass, even for internal/private functions (only exception: truly dynamic/unknown keys)
- **No multi-item tuple returns** — must use dataclass or Pydantic model, even for internal/private functions (no exceptions)
- **Missing type hints** on function parameters and return types
- **Missing `@field_validator`** for datetime fields that should be timezone-aware
### 6. Check for missing tests
For each new or significantly changed function/endpoint/class:
- Check if there is a corresponding test addition or update
- New API endpoints MUST have integration tests
- New utility functions MUST have unit tests
- Bug fixes SHOULD have a regression test
Flag any new logic that lacks test coverage.
### 7. Check API consistency
If any files in `hindsight-api-slim/hindsight_api/api/` were changed:
- Were the OpenAPI specs regenerated? (`./scripts/generate-openapi.sh`)
- Were the client SDKs regenerated? (`./scripts/generate-clients.sh`)
- Were the control plane proxy routes updated? (`hindsight-control-plane/src/app/api/`)
### 8. Check code comments
For each non-trivial change:
- **New non-obvious logic** — is there a comment explaining the reasoning?
- **Changed approach** — does the comment include what was done before and why it changed?
- **Stale comments** — do existing comments near the changed code still accurately describe the behavior?
### 9. Check integration completeness
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.
- **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
If any new MCP tools were added or existing tools renamed in `hindsight-api-slim/hindsight_api/mcp_tools.py`:
- **`_ALL_TOOLS` set** in `mcp_tools.py` — must include the new tool name
- **`tools_to_register` default set** in `register_mcp_tools()` in `mcp_tools.py` — must include the new tool name
- **`_SINGLE_BANK_TOOLS` set** in `hindsight-api-slim/hindsight_api/api/mcp.py` — must include the new tool if it is bank-scoped (not a bank-management tool like `list_banks`/`create_bank`)
- **`MCP_TOOL_GROUPS`** in `hindsight-control-plane/src/components/bank-config-view.tsx` — must include the new tool in the appropriate group for the UI tool selector
- **Tool count assertions** in tests (e.g., `test_mcp_tools.py`) — must be updated to reflect the new count
### 11. Review against other coding standards
Check the diff for violations of the standards listed above:
- Python files at project root (not allowed)
- Missing async patterns (should be async throughout)
- Pydantic models for request/response
- Line length > 120 chars
- New features/code beyond what was asked (over-engineering)
- Unnecessary error handling for impossible scenarios
- Premature abstractions or speculative helpers
- Backwards-compatibility hacks (unused vars, re-exports, "removed" comments)
### 12. Report findings
Present a clear summary organized by severity:
**Must fix** — issues that will break CI or violate hard project rules:
- Unrelated commits on the branch
- Lint failures
- Missing type hints on public functions
- Raw dict usage for structured data (including internal code)
- Multi-item tuple returns (including internal code)
- Missing tests for new endpoints
- New integration missing tests, CI job, or release-integration.sh entry
**Should fix** — issues that hurt code quality:
- Dead code / unused imports missed by linter
- Missing tests for non-trivial utility functions
- Over-engineering beyond the task scope
**Note** — observations that may or may not need action:
- API changes that might need client regeneration
- Patterns that deviate from nearby code style
For each finding, include the file path, line number, and a brief explanation.
Do NOT auto-fix any issues. Report all findings and let the user decide what to address. If there are no findings, confirm the code looks good.
+2 -14
View File
@@ -2,10 +2,10 @@
# Copy this file to .env and fill in your values
# LLM Configuration (Required)
# Supported providers: openai, groq, ollama, gemini, anthropic, lmstudio, vertexai, minimax, volcano
# Supported providers: openai, groq, ollama, gemini, anthropic, lmstudio, vertexai
HINDSIGHT_API_LLM_PROVIDER=openai
HINDSIGHT_API_LLM_API_KEY=your-api-key-here
HINDSIGHT_API_LLM_MODEL=gpt-4o-mini
HINDSIGHT_API_LLM_MODEL=o3-mini
HINDSIGHT_API_LLM_BASE_URL=https://api.openai.com/v1
# Example: Anthropic Claude configuration
@@ -20,11 +20,6 @@ HINDSIGHT_API_LLM_BASE_URL=https://api.openai.com/v1
# HINDSIGHT_API_LLM_VERTEXAI_REGION=us-central1
# HINDSIGHT_API_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY=/path/to/service-account-key.json # Optional, uses ADC if not set
# Example: MiniMax configuration (1M context window)
# HINDSIGHT_API_LLM_PROVIDER=minimax
# HINDSIGHT_API_LLM_API_KEY=your-minimax-api-key
# HINDSIGHT_API_LLM_MODEL=MiniMax-M2.7
# Example: LM Studio local configuration (Qwen 2.5 32B recommended)
# HINDSIGHT_API_LLM_PROVIDER=lmstudio
# HINDSIGHT_API_LLM_API_KEY=lmstudio
@@ -44,15 +39,8 @@ HINDSIGHT_API_LOG_LEVEL=info
# Database (Optional - uses embedded pg0 by default)
# HINDSIGHT_API_DATABASE_URL=postgresql://user:pass@host:5432/db
# HINDSIGHT_API_MIGRATION_DATABASE_URL= # Direct PostgreSQL URL for migrations (bypasses PgBouncer). Falls back to DATABASE_URL.
# HINDSIGHT_API_DATABASE_SCHEMA=public # PostgreSQL schema name (default: public)
# Vector Extension (Optional - uses pgvector by default)
# Options: "pgvector" (default), "vchord", "pgvectorscale" (DiskANN)
# HINDSIGHT_API_VECTOR_EXTENSION=pgvector
# For Azure PostgreSQL with DiskANN:
# HINDSIGHT_API_VECTOR_EXTENSION=pgvectorscale # Auto-detects pg_diskann on Azure
# Embeddings Configuration (Optional - uses local by default)
# Provider: "local" (default) or "tei" (HuggingFace Text Embeddings Inference)
# HINDSIGHT_API_EMBEDDINGS_PROVIDER=local
-6
View File
@@ -1,6 +0,0 @@
version: 2
updates:
- package-ecosystem: "github-actions"
directory: "/"
schedule:
interval: "weekly"
+5 -8
View File
@@ -21,20 +21,17 @@ jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
- uses: actions/setup-node@v6
- uses: actions/checkout@v4
- uses: actions/setup-node@v4
with:
node-version: 20
cache: npm
cache-dependency-path: package-lock.json
- uses: astral-sh/setup-uv@v7
- uses: astral-sh/setup-uv@v4
- run: npm ci --workspace=hindsight-docs
- run: uv run generate-llms-full
- run: npm run build --workspace=hindsight-docs
env:
UMAMI_URL: https://analytics.hindsight.vectorize.io
UMAMI_WEBSITE_ID: ${{ secrets.UMAMI_WEBSITE_ID }}
- uses: actions/upload-pages-artifact@v4
- uses: actions/upload-pages-artifact@v3
with:
path: hindsight-docs/build
deploy:
@@ -44,5 +41,5 @@ jobs:
runs-on: ubuntu-latest
needs: build
steps:
- uses: actions/deploy-pages@v5
- uses: actions/deploy-pages@v4
id: deployment
-120
View File
@@ -1,120 +0,0 @@
name: Release Integration
on:
push:
tags:
- 'integrations/**'
jobs:
publish:
runs-on: ubuntu-latest
permissions:
id-token: write # for PyPI trusted publishing
steps:
- uses: actions/checkout@v6
- name: Extract integration info
id: info
run: |
# refs/tags/integrations/litellm/v0.1.0 → integration=litellm, version=0.1.0
TAG="${GITHUB_REF#refs/tags/}"
INTEGRATION=$(echo "$TAG" | cut -d'/' -f2)
VERSION=$(echo "$TAG" | cut -d'/' -f3 | sed 's/^v//')
echo "integration=$INTEGRATION" >> $GITHUB_OUTPUT
echo "version=$VERSION" >> $GITHUB_OUTPUT
echo "tag=$TAG" >> $GITHUB_OUTPUT
echo "Integration: $INTEGRATION, Version: $VERSION"
- name: Detect integration type
id: type
run: |
if [ -f "hindsight-integrations/${{ steps.info.outputs.integration }}/pyproject.toml" ]; then
echo "type=python" >> $GITHUB_OUTPUT
elif [ -f "hindsight-integrations/${{ steps.info.outputs.integration }}/package.json" ]; then
echo "type=typescript" >> $GITHUB_OUTPUT
else
echo "type=plugin" >> $GITHUB_OUTPUT
fi
# ── Python integrations (litellm, pydantic-ai, crewai) ──────────────────
- name: Install uv
if: steps.type.outputs.type == 'python'
uses: astral-sh/setup-uv@v7
with:
enable-cache: true
- name: Set up Python
if: steps.type.outputs.type == 'python'
uses: actions/setup-python@v6
with:
python-version-file: ".python-version"
- name: Build Python package
if: steps.type.outputs.type == 'python'
working-directory: ./hindsight-integrations/${{ steps.info.outputs.integration }}
run: uv build --out-dir dist
- name: Publish Python package to PyPI
if: steps.type.outputs.type == 'python'
uses: pypa/gh-action-pypi-publish@release/v1
with:
packages-dir: ./hindsight-integrations/${{ steps.info.outputs.integration }}/dist
skip-existing: true
# ── TypeScript integrations (ai-sdk, chat, openclaw) ────────────────────
# ── Plugin integrations (claude-code) — no package to publish ───────────
- name: Plugin release
if: steps.type.outputs.type == 'plugin'
run: |
echo "Plugin integration ${{ steps.info.outputs.integration }} v${{ steps.info.outputs.version }} — no package to publish."
echo "Users install via: claude plugin marketplace add vectorize-io/hindsight --sparse hindsight-integrations"
# ── TypeScript integrations (ai-sdk, chat, openclaw) ────────────────────
- name: Set up Node.js
if: steps.type.outputs.type == 'typescript'
uses: actions/setup-node@v6
with:
node-version: '22'
registry-url: 'https://registry.npmjs.org'
# Guard: fail fast if the integration's lockfile resolves any dep from a
# monorepo workspace (link=true) or a relative file path. The release
# runner has no pre-built workspace `dist/` so `npm run build` would
# later fail at tsc with "Cannot find module". See:
# https://github.com/vectorize-io/hindsight/issues/… (0.6.0 openclaw retry)
- name: Check integration lockfile
if: steps.type.outputs.type == 'typescript'
run: ./scripts/check-integration-lockfiles.sh
- name: Install dependencies
if: steps.type.outputs.type == 'typescript'
working-directory: ./hindsight-integrations/${{ steps.info.outputs.integration }}
run: npm ci
- name: Build TypeScript package
if: steps.type.outputs.type == 'typescript'
working-directory: ./hindsight-integrations/${{ steps.info.outputs.integration }}
run: npm run build
- name: Publish TypeScript package to npm
if: steps.type.outputs.type == 'typescript'
working-directory: ./hindsight-integrations/${{ steps.info.outputs.integration }}
run: |
set +e
OUTPUT=$(npm publish --access public 2>&1)
EXIT_CODE=$?
echo "$OUTPUT"
if [ $EXIT_CODE -ne 0 ]; then
if echo "$OUTPUT" | grep -q "cannot publish over"; then
echo "Package version already published, skipping..."
exit 0
fi
exit $EXIT_CODE
fi
env:
NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
+121 -83
View File
@@ -13,15 +13,15 @@ jobs:
id-token: write
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Install uv
uses: astral-sh/setup-uv@v7
uses: astral-sh/setup-uv@v5
with:
enable-cache: true
- name: Set up Python
uses: actions/setup-python@v6
uses: actions/setup-python@v5
with:
python-version-file: ".python-version"
@@ -30,39 +30,29 @@ jobs:
working-directory: ./hindsight-clients/python
run: uv build --out-dir dist
- name: Build hindsight-api-slim
working-directory: ./hindsight-api-slim
run: uv build --out-dir dist
- name: Build hindsight-api
working-directory: ./hindsight-api
run: uv build --out-dir dist
- name: Build hindsight-all
working-directory: ./hindsight-all
working-directory: ./hindsight
run: uv build --out-dir dist
- name: Build hindsight-all-slim
working-directory: ./hindsight-all-slim
- name: Build hindsight-litellm
working-directory: ./hindsight-integrations/litellm
run: uv build --out-dir dist
- name: Build hindsight-embed
working-directory: ./hindsight-embed
run: uv build --out-dir dist
# Publish in order (client and api-slim first, then api/all wrappers which depend on them)
# Publish in order (client and api first, then hindsight-all which depends on them)
- name: Publish hindsight-client to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
packages-dir: ./hindsight-clients/python/dist
skip-existing: true
- name: Publish hindsight-api-slim to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
packages-dir: ./hindsight-api-slim/dist
skip-existing: true
- name: Publish hindsight-api to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
@@ -72,13 +62,13 @@ jobs:
- name: Publish hindsight-all to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
packages-dir: ./hindsight-all/dist
packages-dir: ./hindsight/dist
skip-existing: true
- name: Publish hindsight-all-slim to PyPI
- name: Publish hindsight-litellm to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
packages-dir: ./hindsight-all-slim/dist
packages-dir: ./hindsight-integrations/litellm/dist
skip-existing: true
- name: Publish hindsight-embed to PyPI
@@ -89,15 +79,14 @@ jobs:
# Upload artifacts for GitHub release
- name: Upload artifacts
uses: actions/upload-artifact@v7
uses: actions/upload-artifact@v4
with:
name: python-packages
path: |
hindsight-clients/python/dist/*
hindsight-api-slim/dist/*
hindsight-api/dist/*
hindsight-all/dist/*
hindsight-all-slim/dist/*
hindsight/dist/*
hindsight-integrations/litellm/dist/*
hindsight-embed/dist/*
retention-days: 1
@@ -106,10 +95,10 @@ jobs:
environment: npm
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Set up Node.js
uses: actions/setup-node@v6
uses: actions/setup-node@v4
with:
node-version: '20'
registry-url: 'https://registry.npmjs.org'
@@ -144,35 +133,35 @@ jobs:
run: npm pack
- name: Upload artifacts
uses: actions/upload-artifact@v7
uses: actions/upload-artifact@v4
with:
name: typescript-client
path: hindsight-clients/typescript/*.tgz
retention-days: 1
release-hindsight-all-npm:
release-openclaw-integration:
runs-on: ubuntu-latest
environment: npm
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Set up Node.js
uses: actions/setup-node@v6
uses: actions/setup-node@v4
with:
node-version: '22'
registry-url: 'https://registry.npmjs.org'
cache: 'npm'
cache-dependency-path: package-lock.json
- name: Install dependencies
run: npm ci --workspace=hindsight-all-npm
working-directory: ./hindsight-integrations/openclaw
run: npm ci
- name: Build
run: npm run build --workspace=hindsight-all-npm
working-directory: ./hindsight-integrations/openclaw
run: npm run build
- name: Publish to npm
working-directory: ./hindsight-all-npm
working-directory: ./hindsight-integrations/openclaw
run: |
set +e
OUTPUT=$(npm publish --access public 2>&1)
@@ -189,14 +178,63 @@ jobs:
NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
- name: Pack for GitHub release
working-directory: ./hindsight-all-npm
working-directory: ./hindsight-integrations/openclaw
run: npm pack
- name: Upload artifacts
uses: actions/upload-artifact@v7
uses: actions/upload-artifact@v4
with:
name: hindsight-all-npm
path: hindsight-all-npm/*.tgz
name: openclaw-integration
path: hindsight-integrations/openclaw/*.tgz
retention-days: 1
release-ai-sdk-integration:
runs-on: ubuntu-latest
environment: npm
steps:
- uses: actions/checkout@v4
- name: Set up Node.js
uses: actions/setup-node@v4
with:
node-version: '22'
registry-url: 'https://registry.npmjs.org'
- name: Install dependencies
working-directory: ./hindsight-integrations/ai-sdk
run: npm ci
- name: Build
working-directory: ./hindsight-integrations/ai-sdk
run: npm run build
- name: Publish to npm
working-directory: ./hindsight-integrations/ai-sdk
run: |
set +e
OUTPUT=$(npm publish --access public 2>&1)
EXIT_CODE=$?
echo "$OUTPUT"
if [ $EXIT_CODE -ne 0 ]; then
if echo "$OUTPUT" | grep -q "cannot publish over"; then
echo "Package version already published, skipping..."
exit 0
fi
exit $EXIT_CODE
fi
env:
NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
- name: Pack for GitHub release
working-directory: ./hindsight-integrations/ai-sdk
run: npm pack
- name: Upload artifacts
uses: actions/upload-artifact@v4
with:
name: ai-sdk-integration
path: hindsight-integrations/ai-sdk/*.tgz
retention-days: 1
release-control-plane:
@@ -204,10 +242,10 @@ jobs:
environment: npm
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Set up Node.js
uses: actions/setup-node@v6
uses: actions/setup-node@v4
with:
node-version: '20'
registry-url: 'https://registry.npmjs.org'
@@ -230,14 +268,11 @@ jobs:
- name: Build
run: npm run build --workspace=hindsight-control-plane
- name: Verify standalone build
run: test -f hindsight-control-plane/standalone/server.js || (echo 'standalone/server.js missing - build failed' && exit 1)
- name: Publish to npm
working-directory: ./hindsight-control-plane
run: |
set +e
OUTPUT=$(npm publish --access public --ignore-scripts 2>&1)
OUTPUT=$(npm publish --access public 2>&1)
EXIT_CODE=$?
echo "$OUTPUT"
if [ $EXIT_CODE -ne 0 ]; then
@@ -255,7 +290,7 @@ jobs:
run: npm pack
- name: Upload artifacts
uses: actions/upload-artifact@v7
uses: actions/upload-artifact@v4
with:
name: control-plane
path: hindsight-control-plane/*.tgz
@@ -278,13 +313,9 @@ jobs:
target: aarch64-apple-darwin
artifact_name: hindsight
asset_name: hindsight-darwin-arm64
- os: ubuntu-24.04-arm
target: aarch64-unknown-linux-gnu
artifact_name: hindsight
asset_name: hindsight-linux-arm64
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
@@ -302,7 +333,7 @@ jobs:
chmod +x artifacts/${{ matrix.asset_name }}
- name: Upload artifacts
uses: actions/upload-artifact@v7
uses: actions/upload-artifact@v4
with:
name: rust-cli-${{ matrix.asset_name }}
path: artifacts/${{ matrix.asset_name }}
@@ -343,7 +374,7 @@ jobs:
PRELOAD_ML_MODELS=false
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Free Disk Space
uses: jlumbroso/free-disk-space@main
@@ -357,13 +388,13 @@ jobs:
swap-storage: true
- name: Set up QEMU
uses: docker/setup-qemu-action@v4
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Log in to GitHub Container Registry
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
@@ -375,7 +406,7 @@ jobs:
- name: Extract metadata for release tags
id: meta
uses: docker/metadata-action@v6
uses: docker/metadata-action@v5
with:
images: ghcr.io/${{ github.repository_owner }}/${{ matrix.image_name }}
flavor: |
@@ -391,7 +422,7 @@ jobs:
# # Step 1: Build for local testing (single platform, no push)
# # This creates an identical image to what will be released, just for one platform
# - name: Build image for testing
# uses: docker/build-push-action@v7
# uses: docker/build-push-action@v6
# with:
# context: .
# file: docker/standalone/Dockerfile
@@ -410,7 +441,7 @@ jobs:
# Build multi-platform and push to release tags
- name: Build and push release images
uses: docker/build-push-action@v7
uses: docker/build-push-action@v6
with:
context: .
file: docker/standalone/Dockerfile
@@ -428,10 +459,10 @@ jobs:
packages: write
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Install Helm
uses: azure/setup-helm@v5
uses: azure/setup-helm@v4
with:
version: 'latest'
@@ -448,7 +479,7 @@ jobs:
run: helm push helm-packages/*.tgz oci://ghcr.io/${{ github.repository_owner }}/charts
- name: Upload artifacts
uses: actions/upload-artifact@v7
uses: actions/upload-artifact@v4
with:
name: helm-chart
path: helm-packages/*.tgz
@@ -456,61 +487,67 @@ jobs:
create-github-release:
runs-on: ubuntu-latest
needs: [release-python-packages, release-typescript-client, release-hindsight-all-npm, release-control-plane, release-rust-cli, release-docker-images, release-helm-chart]
needs: [release-python-packages, release-typescript-client, release-openclaw-integration, release-ai-sdk-integration, release-control-plane, release-rust-cli, release-docker-images, release-helm-chart]
permissions:
contents: write
steps:
- uses: actions/checkout@v6
- uses: actions/checkout@v4
- name: Extract version from tag
id: get_version
run: echo "VERSION=${GITHUB_REF#refs/tags/v}" >> $GITHUB_OUTPUT
- name: Download Python packages
uses: actions/download-artifact@v8
uses: actions/download-artifact@v4
with:
name: python-packages
path: ./artifacts/python-packages
- name: Download TypeScript client
uses: actions/download-artifact@v8
uses: actions/download-artifact@v4
with:
name: typescript-client
path: ./artifacts/typescript-client
- name: Download OpenClaw Integration
uses: actions/download-artifact@v4
with:
name: openclaw-integration
path: ./artifacts/openclaw-integration
- name: Download AI SDK Integration
uses: actions/download-artifact@v4
with:
name: ai-sdk-integration
path: ./artifacts/ai-sdk-integration
- name: Download Control Plane
uses: actions/download-artifact@v8
uses: actions/download-artifact@v4
with:
name: control-plane
path: ./artifacts/control-plane
- name: Download hindsight-embed npm wrapper
uses: actions/download-artifact@v8
with:
name: hindsight-all-npm
path: ./artifacts/hindsight-all-npm
- name: Download Rust CLI (Linux)
uses: actions/download-artifact@v8
uses: actions/download-artifact@v4
with:
name: rust-cli-hindsight-linux-amd64
path: ./artifacts/rust-cli-linux
- name: Download Rust CLI (macOS Intel)
uses: actions/download-artifact@v8
uses: actions/download-artifact@v4
with:
name: rust-cli-hindsight-darwin-amd64
path: ./artifacts/rust-cli-darwin-amd64
- name: Download Rust CLI (macOS ARM)
uses: actions/download-artifact@v8
uses: actions/download-artifact@v4
with:
name: rust-cli-hindsight-darwin-arm64
path: ./artifacts/rust-cli-darwin-arm64
- name: Download Helm chart
uses: actions/download-artifact@v8
uses: actions/download-artifact@v4
with:
name: helm-chart
path: ./artifacts/helm-chart
@@ -520,15 +557,16 @@ jobs:
mkdir -p release-assets
# Python packages
cp artifacts/python-packages/hindsight-clients/python/dist/* release-assets/ || true
cp artifacts/python-packages/hindsight-api-slim/dist/* release-assets/ || true
cp artifacts/python-packages/hindsight-api/dist/* release-assets/ || true
cp artifacts/python-packages/hindsight-all/dist/* release-assets/ || true
cp artifacts/python-packages/hindsight-all-slim/dist/* release-assets/ || true
cp artifacts/python-packages/hindsight/dist/* release-assets/ || true
cp artifacts/python-packages/hindsight-integrations/litellm/dist/* release-assets/ || true
cp artifacts/python-packages/hindsight-embed/dist/* release-assets/ || true
# TypeScript client
cp artifacts/typescript-client/*.tgz release-assets/ || true
# hindsight-embed npm wrapper
cp artifacts/hindsight-all-npm/*.tgz release-assets/ || true
# OpenClaw Integration
cp artifacts/openclaw-integration/*.tgz release-assets/ || true
# AI SDK Integration
cp artifacts/ai-sdk-integration/*.tgz release-assets/ || true
# Control Plane
cp artifacts/control-plane/*.tgz release-assets/ || true
# Rust CLI binaries
@@ -540,7 +578,7 @@ jobs:
ls -la release-assets/
- name: Create GitHub Release
uses: softprops/action-gh-release@v3
uses: softprops/action-gh-release@v2
with:
files: release-assets/*
generate_release_notes: true
+215 -1789
View File
File diff suppressed because it is too large Load Diff
+1 -2
View File
@@ -50,8 +50,7 @@ hindsight-dev/benchmarks/perf/results/
benchmarks/results/
hindsight-cli/target
hindsight-clients/rust/target
.claude/*
!.claude/skills/
.claude
whats-next.md
TASK.md
# Changelog is now tracked in hindsight-docs/src/pages/changelog.md
+69 -52
View File
@@ -11,32 +11,26 @@ Hindsight is an agent memory system that provides long-term memory for AI agents
## Development Commands
### Local Development (API + UI)
```bash
# Start both API server and control plane UI
./scripts/dev/start.sh
```
### API Server (Python/FastAPI)
```bash
# Start API server only (loads .env automatically)
# Start API server (loads .env automatically)
./scripts/dev/start-api.sh
# Run all tests (parallelized with pytest-xdist)
cd hindsight-api-slim && uv run pytest tests/
cd hindsight-api && uv run pytest tests/
# Run specific test file
cd hindsight-api-slim && uv run pytest tests/test_http_api_integration.py -v
cd hindsight-api && uv run pytest tests/test_http_api_integration.py -v
# Run single test function
cd hindsight-api-slim && uv run pytest tests/test_retain.py::test_retain_simple -v
cd hindsight-api && uv run pytest tests/test_retain.py::test_retain_simple -v
# Lint and format
cd hindsight-api-slim && uv run ruff check .
cd hindsight-api-slim && uv run ruff format .
cd hindsight-api && uv run ruff check .
cd hindsight-api && uv run ruff format .
# Type checking (uses ty - extremely fast type checker from Astral)
cd hindsight-api-slim && uv run ty check hindsight_api/
cd hindsight-api && uv run ty check hindsight_api/
```
### Control Plane (Next.js)
@@ -78,17 +72,18 @@ cd hindsight-control-plane && npm run dev
## Architecture
### Monorepo Structure
- **hindsight-api-slim/**: Core FastAPI server with memory engine (Python, uv)
- **hindsight-api/**: Core FastAPI server with memory engine (Python, uv)
- **hindsight/**: Embedded Python bundle (hindsight-all package)
- **hindsight-control-plane/**: Admin UI (Next.js, npm)
- **hindsight-cli/**: CLI tool (Rust, cargo, uses progenitor for API client)
- **hindsight-clients/**: Generated SDK clients (Python, TypeScript, Rust)
- **hindsight-docs/**: Docusaurus documentation site
- **hindsight-integrations/**: Framework integrations (LiteLLM, CrewAI, LangGraph, Pydantic AI, AG2, Claude Code, etc.)
- **hindsight-integrations/**: Framework integrations (LiteLLM, OpenAI)
- **hindsight-dev/**: Development tools and benchmarks
### Core Engine (hindsight-api-slim/hindsight_api/engine/)
- `memory_engine.py`: Main orchestrator for retain/recall/reflect operations
- `llm_wrapper.py`: LLM abstraction supporting OpenAI, Anthropic, Gemini, VertexAI, Groq, MiniMax, Ollama, LM Studio, LiteLLM, Claude Code
### Core Engine (hindsight-api/hindsight_api/engine/)
- `memory_engine.py`: Main orchestrator (~170KB) for retain/recall/reflect operations
- `llm_wrapper.py`: LLM abstraction supporting OpenAI, Anthropic, Gemini, Groq, Ollama, LM Studio
- `embeddings.py`: Embedding generation (local sentence-transformers or TEI)
- `cross_encoder.py`: Reranking (local or TEI)
- `entity_resolver.py`: Entity extraction and normalization
@@ -101,13 +96,13 @@ cd hindsight-control-plane && npm run dev
**search/**: Multi-strategy retrieval
- `retrieval.py`: Main retrieval orchestrator
- `graph_retrieval.py`: Graph retrieval abstract base class
- `link_expansion_retrieval.py`: Link expansion graph retrieval
- `graph_retrieval.py`: Entity/relationship graph traversal
- `mpfp_retrieval.py`: Multi-Path Fact Propagation retrieval
- `fusion.py`: Reciprocal rank fusion for combining results
- `reranking.py`: Cross-encoder reranking
### API Layer (hindsight-api-slim/hindsight_api/api/)
- `http.py`: FastAPI HTTP routers for all REST endpoints
### API Layer (hindsight-api/hindsight_api/api/)
- `http.py`: FastAPI HTTP routers (~80KB) for all REST endpoints
- `mcp.py`: Model Context Protocol server implementation
Main operations:
@@ -116,13 +111,13 @@ Main operations:
- **Reflect**: Disposition-aware reasoning using memories and mental models.
### Database
PostgreSQL with pgvector. Schema managed via Alembic migrations in `hindsight-api-slim/hindsight_api/alembic/`. Migrations run automatically on API startup.
PostgreSQL with pgvector. Schema managed via Alembic migrations in `hindsight-api/hindsight_api/alembic/`. Migrations run automatically on API startup.
Key tables: `banks`, `memory_units`, `documents`, `entities`, `entity_links`
### Adding Database Migrations
1. **Create a new migration file** in `hindsight-api-slim/hindsight_api/alembic/versions/`:
1. **Create a new migration file** in `hindsight-api/hindsight_api/alembic/versions/`:
- File name format: `<revision_id>_<description>.py` (e.g., `f1a2b3c4d5e6_add_new_index.py`)
- Use a unique hex revision ID (12 chars)
- Set `down_revision` to the previous migration's revision ID
@@ -159,7 +154,7 @@ Key tables: `banks`, `memory_units`, `documents`, `entities`, `entity_links`
3. **Run migrations locally**:
```bash
# Set database URL and run migrations for the base schema plus all tenants
# Set database URL and run migrations
uv run hindsight-admin run-db-migration
# Run on a specific tenant schema
@@ -169,17 +164,11 @@ Key tables: `banks`, `memory_units`, `documents`, `entities`, `entity_links`
## Key Conventions
### Code Quality
**Before writing code, read `.claude/skills/code-review/SKILL.md`** for the full coding standards (Python style, type safety, TypeScript style, general principles).
**Always run the lint script after making Python or TypeScript/Node changes:**
```bash
./scripts/hooks/lint.sh
```
**After completing any implementation work, run `/code-review`** to verify your changes against project standards (missing tests, dead code, type safety, etc.). Fix any "must fix" issues before considering the task done.
**MANDATORY: Run `/code-review` before pushing code or creating a pull request.** Do not push or create a PR until all "must fix" issues are resolved.
This runs the same checks as the pre-commit hook (Ruff for Python, ESLint/Prettier for TypeScript).
### Memory Banks
- Each bank is an isolated memory store (like a "brain" for one user/agent)
@@ -211,20 +200,48 @@ When adding or modifying parameters in the dataplane API (hindsight-api), you mu
- Update the client type definition in `lib/api.ts`
- Update any UI components that need to use the new parameter
### Adding New Integrations
### Python Style
- Python 3.11+, type hints required
- Async throughout (asyncpg, async FastAPI)
- Pydantic models for request/response
- Ruff for linting (line-length 120)
- No Python files at project root - maintain clean directory structure
- **Never use multi-item tuple return values** - prefer dataclass or Pydantic model for structured returns
Every new integration in `hindsight-integrations/` must satisfy all of the following before it can be merged:
### Type Safety with Pydantic Models
**NEVER use raw `dict` types for structured data.** Always use Pydantic models:
- Use Pydantic `BaseModel` for all data structures passed between functions
- Add `@field_validator` for type coercion (e.g., ensuring datetimes are timezone-aware)
- Avoid `dict.get()` patterns - use typed model attributes instead
- Parse external data (JSON, API responses) into Pydantic models at the boundary
- This catches type errors at parse time, not deep in business logic
1. **Tests are required** — tests must simulate or exercise the external system (mock the framework's interfaces and verify the integration actually calls Hindsight correctly). Pure unit tests of helper functions are not sufficient.
2. **CI job** — add a test job in `.github/workflows/test.yml` following the existing pattern (e.g., `test-crewai-integration`). The job must build, install deps, and run `uv run pytest tests -v`. Also add the integration to `detect-changes` outputs so it only runs when its files change.
3. **Release process** — add the integration name to the `VALID_INTEGRATIONS` array in `scripts/release-integration.sh` so it can be released via the standard release workflow.
4. **Follow project code standards** — Python style, type safety, no raw dicts for structured data, no multi-item tuple returns (see `.claude/skills/code-review/SKILL.md`).
```python
# BAD - error-prone dict access
def process(data: dict) -> str:
return data.get("name", "") # No validation, silent failures
If any of these are missing, the integration is incomplete and must not be pushed or merged.
# GOOD - typed and validated
class UserData(BaseModel):
name: str
created_at: datetime
### Changelogs
@field_validator("created_at", mode="before")
@classmethod
def ensure_tz_aware(cls, v):
if isinstance(v, str):
v = datetime.fromisoformat(v.replace("Z", "+00:00"))
if v.tzinfo is None:
return v.replace(tzinfo=timezone.utc)
return v
Never add "Unreleased" entries to changelogs (e.g. `hindsight-docs/src/pages/changelog/**`). Changelog entries are written by the release script (`./scripts/release-integration.sh`) when a version is actually cut. If a bug fix or feature needs documenting before release, describe it in the PR/commit — the release tooling will surface it in the published changelog section.
def process(data: UserData) -> str:
return data.name # Type-safe, validated at construction
```
### TypeScript Style
- Next.js App Router for control plane
- Tailwind CSS with shadcn/ui components
### Adding New API Configuration Flags
@@ -234,24 +251,24 @@ Fields must be categorized as either **hierarchical** (can be overridden per-ten
#### Adding a New Configuration Field
1. **config.py** (`hindsight-api-slim/hindsight_api/config.py`):
1. **config.py** (`hindsight-api/hindsight_api/config.py`):
- Add `ENV_*` constant for the environment variable name (e.g., `ENV_MY_SETTING = "HINDSIGHT_API_MY_SETTING"`)
- Add `DEFAULT_*` constant for the default value
- Add field to `HindsightConfig` dataclass with type annotation
- **Mark as configurable** by adding to `_CONFIGURABLE_FIELDS` set if the field should be overridable per-tenant/bank via API
- **Mark as hierarchical or static** by adding to `_HIERARCHICAL_FIELDS` set (hierarchical) or leaving it out (static)
- Add initialization in `from_env()` method
```python
# Configurable field (can be overridden per-tenant/bank via API)
_CONFIGURABLE_FIELDS = {
# Hierarchical field (can be overridden per-bank)
_HIERARCHICAL_FIELDS = {
...,
"my_setting", # Add here for configurable
"my_setting", # Add here for hierarchical
}
# Static field - just don't add to _CONFIGURABLE_FIELDS
# Static field - just don't add to _HIERARCHICAL_FIELDS
```
2. **main.py** (`hindsight-api-slim/hindsight_api/main.py`):
2. **main.py** (`hindsight-api/hindsight_api/main.py`):
- Add field to the manual `HindsightConfig()` constructor call (search for "CLI override")
3. **Use hierarchical config in MemoryEngine**:
@@ -291,19 +308,19 @@ cp .env.example .env
# Edit .env with LLM API key
# Python deps
uv sync --directory hindsight-api-slim/
uv sync --directory hindsight-api/
# Node deps (uses npm workspaces)
npm install
```
Required env vars:
- `HINDSIGHT_API_LLM_PROVIDER`: openai, anthropic, gemini, groq, minimax, ollama, lmstudio
- `HINDSIGHT_API_LLM_PROVIDER`: openai, anthropic, gemini, groq, ollama, lmstudio
- `HINDSIGHT_API_LLM_API_KEY`: Your API key
- `HINDSIGHT_API_LLM_MODEL`: Model name (e.g., gpt-4o-mini, claude-sonnet-4-20250514)
- `HINDSIGHT_API_LLM_MODEL`: Model name (e.g., o3-mini, claude-sonnet-4-20250514)
Optional (uses local models by default):
- `HINDSIGHT_API_EMBEDDINGS_PROVIDER`: local (default) or tei
- `HINDSIGHT_API_RERANKER_PROVIDER`: local (default) or tei
- `HINDSIGHT_API_DATABASE_URL`: External PostgreSQL (uses embedded pg0 by default)
- `HINDSIGHT_API_ENABLE_BANK_CONFIG_API`: Enable per-bank config API (default: true)
- `HINDSIGHT_API_ENABLE_BANK_CONFIG_API`: Enable per-bank config API (default: false, disabled for security)
+4 -8
View File
@@ -7,12 +7,10 @@
[![CI](https://github.com/vectorize-io/hindsight/actions/workflows/release.yml/badge.svg)](https://github.com/vectorize-io/hindsight/actions/workflows/release.yml)
[![Slack Community](https://img.shields.io/badge/Slack-Join%20Community-4A154B?logo=slack)](https://join.slack.com/t/hindsight-space/shared_invite/zt-3nhbm4w29-LeSJ5Ixi6j8PdiYOCPlOgg)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![gitcgr](https://gitcgr.com/badge/vectorize-io/hindsight.svg)](https://gitcgr.com/vectorize-io/hindsight)
![PyPI - Downloads](https://img.shields.io/pypi/dm/hindsight-api?label=PyPI)
![NPM Downloads](https://img.shields.io/npm/dm/%40vectorize-io%2Fhindsight-client?logoColor=orange&label=NPM&color=blue&link=https%3A%2F%2Fwww.npmjs.com%2Fpackage%2F%40vectorize-io%2Fhindsight-client)
<br/>
<a href="https://trendshift.io/repositories/15603" target="_blank"><img src="https://trendshift.io/api/badge/repositories/15603" alt="vectorize-io%2Fhindsight | Trendshift" style="width: 250px; height: 55px;" width="250" height="55"/></a>
</div>
---
@@ -38,7 +36,7 @@ Hindsight is being used in production at Fortune 500 enterprises and by a growin
## Adding Hindsight to Your AI Agents
The easiest way to use Hindsight with an existing agent is with the LLM Wrapper. You can add memory to your agent with 2 lines of code. That will swap your current LLM client out with the Hindsight wrapper. After that, memories will be stored and retrieved automatically as you make LLM calls.
The easiest way use Hindsight with an existing agent is with the LLM Wrapper. You can add memory to your agent with 2 lines of code. That will swap your current LLM client out with the Hindsight wrapper. After that, memories will be stored and retrieved automatically as you make LLM calls.
If you need more control over how and when your agent stores and recalls memories, there's also a simple API you can integrate with using the SDKs or directly via HTTP.
@@ -71,7 +69,7 @@ docker run --rm -it --pull always -p 8888:8888 -p 9999:9999 \
>API: http://localhost:8888
>UI: http://localhost:9999
You can modify the LLM provider by setting `HINDSIGHT_API_LLM_PROVIDER`. Valid options are `openai`, `anthropic`, `gemini`, `groq`, `ollama`, `lmstudio`, and `minimax`. The documentation provides more details on [supported models](https://hindsight.vectorize.io/developer/models).
You can modify the LLM provider by setting `HINDSIGHT_API_LLM_PROVIDER`. Valid options are `openai`, `anthropic`, `gemini`, `groq`, `ollama`, and `lmstudio`. The documentation provides more details on [supported models](https://hindsight.vectorize.io/developer/models).
@@ -183,7 +181,7 @@ Satisfying these requirements in Hindsight is straightforward. When new user inp
![Overview](./hindsight-docs/static/img/hindsight-overview.webp)
Most agent memory implementations rely on basic vector search or sometimes use a knowledge graph. Hindsight uses biomimetic data structures to organize agent memories in a way that is more like how human memory works:
Most agent memory implementation rely on basic vector search or sometimes use a knowledge graph. Hindsight uses biomimetic data structures to organize agent memories in a way that is more like how human memory works:
- **World:** Facts about the world ("The stove gets hot")
- **Experiences:** Agent's own experiences ("I touched the stove and it really hurt")
@@ -309,5 +307,3 @@ MIT — see [LICENSE](./LICENSE)
---
Built by [Vectorize.io](https://vectorize.io)
<img src="https://umami-pixel.chris-latimer.workers.dev/?id=a8b043e6-6964-454d-80df-69b69d3f0d50&host=github.com&url=/vectorize-io/hindsight" width="1" height="1" alt="" />
Generated
-139
View File
@@ -1,139 +0,0 @@
{
"version": "5",
"specifiers": {
"jsr:@std/assert@^1.0.17": "1.0.19",
"jsr:@std/assert@^1.0.19": "1.0.19",
"jsr:@std/expect@*": "1.0.18",
"jsr:@std/internal@^1.0.12": "1.0.12",
"jsr:@std/path@^1.1.4": "1.1.4",
"jsr:@std/testing@*": "1.0.17"
},
"jsr": {
"@std/[email protected]": {
"integrity": "eaada96ee120cb980bc47e040f82814d786fe8162ecc53c91d8df60b8755991e",
"dependencies": [
"jsr:@std/internal"
]
},
"@std/[email protected]": {
"integrity": "8566eab35200466f8609eb7e7aed062ed0db314e9a258d5d201b1b8997ce801a",
"dependencies": [
"jsr:@std/assert@^1.0.19",
"jsr:@std/internal",
"jsr:@std/path"
]
},
"@std/[email protected]": {
"integrity": "972a634fd5bc34b242024402972cd5143eac68d8dffaca5eaa4dba30ce17b027"
},
"@std/[email protected]": {
"integrity": "1d2d43f39efb1b42f0b1882a25486647cb851481862dc7313390b2bb044314b5",
"dependencies": [
"jsr:@std/internal"
]
},
"@std/[email protected]": {
"integrity": "87bdc2700fa98249d48a17cd72413352d3d3680dcfbdb64947fd0982d6bbf681",
"dependencies": [
"jsr:@std/assert@^1.0.17",
"jsr:@std/internal"
]
}
},
"workspace": {
"members": {
"hindsight-clients/typescript": {
"packageJson": {
"dependencies": [
"npm:@hey-api/[email protected]",
"npm:@types/jest@29",
"npm:@types/node@20",
"npm:jest@29",
"npm:ts-jest@29",
"npm:tsup@^8.5.1",
"npm:typescript@5"
]
}
},
"hindsight-control-plane": {
"packageJson": {
"dependencies": [
"npm:@eslint/eslintrc@^3.3.3",
"npm:@eslint/js@^9.39.2",
"npm:@radix-ui/react-alert-dialog@^1.1.15",
"npm:@radix-ui/react-checkbox@^1.3.3",
"npm:@radix-ui/react-dialog@^1.1.15",
"npm:@radix-ui/react-dropdown-menu@^2.1.16",
"npm:@radix-ui/react-label@^2.1.8",
"npm:@radix-ui/react-popover@^1.1.15",
"npm:@radix-ui/react-radio-group@^1.3.8",
"npm:@radix-ui/react-select@^2.2.6",
"npm:@radix-ui/react-slider@^1.3.6",
"npm:@radix-ui/react-slot@^1.2.4",
"npm:@radix-ui/react-switch@^1.2.6",
"npm:@radix-ui/react-tabs@^1.1.13",
"npm:@radix-ui/react-tooltip@^1.2.8",
"npm:@tailwindcss/postcss@^4.1.17",
"npm:@tailwindcss/typography@~0.5.19",
"npm:@types/cytoscape@^3.21.9",
"npm:@types/node@^24.10.0",
"npm:@types/react-dom@^19.2.2",
"npm:@types/react@^19.2.2",
"npm:autoprefixer@^10.4.21",
"npm:class-variance-authority@~0.7.1",
"npm:clsx@^2.1.1",
"npm:cmdk@^1.1.1",
"npm:cytoscape-fcose@^2.2.0",
"npm:cytoscape@^3.33.1",
"npm:eslint-config-next@^16.0.1",
"npm:eslint-plugin-react-hooks@^7.0.1",
"npm:eslint-plugin-react@^7.37.5",
"npm:eslint@^9.39.1",
"npm:[email protected]",
"npm:next-themes@~0.4.6",
"npm:next@^16.1.6",
"npm:postcss@^8.5.6",
"npm:prettier@^3.7.4",
"npm:react-chrono@^2.9.1",
"npm:react-dom@^19.2.0",
"npm:react-markdown@^10.1.0",
"npm:react18-json-view@~0.2.9",
"npm:react@^19.2.0",
"npm:recharts@^3.5.1",
"npm:remark-gfm@^4.0.1",
"npm:sonner@^2.0.7",
"npm:tailwind-merge@^3.4.0",
"npm:tailwindcss-animate@^1.0.7",
"npm:tailwindcss@^4.1.17",
"npm:[email protected]",
"npm:typescript-eslint@^8.50.0",
"npm:typescript@^5.9.3"
]
}
},
"hindsight-docs": {
"packageJson": {
"dependencies": [
"npm:@docusaurus/[email protected]",
"npm:@docusaurus/[email protected]",
"npm:@docusaurus/[email protected]",
"npm:@docusaurus/theme-common@^3.9.2",
"npm:@docusaurus/theme-mermaid@^3.9.2",
"npm:@docusaurus/[email protected]",
"npm:@docusaurus/[email protected]",
"npm:@easyops-cn/docusaurus-search-local@~0.52.2",
"npm:@mdx-js/react@3",
"npm:clsx@2",
"npm:prism-react-renderer@^2.3.0",
"npm:raw-loader@^4.0.2",
"npm:react-dom@19",
"npm:react-icons@^5.6.0",
"npm:react@19",
"npm:redocusaurus@^2.5.0",
"npm:typescript@~5.6.2"
]
}
}
}
}
}
@@ -1,83 +0,0 @@
# Docker Compose file for Hindsight with S3 file storage (SeaweedFS)
#
# SeaweedFS (Apache 2.0) provides an S3-compatible object storage backend
# for storing uploaded files instead of PostgreSQL BYTEA storage.
#
# Make sure to set the required environment variables before running:
# - HINDSIGHT_DB_PASSWORD: Password for the PostgreSQL user
# - Configure LLM provider variables as needed (see below in the hindsight service)
#
# Usage:
# docker compose up -d
#
# Optional environment variables with defaults:
# - HINDSIGHT_VERSION: Hindsight application version (default: latest)
# - HINDSIGHT_DB_USER: PostgreSQL user (default: hindsight_user)
# - HINDSIGHT_DB_NAME: PostgreSQL database name (default: hindsight_db)
# - HINDSIGHT_DB_VERSION: PostgreSQL version (default: 18)
# - SEAWEEDFS_S3_ACCESS_KEY: S3 access key (default: hindsight_s3_key)
# - SEAWEEDFS_S3_SECRET_KEY: S3 secret key (default: hindsight_s3_secret)
services:
db:
image: pgvector/pgvector:pg${HINDSIGHT_DB_VERSION:-18}
container_name: hindsight-db
restart: always
environment:
POSTGRES_USER: ${HINDSIGHT_DB_USER:-hindsight_user}
POSTGRES_PASSWORD: ${HINDSIGHT_DB_PASSWORD:?Please set the HINDSIGHT_DB_PASSWORD env variable}
POSTGRES_DB: ${HINDSIGHT_DB_NAME:-hindsight_db}
volumes:
- pg_data:/var/lib/postgresql/${HINDSIGHT_DB_VERSION:-18}/docker
networks:
- hindsight-net
seaweedfs:
image: chrislusf/seaweedfs:latest
container_name: hindsight-seaweedfs
restart: always
# Single-node mode: master + volume + filer + S3 gateway all in one process
command: >
server
-s3
-s3.port=8333
-s3.config=/etc/seaweedfs/s3.json
-ip.bind=0.0.0.0
volumes:
- seaweedfs_data:/data
- ./s3.json:/etc/seaweedfs/s3.json:ro
# Expose S3 API port (uncomment to access from host)
# ports:
# - "8333:8333"
networks:
- hindsight-net
hindsight:
image: ghcr.io/vectorize-io/hindsight:${HINDSIGHT_VERSION:-latest}
container_name: hindsight-app
ports:
- "8888:8888"
- "9999:9999"
environment:
- HINDSIGHT_API_LLM_API_KEY=${OPENAI_API_KEY?Please set the OPENAI_API_KEY env variable}
- HINDSIGHT_API_DATABASE_URL=postgresql://${HINDSIGHT_DB_USER:-hindsight_user}:${HINDSIGHT_DB_PASSWORD:?Please set the HINDSIGHT_DB_PASSWORD env variable}@db:5432/${HINDSIGHT_DB_NAME:-hindsight_db}
# S3 file storage configuration (SeaweedFS)
- HINDSIGHT_API_FILE_STORAGE_TYPE=s3
- HINDSIGHT_API_FILE_STORAGE_S3_BUCKET=hindsight
- HINDSIGHT_API_FILE_STORAGE_S3_ENDPOINT=http://seaweedfs:8333
- HINDSIGHT_API_FILE_STORAGE_S3_REGION=us-east-1
- HINDSIGHT_API_FILE_STORAGE_S3_ACCESS_KEY_ID=${SEAWEEDFS_S3_ACCESS_KEY:-hindsight_s3_key}
- HINDSIGHT_API_FILE_STORAGE_S3_SECRET_ACCESS_KEY=${SEAWEEDFS_S3_SECRET_KEY:-hindsight_s3_secret}
depends_on:
- db
- seaweedfs
networks:
- hindsight-net
networks:
hindsight-net:
driver: bridge
volumes:
pg_data:
seaweedfs_data:
@@ -1,19 +0,0 @@
{
"identities": [
{
"name": "hindsight",
"credentials": [
{
"accessKey": "hindsight_s3_key",
"secretKey": "hindsight_s3_secret"
}
],
"actions": [
"Admin",
"Read",
"Write",
"List"
]
}
]
}
+13 -20
View File
@@ -42,22 +42,25 @@ RUN apt-get update && apt-get install -y \
&& pip install --no-cache-dir uv
# Copy dependency files and README (required by pyproject.toml)
COPY hindsight-api-slim/pyproject.toml ./api/
COPY hindsight-api-slim/README.md ./api/
COPY hindsight-api/pyproject.toml ./api/
COPY hindsight-api/README.md ./api/
WORKDIR /app/api
# Sync dependencies using appropriate extras based on INCLUDE_LOCAL_MODELS
# local-ml: torch, sentence-transformers, transformers, einops, flashrank, mlx (optional)
# embedded-db: pg0-embedded (always included for embedded PostgreSQL support)
RUN if [ "$INCLUDE_LOCAL_MODELS" = "true" ]; then \
uv sync --extra local-ml --extra embedded-db; \
else \
uv sync --extra embedded-db; \
# Remove local ML model dependencies if INCLUDE_LOCAL_MODELS=false
# This creates a smaller image when using external providers (TEI, OpenAI, Cohere)
RUN if [ "$INCLUDE_LOCAL_MODELS" != "true" ]; then \
echo "Removing local-models dependencies (sentence-transformers, torch, transformers)..." && \
sed -i '/"sentence-transformers/d' pyproject.toml && \
sed -i '/"transformers/d' pyproject.toml && \
sed -i '/"torch/d' pyproject.toml; \
fi
# Sync dependencies (will create lock file if needed)
RUN uv sync
# Copy source code (alembic migrations are inside hindsight_api/)
COPY hindsight-api-slim/hindsight_api ./hindsight_api
COPY hindsight-api/hindsight_api ./hindsight_api
# Install the local package (uv sync only installed dependencies, not the package itself)
RUN uv pip install -e .
@@ -167,11 +170,6 @@ RUN chown -R hindsight:hindsight /app
USER hindsight
# Create pg0 data directory as hindsight user so that Docker seeds new named
# volumes with correct ownership (UID 1000) on first use, avoiding the
# "Permission denied" error when mounting a fresh root-owned volume.
RUN mkdir -p /home/hindsight/.pg0
ENV PATH="/app/api/.venv/bin:${PATH}"
# Pre-download tiktoken encoding (ALWAYS - required for token counting even in air-gapped envs)
@@ -323,11 +321,6 @@ RUN chown -R hindsight:hindsight /app
USER hindsight
# Create pg0 data directory as hindsight user so that Docker seeds new named
# volumes with correct ownership (UID 1000) on first use, avoiding the
# "Permission denied" error when mounting a fresh root-owned volume.
RUN mkdir -p /home/hindsight/.pg0
ENV PATH="/app/api/.venv/bin:${PATH}"
# Pre-download tiktoken encoding (ALWAYS - required for token counting even in air-gapped envs)
+9 -116
View File
@@ -1,28 +1,6 @@
#!/bin/bash
set -e
# =============================================================================
# Embedded pg0 data integrity check (#675)
#
# When using embedded pg0, check if the data directory has existing PostgreSQL
# data before starting. If the directory exists but appears empty/corrupt
# (e.g., missing PG_VERSION file), log a warning. This helps diagnose data
# loss scenarios where a container restart caused the data directory to be
# wiped despite a volume mount being present.
# =============================================================================
PG0_DATA_DIR="${HOME}/.pg0"
if [ -d "$PG0_DATA_DIR" ]; then
# Look for actual PostgreSQL data directories (pg0 creates subdirs per instance)
if compgen -G "$PG0_DATA_DIR"/*/PG_VERSION > /dev/null 2>&1; then
echo "✅ Existing pg0 data directory detected at $PG0_DATA_DIR"
elif [ "$(ls -A "$PG0_DATA_DIR" 2>/dev/null)" ]; then
echo "⚠️ WARNING: pg0 data directory exists at $PG0_DATA_DIR but no PG_VERSION found."
echo " This may indicate data corruption or an incomplete previous shutdown."
echo " If you see all migrations running from scratch after this, your data may have been lost."
echo " See: https://github.com/vectorize-io/hindsight/issues/675"
fi
fi
# Service flags (default to true if not set)
ENABLE_API="${HINDSIGHT_ENABLE_API:-true}"
ENABLE_CP="${HINDSIGHT_ENABLE_CP:-true}"
@@ -93,95 +71,24 @@ if [ "${HINDSIGHT_WAIT_FOR_DEPS:-false}" = "true" ]; then
done
fi
# =============================================================================
# Graceful shutdown handler (#675)
#
# Docker sends SIGTERM on `docker stop`/`docker restart`. Without a trap, child
# processes (hindsight-api + pg0, control-plane) are killed abruptly. For the
# embedded pg0 database this can cause data loss when the data directory is on
# a Docker volume that gets remounted after restart.
#
# The trap forwards SIGTERM to all tracked child PIDs so that:
# - hindsight-api receives the signal and can run its shutdown hooks
# - pg0 gets a clean PostgreSQL shutdown (checkpoint + WAL flush)
# - The control-plane Node.js process exits cleanly
# =============================================================================
# Guard against concurrent cleanup (e.g., child crash + SIGTERM arriving together)
SHUTTING_DOWN=false
cleanup() {
if $SHUTTING_DOWN; then return; fi
SHUTTING_DOWN=true
echo ""
echo "🛑 Received shutdown signal, stopping services gracefully..."
for pid in "${PIDS[@]}"; do
if kill -0 "$pid" 2>/dev/null; then
kill -TERM "$pid" 2>/dev/null
fi
done
# Give processes time to shut down cleanly (pg0 needs to flush WAL).
# NOTE: Docker's default stop_grace_period is 10s. If you use the default,
# either set stop_grace_period: 30s in your compose file / docker stop -t 30,
# or Docker will SIGKILL the container before this timeout expires.
local timeout=30
for ((i=1; i<=timeout; i++)); do
local all_stopped=true
for pid in "${PIDS[@]}"; do
if kill -0 "$pid" 2>/dev/null; then
all_stopped=false
break
fi
done
if $all_stopped; then
echo "✅ All services stopped cleanly"
exit 0
fi
sleep 1
done
# Force kill if still running after timeout
echo "⚠️ Timeout reached, forcing shutdown..."
for pid in "${PIDS[@]}"; do
if kill -0 "$pid" 2>/dev/null; then
kill -9 "$pid" 2>/dev/null
fi
done
exit 1
}
trap cleanup SIGTERM SIGINT
# Track PIDs for wait
PIDS=()
# Start API if enabled
if [ "$ENABLE_API" = "true" ]; then
cd /app/api
API_HEALTH_URL="${HINDSIGHT_API_HEALTH_URL:-http://localhost:8888/health}"
API_STARTUP_WAIT_SECONDS="${HINDSIGHT_API_STARTUP_WAIT_SECONDS:-300}"
# Run API directly - Python's PYTHONUNBUFFERED=1 handles output buffering
hindsight-api &
API_PID=$!
PIDS+=($API_PID)
# Wait for API to be ready
api_ready=false
for ((i=1; i<=API_STARTUP_WAIT_SECONDS; i++)); do
if ! kill -0 "$API_PID" 2>/dev/null; then
wait "$API_PID"
exit $?
fi
if curl -sf "$API_HEALTH_URL" &>/dev/null; then
api_ready=true
for i in {1..60}; do
if curl -sf http://localhost:8888/health &>/dev/null; then
break
fi
sleep 1
done
if [ "$api_ready" != "true" ]; then
echo "❌ API did not become healthy within ${API_STARTUP_WAIT_SECONDS}s"
exit 1
fi
else
echo "API disabled (HINDSIGHT_ENABLE_API=false)"
fi
@@ -190,8 +97,7 @@ fi
if [ "$ENABLE_CP" = "true" ]; then
echo "🎛️ Starting Control Plane..."
cd /app/control-plane
export HOSTNAME="${HINDSIGHT_CP_HOSTNAME:-0.0.0.0}"
PORT="${HINDSIGHT_CP_PORT:-9999}" node server.js &
PORT=9999 node server.js &
CP_PID=$!
PIDS+=($CP_PID)
else
@@ -204,7 +110,7 @@ echo "✅ Hindsight is running!"
echo ""
echo "📍 Access:"
if [ "$ENABLE_CP" = "true" ]; then
echo " Control Plane: http://localhost:${HINDSIGHT_CP_PORT:-9999}"
echo " Control Plane: http://localhost:9999"
fi
if [ "$ENABLE_API" = "true" ]; then
echo " API: http://localhost:8888"
@@ -217,21 +123,8 @@ if [ ${#PIDS[@]} -eq 0 ]; then
exit 1
fi
# Wait for any process to exit (use wait -n with trap-safe loop)
while true; do
# wait -n returns when any child exits; it also returns on signal delivery
# (the trap handler will run and exit, so this loop is just for robustness).
# `&& true` prevents `set -e` from killing the script when wait -n returns
# non-zero (child exited with error or no backgrounded children remain).
wait -n && true
# Check if any tracked PID has exited
for pid in "${PIDS[@]}"; do
if ! kill -0 "$pid" 2>/dev/null; then
wait "$pid" 2>/dev/null
exit_code=$?
echo "⚠️ Service (PID $pid) exited with code $exit_code"
# Trigger cleanup for remaining services
cleanup
fi
done
done
# Wait for any process to exit
wait -n
# Exit with status of first exited process
exit $?
+10 -44
View File
@@ -13,9 +13,9 @@
# target - Optional: 'cp-only' for control plane, otherwise assumes API image (default: api)
#
# Environment variables:
# HINDSIGHT_API_LLM_API_KEY - Required for API/standalone images (LLM verification)
# HINDSIGHT_API_LLM_PROVIDER - LLM provider (default: openai)
# HINDSIGHT_API_LLM_MODEL - LLM model (default: gpt-4o-mini)
# GROQ_API_KEY - Required for API/standalone images (LLM verification)
# HINDSIGHT_API_LLM_PROVIDER - LLM provider (default: groq)
# HINDSIGHT_API_LLM_MODEL - LLM model (default: llama-3.3-70b-versatile)
# HINDSIGHT_API_EMBEDDINGS_PROVIDER - Embeddings provider (optional, for slim images: openai, cohere, tei)
# HINDSIGHT_API_EMBEDDINGS_OPENAI_API_KEY - OpenAI API key for embeddings (optional)
# HINDSIGHT_API_RERANKER_PROVIDER - Reranker provider (optional, for slim images: cohere, tei)
@@ -34,7 +34,7 @@
# ./docker/test-image.sh hindsight-control-plane:test cp-only
#
# # Test slim image with external providers
# export HINDSIGHT_API_LLM_API_KEY=sk_xxx
# export GROQ_API_KEY=gsk_xxx
# export HINDSIGHT_API_EMBEDDINGS_PROVIDER=openai
# export HINDSIGHT_API_EMBEDDINGS_OPENAI_API_KEY=sk-xxx
# export HINDSIGHT_API_RERANKER_PROVIDER=cohere
@@ -49,9 +49,6 @@
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(dirname "$SCRIPT_DIR")"
# Colors for output
RED='\033[0;31m'
GREEN='\033[0;32m'
@@ -63,8 +60,8 @@ IMAGE="${1:-}"
TARGET="${2:-api}"
TIMEOUT="${SMOKE_TEST_TIMEOUT:-120}"
CONTAINER_NAME="${SMOKE_TEST_CONTAINER_NAME:-hindsight-smoke-test}"
LLM_PROVIDER="${HINDSIGHT_API_LLM_PROVIDER:-openai}"
LLM_MODEL="${HINDSIGHT_API_LLM_MODEL:-gpt-4o-mini}"
LLM_PROVIDER="${HINDSIGHT_API_LLM_PROVIDER:-groq}"
LLM_MODEL="${HINDSIGHT_API_LLM_MODEL:-llama-3.3-70b-versatile}"
# Validate arguments
if [ -z "$IMAGE" ]; then
@@ -91,9 +88,9 @@ else
fi
# Check for required environment variables
if [ "$NEEDS_LLM" = true ] && [ "$LLM_PROVIDER" != "vertexai" ] && [ -z "${HINDSIGHT_API_LLM_API_KEY:-}" ]; then
echo -e "${RED}Error: HINDSIGHT_API_LLM_API_KEY environment variable is required for API/standalone images${NC}"
echo "Set it with: export HINDSIGHT_API_LLM_API_KEY=your-api-key"
if [ "$NEEDS_LLM" = true ] && [ -z "${GROQ_API_KEY:-}" ]; then
echo -e "${RED}Error: GROQ_API_KEY environment variable is required for API/standalone images${NC}"
echo "Set it with: export GROQ_API_KEY=your-api-key"
exit 2
fi
@@ -126,25 +123,9 @@ else
# Build docker run command with required and optional env vars
DOCKER_CMD="docker run -d --name $CONTAINER_NAME"
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_LLM_PROVIDER=$LLM_PROVIDER"
if [ -n "${HINDSIGHT_API_LLM_API_KEY:-}" ]; then
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_LLM_API_KEY=${HINDSIGHT_API_LLM_API_KEY}"
fi
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_LLM_API_KEY=${GROQ_API_KEY}"
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_LLM_MODEL=$LLM_MODEL"
# Add Vertex AI config if provider is vertexai
if [ "$LLM_PROVIDER" = "vertexai" ]; then
if [ -n "${HINDSIGHT_API_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY:-}" ]; then
DOCKER_CMD="$DOCKER_CMD -v ${HINDSIGHT_API_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY}:/tmp/gcp-credentials.json:ro"
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY=/tmp/gcp-credentials.json"
fi
if [ -n "${HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID:-}" ]; then
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID=${HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID}"
fi
if [ -n "${HINDSIGHT_API_LLM_VERTEXAI_REGION:-}" ]; then
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_LLM_VERTEXAI_REGION=${HINDSIGHT_API_LLM_VERTEXAI_REGION}"
fi
fi
# Add optional embeddings provider config
if [ -n "${HINDSIGHT_API_EMBEDDINGS_PROVIDER:-}" ]; then
DOCKER_CMD="$DOCKER_CMD -e HINDSIGHT_API_EMBEDDINGS_PROVIDER=${HINDSIGHT_API_EMBEDDINGS_PROVIDER}"
@@ -181,21 +162,6 @@ for i in $(seq 1 "$TIMEOUT"); do
echo "=== Health Response ==="
curl -s "http://localhost:${HEALTH_PORT}${HEALTH_PATH}" | python3 -m json.tool 2>/dev/null || curl -s "http://localhost:${HEALTH_PORT}${HEALTH_PATH}"
echo ""
# Run retain/recall smoke test for API targets
if [ "$TARGET" != "cp-only" ]; then
echo ""
echo "=== Retain/Recall Smoke Test ==="
if ! "$REPO_ROOT/scripts/smoke-test-slim.sh" "http://localhost:${HEALTH_PORT}"; then
echo ""
echo "=== Container Logs (last 50 lines) ==="
docker logs "$CONTAINER_NAME" 2>&1 | tail -50
echo ""
echo -e "${RED}Smoke test FAILED${NC}"
exit 1
fi
fi
echo ""
echo "=== Container Logs (last 50 lines) ==="
docker logs "$CONTAINER_NAME" 2>&1 | tail -50
+9 -5
View File
@@ -6,17 +6,24 @@
# It expects API keys to be set in environment variables.
#
# Usage:
# export GROQ_API_KEY=gsk_xxx
# export OPENAI_API_KEY=sk-xxx
# export COHERE_API_KEY=xxx
# ./docker/test-slim-local.sh
#
# Or inline:
# OPENAI_API_KEY=sk_xxx COHERE_API_KEY=xxx ./docker/test-slim-local.sh
# GROQ_API_KEY=gsk_xxx OPENAI_API_KEY=sk_xxx COHERE_API_KEY=xxx ./docker/test-slim-local.sh
#
set -euo pipefail
# Check for required API keys
if [ -z "${GROQ_API_KEY:-}" ]; then
echo "❌ Error: GROQ_API_KEY environment variable is required"
echo "Set it with: export GROQ_API_KEY=gsk_xxx"
exit 1
fi
if [ -z "${OPENAI_API_KEY:-}" ]; then
echo "❌ Error: OPENAI_API_KEY environment variable is required"
echo "Set it with: export OPENAI_API_KEY=sk-xxx"
@@ -34,10 +41,7 @@ IMAGE="${1:-hindsight-slim:test}"
echo "Testing image: $IMAGE"
echo ""
# Set up LLM and external providers
export HINDSIGHT_API_LLM_PROVIDER=openai
export HINDSIGHT_API_LLM_API_KEY=$OPENAI_API_KEY
export HINDSIGHT_API_LLM_MODEL=gpt-4o-mini
# Set up external providers
export HINDSIGHT_API_EMBEDDINGS_PROVIDER=openai
export HINDSIGHT_API_EMBEDDINGS_OPENAI_API_KEY=$OPENAI_API_KEY
export HINDSIGHT_API_RERANKER_PROVIDER=cohere
+2 -2
View File
@@ -2,8 +2,8 @@ apiVersion: v2
name: hindsight
description: Hindsight helm chart
type: application
version: 0.5.1
appVersion: "0.5.1"
version: 0.4.11
appVersion: "0.4.11"
keywords:
- ai
- memory
@@ -95,27 +95,6 @@ spec:
{{- toYaml .Values.api.readinessProbe | nindent 10 }}
resources:
{{- toYaml .Values.api.resources | nindent 10 }}
{{- if or .Values.api.persistence.modelCache.enabled .Values.api.extraVolumeMounts }}
volumeMounts:
{{- if .Values.api.persistence.modelCache.enabled }}
- name: model-cache
mountPath: /home/hindsight/.cache
{{- end }}
{{- with .Values.api.extraVolumeMounts }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- end }}
{{- if or .Values.api.persistence.modelCache.enabled .Values.api.extraVolumes }}
volumes:
{{- if .Values.api.persistence.modelCache.enabled }}
- name: model-cache
persistentVolumeClaim:
claimName: {{ include "hindsight.fullname" . }}-api-model-cache
{{- end }}
{{- with .Values.api.extraVolumes }}
{{- toYaml . | nindent 6 }}
{{- end }}
{{- end }}
{{- with .Values.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
@@ -1,21 +0,0 @@
{{- if and .Values.api.enabled .Values.api.persistence.modelCache.enabled }}
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: {{ include "hindsight.fullname" . }}-api-model-cache
labels:
{{- include "hindsight.api.labels" . | nindent 4 }}
{{- with .Values.api.persistence.modelCache.annotations }}
annotations:
{{- toYaml . | nindent 4 }}
{{- end }}
spec:
accessModes:
{{- toYaml .Values.api.persistence.modelCache.accessModes | nindent 4 }}
{{- if .Values.api.persistence.modelCache.storageClass }}
storageClassName: {{ .Values.api.persistence.modelCache.storageClass }}
{{- end }}
resources:
requests:
storage: {{ .Values.api.persistence.modelCache.size }}
{{- end }}
@@ -95,16 +95,6 @@ spec:
{{- toYaml .Values.worker.readinessProbe | nindent 10 }}
resources:
{{- toYaml .Values.worker.resources | nindent 10 }}
{{- if or .Values.worker.persistence.modelCache.enabled .Values.worker.extraVolumeMounts }}
volumeMounts:
{{- if .Values.worker.persistence.modelCache.enabled }}
- name: model-cache
mountPath: /home/hindsight/.cache
{{- end }}
{{- with .Values.worker.extraVolumeMounts }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- end }}
{{- with .Values.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
@@ -117,26 +107,4 @@ spec:
tolerations:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.worker.extraVolumes }}
volumes:
{{- toYaml . | nindent 6 }}
{{- end }}
{{- if .Values.worker.persistence.modelCache.enabled }}
volumeClaimTemplates:
- metadata:
name: model-cache
{{- with .Values.worker.persistence.modelCache.annotations }}
annotations:
{{- toYaml . | nindent 8 }}
{{- end }}
spec:
accessModes:
{{- toYaml .Values.worker.persistence.modelCache.accessModes | nindent 8 }}
{{- if .Values.worker.persistence.modelCache.storageClass }}
storageClassName: {{ .Values.worker.persistence.modelCache.storageClass }}
{{- end }}
resources:
requests:
storage: {{ .Values.worker.persistence.modelCache.size }}
{{- end }}
{{- end }}
-53
View File
@@ -67,33 +67,6 @@ api:
# Pod affinity/anti-affinity (overrides global affinity for this component)
# affinity: {}
# Persistent volume for local model cache (reranker, embeddings)
# Models are downloaded to /home/hindsight/.cache on first use.
# Without persistence, models are re-downloaded on every pod restart.
persistence:
modelCache:
enabled: false
size: 5Gi
storageClass: ""
accessModes:
- ReadWriteOnce
annotations: {}
# Extra volume mounts for the api container
# e.g.
# extraVolumeMounts:
# - name: my-volume
# mountPath: /mnt/my-volume
extraVolumeMounts: []
# Extra volumes for the api pod
# e.g.
# extraVolumes:
# - name: my-volume
# configMap:
# name: my-configmap
extraVolumes: []
# Environment variables
env:
#HINDSIGHT_API_LLM_PROVIDER: "groq"
@@ -167,32 +140,6 @@ worker:
# Pod affinity/anti-affinity (overrides global affinity for this component)
# affinity: {}
# Persistent volume for local model cache (reranker, embeddings)
# Uses volumeClaimTemplates since worker is a StatefulSet.
persistence:
modelCache:
enabled: false
size: 5Gi
storageClass: ""
accessModes:
- ReadWriteOnce
annotations: {}
# Extra volume mounts for the worker container
# e.g.
# extraVolumeMounts:
# - name: my-volume
# mountPath: /mnt/my-volume
extraVolumeMounts: []
# Extra volumes for the worker pod
# e.g.
# extraVolumes:
# - name: my-volume
# configMap:
# name: my-configmap
extraVolumes: []
# Secret environment variables (inherited from api.secrets if not specified)
secrets: {}
-4
View File
@@ -1,4 +0,0 @@
node_modules
dist
*.tgz
.DS_Store
-80
View File
@@ -1,80 +0,0 @@
# @vectorize-io/hindsight-all
Node.js equivalent of the Python [`hindsight-all`](https://pypi.org/project/hindsight-all/) package — programmatic lifecycle manager for a local Hindsight daemon. Use this when you want to embed Hindsight in a Node application without hand-rolling subprocess management.
This package deliberately does **not** ship an HTTP client. Once the daemon is running, talk to it with [`@vectorize-io/hindsight-client`](https://www.npmjs.com/package/@vectorize-io/hindsight-client) against `server.getBaseUrl()`. The two packages compose — one owns the daemon process, the other owns the HTTP API surface.
## Requirements
- **Node.js >= 22** — uses global `fetch` and `AbortSignal.timeout`.
- **`uv` / `uvx`** on `PATH` — used to download and run the underlying `hindsight-embed` daemon on first use. Install via <https://docs.astral.sh/uv/>.
## Install
```bash
npm install @vectorize-io/hindsight-all @vectorize-io/hindsight-client
```
## Example
```ts
import { HindsightServer, consoleLogger } from '@vectorize-io/hindsight-all';
import { HindsightClient } from '@vectorize-io/hindsight-client';
const server = new HindsightServer({
profile: 'my-app',
port: 9077,
env: {
HINDSIGHT_API_LLM_PROVIDER: 'anthropic',
HINDSIGHT_API_LLM_API_KEY: process.env.ANTHROPIC_API_KEY,
HINDSIGHT_API_LLM_MODEL: 'claude-sonnet-4-20250514',
HINDSIGHT_EMBED_DAEMON_IDLE_TIMEOUT: '0',
},
logger: consoleLogger,
});
await server.start();
const client = new HindsightClient({ baseUrl: server.getBaseUrl() });
await client.retain('user-123', 'User prefers dark mode and concise answers.', {
documentId: 'pref-2026-04-01',
});
const recall = await client.recall('user-123', 'what are the user preferences?');
console.log(recall.results);
await server.stop();
```
For a remote Hindsight API, skip `HindsightServer` entirely and just point `HindsightClient` at the remote URL.
## Open config — forward-compatible with new daemon flags
`HindsightServerOptions` is designed so every new environment variable or CLI flag in the underlying Hindsight daemon can be used without waiting for a wrapper release:
- **`env`** accepts an arbitrary `Record<string, string>`. Every entry is exported into the daemon process and written into the profile config via `--env KEY=VALUE`.
- **`extraProfileCreateArgs`** / **`extraDaemonStartArgs`** append raw args to the respective commands.
## Development against a local checkout
If you're hacking on the Python `hindsight-embed` package in the same monorepo, point the server at the local path — it'll use `uv run --directory <path>` instead of `uvx`:
```ts
new HindsightServer({
embedPackagePath: '/path/to/hindsight-embed',
// ...
});
```
## API surface
- `HindsightServer` — daemon lifecycle (`start`, `stop`, `checkHealth`, `getBaseUrl`, `getProfile`).
- `Logger` interface plus `silentLogger` (default) and `consoleLogger` helpers.
- `getEmbedCommand(opts)` — low-level helper that returns the `[cmd, ...args]` tuple used to invoke the underlying Python CLI.
For memory operations (retain, recall, reflect, bank management, stats) use [`@vectorize-io/hindsight-client`](https://www.npmjs.com/package/@vectorize-io/hindsight-client).
## License
MIT
-57
View File
@@ -1,57 +0,0 @@
{
"name": "@vectorize-io/hindsight-all",
"version": "0.5.1",
"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",
"type": "module",
"exports": {
".": {
"types": "./dist/index.d.ts",
"import": "./dist/index.js"
}
},
"keywords": [
"hindsight",
"hindsight-all",
"memory",
"ai",
"agent",
"long-term-memory",
"llm",
"embedded-server"
],
"author": "Vectorize <[email protected]>",
"license": "MIT",
"repository": {
"type": "git",
"url": "https://github.com/vectorize-io/hindsight.git",
"directory": "hindsight-all-npm"
},
"files": [
"dist",
"README.md"
],
"scripts": {
"build": "tsup",
"dev": "tsup --watch",
"clean": "rm -rf dist",
"test": "vitest run src",
"test:watch": "vitest src",
"prepublishOnly": "npm run clean && npm run build"
},
"devDependencies": {
"@types/node": "^22.0.0",
"tsup": "^8.5.1",
"typescript": "^5.7.0",
"vitest": "^4.1.2"
},
"engines": {
"node": ">=22"
},
"overrides": {
"rollup": "^4.59.0",
"picomatch": ">=2.3.2 <3.0.0 || >=4.0.4",
"vite": ">=8.0.5"
}
}
-32
View File
@@ -1,32 +0,0 @@
import { describe, it, expect } from 'vitest';
import { getEmbedCommand } from './command.js';
describe('getEmbedCommand', () => {
it('defaults to uvx hindsight-embed@latest', () => {
expect(getEmbedCommand()).toEqual(['uvx', 'hindsight-embed@latest']);
});
it('honours an explicit version', () => {
expect(getEmbedCommand({ embedVersion: '0.5.0' })).toEqual(['uvx', '[email protected]']);
});
it('treats an empty version as latest', () => {
expect(getEmbedCommand({ embedVersion: '' })).toEqual(['uvx', 'hindsight-embed@latest']);
});
it('uses uv run --directory when a local path is given', () => {
expect(getEmbedCommand({ embedPackagePath: '/abs/path' })).toEqual([
'uv',
'run',
'--directory',
'/abs/path',
'hindsight-embed',
]);
});
it('local path takes precedence over version', () => {
expect(
getEmbedCommand({ embedPackagePath: '/abs/path', embedVersion: '0.5.0' }),
).toEqual(['uv', 'run', '--directory', '/abs/path', 'hindsight-embed']);
});
});
-25
View File
@@ -1,25 +0,0 @@
/**
* Resolve the command that invokes the `hindsight-embed` Python CLI.
*
* - If `embedPackagePath` is set, runs the package from a local checkout via
* `uv run --directory <path> hindsight-embed`. Used for in-repo development.
* - Otherwise runs it via `uvx hindsight-embed@<version>` so no global install
* is required.
*
* Returns the argv as `[command, ...baseArgs]` suitable for `spawn()` /
* `execFile()` (never shell-interpolated).
*/
export interface EmbedCommandOptions {
/** Version spec passed to uvx (e.g. "latest", "0.5.0"). Default: "latest". */
embedVersion?: string;
/** Local checkout path. When set, overrides `embedVersion` and uses `uv run`. */
embedPackagePath?: string;
}
export function getEmbedCommand(opts: EmbedCommandOptions = {}): string[] {
if (opts.embedPackagePath) {
return ['uv', 'run', '--directory', opts.embedPackagePath, 'hindsight-embed'];
}
const version = opts.embedVersion && opts.embedVersion.length > 0 ? opts.embedVersion : 'latest';
return ['uvx', `hindsight-embed@${version}`];
}
-7
View File
@@ -1,7 +0,0 @@
export { HindsightServer } from './server.js';
export { getEmbedCommand } from './command.js';
export { silentLogger, consoleLogger } from './logger.js';
export type { Logger } from './logger.js';
export type { EmbedCommandOptions } from './command.js';
export type { HindsightServerOptions } from './types.js';
-29
View File
@@ -1,29 +0,0 @@
/**
* Pluggable logger interface.
*
* This package does not own any logging infrastructure — consumers inject
* whatever they want (console, pino, openclaw's logger, a no-op). The default
* is silent so embedding this package never adds noise to an unrelated app.
*/
export interface Logger {
debug(msg: string): void;
info(msg: string): void;
warn(msg: string): void;
error(msg: string): void;
}
/** Logger that drops every call. Used when no logger is passed. */
export const silentLogger: Logger = {
debug: () => {},
info: () => {},
warn: () => {},
error: () => {},
};
/** Logger that writes to the standard console. Handy for CLIs and tests. */
export const consoleLogger: Logger = {
debug: (msg) => console.debug(msg),
info: (msg) => console.log(msg),
warn: (msg) => console.warn(msg),
error: (msg) => console.error(msg),
};
-35
View File
@@ -1,35 +0,0 @@
import { describe, it, expect } from 'vitest';
import { HindsightServer } from './server.js';
describe('HindsightServer construction', () => {
it('defaults base URL to http://127.0.0.1:8888', () => {
const server = new HindsightServer();
expect(server.getBaseUrl()).toBe('http://127.0.0.1:8888');
expect(server.getProfile()).toBe('default');
});
it('honours custom profile, port, and host', () => {
const server = new HindsightServer({ profile: 'app', port: 9077, host: '0.0.0.0' });
expect(server.getProfile()).toBe('app');
expect(server.getBaseUrl()).toBe('http://0.0.0.0:9077');
});
it('accepts open env pass-through without complaining about unknown keys', () => {
const server = new HindsightServer({
env: {
HINDSIGHT_API_LLM_PROVIDER: 'openai',
HINDSIGHT_API_LLM_MODEL: 'gpt-4o-mini',
// A field that does not exist today — should still be accepted
HINDSIGHT_FUTURE_FLAG: 'enabled',
},
});
expect(server).toBeInstanceOf(HindsightServer);
});
it('exposes checkHealth that returns false when no daemon is running', async () => {
// Random high port that nothing is listening on.
const server = new HindsightServer({ port: 1, readyTimeoutMs: 100 });
const healthy = await server.checkHealth();
expect(healthy).toBe(false);
});
});
-322
View File
@@ -1,322 +0,0 @@
import { spawn } from 'child_process';
import { getEmbedCommand } from './command.js';
import { silentLogger } from './logger.js';
import type { Logger } from './logger.js';
import type { HindsightServerOptions } from './types.js';
const DEFAULT_PORT = 8888;
const DEFAULT_HOST = '127.0.0.1';
const DEFAULT_PROFILE = 'default';
const DEFAULT_READY_TIMEOUT_MS = 30_000;
const DEFAULT_READY_POLL_INTERVAL_MS = 1_000;
/**
* Manages the lifecycle of a local Hindsight daemon from a Node.js process.
*
* On {@link start}, this class:
* 1. Resolves the `hindsight-embed` command (via `uvx` or a local `uv run`).
* 2. Runs `profile create <name> --merge --port <port> [--env K=V ...]`
* with every entry in {@link HindsightServerOptions.env} forwarded as
* an `--env` flag.
* 3. Runs `daemon --profile <name> start` and waits for the start command
* to exit.
* 4. Polls `http://host:port/health` until it returns `200` or the
* `readyTimeoutMs` budget is exhausted.
*
* On {@link stop}, it runs `daemon --profile <name> stop` and returns once
* the command exits (or after a short grace period).
*
* This is the Node.js equivalent of the Python `hindsight-all` package's
* `HindsightServer`: a thin programmatic lifecycle wrapper around the
* Hindsight daemon. It does NOT ship an HTTP client — once `start()`
* resolves, use `@vectorize-io/hindsight-client` against `getBaseUrl()` for
* retain / recall / reflect.
*
* The class is deliberately transparent about the daemon: new CLI flags or
* environment variables never require a code change here — callers can pass
* them via `env`, `extraProfileCreateArgs`, or `extraDaemonStartArgs`.
*/
export class HindsightServer {
private readonly profile: string;
private readonly port: number;
private readonly host: string;
private readonly baseUrl: string;
private readonly embedVersion: string | undefined;
private readonly embedPackagePath: string | undefined;
private readonly userEnv: Record<string, string | undefined>;
private readonly extraProfileCreateArgs: string[];
private readonly extraDaemonStartArgs: string[];
private readonly platformCpuWorkaround: boolean;
private readonly readyTimeoutMs: number;
private readonly readyPollIntervalMs: number;
private readonly logger: Logger;
constructor(opts: HindsightServerOptions = {}) {
this.profile = opts.profile ?? DEFAULT_PROFILE;
this.port = opts.port ?? DEFAULT_PORT;
this.host = opts.host ?? DEFAULT_HOST;
this.baseUrl = `http://${this.host}:${this.port}`;
this.embedVersion = opts.embedVersion;
this.embedPackagePath = opts.embedPackagePath;
this.userEnv = opts.env ?? {};
this.extraProfileCreateArgs = opts.extraProfileCreateArgs ?? [];
this.extraDaemonStartArgs = opts.extraDaemonStartArgs ?? [];
this.platformCpuWorkaround = opts.platformCpuWorkaround ?? (process.platform === 'darwin');
this.readyTimeoutMs = opts.readyTimeoutMs ?? DEFAULT_READY_TIMEOUT_MS;
this.readyPollIntervalMs = opts.readyPollIntervalMs ?? DEFAULT_READY_POLL_INTERVAL_MS;
this.logger = opts.logger ?? silentLogger;
}
/** The base URL the daemon listens on (`http://host:port`). */
getBaseUrl(): string {
return this.baseUrl;
}
/** The profile name this server operates on. */
getProfile(): string {
return this.profile;
}
/**
* Ensure the daemon is configured and running. Idempotent — the underlying
* `profile create --merge` and `daemon start` commands tolerate re-runs.
*/
async start(): Promise<void> {
this.logger.info(`[hindsight] starting daemon for profile "${this.profile}"`);
const env = this.buildEnv();
await this.configureProfile(env);
await this.startDaemon(env);
await this.waitForReady();
this.logger.info(`[hindsight] daemon ready at ${this.baseUrl}`);
}
/** Stop the daemon. Never throws — logs and resolves even on failure. */
async stop(): Promise<void> {
this.logger.info(`[hindsight] stopping daemon for profile "${this.profile}"`);
const [cmd, ...baseArgs] = getEmbedCommand({
embedVersion: this.embedVersion,
embedPackagePath: this.embedPackagePath,
});
const args = [...baseArgs, 'daemon', '--profile', this.profile, 'stop'];
const child = spawn(cmd, args, { stdio: 'pipe' });
this.pipeOutput(child, 'daemon.stop');
await new Promise<void>((resolve) => {
const timeout = setTimeout(() => {
this.logger.warn(`[hindsight] daemon stop timed out after 5s`);
resolve();
}, 5_000);
child.on('exit', () => {
clearTimeout(timeout);
this.logger.info(`[hindsight] daemon stopped`);
resolve();
});
child.on('error', (err) => {
clearTimeout(timeout);
this.logger.warn(`[hindsight] error stopping daemon: ${err.message}`);
resolve();
});
});
}
/** Probe `/health` once with a short timeout. */
async checkHealth(): Promise<boolean> {
try {
const res = await fetch(`${this.baseUrl}/health`, {
signal: AbortSignal.timeout(2_000),
});
return res.ok;
} catch {
return false;
}
}
// -------------------------------------------------------------------------
// Internal
// -------------------------------------------------------------------------
/**
* Merge the process env, the caller-supplied `env`, and (on macOS) the
* embeddings CPU workaround. Caller-supplied values always win over the
* workaround; undefined values are dropped.
*/
private buildEnv(): NodeJS.ProcessEnv {
const merged: NodeJS.ProcessEnv = { ...process.env };
if (this.platformCpuWorkaround && process.platform === 'darwin') {
merged['HINDSIGHT_API_EMBEDDINGS_LOCAL_FORCE_CPU'] = '1';
merged['HINDSIGHT_API_RERANKER_LOCAL_FORCE_CPU'] = '1';
}
for (const [key, value] of Object.entries(this.userEnv)) {
if (value !== undefined) {
merged[key] = value;
}
}
return merged;
}
/**
* Run `profile create <name> --merge --port <port> [--env K=V ...]`.
* Every entry in the merged env that was passed via {@link userEnv} (or
* auto-applied by the CPU workaround) is forwarded as `--env`.
*/
private async configureProfile(env: NodeJS.ProcessEnv): Promise<void> {
this.logger.info(`[hindsight] configuring profile "${this.profile}"`);
const [cmd, ...baseArgs] = getEmbedCommand({
embedVersion: this.embedVersion,
embedPackagePath: this.embedPackagePath,
});
const createArgs = [
...baseArgs,
'profile',
'create',
this.profile,
'--merge',
'--port',
String(this.port),
];
// Forward every env var that the caller intended for the daemon as --env.
// We only forward keys the caller explicitly set (userEnv) plus the CPU
// workaround values — not the entire process.env, to avoid leaking random
// host state into profile config.
const envForProfile = this.collectProfileEnv(env);
for (const [key, value] of Object.entries(envForProfile)) {
createArgs.push('--env', `${key}=${value}`);
}
createArgs.push(...this.extraProfileCreateArgs);
await this.runCommand(cmd, createArgs, env, 'profile.create');
}
/** Collect only the env vars that should be written into the profile file. */
private collectProfileEnv(env: NodeJS.ProcessEnv): Record<string, string> {
const out: Record<string, string> = {};
// 1. User-supplied env — always forwarded.
for (const [key, value] of Object.entries(this.userEnv)) {
if (value !== undefined) {
out[key] = value;
}
}
// 2. CPU workaround — only if auto-applied and not already overridden.
if (this.platformCpuWorkaround && process.platform === 'darwin') {
const cpuKeys = [
'HINDSIGHT_API_EMBEDDINGS_LOCAL_FORCE_CPU',
'HINDSIGHT_API_RERANKER_LOCAL_FORCE_CPU',
];
for (const key of cpuKeys) {
if (!(key in out) && env[key] !== undefined) {
out[key] = env[key] as string;
}
}
}
return out;
}
private async startDaemon(env: NodeJS.ProcessEnv): Promise<void> {
const [cmd, ...baseArgs] = getEmbedCommand({
embedVersion: this.embedVersion,
embedPackagePath: this.embedPackagePath,
});
const args = [
...baseArgs,
'daemon',
'--profile',
this.profile,
'start',
...this.extraDaemonStartArgs,
];
await this.runCommand(cmd, args, env, 'daemon.start');
}
/**
* Spawn `cmd` with `args`, pipe its output through the logger, and resolve
* once it exits with code 0. Rejects on non-zero exit or spawn error.
*/
private async runCommand(
cmd: string,
args: string[],
env: NodeJS.ProcessEnv,
label: string,
): Promise<void> {
const child = spawn(cmd, args, { stdio: 'pipe', env });
let output = '';
child.stdout?.on('data', (data: Buffer) => {
const text = data.toString();
output += text;
for (const line of text.trimEnd().split('\n')) {
if (line) this.logger.info(`[hindsight:${label}] ${line}`);
}
});
child.stderr?.on('data', (data: Buffer) => {
const text = data.toString();
output += text;
for (const line of text.trimEnd().split('\n')) {
if (line) this.logger.warn(`[hindsight:${label}] ${line}`);
}
});
await new Promise<void>((resolve, reject) => {
child.on('exit', (code) => {
if (code === 0) {
resolve();
} else {
reject(new Error(`${label} failed with code ${code}: ${output.trim()}`));
}
});
child.on('error', (err) => {
reject(new Error(`${label} failed to spawn: ${err.message}`, { cause: err }));
});
});
}
/** Stream a spawned child's stdout/stderr through the logger without blocking. */
private pipeOutput(child: ReturnType<typeof spawn>, label: string): void {
child.stdout?.on('data', (data: Buffer) => {
for (const line of data.toString().trimEnd().split('\n')) {
if (line) this.logger.info(`[hindsight:${label}] ${line}`);
}
});
child.stderr?.on('data', (data: Buffer) => {
for (const line of data.toString().trimEnd().split('\n')) {
if (line) this.logger.warn(`[hindsight:${label}] ${line}`);
}
});
}
/** Poll `/health` until it succeeds or `readyTimeoutMs` elapses. */
private async waitForReady(): Promise<void> {
const deadline = Date.now() + this.readyTimeoutMs;
let attempt = 0;
while (Date.now() < deadline) {
attempt++;
try {
const res = await fetch(`${this.baseUrl}/health`, {
signal: AbortSignal.timeout(this.readyPollIntervalMs),
});
if (res.ok) {
this.logger.debug(`[hindsight] health check passed (attempt ${attempt})`);
return;
}
} catch {
// expected while the daemon is still booting
}
await new Promise((resolve) => setTimeout(resolve, this.readyPollIntervalMs));
}
throw new Error(
`Hindsight daemon did not become ready within ${this.readyTimeoutMs}ms at ${this.baseUrl}`,
);
}
}
-54
View File
@@ -1,54 +0,0 @@
import type { Logger } from './logger.js';
/**
* Options for {@link HindsightServer}.
*
* The server is intentionally thin and pass-through: anything configurable
* on the daemon side (env vars or CLI flags) can be set here without needing
* a new dedicated option. Use {@link env} for `HINDSIGHT_*` / `OPENAI_API_KEY` /
* custom provider settings, and the two `extra*` arrays to append raw CLI
* args to `profile create` or `daemon start`.
*
* For talking to the daemon after `start()`, use `@vectorize-io/hindsight-client`
* against `server.getBaseUrl()`. This package does not ship its own HTTP
* client.
*/
export interface HindsightServerOptions {
/** Profile name used for `--profile <name>` on every sub-command. Default: `"default"`. */
profile?: string;
/** TCP port the daemon listens on. Default: `8888`. */
port?: number;
/** Hostname the daemon binds to (for health checks). Default: `127.0.0.1`. */
host?: string;
/** Version of the underlying `hindsight-embed` PyPI package to run via `uvx`. Default: `"latest"`. */
embedVersion?: string;
/** Local path to a `hindsight-embed` checkout — takes precedence over `embedVersion`. */
embedPackagePath?: string;
/**
* Environment variables passed to the daemon process AND written into the
* profile via repeated `--env KEY=VALUE` flags. This is the preferred way
* to surface any `HINDSIGHT_API_*` / `HINDSIGHT_EMBED_*` setting — adding a
* new daemon env var never requires a wrapper update.
*
* Values of `undefined` are dropped (so you can spread conditionally).
*/
env?: Record<string, string | undefined>;
/** Extra args appended verbatim to `hindsight-embed profile create <name> --merge ...`. */
extraProfileCreateArgs?: string[];
/** Extra args appended verbatim to `hindsight-embed daemon --profile <name> start ...`. */
extraDaemonStartArgs?: string[];
/**
* On macOS, automatically set
* `HINDSIGHT_API_EMBEDDINGS_LOCAL_FORCE_CPU=1` and
* `HINDSIGHT_API_RERANKER_LOCAL_FORCE_CPU=1` to avoid Metal/MPS crashes in
* daemon mode. Default: `true` on `darwin`, ignored elsewhere. Any value set
* explicitly in {@link env} wins over the auto-applied value.
*/
platformCpuWorkaround?: boolean;
/** Max time (ms) to wait for `/health` to return 200. Default: `30_000`. */
readyTimeoutMs?: number;
/** Polling interval (ms) while waiting for `/health`. Default: `1_000`. */
readyPollIntervalMs?: number;
/** Optional pluggable logger. Default: silent. */
logger?: Logger;
}
-18
View File
@@ -1,18 +0,0 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "ES2022",
"lib": ["ES2022"],
"moduleResolution": "node",
"declaration": true,
"outDir": "./dist",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist", "src/**/*.test.ts"]
}
-11
View File
@@ -1,11 +0,0 @@
import { defineConfig } from 'tsup';
export default defineConfig({
entry: ['src/index.ts'],
format: ['esm'],
dts: true,
outDir: 'dist',
clean: true,
sourcemap: true,
bundle: true,
});
-8
View File
@@ -1,8 +0,0 @@
import { defineConfig } from 'vitest/config';
export default defineConfig({
test: {
include: ['src/**/*.test.ts'],
environment: 'node',
},
});
-33
View File
@@ -1,33 +0,0 @@
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[project]
name = "hindsight-all-slim"
version = "0.5.1"
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.4.17",
"hindsight-client>=0.0.7",
"hindsight-embed>=0.1.0",
]
[tool.uv.sources]
hindsight-api-slim = { workspace = true }
hindsight-client = { workspace = true }
hindsight-embed = { workspace = true }
[project.optional-dependencies]
test = [
"pytest>=7.0.0",
"pytest-asyncio>=0.21.0",
]
[tool.setuptools]
packages = []
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
-48
View File
@@ -1,48 +0,0 @@
# hindsight-all
All-in-one package for Hindsight - Agent Memory That Works Like Human Memory
## Quick Start
```python
from hindsight import start_server, HindsightClient
# Start server with embedded PostgreSQL
server = start_server(
llm_provider="groq",
llm_api_key="your-api-key",
llm_model="openai/gpt-oss-120b"
)
# Create client
client = HindsightClient(base_url=server.url)
# Store memories
client.put(agent_id="assistant", content="User prefers Python for data analysis")
# Search memories
results = client.search(agent_id="assistant", query="programming preferences")
# Generate contextual response
response = client.think(agent_id="assistant", query="What languages should I recommend?")
# Stop server when done
server.stop()
```
## Using Context Manager
```python
from hindsight import HindsightServer, HindsightClient
with HindsightServer(llm_provider="groq", llm_api_key="...") as server:
client = HindsightClient(base_url=server.url)
# ... use client ...
# Server automatically stops
```
## Installation
```bash
pip install hindsight-all
```
-423
View File
@@ -1,423 +0,0 @@
"""
Wrapper for Hindsight client that adds API namespaces.
Provides organized access to different parts of the Hindsight API through
namespaces like .banks, .mental_models, etc.
"""
from __future__ import annotations
from typing import Any
from hindsight_client import Hindsight
class BanksAPI:
"""Namespace for bank-related operations.
Provides methods to create, delete, and manage memory banks.
"""
def __init__(self, client: Hindsight):
self._client = client
def create(
self,
bank_id: str,
name: str | None = None,
mission: str | None = None,
disposition: dict[str, Any] | None = None,
) -> Any:
"""Create a new bank.
Args:
bank_id: Unique identifier for the bank.
name: Optional display name for the bank.
mission: Optional mission statement for the bank.
disposition: Optional disposition configuration dict.
Returns:
Bank creation response from the API.
"""
return self._client.create_bank(
bank_id=bank_id,
name=name,
mission=mission,
disposition=disposition,
)
def delete(self, bank_id: str) -> Any:
"""Delete a bank.
Args:
bank_id: The ID of the bank to delete.
Returns:
Deletion response from the API.
"""
return self._client.delete_bank(bank_id=bank_id)
def set_mission(self, bank_id: str, mission: str) -> Any:
"""Set or update the mission for a bank.
Args:
bank_id: The ID of the bank.
mission: The mission statement to set.
Returns:
API response confirming the update.
"""
return self._client.set_mission(bank_id=bank_id, mission=mission)
def set_disposition(self, bank_id: str, disposition: dict[str, Any]) -> Any:
"""Set or update the disposition for a bank.
Args:
bank_id: The ID of the bank.
disposition: The disposition configuration dict.
Returns:
API response confirming the update.
"""
return self._client.set_disposition(bank_id=bank_id, disposition=disposition)
def list(self) -> Any:
"""List all banks.
Returns:
List of banks from the API.
"""
from hindsight_client.hindsight_client import _run_async
return _run_async(self._client._banks_api.list_banks())
class MentalModelsAPI:
"""Namespace for mental model operations.
Mental models are reusable knowledge structures that guide agent behavior.
"""
def __init__(self, client: Hindsight):
self._client = client
def create(
self,
bank_id: str,
name: str,
content: str,
tags: list[str] | None = None,
) -> Any:
"""Create a new mental model.
Args:
bank_id: The ID of the bank to add the model to.
name: Name for the mental model.
content: The content/instructions for the mental model.
tags: Optional list of tags for categorization.
Returns:
Creation response from the API.
"""
return self._client.create_mental_model(
bank_id=bank_id,
name=name,
content=content,
tags=tags,
)
def list(self, bank_id: str, tags: list[str] | None = None) -> Any:
"""List all mental models for a bank.
Args:
bank_id: The ID of the bank.
tags: Optional filter by tags.
Returns:
List of mental models.
"""
return self._client.list_mental_models(bank_id=bank_id, tags=tags)
def get(self, bank_id: str, mental_model_id: str) -> Any:
"""Get a specific mental model.
Args:
bank_id: The ID of the bank.
mental_model_id: The ID of the mental model.
Returns:
The mental model details.
"""
return self._client.get_mental_model(bank_id=bank_id, mental_model_id=mental_model_id)
def refresh(self, bank_id: str, mental_model_id: str) -> Any:
"""Refresh a mental model.
Args:
bank_id: The ID of the bank.
mental_model_id: The ID of the mental model to refresh.
Returns:
Refresh response from the API.
"""
return self._client.refresh_mental_model(bank_id=bank_id, mental_model_id=mental_model_id)
def update(
self,
bank_id: str,
mental_model_id: str,
name: str | None = None,
content: str | None = None,
tags: list[str] | None = None,
) -> Any:
"""Update a mental model.
Args:
bank_id: The ID of the bank.
mental_model_id: The ID of the mental model to update.
name: Optional new name.
content: Optional new content.
tags: Optional new tags list.
Returns:
Update response from the API.
"""
return self._client.update_mental_model(
bank_id=bank_id,
mental_model_id=mental_model_id,
name=name,
content=content,
tags=tags,
)
def delete(self, bank_id: str, mental_model_id: str) -> Any:
"""Delete a mental model.
Args:
bank_id: The ID of the bank.
mental_model_id: The ID of the mental model to delete.
Returns:
Deletion response from the API.
"""
return self._client.delete_mental_model(bank_id=bank_id, mental_model_id=mental_model_id)
class DirectivesAPI:
"""Namespace for directive operations.
Directives are explicit instructions that guide agent behavior.
"""
def __init__(self, client: Hindsight):
self._client = client
def create(
self,
bank_id: str,
name: str,
content: str,
tags: list[str] | None = None,
) -> Any:
"""Create a new directive.
Args:
bank_id: The ID of the bank to add the directive to.
name: Name for the directive.
content: The directive content/instructions.
tags: Optional list of tags for categorization.
Returns:
Creation response from the API.
"""
return self._client.create_directive(
bank_id=bank_id,
name=name,
content=content,
tags=tags,
)
def list(self, bank_id: str, tags: list[str] | None = None) -> Any:
"""List all directives for a bank.
Args:
bank_id: The ID of the bank.
tags: Optional filter by tags.
Returns:
List of directives.
"""
return self._client.list_directives(bank_id=bank_id, tags=tags)
def get(self, bank_id: str, directive_id: str) -> Any:
"""Get a specific directive.
Args:
bank_id: The ID of the bank.
directive_id: The ID of the directive.
Returns:
The directive details.
"""
return self._client.get_directive(bank_id=bank_id, directive_id=directive_id)
def update(
self,
bank_id: str,
directive_id: str,
name: str | None = None,
content: str | None = None,
tags: list[str] | None = None,
) -> Any:
"""Update a directive.
Args:
bank_id: The ID of the bank.
directive_id: The ID of the directive to update.
name: Optional new name.
content: Optional new content.
tags: Optional new tags list.
Returns:
Update response from the API.
"""
return self._client.update_directive(
bank_id=bank_id,
directive_id=directive_id,
name=name,
content=content,
tags=tags,
)
def delete(self, bank_id: str, directive_id: str) -> Any:
"""Delete a directive.
Args:
bank_id: The ID of the bank.
directive_id: The ID of the directive to delete.
Returns:
Deletion response from the API.
"""
return self._client.delete_directive(bank_id=bank_id, directive_id=directive_id)
class MemoriesAPI:
"""Namespace for memory operations.
Provides methods to query and retrieve stored memories.
"""
def __init__(self, client: Hindsight):
self._client = client
def list(
self,
bank_id: str,
type: str | None = None,
search_query: str | None = None,
limit: int = 100,
offset: int = 0,
) -> Any:
"""List memories in a bank.
Args:
bank_id: The ID of the bank to query.
type: Optional filter by memory type.
search_query: Optional search query for filtering.
limit: Maximum number of results to return (default: 100).
offset: Number of results to skip for pagination (default: 0).
Returns:
List of memories matching the criteria.
"""
return self._client.list_memories(
bank_id=bank_id,
type=type,
search_query=search_query,
limit=limit,
offset=offset,
)
class HindsightClient(Hindsight):
"""
Enhanced Hindsight client with organized API namespaces.
This wrapper extends the auto-generated Hindsight client with organized
access to different parts of the API through namespaces.
Example:
```python
from hindsight import HindsightClient
client = HindsightClient(base_url="http://localhost:8888")
# Core operations (inherited from Hindsight)
client.retain(bank_id="test", content="Hello")
results = client.recall(bank_id="test", query="Hello")
# Organized API access through namespaces
client.banks.create(bank_id="test", name="Test Bank")
models = client.mental_models.list(bank_id="test")
directives = client.directives.list(bank_id="test")
memories = client.memories.list(bank_id="test")
```
Attributes:
banks: Namespace for bank management operations.
mental_models: Namespace for mental model operations.
directives: Namespace for directive operations.
memories: Namespace for memory listing operations.
"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self._banks_namespace: BanksAPI | None = None
self._mental_models_namespace: MentalModelsAPI | None = None
self._directives_namespace: DirectivesAPI | None = None
self._memories_namespace: MemoriesAPI | None = None
@property
def banks(self) -> BanksAPI:
"""Access bank management operations.
Returns:
BanksAPI instance for bank operations.
"""
if self._banks_namespace is None:
self._banks_namespace = BanksAPI(self)
return self._banks_namespace
@property
def mental_models(self) -> MentalModelsAPI:
"""Access mental model operations.
Returns:
MentalModelsAPI instance for mental model operations.
"""
if self._mental_models_namespace is None:
self._mental_models_namespace = MentalModelsAPI(self)
return self._mental_models_namespace
@property
def directives(self) -> DirectivesAPI:
"""Access directive operations.
Returns:
DirectivesAPI instance for directive operations.
"""
if self._directives_namespace is None:
self._directives_namespace = DirectivesAPI(self)
return self._directives_namespace
@property
def memories(self) -> MemoriesAPI:
"""Access memory listing operations.
Returns:
MemoriesAPI instance for memory operations.
"""
if self._memories_namespace is None:
self._memories_namespace = MemoriesAPI(self)
return self._memories_namespace
View File
@@ -1,56 +0,0 @@
"""
Unit test for _cleanup lock timeout behavior.
Verifies that _cleanup completes even when the lock is held by another thread,
instead of hanging indefinitely (fixes #952).
"""
import threading
import time
from unittest.mock import MagicMock, patch
import pytest
def test_cleanup_completes_when_lock_held():
"""
_cleanup should complete (best-effort) even when self._lock is held
by another thread, e.g. during a long _ensure_started call.
"""
with patch.dict("sys.modules", {
"hindsight_client": MagicMock(),
"hindsight_embed": MagicMock(),
"hindsight.api_namespaces": MagicMock(),
}):
from hindsight.embedded import HindsightEmbedded
client = HindsightEmbedded.__new__(HindsightEmbedded)
client.profile = "test"
client._lock = threading.Lock()
client._closed = False
client._client = None
client._started = False
client._ui = False
# Simulate another thread holding the lock
client._lock.acquire()
cleanup_done = threading.Event()
def run_cleanup():
client._cleanup()
cleanup_done.set()
t = threading.Thread(target=run_cleanup)
t.start()
# Cleanup should complete within the timeout (5s) + margin
assert cleanup_done.wait(timeout=8.0), (
"_cleanup hung instead of timing out on lock acquisition"
)
# Release the lock from the simulating thread
client._lock.release()
t.join(timeout=1.0)
assert client._closed, "Client should be marked as closed after cleanup"
-137
View File
@@ -1,137 +0,0 @@
# Hindsight API
**Memory System for AI Agents** — Temporal + Semantic + Entity Memory Architecture using PostgreSQL with pgvector.
Hindsight gives AI agents persistent memory that works like human memory: it stores facts, tracks entities and relationships, handles temporal reasoning ("what happened last spring?"), and forms opinions based on configurable disposition traits.
## Installation
```bash
pip install hindsight-api
```
## Quick Start
### Run the Server
```bash
# Set your LLM provider
export HINDSIGHT_API_LLM_PROVIDER=openai
export HINDSIGHT_API_LLM_API_KEY=sk-xxxxxxxxxxxx
# Start the server (uses embedded PostgreSQL by default)
hindsight-api
```
The server starts at http://localhost:8888 with:
- REST API for memory operations
- MCP server at `/mcp` for tool-use integration
### Use the Python API
```python
from hindsight_api import MemoryEngine
# Create and initialize the memory engine
memory = MemoryEngine()
await memory.initialize()
# Create a memory bank for your agent
bank = await memory.create_memory_bank(
name="my-assistant",
background="A helpful coding assistant"
)
# Store a memory
await memory.retain(
memory_bank_id=bank.id,
content="The user prefers Python for data science projects"
)
# Recall memories
results = await memory.recall(
memory_bank_id=bank.id,
query="What programming language does the user prefer?"
)
# Reflect with reasoning
response = await memory.reflect(
memory_bank_id=bank.id,
query="Should I recommend Python or R for this ML project?"
)
```
## CLI Options
```bash
hindsight-api --help
# Common options
hindsight-api --port 9000 # Custom port (default: 8888)
hindsight-api --host 127.0.0.1 # Bind to localhost only
hindsight-api --workers 4 # Multiple worker processes
hindsight-api --log-level debug # Verbose logging
```
## Configuration
Configure via environment variables:
| Variable | Description | Default |
|----------|-------------|---------|
| `HINDSIGHT_API_DATABASE_URL` | PostgreSQL connection string | `pg0` (embedded) |
| `HINDSIGHT_API_LLM_PROVIDER` | `openai`, `anthropic`, `gemini`, `groq`, `ollama`, `lmstudio` | `openai` |
| `HINDSIGHT_API_LLM_API_KEY` | API key for LLM provider | - |
| `HINDSIGHT_API_LLM_MODEL` | Model name | `gpt-4o-mini` |
| `HINDSIGHT_API_HOST` | Server bind address | `0.0.0.0` |
| `HINDSIGHT_API_PORT` | Server port | `8888` |
### Example with External PostgreSQL
```bash
export HINDSIGHT_API_DATABASE_URL=postgresql://user:pass@localhost:5432/hindsight
export HINDSIGHT_API_LLM_PROVIDER=groq
export HINDSIGHT_API_LLM_API_KEY=gsk_xxxxxxxxxxxx
hindsight-api
```
## Docker
```bash
docker run --rm -it -p 8888:8888 \
-e HINDSIGHT_API_LLM_API_KEY=$OPENAI_API_KEY \
-v $HOME/.hindsight-docker:/home/hindsight/.pg0 \
ghcr.io/vectorize-io/hindsight:latest
```
## MCP Server
For local MCP integration without running the full API server:
```bash
hindsight-local-mcp
```
This runs a stdio-based MCP server that can be used directly with MCP-compatible clients.
## Key Features
- **Multi-Strategy Retrieval (TEMPR)** — Semantic, keyword, graph, and temporal search combined with RRF fusion
- **Entity Graph** — Automatic entity extraction and relationship tracking
- **Temporal Reasoning** — Native support for time-based queries
- **Disposition Traits** — Configurable skepticism, literalism, and empathy influence opinion formation
- **Three Memory Types** — World facts, bank actions, and formed opinions with confidence scores
## Documentation
Full documentation: [https://hindsight.vectorize.io](https://hindsight.vectorize.io)
- [Installation Guide](https://hindsight.vectorize.io/developer/installation)
- [Configuration Reference](https://hindsight.vectorize.io/developer/configuration)
- [API Reference](https://hindsight.vectorize.io/api-reference)
- [Python SDK](https://hindsight.vectorize.io/sdks/python)
## License
Apache 2.0
@@ -1,45 +0,0 @@
"""Recreate entities trigram index on LOWER(canonical_name) for case-insensitive matching
The previous GIN trigram index on canonical_name was case-sensitive, causing
"Alice" and "alice" to have different trigram sets. This recreates it on
LOWER(canonical_name) so the % operator matches case-insensitively.
Revision ID: d6e7f8a9b0c1
Revises: c5d6e7f8a9b0
Create Date: 2026-03-31
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "d6e7f8a9b0c1"
down_revision: str | Sequence[str] | None = "c5d6e7f8a9b0"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# Drop the old case-sensitive trigram index
op.execute("DROP INDEX IF EXISTS entities_canonical_name_trgm_idx")
# Create case-insensitive trigram index on LOWER(canonical_name)
op.execute(
f"CREATE INDEX IF NOT EXISTS entities_canonical_name_lower_trgm_idx "
f"ON {schema}entities USING GIN (LOWER(canonical_name) gin_trgm_ops)"
)
def downgrade() -> None:
op.execute("DROP INDEX IF EXISTS entities_canonical_name_lower_trgm_idx")
schema = _get_schema_prefix()
# Restore original case-sensitive index
op.execute(
f"CREATE INDEX IF NOT EXISTS entities_canonical_name_trgm_idx "
f"ON {schema}entities USING GIN (canonical_name gin_trgm_ops)"
)
@@ -1,70 +0,0 @@
"""Add file_storage table for BYTEA-based file storage
Revision ID: a1b2c3d4e5f6
Revises: y0t1u2v3w4x5
Create Date: 2026-02-16
Creates a dedicated table for storing uploaded files using BYTEA.
This provides zero-config file storage that "just works" for development
and small deployments. For production/scale, use S3-compatible storage.
Files are stored in a separate table to avoid bloating the documents table.
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "a1b2c3d4e5f6"
down_revision: str | Sequence[str] | None = "y0t1u2v3w4x5"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
"""Create file_storage table for BYTEA storage."""
schema = _get_schema_prefix()
# Create file_storage table (minimal: just key + data)
op.execute(
f"""
CREATE TABLE IF NOT EXISTS {schema}file_storage (
storage_key TEXT PRIMARY KEY,
data BYTEA NOT NULL
)
"""
)
# Add file tracking columns to documents table
op.execute(
f"""
ALTER TABLE {schema}documents
ADD COLUMN IF NOT EXISTS file_storage_key TEXT,
ADD COLUMN IF NOT EXISTS file_original_name TEXT,
ADD COLUMN IF NOT EXISTS file_content_type TEXT
"""
)
def downgrade() -> None:
"""Remove file_storage table and related columns."""
schema = _get_schema_prefix()
# Drop columns from documents table
op.execute(
f"""
ALTER TABLE {schema}documents
DROP COLUMN IF EXISTS file_storage_key,
DROP COLUMN IF EXISTS file_original_name,
DROP COLUMN IF EXISTS file_content_type
"""
)
# Drop file_storage table
op.execute(f"DROP TABLE IF EXISTS {schema}file_storage")
@@ -1,88 +0,0 @@
"""Add text_signals column to memory_units for enriched BM25 indexing.
text_signals stores a denormalized space-separated string of entity names
(and future signals) to improve full-text search recall without polluting
the stored fact text.
- vchord: text_signals included in tokenize() at insert time
- native: search_vector GENERATED column regenerated to include text_signals
- pg_textsearch: no change (index only supports a single base column)
Revision ID: a2b3c4d5e6f7
Revises: z1u2v3w4x5y6
Create Date: 2026-02-28
"""
import os
from collections.abc import Sequence
from alembic import context, op
revision: str = "a2b3c4d5e6f7"
down_revision: str | Sequence[str] | None = "aa2b3c4d5e6f"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def _detect_text_search_extension() -> str:
return os.getenv("HINDSIGHT_API_TEXT_SEARCH_EXTENSION", "native").lower()
def upgrade() -> None:
schema = _get_schema_prefix()
table = f"{schema}memory_units"
text_search_ext = _detect_text_search_extension()
# Add text_signals column (nullable TEXT, populated at retain time)
op.execute(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS text_signals TEXT")
if text_search_ext == "native":
# Native PostgreSQL: drop and recreate the GENERATED tsvector column to include text_signals
op.execute(f"ALTER TABLE {table} DROP COLUMN IF EXISTS search_vector")
op.execute(f"""
ALTER TABLE {table}
ADD COLUMN search_vector tsvector
GENERATED ALWAYS AS (
to_tsvector('english',
COALESCE(text, '') || ' ' ||
COALESCE(context, '') || ' ' ||
COALESCE(text_signals, '')
)
) STORED
""")
# Recreate GIN index (was dropped with the column)
op.execute(f"""
CREATE INDEX IF NOT EXISTS idx_memory_units_text_search
ON {table} USING gin(search_vector)
""")
# vchord: tokenize() call in fact_storage.py is updated to include text_signals at insert time
# pg_textsearch: no change — index operates on the base `text` column only
def downgrade() -> None:
schema = _get_schema_prefix()
table = f"{schema}memory_units"
text_search_ext = _detect_text_search_extension()
if text_search_ext == "native":
op.execute(f"DROP INDEX IF EXISTS {schema}idx_memory_units_text_search")
op.execute(f"ALTER TABLE {table} DROP COLUMN IF EXISTS search_vector")
op.execute(f"""
ALTER TABLE {table}
ADD COLUMN search_vector tsvector
GENERATED ALWAYS AS (
to_tsvector('english', COALESCE(text, '') || ' ' || COALESCE(context, ''))
) STORED
""")
op.execute(f"""
CREATE INDEX idx_memory_units_text_search
ON {table} USING gin(search_vector)
""")
op.execute(f"ALTER TABLE {table} DROP COLUMN IF EXISTS text_signals")
@@ -1,54 +0,0 @@
"""Add GIN index on source_memory_ids for observation lookup performance
Without this index, queries using the array overlap operator (&&) or array
containment (@>) on source_memory_ids require a full sequential scan over all
observation memory_units. At ~77k observations this was measured at 45ms per
query, becoming a bottleneck during consolidation recall (57-64s timeouts) and
user recall (18-27s average).
The GIN index reduces these queries to index scans: 45ms → 0.049ms (927x
speedup). Recall dropped from 18-27s to ~6s, and consolidation recall
stabilised from timeout to ~15s.
Created with CONCURRENTLY so the migration does not block reads or writes.
CONCURRENTLY requires running outside a transaction block, so the migration
emits an explicit COMMIT before the statement and uses IF NOT EXISTS for
idempotency.
Revision ID: a2b3c4d5e6f8
Revises: f7g8h9i0j1k2
Create Date: 2026-03-04
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "a2b3c4d5e6f8"
down_revision: str | Sequence[str] | None = "f7g8h9i0j1k2"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# CREATE INDEX CONCURRENTLY cannot run inside a transaction block.
# Commit the current Alembic transaction first.
op.execute("COMMIT")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_units_source_memory_ids "
f"ON {schema}memory_units USING GIN (source_memory_ids) "
f"WHERE source_memory_ids IS NOT NULL"
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_units_source_memory_ids")
@@ -1,52 +0,0 @@
"""Add consolidation_failed_at column to memory_units for tracking persistent LLM failures.
When all LLM retries are exhausted on a single-memory batch, the memory is marked
with consolidation_failed_at instead of consolidated_at, so it is not silently lost
and can be retried later via the API.
Revision ID: a3b4c5d6e7f8
Revises: g7h8i9j0k1l2
Create Date: 2026-03-17
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "a3b4c5d6e7f8"
down_revision: str | Sequence[str] | None = "g7h8i9j0k1l2"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(
f"""
ALTER TABLE {schema}memory_units
ADD COLUMN IF NOT EXISTS consolidation_failed_at TIMESTAMPTZ DEFAULT NULL
"""
)
# Index to efficiently query memories that failed consolidation for a given bank
op.execute(
f"""
CREATE INDEX IF NOT EXISTS idx_memory_units_consolidation_failed
ON {schema}memory_units (bank_id, consolidation_failed_at)
WHERE consolidation_failed_at IS NOT NULL AND fact_type IN ('experience', 'world')
"""
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"DROP INDEX IF EXISTS {schema}idx_memory_units_consolidation_failed")
op.execute(f"ALTER TABLE {schema}memory_units DROP COLUMN IF EXISTS consolidation_failed_at")
@@ -1,142 +0,0 @@
"""Fix per-bank vector indexes to match configured extension
Revision ID: a4b5c6d7e8f9
Revises: d6e7f8a9b0c1
Create Date: 2026-04-01
Migration d5e6f7a8b9c0 hardcoded HNSW when creating per-bank partial vector
indexes, ignoring HINDSIGHT_API_VECTOR_EXTENSION. Banks that existed when that
migration ran got HNSW indexes even when pgvectorscale (DiskANN) or vchord
was configured.
This migration detects the mismatch and recreates the affected indexes with
the correct type. Skipped entirely when the configured extension is pgvector
(the default), since those indexes are already correct.
"""
import os
from collections.abc import Sequence
from alembic import context, op
from sqlalchemy import text
revision: str = "a4b5c6d7e8f9"
down_revision: str | Sequence[str] | None = "d6e7f8a9b0c1"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_FACT_TYPES: dict[str, str] = {
"world": "worl",
"experience": "expr",
"observation": "obsv",
}
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def _target_index_type() -> str | None:
"""Return the target index type, or None if pgvector (no fix needed)."""
ext = os.getenv("HINDSIGHT_API_VECTOR_EXTENSION", "pgvector").lower()
if ext == "pgvectorscale":
return "diskann"
elif ext == "vchord":
return "vchordrq"
return None
def _vector_index_using_clause() -> str:
"""Return the USING clause based on the configured vector extension."""
ext = os.getenv("HINDSIGHT_API_VECTOR_EXTENSION", "pgvector").lower()
if ext == "pgvectorscale":
return "USING diskann (embedding vector_cosine_ops) WITH (num_neighbors = 50)"
elif ext == "vchord":
return "USING vchordrq (embedding vector_l2_ops)"
else:
return "USING hnsw (embedding vector_cosine_ops)"
def upgrade() -> None:
target = _target_index_type()
if target is None:
# pgvector — indexes are already HNSW, nothing to fix
return
bind = op.get_bind()
schema_name = context.config.get_main_option("target_schema")
schema = _get_schema_prefix()
table_ref = f'"{schema_name}".memory_units' if schema_name else "memory_units"
banks_ref = f'"{schema_name}".banks' if schema_name else "banks"
using_clause = _vector_index_using_clause()
pg_schema = schema_name or "public"
rows = bind.execute(text(f"SELECT bank_id, internal_id FROM {banks_ref}")).fetchall() # noqa: S608
for row in rows:
bank_id = row[0]
internal_id = str(row[1]).replace("-", "")[:16]
escaped_bank_id = bank_id.replace("'", "''")
for ft, ft_short in _FACT_TYPES.items():
idx_name = f"idx_mu_emb_{ft_short}_{internal_id}"
# Check if this index exists and what type it is
idx_info = bind.execute(
text("SELECT indexdef FROM pg_indexes WHERE schemaname = :schema AND indexname = :idx"),
{"schema": pg_schema, "idx": idx_name},
).fetchone()
if idx_info is None:
# Index doesn't exist — create it with the correct type
bind.execute(
text(
f"CREATE INDEX IF NOT EXISTS {idx_name} "
f"ON {table_ref} {using_clause} "
f"WHERE fact_type = '{ft}' AND bank_id = '{escaped_bank_id}'"
)
)
continue
indexdef = idx_info[0].lower()
if target in indexdef:
# Already the correct type
continue
# Wrong type — drop and recreate
bind.execute(text(f"DROP INDEX IF EXISTS {schema}{idx_name}"))
bind.execute(
text(
f"CREATE INDEX IF NOT EXISTS {idx_name} "
f"ON {table_ref} {using_clause} "
f"WHERE fact_type = '{ft}' AND bank_id = '{escaped_bank_id}'"
)
)
def downgrade() -> None:
# Downgrade recreates indexes as HNSW (the original hardcoded behavior)
target = _target_index_type()
if target is None:
return
bind = op.get_bind()
schema_name = context.config.get_main_option("target_schema")
schema = _get_schema_prefix()
table_ref = f'"{schema_name}".memory_units' if schema_name else "memory_units"
banks_ref = f'"{schema_name}".banks' if schema_name else "banks"
rows = bind.execute(text(f"SELECT bank_id, internal_id FROM {banks_ref}")).fetchall() # noqa: S608
for row in rows:
bank_id = row[0]
internal_id = str(row[1]).replace("-", "")[:16]
escaped_bank_id = bank_id.replace("'", "''")
for ft, ft_short in _FACT_TYPES.items():
idx_name = f"idx_mu_emb_{ft_short}_{internal_id}"
bind.execute(text(f"DROP INDEX IF EXISTS {schema}{idx_name}"))
bind.execute(
text(
f"CREATE INDEX IF NOT EXISTS {idx_name} "
f"ON {table_ref} USING hnsw (embedding vector_cosine_ops) "
f"WHERE fact_type = '{ft}' AND bank_id = '{escaped_bank_id}'"
)
)
@@ -1,36 +0,0 @@
"""Make event_date nullable in memory_units to support timestamp-free content
Revision ID: aa2b3c4d5e6f
Revises: z1u2v3w4x5y6
Create Date: 2026-03-02
When callers retain content without a timestamp (e.g. fictional documents, static text),
the event_date column should be allowed to be NULL rather than defaulting to utcnow().
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "aa2b3c4d5e6f"
down_revision: str | Sequence[str] | None = "z1u2v3w4x5y6"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}memory_units ALTER COLUMN event_date DROP NOT NULL")
def downgrade() -> None:
schema = _get_schema_prefix()
# Backfill NULLs with now() before restoring the NOT NULL constraint
op.execute(f"UPDATE {schema}memory_units SET event_date = now() WHERE event_date IS NULL")
op.execute(f"ALTER TABLE {schema}memory_units ALTER COLUMN event_date SET NOT NULL")
@@ -1,32 +0,0 @@
"""add content_hash to chunks table for delta retain
Revision ID: b3c4d5e6f7a8
Revises: a3b4c5d6e7f8
Create Date: 2026-03-25
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "b3c4d5e6f7a8"
down_revision: str | Sequence[str] | None = "a3b4c5d6e7f8"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# Add content_hash column to chunks table for delta comparison
op.execute(f"ALTER TABLE {schema}chunks ADD COLUMN IF NOT EXISTS content_hash TEXT")
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}chunks DROP COLUMN IF EXISTS content_hash")
@@ -1,68 +0,0 @@
"""Add partial indexes on memory_units temporal date fields for fast temporal retrieval
Revision ID: b3c4d5e6f7g8
Revises: c1a2b3d4e5f6
Create Date: 2026-03-02
The temporal retrieval entry-point query filters memory_units by occurred_start,
occurred_end, and mentioned_at using OR conditions. Without dedicated indexes the
planner falls back to a sequential scan of all bank rows after applying the
(bank_id, fact_type) index, then re-checks each date field.
These three partial indexes give the planner bitmap-index scan options for the
three most common date predicates, dramatically reducing the row set before any
embedding computation is required.
All indexes are created CONCURRENTLY so the migration does not block writes on
memory_units during production deployments. CONCURRENTLY requires running outside
a transaction block; see migrations.py for how this is handled safely.
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "b3c4d5e6f7g8"
down_revision: str | Sequence[str] | None = "c1a2b3d4e5f6"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# Partial index on occurred_start (covers "occurred_start BETWEEN $4 AND $5")
op.execute("COMMIT")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_units_bank_occurred_start "
f"ON {schema}memory_units(bank_id, fact_type, occurred_start) "
f"WHERE occurred_start IS NOT NULL"
)
# Partial index on occurred_end (covers "occurred_end BETWEEN $4 AND $5")
op.execute("COMMIT")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_units_bank_occurred_end "
f"ON {schema}memory_units(bank_id, fact_type, occurred_end) "
f"WHERE occurred_end IS NOT NULL"
)
# Partial index on mentioned_at (covers "mentioned_at BETWEEN $4 AND $5")
op.execute("COMMIT")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_units_bank_mentioned_at "
f"ON {schema}memory_units(bank_id, fact_type, mentioned_at) "
f"WHERE mentioned_at IS NOT NULL"
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_units_bank_mentioned_at")
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_units_bank_occurred_end")
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_units_bank_occurred_start")
@@ -1,34 +0,0 @@
"""Backfill observation_scopes column if missing.
This migration ensures observation_scopes exists even on databases that had
revision z1u2v3w4x5y6 applied when it referred to the old text_signals migration
(before it was renamed to a2b3c4d5e6f7). The ADD COLUMN IF NOT EXISTS makes this
a no-op on databases that already have the column.
Revision ID: b4c5d6e7f8a9
Revises: a2b3c4d5e6f7
Create Date: 2026-03-02
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "b4c5d6e7f8a9"
down_revision: str | Sequence[str] | None = "a2b3c4d5e6f7"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}memory_units ADD COLUMN IF NOT EXISTS observation_scopes JSONB")
def downgrade() -> None:
pass # intentionally no-op — safe to leave the column in place
@@ -1,59 +0,0 @@
"""Enable pg_trgm extension and add GIN trigram index on entities.canonical_name
Revision ID: c1a2b3d4e5f6
Revises: b4c5d6e7f8a9
Create Date: 2026-03-02
Index is created CONCURRENTLY so the migration does not block writes on entities
during production deployments. CONCURRENTLY requires running outside a transaction
block; see migrations.py for how this is handled safely.
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import context, op
revision: str = "c1a2b3d4e5f6"
down_revision: str | Sequence[str] | None = "b4c5d6e7f8a9"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
# pg_trgm ships with most PostgreSQL installations as a contrib module.
# It enables fast similarity lookups via GIN indexes, used for entity name matching.
# On managed services (e.g. Azure Flexible Server), the extension may not be
# available or may require manual enablement. We gracefully skip the index
# creation if the extension cannot be loaded — the entity resolver will
# auto-detect and fall back to the "full" lookup strategy at runtime. See #626.
conn = op.get_bind()
try:
conn.execute(sa.text("CREATE EXTENSION IF NOT EXISTS pg_trgm"))
except Exception:
# Extension not available (managed Postgres, insufficient privileges, etc.)
# Roll back the failed statement and skip index creation.
conn.execute(sa.text("ROLLBACK"))
conn.execute(sa.text("BEGIN"))
return
schema = _get_schema_prefix()
# GIN index on canonical_name enables sub-millisecond trigram similarity queries
# (% operator, similarity()) instead of full-table scans across all bank entities.
op.execute("COMMIT")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS entities_canonical_name_trgm_idx "
f"ON {schema}entities USING GIN (canonical_name gin_trgm_ops)"
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}entities_canonical_name_trgm_idx")
# Note: not dropping pg_trgm extension as other indexes may depend on it
@@ -1,61 +0,0 @@
"""Add audit_log table for feature usage tracking.
Merge migration that combines the two existing heads (a3b4c5d6e7f8 + c8e5f2a3b4d1).
Stores raw request/response as JSONB for expandability without future migrations.
The metadata JSONB column allows adding arbitrary fields in the future.
Revision ID: c2d3e4f5g6h7
Revises: a3b4c5d6e7f8, c8e5f2a3b4d1
Create Date: 2026-03-26
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "c2d3e4f5g6h7"
down_revision: str | Sequence[str] | None = ("a3b4c5d6e7f8", "c8e5f2a3b4d1")
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(
f"""
CREATE TABLE IF NOT EXISTS {schema}audit_log (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
action TEXT NOT NULL,
transport TEXT NOT NULL,
bank_id TEXT,
started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
ended_at TIMESTAMPTZ,
request JSONB,
response JSONB,
metadata JSONB DEFAULT '{{}}'::jsonb
)
"""
)
op.execute(
f"CREATE INDEX IF NOT EXISTS idx_audit_log_action_started ON {schema}audit_log (action, started_at DESC)"
)
op.execute(f"CREATE INDEX IF NOT EXISTS idx_audit_log_bank_started ON {schema}audit_log (bank_id, started_at DESC)")
op.execute(f"CREATE INDEX IF NOT EXISTS idx_audit_log_started ON {schema}audit_log (started_at DESC)")
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"DROP INDEX IF EXISTS {schema}idx_audit_log_started")
op.execute(f"DROP INDEX IF EXISTS {schema}idx_audit_log_bank_started")
op.execute(f"DROP INDEX IF EXISTS {schema}idx_audit_log_action_started")
op.execute(f"DROP TABLE IF EXISTS {schema}audit_log")
@@ -1,30 +0,0 @@
"""Add history column to mental_models
Revision ID: c3d4e5f6g7h8
Revises: a2b3c4d5e6f7, a2b3c4d5e6f8
Create Date: 2026-03-06
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "c3d4e5f6g7h8"
down_revision: str | Sequence[str] | None = ("a2b3c4d5e6f7", "a2b3c4d5e6f8")
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}mental_models ADD COLUMN IF NOT EXISTS history JSONB DEFAULT '[]'::jsonb")
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}mental_models DROP COLUMN IF EXISTS history")
@@ -1,48 +0,0 @@
"""Add bank_id column to memory_links for direct filtering
The stats endpoint JOINs memory_links to memory_units just to filter by
bank_id. With millions of links this takes 18+ seconds. Adding bank_id
directly to memory_links lets Postgres push the filter down before the JOIN.
Revision ID: c5d6e7f8a9b0
Revises: b3c4d5e6f7a8
Create Date: 2026-03-26
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "c5d6e7f8a9b0"
down_revision: str | Sequence[str] | None = "b3c4d5e6f7a8"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# 1. Add nullable column
op.execute(f"ALTER TABLE {schema}memory_links ADD COLUMN IF NOT EXISTS bank_id TEXT")
# 2. Backfill from memory_units
op.execute(f"""
UPDATE {schema}memory_links ml
SET bank_id = mu.bank_id
FROM {schema}memory_units mu
WHERE ml.from_unit_id = mu.id
AND ml.bank_id IS NULL
""")
# 3. Set NOT NULL
op.execute(f"ALTER TABLE {schema}memory_links ALTER COLUMN bank_id SET NOT NULL")
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}memory_links DROP COLUMN IF EXISTS bank_id")
@@ -1,83 +0,0 @@
"""Add covering and composite indexes to speed up link expansion graph retrieval.
Two indexes target the two bottlenecks identified by EXPLAIN ANALYZE on a 17M-row
memory_links table:
1. idx_memory_links_to_type_weight (to_unit_id, link_type, weight DESC)
The semantic incoming direction — finding facts that consider seeds as their
nearest neighbour — currently hits an expensive BitmapAnd of two separate
bitmap scans (to_unit_id bitmap ∩ link_type bitmap). A composite index
on (to_unit_id, link_type) turns this into a single index scan and reduces
latency from ~36 ms to < 5 ms per query.
2. idx_memory_links_entity_covering (from_unit_id) INCLUDE (to_unit_id, entity_id)
WHERE link_type = 'entity'
The entity co-occurrence expansion uses COUNT(DISTINCT ml.entity_id) and
joins on ml.to_unit_id. Without a covering index the planner must read
~2 500 heap pages to fetch entity_id and to_unit_id after the bitmap index
scan, adding ~230 ms of random I/O. INCLUDE adds those two columns to the
index leaf pages so the entire query can be served from the index (index-only
scan), eliminating the heap reads entirely.
Partial index (WHERE link_type = 'entity') keeps index size ~40 % smaller.
Both indexes are created with CONCURRENTLY so the migration does not block
concurrent reads or writes on memory_links. CONCURRENTLY requires running
outside a transaction block, so the migration emits an explicit COMMIT before
each statement and uses IF NOT EXISTS for idempotency.
Revision ID: d2e3f4a5b6c7
Revises: b3c4d5e6f7g8
Create Date: 2026-03-02
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "d2e3f4a5b6c7"
down_revision: str | Sequence[str] | None = "b3c4d5e6f7g8"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# CREATE INDEX CONCURRENTLY cannot run inside a transaction block.
# Commit the current Alembic transaction, then issue each CONCURRENTLY
# statement in its own implicit autocommit transaction.
# IF NOT EXISTS makes each statement idempotent if the migration is retried.
# Index for the semantic *incoming* direction in link_expansion_retrieval.py.
# Replaces the BitmapAnd of idx_memory_links_to_unit ∩ idx_memory_links_link_type
# with a single composite index scan.
op.execute("COMMIT")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_links_to_type_weight "
f"ON {schema}memory_links(to_unit_id, link_type, weight DESC)"
)
# Covering index for entity co-occurrence expansion.
# Enables an index-only scan: entity_id and to_unit_id are read from the
# index leaf pages instead of the heap, eliminating ~2 500 random heap-page
# reads per expansion query.
op.execute("COMMIT")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_links_entity_covering "
f"ON {schema}memory_links(from_unit_id) "
f"INCLUDE (to_unit_id, entity_id) "
f"WHERE link_type = 'entity'"
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_links_entity_covering")
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_links_to_type_weight")
@@ -1,53 +0,0 @@
"""Recreate idx_memory_units_source_memory_ids GIN index with fastupdate=off
GIN indexes use a "fastupdate" pending list by default: small writes are
buffered there and flushed to the main GIN tree in bulk. Flushing requires
AccessExclusiveLock on the index. Under high insert concurrency (e.g. 8
parallel pytest-xdist workers all calling retain_async) two transactions can
each trigger a flush simultaneously and deadlock.
Disabling fastupdate makes every insert write directly to the GIN tree
(slightly slower per insert, but no pending-list lock cycles).
Revision ID: d4e5f6g7h8i9
Revises: d5e6f7a8b9c0
Create Date: 2026-03-11
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "d4e5f6g7h8i9"
down_revision: str | Sequence[str] | None = "d5e6f7a8b9c0"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# DROP + CREATE CONCURRENTLY must run outside a transaction block.
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_units_source_memory_ids")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_units_source_memory_ids "
f"ON {schema}memory_units USING GIN (source_memory_ids) "
f"WITH (fastupdate=off) "
f"WHERE source_memory_ids IS NOT NULL"
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute("COMMIT")
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_memory_units_source_memory_ids")
op.execute(
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_memory_units_source_memory_ids "
f"ON {schema}memory_units USING GIN (source_memory_ids) "
f"WHERE source_memory_ids IS NOT NULL"
)
@@ -1,139 +0,0 @@
"""Add internal_id to banks and per-(bank, fact_type) partial vector indexes
Revision ID: d5e6f7a8b9c0
Revises: a3b4c5d6e7f8
Create Date: 2026-03-11
This migration:
1. Adds internal_id UUID column to banks (stable identifier for index naming)
2. Drops the global vector index (competes with per-bank partial indexes)
3. Creates per-(bank_id, fact_type) partial vector indexes for all existing banks
using the configured vector extension (HNSW for pgvector, DiskANN for
pgvectorscale, vchordrq for vchord).
(new banks get indexes created at bank-creation time via bank_utils.create_bank_vector_indexes)
Why per-(bank, fact_type) indexes:
- fact_type-only partial indexes are never chosen by the planner when bank_id is in the WHERE
clause, because the idx_memory_units_bank_id B-tree index always wins at planning time.
- Per-(bank, fact_type) partial indexes have both predicates matching → planner selects them.
- The global vector index competes for larger partitions (world, observation) and must be dropped.
"""
import os
from collections.abc import Sequence
from alembic import context, op
from sqlalchemy import text
revision: str = "d5e6f7a8b9c0"
down_revision: str | Sequence[str] | None = "c3d4e5f6g7h8"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_FACT_TYPES: dict[str, str] = {
"world": "worl",
"experience": "expr",
"observation": "obsv",
}
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def _vector_index_using_clause() -> str:
"""Return the USING clause based on the configured vector extension."""
ext = os.getenv("HINDSIGHT_API_VECTOR_EXTENSION", "pgvector").lower()
if ext == "pgvectorscale":
return "USING diskann (embedding vector_cosine_ops) WITH (num_neighbors = 50)"
elif ext == "vchord":
return "USING vchordrq (embedding vector_l2_ops)"
else:
return "USING hnsw (embedding vector_cosine_ops)"
def upgrade() -> None:
schema = _get_schema_prefix()
# 1. Add internal_id column to banks
op.execute(
f"ALTER TABLE {schema}banks ADD COLUMN IF NOT EXISTS internal_id UUID DEFAULT gen_random_uuid() NOT NULL"
)
op.execute(f"ALTER TABLE {schema}banks ADD CONSTRAINT banks_internal_id_unique UNIQUE (internal_id)")
# 2. Drop any fact_type-only partial indexes that may exist from prior migrations
# (bank_id B-tree always wins over them when bank_id is in the WHERE clause)
op.execute(f"DROP INDEX IF EXISTS {schema}idx_mu_emb_world")
op.execute(f"DROP INDEX IF EXISTS {schema}idx_mu_emb_observation")
op.execute(f"DROP INDEX IF EXISTS {schema}idx_mu_emb_experience")
# 4. Drop global vector index (competes with per-bank partial indexes)
op.execute(f"DROP INDEX IF EXISTS {schema}idx_memory_units_embedding")
# 5. Create per-(bank, fact_type) partial vector indexes for all existing banks
# using the configured extension (HNSW / DiskANN / vchordrq)
bind = op.get_bind()
schema_name = context.config.get_main_option("target_schema")
table_ref = f'"{schema_name}".memory_units' if schema_name else "memory_units"
banks_ref = f'"{schema_name}".banks' if schema_name else "banks"
using_clause = _vector_index_using_clause()
rows = bind.execute(text(f"SELECT bank_id, internal_id FROM {banks_ref}")).fetchall() # noqa: S608
for row in rows:
bank_id = row[0]
internal_id = str(row[1]).replace("-", "")[:16]
escaped_bank_id = bank_id.replace("'", "''")
for ft, ft_short in _FACT_TYPES.items():
idx_name = f"idx_mu_emb_{ft_short}_{internal_id}"
# Index name is schema-unqualified (indexes live in the schema of their table)
bind.execute(
text(
f"CREATE INDEX IF NOT EXISTS {idx_name} "
f"ON {table_ref} {using_clause} "
f"WHERE fact_type = '{ft}' AND bank_id = '{escaped_bank_id}'"
)
)
def downgrade() -> None:
schema = _get_schema_prefix()
# Drop per-bank HNSW indexes (iterate existing banks)
bind = op.get_bind()
schema_name = context.config.get_main_option("target_schema")
banks_ref = f'"{schema_name}".banks' if schema_name else "banks"
rows = bind.execute(text(f"SELECT internal_id FROM {banks_ref}")).fetchall() # noqa: S608
for row in rows:
internal_id = str(row[0]).replace("-", "")[:16]
for ft_short in _HNSW_FACT_TYPES.values():
idx_name = f"idx_mu_emb_{ft_short}_{internal_id}"
bind.execute(text(f"DROP INDEX IF EXISTS {schema}{idx_name}"))
# Restore the global HNSW index
table_ref = f'"{schema_name}".memory_units' if schema_name else "memory_units"
op.execute(
f"CREATE INDEX IF NOT EXISTS idx_memory_units_embedding ON {table_ref} USING hnsw (embedding vector_cosine_ops)"
)
# Restore old fact_type-only partial indexes
op.execute(
f"CREATE INDEX IF NOT EXISTS idx_mu_emb_world "
f"ON {table_ref} USING hnsw (embedding vector_cosine_ops) "
f"WHERE fact_type = 'world'"
)
op.execute(
f"CREATE INDEX IF NOT EXISTS idx_mu_emb_observation "
f"ON {table_ref} USING hnsw (embedding vector_cosine_ops) "
f"WHERE fact_type = 'observation'"
)
op.execute(
f"CREATE INDEX IF NOT EXISTS idx_mu_emb_experience "
f"ON {table_ref} USING hnsw (embedding vector_cosine_ops) "
f"WHERE fact_type = 'experience'"
)
# Drop internal_id column
op.execute(f"ALTER TABLE {schema}banks DROP CONSTRAINT IF EXISTS banks_internal_id_unique")
op.execute(f"ALTER TABLE {schema}banks DROP COLUMN IF EXISTS internal_id")
@@ -1,62 +0,0 @@
"""Add webhooks table and next_retry_at to async_operations.
Webhook deliveries are handled as async_operations tasks (operation_type='webhook_delivery')
rather than a dedicated webhook_deliveries table.
Revision ID: e4f5a6b7c8d9
Revises: d2e3f4a5b6c7
Create Date: 2026-03-04
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "e4f5a6b7c8d9"
down_revision: str | Sequence[str] | None = "d2e3f4a5b6c7"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(
f"""
CREATE TABLE IF NOT EXISTS {schema}webhooks (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
bank_id TEXT,
url TEXT NOT NULL,
secret TEXT,
event_types TEXT[] NOT NULL DEFAULT '{{}}',
enabled BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
)
"""
)
# Index for bank-scoped webhook lookup
op.execute(f"CREATE INDEX IF NOT EXISTS idx_webhooks_bank_id ON {schema}webhooks(bank_id)")
# Add next_retry_at to async_operations for task-owned retry scheduling
op.execute(f"ALTER TABLE {schema}async_operations ADD COLUMN IF NOT EXISTS next_retry_at TIMESTAMPTZ NULL")
# Index for polling: status + next_retry_at
op.execute(
f"CREATE INDEX IF NOT EXISTS idx_async_operations_status_retry "
f"ON {schema}async_operations(status, next_retry_at)"
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"DROP INDEX IF EXISTS {schema}idx_async_operations_status_retry")
op.execute(f"ALTER TABLE {schema}async_operations DROP COLUMN IF EXISTS next_retry_at")
op.execute(f"DROP INDEX IF EXISTS {schema}idx_webhooks_bank_id")
op.execute(f"DROP TABLE IF EXISTS {schema}webhooks")
@@ -1,73 +0,0 @@
"""Add CASCADE DELETE FK from async_operations and webhooks to banks.
When a bank is deleted, all its async_operations and webhooks rows are
automatically deleted by the database. This ensures that any in-flight
worker tasks detect the deletion via _check_op_alive() and abort early.
Revision ID: e5f6g7h8i9j0
Revises: d4e5f6g7h8i9
Create Date: 2026-03-11
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "e5f6g7h8i9j0"
down_revision: str | Sequence[str] | None = "d4e5f6g7h8i9"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# Remove orphaned async_operations rows whose bank no longer exists
# (can happen because there was no FK before this migration).
op.execute(
f"""
DELETE FROM {schema}async_operations
WHERE bank_id IS NOT NULL
AND bank_id NOT IN (SELECT bank_id FROM {schema}banks)
"""
)
# Remove orphaned webhooks rows whose bank no longer exists.
op.execute(
f"""
DELETE FROM {schema}webhooks
WHERE bank_id IS NOT NULL
AND bank_id NOT IN (SELECT bank_id FROM {schema}banks)
"""
)
# Add FK with ON DELETE CASCADE so that deleting a bank automatically
# cleans up all its pending/processing operations and webhook configs.
op.execute(
f"""
ALTER TABLE {schema}async_operations
ADD CONSTRAINT fk_async_operations_bank_id
FOREIGN KEY (bank_id) REFERENCES {schema}banks(bank_id)
ON DELETE CASCADE
"""
)
op.execute(
f"""
ALTER TABLE {schema}webhooks
ADD CONSTRAINT fk_webhooks_bank_id
FOREIGN KEY (bank_id) REFERENCES {schema}banks(bank_id)
ON DELETE CASCADE
"""
)
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}async_operations DROP CONSTRAINT IF EXISTS fk_async_operations_bank_id")
op.execute(f"ALTER TABLE {schema}webhooks DROP CONSTRAINT IF EXISTS fk_webhooks_bank_id")
@@ -1,57 +0,0 @@
"""chunk_fk_cascade_delete
Revision ID: f6g7h8i9j0k1
Revises: e5f6g7h8i9j0
Create Date: 2026-03-16 00:00:00.000000
"""
from collections.abc import Sequence
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "f6g7h8i9j0k1"
down_revision: str | Sequence[str] | None = "e5f6g7h8i9j0"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Change memory_units.chunk_id FK from SET NULL to CASCADE.
When a document is deleted the CASCADE reaches chunks first; with SET NULL
the memory_units rows survived with chunk_id = NULL, leaving ghost records.
Switching to CASCADE ensures they are removed together with their chunk.
"""
from alembic import context
schema = context.config.get_main_option("target_schema")
schema_prefix = f'"{schema}".' if schema else ""
# Use raw SQL with IF EXISTS so this is safe on schemas where the FK was
# already dropped or never existed under this name.
op.execute(f"ALTER TABLE {schema_prefix}memory_units DROP CONSTRAINT IF EXISTS memory_units_chunk_fkey")
# Use a DO block so the ADD is also idempotent: if the FK already exists (e.g.
# the schema was provisioned after the base migration already added it) the
# duplicate_object exception is swallowed rather than failing the migration.
op.execute(
f"""
DO $$ BEGIN
ALTER TABLE {schema_prefix}memory_units
ADD CONSTRAINT memory_units_chunk_fkey
FOREIGN KEY (chunk_id)
REFERENCES {schema_prefix}chunks (chunk_id)
ON DELETE CASCADE;
EXCEPTION
WHEN duplicate_object THEN NULL;
END $$;
"""
)
def downgrade() -> None:
"""Revert to SET NULL behaviour."""
op.drop_constraint("memory_units_chunk_fkey", "memory_units", type_="foreignkey")
op.create_foreign_key(
"memory_units_chunk_fkey", "memory_units", "chunks", ["chunk_id"], ["chunk_id"], ondelete="SET NULL"
)
@@ -1,33 +0,0 @@
"""Add http_config JSONB column to webhooks table.
Stores HTTP delivery configuration (method, timeout, headers, params) as a
single JSONB column rather than separate columns.
Revision ID: f7g8h9i0j1k2
Revises: e4f5a6b7c8d9
Create Date: 2026-03-04
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "f7g8h9i0j1k2"
down_revision: str | Sequence[str] | None = "e4f5a6b7c8d9"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}webhooks ADD COLUMN IF NOT EXISTS http_config JSONB NOT NULL DEFAULT '{{}}'")
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}webhooks DROP COLUMN IF EXISTS http_config")
@@ -1,83 +0,0 @@
"""remove_opinion_fact_type
Revision ID: g2h3i4j5k6l7
Revises: f1a2b3c4d5e6
Create Date: 2026-04-02
Remove the deprecated 'opinion' fact type: drop opinion-specific indexes,
update CHECK constraints, delete any remaining opinion rows, and drop the
confidence_score column (was only used for opinions, always NULL otherwise).
"""
from collections.abc import Sequence
from alembic import context, op
# revision identifiers, used by Alembic.
revision: str = "g2h3i4j5k6l7"
down_revision: str | Sequence[str] | None = "f1a2b3c4d5e6"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# 1. Delete any remaining opinion rows
op.execute(f"DELETE FROM {schema}memory_units WHERE fact_type = 'opinion'")
# 2. Drop opinion-specific indexes
op.execute(f"DROP INDEX IF EXISTS {schema}idx_memory_units_opinion_confidence")
op.execute(f"DROP INDEX IF EXISTS {schema}idx_memory_units_opinion_date")
# 3. Drop confidence_score constraints and column (only used for opinions, always NULL otherwise)
op.execute(f"ALTER TABLE {schema}memory_units DROP CONSTRAINT IF EXISTS confidence_score_fact_type_check")
op.execute(f"ALTER TABLE {schema}memory_units DROP CONSTRAINT IF EXISTS memory_units_confidence_score_check")
op.execute(f"ALTER TABLE {schema}memory_units DROP COLUMN IF EXISTS confidence_score")
# 4. Replace fact_type CHECK constraint
op.execute(f"ALTER TABLE {schema}memory_units DROP CONSTRAINT IF EXISTS memory_units_fact_type_check")
op.execute(
f"ALTER TABLE {schema}memory_units ADD CONSTRAINT memory_units_fact_type_check "
f"CHECK (fact_type IN ('world', 'experience', 'observation'))"
)
def downgrade() -> None:
schema = _get_schema_prefix()
# Restore confidence_score column
op.execute(f"ALTER TABLE {schema}memory_units ADD COLUMN IF NOT EXISTS confidence_score float")
op.execute(
f"ALTER TABLE {schema}memory_units ADD CONSTRAINT memory_units_confidence_score_check "
f"CHECK (confidence_score IS NULL OR (confidence_score >= 0.0 AND confidence_score <= 1.0))"
)
op.execute(
f"ALTER TABLE {schema}memory_units ADD CONSTRAINT confidence_score_fact_type_check "
f"CHECK ((fact_type = 'opinion' AND confidence_score IS NOT NULL) OR "
f"(fact_type = 'observation') OR "
f"(fact_type NOT IN ('opinion', 'observation') AND confidence_score IS NULL))"
)
# Restore original fact_type CHECK constraint (with opinion)
op.execute(f"ALTER TABLE {schema}memory_units DROP CONSTRAINT IF EXISTS memory_units_fact_type_check")
op.execute(
f"ALTER TABLE {schema}memory_units ADD CONSTRAINT memory_units_fact_type_check "
f"CHECK (fact_type IN ('world', 'experience', 'opinion', 'observation'))"
)
# Recreate opinion indexes
op.execute(
f"CREATE INDEX idx_memory_units_opinion_confidence ON {schema}memory_units "
f"(bank_id, confidence_score DESC) WHERE fact_type = 'opinion'"
)
op.execute(
f"CREATE INDEX idx_memory_units_opinion_date ON {schema}memory_units "
f"(bank_id, event_date DESC) WHERE fact_type = 'opinion'"
)
@@ -1,71 +0,0 @@
"""backsweep_orphan_memory_units
Two-pass cleanup of memory_units rows that were never removed by earlier bugs:
Pass 1 — any fact_type, bank gone:
memory_units whose bank_id no longer exists in banks. These accumulate when
a bank is deleted without a proper cascade (no FK from memory_units to banks
exists in the schema).
Pass 2 — observations only, all sources gone:
observation rows whose bank still exists but every source_memory_id points
to a deleted memory unit. These were left behind before PR #580 fixed the
chunk FK cascade and before delete_document() called
_delete_stale_observations_for_memories.
Revision ID: g7h8i9j0k1l2
Revises: f6g7h8i9j0k1
Create Date: 2026-03-16
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "g7h8i9j0k1l2"
down_revision: str | Sequence[str] | None = "f6g7h8i9j0k1"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
mu = f"{schema}memory_units"
banks = f"{schema}banks"
# Pass 1: delete all memory_units (any fact_type) whose bank no longer exists.
# There is no FK from memory_units to banks, so these never cascade away.
op.execute(
f"""
DELETE FROM {mu}
WHERE NOT EXISTS (
SELECT 1 FROM {banks} b WHERE b.bank_id = {mu}.bank_id
)
"""
)
# Pass 2: delete orphaned observations whose bank still exists but every
# source_memory_id refers to a now-deleted memory unit (or the array is
# empty). Observations with at least one surviving source are left alone.
op.execute(
f"""
DELETE FROM {mu} orphan
WHERE orphan.fact_type = 'observation'
AND NOT EXISTS (
SELECT 1
FROM {mu} src
WHERE src.id = ANY(orphan.source_memory_ids)
AND src.bank_id = orphan.bank_id
)
"""
)
def downgrade() -> None:
# Deleted rows cannot be restored.
pass
@@ -1,42 +0,0 @@
"""Merge 3 migration heads and add unit_entities composite index
Revision ID: h3i4j5k6l7m8
Revises: a4b5c6d7e8f9, c2d3e4f5g6h7, g2h3i4j5k6l7
Create Date: 2026-04-07
Merges three unmerged migration heads into one, and adds a composite index
(entity_id, unit_id) on unit_entities for index-only scans in the LATERAL
entity expansion query.
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "h3i4j5k6l7m8"
down_revision: str | Sequence[str] | None = ("a4b5c6d7e8f9", "c2d3e4f5g6h7", "g2h3i4j5k6l7")
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
# Composite index enables index-only scans for entity_id -> unit_id lookups
op.execute(
f"CREATE INDEX IF NOT EXISTS idx_unit_entities_entity_unit ON {schema}unit_entities (entity_id, unit_id)"
)
# Drop the now-redundant single-column index
op.execute(f"DROP INDEX IF EXISTS {schema}idx_unit_entities_entity")
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"DROP INDEX IF EXISTS {schema}idx_unit_entities_entity_unit")
# Restore the single-column index
op.execute(f"CREATE INDEX IF NOT EXISTS idx_unit_entities_entity ON {schema}unit_entities (entity_id)")
@@ -1,35 +0,0 @@
"""Add observation_scopes column to memory_units table
Revision ID: z1u2v3w4x5y6
Revises: a1b2c3d4e5f6
Create Date: 2026-02-25
Adds observation_scopes JSONB column to memory_units to control how observations
are scoped during consolidation. Accepts "per_tag", "combined", or an explicit
list of tag-set lists for custom multi-pass consolidation.
"""
from collections.abc import Sequence
from alembic import context, op
revision: str = "z1u2v3w4x5y6"
down_revision: str | Sequence[str] | None = "a1b2c3d4e5f6"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _get_schema_prefix() -> str:
"""Get schema prefix for table names (required for multi-tenant support)."""
schema = context.config.get_main_option("target_schema")
return f'"{schema}".' if schema else ""
def upgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}memory_units ADD COLUMN IF NOT EXISTS observation_scopes JSONB")
def downgrade() -> None:
schema = _get_schema_prefix()
op.execute(f"ALTER TABLE {schema}memory_units DROP COLUMN IF EXISTS observation_scopes")
@@ -1,209 +0,0 @@
"""Audit logging for feature usage tracking.
Provides fire-and-forget audit logging of all mutating and core operations
(retain, recall, reflect, bank CRUD, etc.) across HTTP, MCP, and system transports.
"""
from __future__ import annotations
import asyncio
import json
import logging
import uuid
from collections.abc import Callable
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
import asyncpg
from ..engine.db_utils import acquire_with_retry
logger = logging.getLogger(__name__)
@dataclass
class AuditEntry:
"""A single audit log entry."""
action: str
transport: str # "http", "mcp", "system"
bank_id: str | None = None
started_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
ended_at: datetime | None = None
request: dict[str, Any] | None = None
response: dict[str, Any] | None = None
metadata: dict[str, Any] = field(default_factory=dict)
def _json_default(obj: Any) -> str:
"""JSON serializer for objects not serializable by default."""
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, uuid.UUID):
return str(obj)
if isinstance(obj, bytes):
return "<bytes>"
if isinstance(obj, set):
return list(obj)
return str(obj)
def _safe_json(data: Any) -> str | None:
"""Serialize data to JSON string, returning None on failure."""
if data is None:
return None
try:
return json.dumps(data, default=_json_default)
except Exception:
logger.debug("Failed to serialize audit data", exc_info=True)
return None
_SWEEP_INTERVAL_SECONDS = 3600 # Run retention sweep every hour
class AuditLogger:
"""Fire-and-forget audit log writer with optional retention sweep."""
def __init__(
self,
pool_getter: Callable[[], asyncpg.Pool | None],
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."""
if not self._enabled:
return False
if self._allowed_actions is not None:
return action in self._allowed_actions
return True
def log_fire_and_forget(self, entry: AuditEntry) -> None:
"""Schedule an audit write as a background task."""
if not self.is_enabled(entry.action):
return
try:
asyncio.create_task(self._safe_log(entry))
except RuntimeError:
# No running event loop (e.g. during shutdown)
logger.debug("Cannot schedule audit log write: no running event loop")
async def _safe_log(self, entry: AuditEntry) -> None:
"""Write audit entry to DB. Errors are logged, never raised."""
pool = self._pool_getter()
if pool is None:
logger.debug("Audit log skipped: pool not available")
return
try:
schema = self._schema_getter()
table = f"{schema}.audit_log"
async with acquire_with_retry(pool, max_retries=1) as conn:
await conn.execute(
f"""
INSERT INTO {table}
(id, action, transport, bank_id, started_at, ended_at, request, response, metadata)
VALUES
($1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9::jsonb)
""",
uuid.uuid4(),
entry.action,
entry.transport,
entry.bank_id,
entry.started_at,
entry.ended_at,
_safe_json(entry.request),
_safe_json(entry.response),
_safe_json(entry.metadata) or "{}",
)
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(
audit_logger: AuditLogger | None,
action: str,
transport: str,
bank_id: str | None = None,
request: dict[str, Any] | None = None,
metadata: dict[str, Any] | None = None,
):
"""Async context manager that times the operation and writes audit on exit.
Usage:
async with audit_context(logger, "retain", "http", bank_id, request_dict) as entry:
result = await do_work()
entry.response = result_dict
"""
if audit_logger is None or not audit_logger.is_enabled(action):
entry = AuditEntry(action=action, transport=transport, bank_id=bank_id)
yield entry
return
entry = AuditEntry(
action=action,
transport=transport,
bank_id=bank_id,
started_at=datetime.now(timezone.utc),
request=request,
metadata=metadata or {},
)
try:
yield entry
finally:
entry.ended_at = datetime.now(timezone.utc)
audit_logger.log_fire_and_forget(entry)
File diff suppressed because it is too large Load Diff
@@ -1,104 +0,0 @@
"""Prompts for the consolidation engine."""
# Default mission when no bank-specific mission is set
_DEFAULT_MISSION = "Track every detail: names, numbers, dates, places, and relationships. Prefer specifics over abstractions, never generalise."
# Processing rules — always present regardless of mission
_PROCESSING_RULES = """Processing rules (always apply):
1. ONE OBSERVATION PER DISTINCT FACET: each observation tracks exactly one specific facet — a count ("has 3 items"), a named entity ("has a dog named Rex"), a relationship ("works at Google"), etc. Never merge different facets into one observation.
2. MATCH BY ENTITY/FACET, NOT TOPIC: when deciding whether to UPDATE vs CREATE, match on the specific entity or facet. "Sold item X" updates only the X observation. "Now has 5 items" updates only the count observation. Do not update observations about different entities just because they share a general topic.
3. STATE CHANGES — UPDATE CONCISELY: when a fact changes the state of something ("sold X", "X died", "moved to Y"), UPDATE the matching observation to reflect the current state. Include dates when available. Keep it concise — only information about THAT specific facet. Example: "User owned a dog named Rex who died on March 15, 2025". Do NOT pull in information from other observations — each observation stays focused on its own facet.
4. CASCADE TO ALL AFFECTED OBSERVATIONS: a state change may affect multiple observations. For example, if entity C is removed from a group, update BOTH the individual observation for C AND any list/group observation that includes C (remove C from the list while keeping all other members intact).
5. NO COMPUTATION: you do not have the full picture — never calculate, derive, or adjust numeric values. If the user says "I have 2 dogs" and then "I have a dog named Rex", do NOT update the count to 3 — you don't know if Rex is one of the 2 or a new one. If the user says "I sold X", do NOT decrement a count. Only update a count when the user explicitly states a new count. Synthesize and consolidate what was stated, but never do arithmetic or logical deductions.
6. SAME FACET → UPDATE, NOT CREATE: a new count supersedes the old count — UPDATE the existing count observation, don't create a second one. If there's an existing observation for the same specific facet, always UPDATE it rather than creating a duplicate.
7. PRESERVE HISTORY: observations that record significant events (sold, died, moved, changed) are important history — never DELETE them. Only delete an observation when it is restated identically or truly meaningless. Be very conservative with deletes.
8. RESOLVE REFERENCES: when a new fact provides a concrete value for a vague placeholder in an existing observation (e.g., "home country""Sweden"), UPDATE to embed the resolved value.
9. NEVER merge observations about different people or unrelated topics."""
# Data section — format placeholders {facts_text} and {observations_text} are substituted at call time
_BATCH_DATA_SECTION = """
NEW FACTS:
{facts_text}
EXISTING OBSERVATIONS (JSON array, pooled from recalls across all facts above):
{observations_text}
Each observation includes:
- id: unique identifier for updating
- text: the observation content
- proof_count: number of supporting memories
- occurred_start/occurred_end: temporal range of source facts
- source_memories: array of supporting facts with their text and dates
Compare the facts against existing observations:
- Same facet as an existing observation → UPDATE it (observation_id + source_fact_ids)
- New facet with durable knowledge → CREATE a new observation (source_fact_ids)
- Cross-reference facts within the batch: a later fact may resolve a vague reference in an earlier one
- Purely ephemeral facts → omit them unless the MISSION above explicitly targets such data (e.g. timestamped events, session state, screen content)"""
# Output format — JSON braces escaped as {{ }} so .format() leaves them literal
_BATCH_OUTPUT_FORMAT = """
Output a JSON object with three arrays.
## EXAMPLE
Input facts:
[a1b2c3d4-e5f6-7890-abcd-ef1234567890] Alice mentioned she works long hours, often past midnight | Involving: Alice (occurred_start=2024-01-15, mentioned_at=2024-01-15)
[b2c3d4e5-f6a7-8901-bcde-f12345678901] Alice said she's exhausted from the project deadlines | Involving: Alice (occurred_start=2024-01-20, mentioned_at=2024-01-20)
Good observation text — clean prose, no metadata, each fact tracked distinctly:
"Alice works long hours, often past midnight."
"Alice feels exhausted from project deadlines."
Bad observation text — NEVER do this (verbatim copy of fact text with metadata):
"Alice mentioned she works long hours, often past midnight | Involving: Alice (occurred_start=2024-01-15, mentioned_at=2024-01-15)"
Observation text rules:
- Write clean prose — NEVER copy raw fact lines or their metadata (temporal fields, "Involving:", "When:" labels, UUIDs).
- Parenthesized metadata like (occurred_start=...) and pipe-separated labels like "| Involving: ..." are fact formatting — strip them entirely from observation text.
- How many observations to create and how much to aggregate is driven by the MISSION above.
{{"creates": [{{"text": "Alice works long hours, often past midnight.", "source_fact_ids": ["a1b2c3d4-e5f6-7890-abcd-ef1234567890"]}}, {{"text": "Alice feels exhausted from project deadlines.", "source_fact_ids": ["b2c3d4e5-f6a7-8901-bcde-f12345678901"]}}],
"updates": [{{"text": "Alice works at Acme Corp as a senior engineer", "observation_id": "c3d4e5f6-a7b8-9012-cdef-123456789012", "source_fact_ids": ["d4e5f6a7-b8c9-0123-defa-234567890123"]}}],
"deletes": [{{"observation_id": "e5f6a7b8-c9d0-1234-efab-345678901234"}}]}}
Rules:
- "source_fact_ids": copy the EXACT UUID strings shown in brackets [uuid] from NEW FACTS — never use integers or positions.
- "observation_id": copy the EXACT "id" UUID string from EXISTING OBSERVATIONS.
- One create/update may reference multiple facts when they jointly support the observation.
- "deletes": only when an observation is directly superseded or contradicted by new facts.
- Do NOT include "tags" — handled automatically.
- Return {{"creates": [], "updates": [], "deletes": []}} if nothing durable is found."""
def build_batch_consolidation_prompt(
observations_mission: str | None = None,
observation_capacity_note: str | None = None,
) -> str:
"""
Build the consolidation prompt for batch mode (multiple facts per LLM call).
The mission defines *what* to track (customisable per bank).
Processing rules and output format are always present regardless of mission.
"""
mission = observations_mission or _DEFAULT_MISSION
capacity_section = ""
if observation_capacity_note:
capacity_section = f"\n\n## CAPACITY CONSTRAINT\n{observation_capacity_note}"
return (
"You are a memory consolidation system. Synthesize facts into observations "
"and merge with existing observations when appropriate.\n\n"
f"## MISSION\n{mission}{capacity_section}\n\n"
f"{_PROCESSING_RULES}" + _BATCH_DATA_SECTION + _BATCH_OUTPUT_FORMAT
)
@@ -1,144 +0,0 @@
"""
MLX implementation of jina-reranker-v3 for Apple Silicon.
This file is adapted from the official model repository:
https://huggingface.co/jinaai/jina-reranker-v3-mlx/blob/main/rerank.py
License: CC BY-NC 4.0 (contact Jina AI for commercial usage)
Changes from upstream:
- Removed the __main__ example block
- Type annotations added to public methods
- top_n parameter added to rerank() (upstream only exposed it implicitly)
"""
import numpy as np
class _MLPProjector:
def __init__(self):
import mlx.nn as nn
self.linear1 = nn.Linear(1024, 512, bias=False)
self.linear2 = nn.Linear(512, 512, bias=False)
def __call__(self, x):
import mlx.nn as nn
x = self.linear1(x)
x = nn.relu(x)
x = self.linear2(x)
return x
def _load_projector(projector_path: str) -> _MLPProjector:
import mlx.core as mx
from safetensors import safe_open
projector = _MLPProjector()
with safe_open(projector_path, framework="numpy") as f:
projector.linear1.weight = mx.array(f.get_tensor("linear1.weight"))
projector.linear2.weight = mx.array(f.get_tensor("linear2.weight"))
return projector
def _sanitize(text: str, special_tokens: dict[str, str]) -> str:
for token in special_tokens.values():
text = text.replace(token, "")
return text
def _format_prompt(query: str, docs: list[str], special_tokens: dict[str, str]) -> str:
query = _sanitize(query, special_tokens)
docs = [_sanitize(d, special_tokens) for d in docs]
doc_token = special_tokens["doc_embed_token"]
query_token = special_tokens["query_embed_token"]
prefix = (
"<|im_start|>system\n"
"You are a search relevance expert who can determine a ranking of the passages based on how relevant they are to the query. "
"If the query is a question, how relevant a passage is depends on how well it answers the question. "
"If not, try to analyze the intent of the query and assess how well each passage satisfies the intent. "
"If an instruction is provided, you should follow the instruction when determining the ranking."
"<|im_end|>\n<|im_start|>user\n"
)
suffix = "<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
body = (
f"I will provide you with {len(docs)} passages, each indicated by a numerical identifier. "
f"Rank the passages based on their relevance to query: {query}\n"
)
body += "\n".join(f'<passage id="{i}">\n{doc}{doc_token}\n</passage>' for i, doc in enumerate(docs))
body += f"\n<query>\n{query}{query_token}\n</query>"
return prefix + body + suffix
class MLXReranker:
"""
MLX-accelerated jina-reranker-v3 for Apple Silicon.
Loads the model from a local directory (use huggingface_hub.snapshot_download
to fetch jinaai/jina-reranker-v3-mlx if you don't have it already).
"""
_SPECIAL_TOKENS = {
"query_embed_token": "<|rerank_token|>",
"doc_embed_token": "<|embed_token|>",
}
_DOC_TOKEN_ID = 151670
_QUERY_TOKEN_ID = 151671
def __init__(self, model_path: str, projector_path: str):
from mlx_lm import load
self.model, self.tokenizer = load(model_path)
self.model.eval()
self.projector = _load_projector(projector_path)
def rerank(self, query: str, documents: list[str], top_n: int | None = None) -> list[dict]:
"""
Rank documents by relevance to a query.
Returns a list of dicts with keys: document, relevance_score, index.
Sorted by descending relevance_score.
"""
import mlx.core as mx
prompt = _format_prompt(query, documents, self._SPECIAL_TOKENS)
input_ids = self.tokenizer.encode(prompt)
hidden_states = self.model.model([input_ids])[0] # [seq_len, hidden_size]
input_ids_np = np.array(input_ids)
query_positions = np.where(input_ids_np == self._QUERY_TOKEN_ID)[0]
doc_positions = np.where(input_ids_np == self._DOC_TOKEN_ID)[0]
if len(query_positions) == 0:
raise ValueError("Query embed token not found in prompt")
if len(doc_positions) == 0:
raise ValueError("Document embed tokens not found in prompt")
query_hidden = mx.expand_dims(hidden_states[int(query_positions[0])], axis=0)
doc_hidden = mx.stack([hidden_states[int(p)] for p in doc_positions])
query_emb = self.projector(query_hidden) # [1, 512]
doc_emb = self.projector(doc_hidden) # [num_docs, 512]
query_exp = mx.broadcast_to(mx.expand_dims(query_emb, 0), (1, len(documents), 512))
doc_exp = mx.expand_dims(doc_emb, 0)
scores = mx.sum(doc_exp * query_exp, axis=-1) / (
mx.sqrt(mx.sum(doc_exp * doc_exp, axis=-1)) * mx.sqrt(mx.sum(query_exp * query_exp, axis=-1))
) # [1, num_docs]
scores_np = np.array(scores[0])
order = np.argsort(scores_np)[::-1]
n = min(top_n, len(documents)) if top_n is not None else len(documents)
return [
{
"document": documents[order[i]],
"relevance_score": float(scores_np[order[i]]),
"index": int(order[i]),
}
for i in range(n)
]
@@ -1,128 +0,0 @@
"""File parser implementations."""
import logging
from dataclasses import dataclass
from .base import FileParser, UnsupportedFileTypeError
from .iris import IrisParser
from .markitdown import MarkitdownParser
__all__ = [
"FileParser",
"UnsupportedFileTypeError",
"IrisParser",
"MarkitdownParser",
"FileParserRegistry",
"ConvertResult",
]
@dataclass
class ConvertResult:
"""Result of a successful file conversion."""
content: str
parser_name: str
logger = logging.getLogger(__name__)
class FileParserRegistry:
"""Registry for file parsers with auto-detection."""
def __init__(self):
"""Initialize empty parser registry."""
self._parsers: dict[str, FileParser] = {}
def register(self, parser: FileParser):
"""
Register a parser.
Args:
parser: FileParser instance
"""
self._parsers[parser.name()] = parser
def get_parser(
self,
name: str | None,
filename: str,
content_type: str | None = None,
) -> FileParser:
"""
Get parser by name or auto-detect.
Args:
name: Parser name (e.g., "markitdown") or None for auto-detect
filename: File name for auto-detection
content_type: MIME type (optional)
Returns:
FileParser instance
Raises:
ValueError: If no suitable parser found
"""
if name:
# Explicit parser requested — return it directly, let the parser
# raise UnsupportedFileTypeError from convert() if needed
if name not in self._parsers:
raise ValueError(f"Parser '{name}' not found. Available: {list(self._parsers.keys())}")
return self._parsers[name]
# Auto-detect parser
for parser in self._parsers.values():
if parser.supports(filename, content_type):
return parser
raise ValueError(f"No parser found for {filename}. Available parsers: {list(self._parsers.keys())}")
async def convert_with_fallback(
self,
parsers: list[str],
file_data: bytes,
filename: str,
content_type: str | None = None,
) -> ConvertResult:
"""
Try each parser in order, falling back on failure or empty content.
Moves to the next parser if the current one raises UnsupportedFileTypeError
or returns empty content. Any other exception (RuntimeError, network error,
etc.) also triggers a fallback so the chain is exhausted before failing.
Args:
parsers: Ordered list of parser names to try
file_data: Raw file bytes
filename: Original filename
content_type: MIME type (optional)
Returns:
ConvertResult with the parsed content and the name of the parser that succeeded
Raises:
ValueError: If a parser name is not registered
RuntimeError: If all parsers fail or return empty content
"""
last_error: Exception | None = None
for name in parsers:
parser = self.get_parser(name, filename, content_type)
try:
content = await parser.convert(file_data, filename)
if content and content.strip():
return ConvertResult(content=content, parser_name=name)
logger.warning(f"Parser '{name}' returned empty content for '{filename}', trying next")
last_error = RuntimeError(f"Parser '{name}' returned no content for '{filename}'")
except UnsupportedFileTypeError as e:
logger.warning(f"Parser '{name}' does not support '{filename}', trying next: {e}")
last_error = e
except Exception as e:
logger.warning(f"Parser '{name}' failed for '{filename}', trying next: {e}")
last_error = e
raise last_error or RuntimeError(f"No parsers available for '{filename}'")
def list_parsers(self) -> list[str]:
"""Get list of registered parser names."""
return list(self._parsers.keys())
@@ -1,58 +0,0 @@
"""Abstract base class for file parsers."""
from abc import ABC, abstractmethod
class UnsupportedFileTypeError(Exception):
"""Raised by a parser when it does not support the given file type."""
pass
class FileParser(ABC):
"""Abstract base for file to markdown parsers."""
@abstractmethod
async def convert(self, file_data: bytes, filename: str) -> str:
"""
Parse file to markdown.
Args:
file_data: Raw file bytes
filename: Original filename (used for format detection)
Returns:
Markdown content as string
Raises:
UnsupportedFileTypeError: If the file type is not supported by this parser
RuntimeError: If parsing fails for another reason
"""
pass
def supports(self, filename: str, content_type: str | None = None) -> bool:
"""
Check if parser supports this file type.
Override this for local/static extension-based filtering.
Parsers that delegate to a remote service should leave this as True
and raise UnsupportedFileTypeError from convert() instead.
Args:
filename: File name (used for extension check)
content_type: MIME type (optional)
Returns:
True if this parser can handle the file (default: True)
"""
return True
@abstractmethod
def name(self) -> str:
"""
Get parser name.
Returns:
Parser name (e.g., "markitdown")
"""
pass
@@ -1,138 +0,0 @@
"""Iris parser implementation using the Vectorize Iris HTTP API."""
import asyncio
import logging
import mimetypes
import time
import httpx
from .base import FileParser, UnsupportedFileTypeError
logger = logging.getLogger(__name__)
_IRIS_BASE_URL = "https://api.vectorize.io/v1"
_DEFAULT_POLL_INTERVAL = 2.0 # seconds
_DEFAULT_TIMEOUT = 300.0 # seconds
class IrisParser(FileParser):
"""
Iris file parser using the Vectorize Iris cloud extraction service.
Uploads files to the Vectorize Iris API, starts an extraction job,
and polls until the text is ready. The API determines which file types
are supported — UnsupportedFileTypeError is raised if the file is rejected.
Authentication:
Requires HINDSIGHT_API_FILE_PARSER_IRIS_TOKEN and
HINDSIGHT_API_FILE_PARSER_IRIS_ORG_ID environment variables,
or pass them explicitly via the constructor.
"""
def __init__(
self,
token: str,
org_id: str,
poll_interval: float = _DEFAULT_POLL_INTERVAL,
timeout: float = _DEFAULT_TIMEOUT,
):
"""
Initialize iris parser.
Args:
token: Vectorize API token
org_id: Vectorize organization ID
poll_interval: Seconds between status poll requests (default: 2)
timeout: Maximum seconds to wait for extraction (default: 300)
"""
self._token = token
self._org_id = org_id
self._poll_interval = poll_interval
self._timeout = timeout
self._auth_headers = {"Authorization": f"Bearer {token}"}
async def convert(self, file_data: bytes, filename: str) -> str:
"""
Parse file to text using the Vectorize Iris API.
Raises:
UnsupportedFileTypeError: If the Iris API rejects the file type (4xx)
RuntimeError: If extraction fails for another reason
"""
content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0, read=120.0)) as client:
# Step 1: Request a presigned upload URL
init_resp = await client.post(
f"{_IRIS_BASE_URL}/org/{self._org_id}/files",
headers=self._auth_headers,
json={"name": filename, "contentType": content_type},
)
_raise_for_status(init_resp, filename, "file upload init")
init_data = init_resp.json()
file_id: str = init_data["fileId"]
upload_url: str = init_data["uploadUrl"]
# Step 2: Upload the file bytes to the presigned URL (no auth header)
# Ensure file_data is plain bytes (GCS storage may return obstore.Bytes)
upload_resp = await client.put(
upload_url,
content=bytes(file_data),
headers={"Content-Type": content_type},
)
_raise_for_status(upload_resp, filename, "file upload")
# Step 3: Start extraction
extract_resp = await client.post(
f"{_IRIS_BASE_URL}/org/{self._org_id}/extraction",
headers=self._auth_headers,
json={"fileId": file_id},
)
_raise_for_status(extract_resp, filename, "start extraction")
extraction_id: str = extract_resp.json()["extractionId"]
# Step 4: Poll until ready or timeout
deadline = time.monotonic() + self._timeout
while True:
status_resp = await client.get(
f"{_IRIS_BASE_URL}/org/{self._org_id}/extraction/{extraction_id}",
headers=self._auth_headers,
)
_raise_for_status(status_resp, filename, "poll extraction status")
status_data = status_resp.json()
if status_data.get("ready"):
data = status_data.get("data", {})
if not data.get("success"):
error = data.get("error", "unknown error")
raise RuntimeError(f"Iris extraction failed for '{filename}': {error}")
text = data.get("text")
if not text:
raise RuntimeError(f"No content extracted from '{filename}'")
return text
if time.monotonic() >= deadline:
raise RuntimeError(f"Iris extraction timed out after {self._timeout}s for '{filename}'")
await asyncio.sleep(self._poll_interval)
def name(self) -> str:
"""Get parser name."""
return "iris"
def _raise_for_status(response: httpx.Response, filename: str, step: str) -> None:
"""
Raise an appropriate error including the response body on HTTP errors.
Raises UnsupportedFileTypeError for 4xx responses (file rejected by the API),
RuntimeError for other HTTP errors.
"""
if not response.is_error:
return
body = response.text or "<empty>"
msg = f"Iris API error during {step} for '{filename}': {response.status_code} {response.reason_phrase}{body}"
if response.is_client_error:
raise UnsupportedFileTypeError(msg)
raise RuntimeError(msg)
@@ -1,109 +0,0 @@
"""Markitdown parser implementation."""
import asyncio
import logging
import tempfile
from pathlib import Path
from .base import FileParser
logger = logging.getLogger(__name__)
class MarkitdownParser(FileParser):
"""
Markitdown file parser.
Uses Microsoft's markitdown library to convert various file formats
to markdown including PDF, Office docs, images (via OCR), audio, HTML.
Supported formats:
- PDF (.pdf)
- Word (.docx, .doc)
- PowerPoint (.pptx, .ppt)
- Excel (.xlsx, .xls)
- Images (.jpg, .jpeg, .png) - with OCR
- HTML (.html, .htm)
- Text (.txt, .md)
- Audio (.mp3, .wav) - with transcription
"""
def __init__(self):
"""Initialize markitdown parser."""
# Lazy import to avoid requiring markitdown for all users
try:
from markitdown import MarkItDown
self._markitdown = MarkItDown()
except ImportError as e:
raise ImportError(
"markitdown package is required for file parsing. Install with: pip install markitdown"
) from e
async def convert(self, file_data: bytes, filename: str) -> str:
"""Parse file to markdown using markitdown."""
# markitdown is synchronous, so we run it in executor to avoid blocking
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, self._convert_sync, file_data, filename)
def _convert_sync(self, file_data: bytes, filename: str) -> str:
"""Synchronous parsing (runs in thread pool)."""
# Write to temp file (markitdown requires file path)
with tempfile.NamedTemporaryFile(suffix=Path(filename).suffix, delete=False) as tmp:
tmp.write(file_data)
tmp_path = tmp.name
try:
# Parse using markitdown
result = self._markitdown.convert(tmp_path)
if not result or not result.text_content:
raise RuntimeError(f"No content extracted from '{filename}'")
return result.text_content
except Exception as e:
logger.error(f"Markitdown parsing failed for {filename}: {e}")
raise RuntimeError(f"Failed to parse '{filename}': {e}") from e
finally:
# Clean up temp file
try:
Path(tmp_path).unlink()
except Exception:
pass
def supports(self, filename: str, content_type: str | None = None) -> bool:
"""Check if markitdown supports this file type."""
# Supported extensions (from markitdown docs)
supported_extensions = {
# Documents
".pdf",
".docx",
".doc",
".pptx",
".ppt",
".xlsx",
".xls",
# Images (with OCR)
".jpg",
".jpeg",
".png",
# Web
".html",
".htm",
# Text
".txt",
".md",
".csv",
# Audio (with transcription)
".mp3",
".wav",
}
ext = Path(filename).suffix.lower()
return ext in supported_extensions
def name(self) -> str:
"""Get parser name."""
return "markitdown"
@@ -1,385 +0,0 @@
"""
LiteLLM LLM provider for universal model support.
This provider enables using 100+ LLM providers via the LiteLLM SDK, including:
- AWS Bedrock (bedrock/anthropic.claude-3-5-sonnet-...)
- Azure OpenAI (azure/gpt-4o)
- Together AI (together_ai/meta-llama/...)
- Any other LiteLLM-supported provider
Uses litellm.acompletion() for async chat completions.
Authentication for cloud providers (e.g., AWS Bedrock via boto3 credential chain)
is handled automatically by LiteLLM.
"""
import asyncio
import json
import logging
import time
from typing import Any
from hindsight_api.engine.llm_interface import LLMInterface, OutputTooLongError
from hindsight_api.engine.response_models import LLMToolCall, LLMToolCallResult, TokenUsage
from hindsight_api.metrics import get_metrics_collector
from hindsight_api.worker.stage import set_stage
logger = logging.getLogger(__name__)
class LiteLLMLLM(LLMInterface):
"""
LLM provider using the LiteLLM SDK for universal model support.
Supports any model accessible via litellm.acompletion(), including AWS Bedrock,
Azure OpenAI, Together AI, Fireworks AI, and more.
Model names follow LiteLLM conventions with provider prefixes:
- bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0
- azure/gpt-4o
- together_ai/meta-llama/Llama-3-70b-chat-hf
- fireworks_ai/accounts/fireworks/models/llama-v3p1-70b-instruct
"""
def __init__(
self,
provider: str,
api_key: str,
base_url: str,
model: str,
reasoning_effort: str = "low",
timeout: float = 300.0,
**kwargs: Any,
):
super().__init__(provider, api_key, base_url, model, reasoning_effort, **kwargs)
self.timeout = timeout
self._litellm: Any = None
try:
import litellm
self._litellm = litellm
# Suppress LiteLLM's verbose logging
litellm.suppress_debug_info = True # type: ignore[assignment]
# Drop unsupported params instead of raising errors (e.g. tool_choice on some Bedrock models)
litellm.drop_params = True # type: ignore[assignment]
logging.getLogger("LiteLLM").setLevel(logging.WARNING)
logger.info(f"LiteLLM SDK initialized for model: {self.model}")
except ImportError as e:
raise RuntimeError("LiteLLM SDK not installed. Run: uv add litellm or pip install litellm") from e
async def verify_connection(self) -> None:
try:
test_messages = [{"role": "user", "content": "test"}]
await self.call(
messages=test_messages,
max_completion_tokens=50,
temperature=0.0,
scope="verification",
max_retries=0,
)
logger.info("LiteLLM connection verified successfully")
except OutputTooLongError:
# Truncation is fine for verification — it means the connection works
logger.info("LiteLLM connection verified successfully (response truncated)")
except Exception as e:
logger.error(f"LiteLLM connection verification failed: {e}")
raise RuntimeError(f"Failed to verify LiteLLM connection: {e}") from e
def _build_common_kwargs(
self,
messages: list[dict[str, Any]],
max_completion_tokens: int | None = None,
temperature: float | None = None,
) -> dict[str, Any]:
"""Build common kwargs for litellm calls."""
kwargs: dict[str, Any] = {
"model": self.model,
"messages": messages,
"timeout": self.timeout,
}
if self.api_key:
kwargs["api_key"] = self.api_key
if self.base_url:
kwargs["api_base"] = self.base_url
if max_completion_tokens is not None:
kwargs["max_completion_tokens"] = max_completion_tokens
if temperature is not None:
kwargs["temperature"] = temperature
return kwargs
async def call(
self,
messages: list[dict[str, str]],
response_format: Any | None = None,
max_completion_tokens: int | None = None,
temperature: float | None = None,
scope: str = "memory",
max_retries: int = 10,
initial_backoff: float = 1.0,
max_backoff: float = 60.0,
skip_validation: bool = False,
strict_schema: bool = False,
return_usage: bool = False,
) -> Any:
start_time = time.time()
call_kwargs = self._build_common_kwargs(messages, max_completion_tokens, temperature)
# Add JSON schema response format if provided
if response_format is not None and hasattr(response_format, "model_json_schema"):
schema = response_format.model_json_schema()
call_kwargs["response_format"] = {
"type": "json_schema",
"json_schema": {
"name": response_format.__name__ if hasattr(response_format, "__name__") else "response",
"schema": schema,
"strict": strict_schema,
},
}
last_exception = None
for attempt in range(max_retries + 1):
if attempt > 0:
set_stage(f"llm.litellm.{scope}.attempt={attempt + 1}/{max_retries + 1}")
try:
response = await self._litellm.acompletion(**call_kwargs)
content = response.choices[0].message.content or ""
finish_reason = response.choices[0].finish_reason
# Check for length-limited output
if finish_reason == "length":
raise OutputTooLongError("LiteLLM response was truncated due to token limit")
if response_format is not None:
# Strip markdown code fences if present
clean_content = content
if "```json" in content:
clean_content = content.split("```json")[1].split("```")[0].strip()
elif "```" in content:
clean_content = content.split("```")[1].split("```")[0].strip()
try:
json_data = json.loads(clean_content)
except json.JSONDecodeError:
json_data = json.loads(content)
if skip_validation:
result = json_data
else:
result = response_format.model_validate(json_data)
else:
result = content
# Extract usage
input_tokens = getattr(response.usage, "prompt_tokens", 0) or 0
output_tokens = getattr(response.usage, "completion_tokens", 0) or 0
total_tokens = input_tokens + output_tokens
# Record metrics
duration = time.time() - start_time
metrics = get_metrics_collector()
metrics.record_llm_call(
provider=self.provider,
model=self.model,
scope=scope,
duration=duration,
input_tokens=input_tokens,
output_tokens=output_tokens,
success=True,
)
# Record trace span
from hindsight_api.tracing import _serialize_for_span, get_span_recorder
span_recorder = get_span_recorder()
span_recorder.record_llm_call(
provider=self.provider,
model=self.model,
scope=scope,
messages=messages,
response_content=_serialize_for_span(result),
input_tokens=input_tokens,
output_tokens=output_tokens,
duration=duration,
finish_reason=finish_reason,
error=None,
)
if duration > 10.0:
logger.info(
f"slow llm call: scope={scope}, model={self.provider}/{self.model}, "
f"input_tokens={input_tokens}, output_tokens={output_tokens}, "
f"time={duration:.3f}s"
)
if return_usage:
token_usage = TokenUsage(
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=total_tokens,
)
return result, token_usage
return result
except OutputTooLongError:
raise
except json.JSONDecodeError as e:
last_exception = e
if attempt < max_retries:
logger.warning("LiteLLM returned invalid JSON, retrying...")
backoff = min(initial_backoff * (2**attempt), max_backoff)
await asyncio.sleep(backoff)
continue
else:
logger.error(f"LiteLLM returned invalid JSON after {max_retries + 1} attempts")
raise
except Exception as e:
error_str = str(e).lower()
# Fast fail on auth errors
if "401" in error_str or "403" in error_str or "unauthorized" in error_str:
logger.error(f"LiteLLM auth error, not retrying: {e}")
raise
last_exception = e
if attempt < max_retries:
# Retry on rate limits, connection errors, server errors
is_retryable = any(
keyword in error_str
for keyword in ("rate", "limit", "timeout", "connection", "500", "502", "503", "529")
)
if is_retryable:
backoff = min(initial_backoff * (2**attempt), max_backoff)
jitter = backoff * 0.2 * (2 * (time.time() % 1) - 1)
await asyncio.sleep(backoff + jitter)
continue
logger.error(f"LiteLLM API error after {attempt + 1} attempts: {e}")
raise
if last_exception:
raise last_exception
raise RuntimeError("LiteLLM call failed after all retries")
async def call_with_tools(
self,
messages: list[dict[str, Any]],
tools: list[dict[str, Any]],
max_completion_tokens: int | None = None,
temperature: float | None = None,
scope: str = "tools",
max_retries: int = 5,
initial_backoff: float = 1.0,
max_backoff: float = 30.0,
tool_choice: str | dict[str, Any] = "auto",
) -> LLMToolCallResult:
start_time = time.time()
call_kwargs = self._build_common_kwargs(messages, max_completion_tokens, temperature)
call_kwargs["tools"] = tools
call_kwargs["tool_choice"] = tool_choice
last_exception = None
for attempt in range(max_retries + 1):
if attempt > 0:
set_stage(f"llm.litellm.tools.attempt={attempt + 1}/{max_retries + 1}")
try:
response = await self._litellm.acompletion(**call_kwargs)
message = response.choices[0].message
content = message.content
finish_reason = response.choices[0].finish_reason
# Extract tool calls
tool_calls: list[LLMToolCall] = []
if message.tool_calls:
for tc in message.tool_calls:
arguments = tc.function.arguments
if isinstance(arguments, str):
arguments = json.loads(arguments)
tool_calls.append(
LLMToolCall(
id=tc.id,
name=tc.function.name,
arguments=arguments,
)
)
# Extract usage
input_tokens = getattr(response.usage, "prompt_tokens", 0) or 0
output_tokens = getattr(response.usage, "completion_tokens", 0) or 0
# Record metrics
duration = time.time() - start_time
metrics = get_metrics_collector()
metrics.record_llm_call(
provider=self.provider,
model=self.model,
scope=scope,
duration=duration,
input_tokens=input_tokens,
output_tokens=output_tokens,
success=True,
)
# Record trace span
from hindsight_api.tracing import get_span_recorder
span_recorder = get_span_recorder()
tool_calls_dict = (
[{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in tool_calls]
if tool_calls
else None
)
span_recorder.record_llm_call(
provider=self.provider,
model=self.model,
scope=scope,
messages=messages,
response_content=content,
input_tokens=input_tokens,
output_tokens=output_tokens,
duration=duration,
finish_reason=finish_reason,
error=None,
tool_calls=tool_calls_dict,
)
return LLMToolCallResult(
content=content,
tool_calls=tool_calls,
finish_reason=finish_reason or ("tool_calls" if tool_calls else "stop"),
input_tokens=input_tokens,
output_tokens=output_tokens,
)
except Exception as e:
error_str = str(e).lower()
if "401" in error_str or "403" in error_str or "unauthorized" in error_str:
raise
last_exception = e
if attempt < max_retries:
is_retryable = any(
keyword in error_str
for keyword in ("rate", "limit", "timeout", "connection", "500", "502", "503", "529")
)
if is_retryable:
await asyncio.sleep(min(initial_backoff * (2**attempt), max_backoff))
continue
logger.error(f"LiteLLM tool call error after {attempt + 1} attempts: {e}")
raise
if last_exception:
raise last_exception
raise RuntimeError("LiteLLM tool call failed after all retries")
async def cleanup(self) -> None:
"""Clean up resources."""
pass
@@ -1,428 +0,0 @@
"""
Built-in llama.cpp LLM provider for fully offline operation.
Manages a llama-cpp-python server as a subprocess, downloads GGUF models
from HuggingFace on first use, and delegates inference to the OpenAI-compatible API.
Usage:
HINDSIGHT_API_LLM_PROVIDER=llamacpp
HINDSIGHT_API_LLAMACPP_MODEL_PATH=~/.hindsight/models/gemma-4-E2B-it-Q4_K_M.gguf
HINDSIGHT_API_LLAMACPP_GPU_LAYERS=-1 # -1 = all layers on GPU
HINDSIGHT_API_LLAMACPP_CONTEXT_SIZE=8192
"""
import asyncio
import logging
import os
import signal
import socket
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
from hindsight_api.engine.llm_interface import LLMInterface
from hindsight_api.engine.response_models import LLMToolCallResult
logger = logging.getLogger(__name__)
# Default GGUF model for offline mode
DEFAULT_LLAMACPP_HF_REPO = "bartowski/google_gemma-4-E2B-it-GGUF"
DEFAULT_LLAMACPP_HF_FILENAME = "google_gemma-4-E2B-it-Q4_K_M.gguf"
DEFAULT_LLAMACPP_MODEL_ALIAS = "gemma-4-e2b-it"
MODELS_DIR = Path.home() / ".hindsight" / "models"
# Singleton server instance — shared across all LlamaCppLLM instances
# (retain, reflect, consolidation each create their own LLMProvider,
# but they should all share one llama.cpp server process)
_shared_server: "LlamaCppServer | None" = None
_shared_server_lock = asyncio.Lock()
def _find_free_port() -> int:
"""Find a free TCP port on localhost."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def _download_default_model() -> Path:
"""Download the default GGUF model from HuggingFace if not already cached.
Returns:
Path to the downloaded GGUF file.
"""
try:
from huggingface_hub import hf_hub_download
except ImportError:
raise ImportError(
"huggingface-hub is required for automatic model download. "
"Install with: pip install 'hindsight-api-slim[local-llm]'"
)
MODELS_DIR.mkdir(parents=True, exist_ok=True)
target = MODELS_DIR / DEFAULT_LLAMACPP_HF_FILENAME
if target.exists():
logger.info(f"Using cached model: {target}")
return target
logger.info(
f"Downloading {DEFAULT_LLAMACPP_HF_FILENAME} from {DEFAULT_LLAMACPP_HF_REPO} (~3.5 GB, first run only)..."
)
downloaded = hf_hub_download(
repo_id=DEFAULT_LLAMACPP_HF_REPO,
filename=DEFAULT_LLAMACPP_HF_FILENAME,
local_dir=str(MODELS_DIR),
)
logger.info(f"Model downloaded: {downloaded}")
return Path(downloaded)
def _resolve_model_path(model_path: str | None) -> Path:
"""Resolve the model path, downloading the default if needed.
Args:
model_path: Explicit path to a GGUF file, or None to use the default.
Returns:
Resolved Path to the GGUF file.
"""
if model_path:
p = Path(model_path).expanduser()
if not p.exists():
raise FileNotFoundError(
f"GGUF model not found: {p}\n"
f"Set HINDSIGHT_API_LLAMACPP_MODEL_PATH to a valid .gguf file, "
f"or remove the setting to auto-download the default model."
)
return p
return _download_default_model()
class LlamaCppServer:
"""Manages a llama-cpp-python OpenAI-compatible server as a subprocess."""
def __init__(
self,
model_path: Path,
port: int,
gpu_layers: int = -1,
context_size: int = 8192,
chat_format: str | None = None,
extra_args: str | None = None,
):
self.model_path = model_path
self.port = port
self.gpu_layers = gpu_layers
self.context_size = context_size
self.chat_format = chat_format
self.extra_args = extra_args
self._process: subprocess.Popen | None = None
@property
def base_url(self) -> str:
return f"http://127.0.0.1:{self.port}/v1"
async def start(self) -> None:
"""Start the llama.cpp server subprocess."""
cmd = [
sys.executable,
"-m",
"llama_cpp.server",
"--model",
str(self.model_path),
"--host",
"127.0.0.1",
"--port",
str(self.port),
"--n_gpu_layers",
str(self.gpu_layers),
"--n_ctx",
str(self.context_size),
"--flash_attn",
"true",
"--n_batch",
"2048",
# Prompt cache: reuse KV cache for repeated system prompts
"--cache",
"true",
]
# Only pass chat_format if explicitly set (most GGUF models have it embedded)
if self.chat_format:
cmd.extend(["--chat_format", self.chat_format])
# User-provided extra args (e.g. "--type_k 1 --type_v 1 --n_threads 8")
if self.extra_args:
cmd.extend(self.extra_args.split())
logger.info(f"Starting llama.cpp server: {' '.join(cmd)}")
# Write stderr to a log file to avoid pipe buffer deadlock
# (llama.cpp outputs a lot of model metadata on stderr during loading)
self._log_path = MODELS_DIR / "llamacpp_server.log"
self._log_file = open(self._log_path, "w")
self._process = subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=self._log_file,
# Ensure the subprocess is killed when the parent exits
preexec_fn=os.setsid if hasattr(os, "setsid") else None,
)
# Wait for the server to be ready
await self._wait_for_ready()
async def _wait_for_ready(self, timeout: float = 120.0) -> None:
"""Wait for the llama.cpp server to accept connections."""
import httpx
start = time.monotonic()
url = f"http://127.0.0.1:{self.port}/v1/models"
last_log = start
while time.monotonic() - start < timeout:
# Check if process died
if self._process and self._process.poll() is not None:
stderr = ""
try:
stderr = self._log_path.read_text()[-2000:]
except Exception:
pass
raise RuntimeError(f"llama.cpp server exited with code {self._process.returncode}.\nstderr: {stderr}")
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=5.0)
if resp.status_code == 200:
logger.info(f"llama.cpp server ready on port {self.port}")
return
except (httpx.ConnectError, httpx.TimeoutException, httpx.ConnectTimeout):
pass
# Log progress every 15s
now = time.monotonic()
if now - last_log > 15:
elapsed = int(now - start)
logger.info(f"Waiting for llama.cpp server to load model... ({elapsed}s)")
last_log = now
await asyncio.sleep(1.0)
# Timeout — read the log to help debug
stderr = ""
try:
stderr = self._log_path.read_text()[-2000:]
except Exception:
pass
raise TimeoutError(
f"llama.cpp server did not become ready within {timeout}s.\n"
f"Check model compatibility and available memory.\n"
f"Server log: {stderr}"
)
async def stop(self) -> None:
"""Stop the llama.cpp server subprocess."""
if self._process is None:
return
logger.info("Stopping llama.cpp server...")
try:
# Send SIGTERM to the process group
if hasattr(os, "killpg"):
os.killpg(os.getpgid(self._process.pid), signal.SIGTERM)
else:
self._process.terminate()
# Wait up to 10s for graceful shutdown
try:
self._process.wait(timeout=10)
except subprocess.TimeoutExpired:
if hasattr(os, "killpg"):
os.killpg(os.getpgid(self._process.pid), signal.SIGKILL)
else:
self._process.kill()
self._process.wait(timeout=5)
except (ProcessLookupError, OSError):
pass # Process already exited
finally:
self._process = None
if hasattr(self, "_log_file") and self._log_file:
self._log_file.close()
self._log_file = None
logger.info("llama.cpp server stopped")
class LlamaCppLLM(LLMInterface):
"""
Built-in llama.cpp provider.
Manages a llama-cpp-python server subprocess and delegates to OpenAICompatibleLLM
for actual inference calls. Handles model downloading and server lifecycle.
"""
def __init__(
self,
provider: str,
api_key: str,
base_url: str,
model: str,
reasoning_effort: str = "low",
model_path: str | None = None,
gpu_layers: int = -1,
context_size: int = 8192,
chat_format: str | None = None,
no_grammar: bool = False,
extra_args: str | None = None,
**kwargs: Any,
):
super().__init__(
provider=provider,
api_key=api_key or "llamacpp",
base_url=base_url or "",
model=model or DEFAULT_LLAMACPP_MODEL_ALIAS,
reasoning_effort=reasoning_effort,
)
self._model_path_str = model_path
self._gpu_layers = gpu_layers
self._context_size = context_size
self._chat_format = chat_format
self._no_grammar = no_grammar
self._extra_args = extra_args
self._server: LlamaCppServer | None = None
self._delegate: Any = None # OpenAICompatibleLLM, created after server starts
self._initialized = False
async def _ensure_initialized(self) -> None:
"""Lazy initialization: download model + start shared server on first use."""
if self._initialized:
return
global _shared_server
from .openai_compatible_llm import OpenAICompatibleLLM
async with _shared_server_lock:
if _shared_server is None:
# Resolve and potentially download the model
model_path = _resolve_model_path(self._model_path_str)
logger.info(f"Using GGUF model: {model_path}")
# Start the shared llama.cpp server
port = _find_free_port()
_shared_server = LlamaCppServer(
model_path=model_path,
port=port,
gpu_layers=self._gpu_layers,
context_size=self._context_size,
chat_format=self._chat_format,
extra_args=self._extra_args,
)
await _shared_server.start()
self._server = _shared_server
# Create the delegate that talks to the shared server's OpenAI-compatible API
if self._no_grammar:
logger.info("Grammar enforcement disabled (HINDSIGHT_API_LLAMACPP_NO_GRAMMAR=true)")
self._delegate = OpenAICompatibleLLM(
provider="llamacpp",
api_key="llamacpp",
base_url=self._server.base_url,
model=self.model,
reasoning_effort=self.reasoning_effort,
)
self._initialized = True
async def verify_connection(self) -> None:
"""Verify the llama.cpp server is running and can generate text."""
await self._ensure_initialized()
# Make a simple test call to verify the model can actually generate
await self._delegate.call(
messages=[{"role": "user", "content": "Say 'ok'"}],
max_completion_tokens=10,
max_retries=2,
initial_backoff=0.5,
max_backoff=2.0,
scope="verification",
)
logger.info("llama.cpp LLM verification passed")
async def call(
self,
messages: list[dict[str, str]],
response_format: Any | None = None,
max_completion_tokens: int | None = None,
temperature: float | None = None,
scope: str = "memory",
max_retries: int = 10,
initial_backoff: float = 1.0,
max_backoff: float = 60.0,
skip_validation: bool = False,
strict_schema: bool = False,
return_usage: bool = False,
) -> Any:
"""Delegate call to the OpenAI-compatible API."""
await self._ensure_initialized()
return await self._delegate.call(
messages=messages,
response_format=response_format,
max_completion_tokens=max_completion_tokens,
temperature=temperature,
scope=scope,
max_retries=max_retries,
initial_backoff=initial_backoff,
max_backoff=max_backoff,
skip_validation=skip_validation,
strict_schema=strict_schema,
return_usage=return_usage,
)
async def call_with_tools(
self,
messages: list[dict[str, Any]],
tools: list[dict[str, Any]],
max_completion_tokens: int | None = None,
temperature: float | None = None,
scope: str = "tools",
max_retries: int = 5,
initial_backoff: float = 1.0,
max_backoff: float = 30.0,
tool_choice: str | dict[str, Any] = "auto",
) -> LLMToolCallResult:
"""Delegate tool calls to the OpenAI-compatible API."""
await self._ensure_initialized()
return await self._delegate.call_with_tools(
messages=messages,
tools=tools,
max_completion_tokens=max_completion_tokens,
temperature=temperature,
scope=scope,
max_retries=max_retries,
initial_backoff=initial_backoff,
max_backoff=max_backoff,
tool_choice=tool_choice,
)
async def cleanup(self) -> None:
"""Stop the shared llama.cpp server."""
global _shared_server
if self._delegate:
await self._delegate.cleanup()
self._delegate = None
# Stop the shared server (only the first cleanup call actually stops it)
async with _shared_server_lock:
if _shared_server is not None:
await _shared_server.stop()
_shared_server = None
self._server = None
self._initialized = False
@@ -1,78 +0,0 @@
"""
No-op LLM provider for chunk-only storage mode.
When the LLM provider is set to "none", the system operates without any LLM dependency.
Retain uses chunks mode (no fact extraction), and reflect/consolidation are disabled.
This provider acts as a safety net — if any code path unexpectedly tries to call the LLM,
it raises a clear error instead of a confusing connection failure.
"""
import logging
from typing import Any
from ..llm_interface import LLMInterface
from ..response_models import LLMToolCallResult
logger = logging.getLogger(__name__)
class LLMNotAvailableError(Exception):
"""Raised when an operation requires an LLM but the provider is set to 'none'."""
pass
class NoneLLM(LLMInterface):
"""
No-op LLM provider that rejects all LLM calls.
Used when HINDSIGHT_API_LLM_PROVIDER=none to run Hindsight as a chunk store
with semantic search but without LLM-based features (fact extraction, reflect,
consolidation).
"""
async def verify_connection(self) -> None:
"""No-op — no LLM connection to verify."""
logger.debug("NoneLLM: no LLM connection to verify (provider=none)")
async def call(
self,
messages: list[dict[str, str]],
response_format: Any | None = None,
max_completion_tokens: int | None = None,
temperature: float | None = None,
scope: str = "memory",
max_retries: int = 10,
initial_backoff: float = 1.0,
max_backoff: float = 60.0,
skip_validation: bool = False,
strict_schema: bool = False,
return_usage: bool = False,
) -> Any:
"""Raise LLMNotAvailableError — no LLM is configured."""
raise LLMNotAvailableError(
"LLM provider is set to 'none'. This operation requires an LLM. "
"Set HINDSIGHT_API_LLM_PROVIDER to a real provider (e.g., openai, anthropic, gemini)."
)
async def call_with_tools(
self,
messages: list[dict[str, Any]],
tools: list[dict[str, Any]],
max_completion_tokens: int | None = None,
temperature: float | None = None,
scope: str = "tools",
max_retries: int = 5,
initial_backoff: float = 1.0,
max_backoff: float = 30.0,
tool_choice: str | dict[str, Any] = "auto",
) -> LLMToolCallResult:
"""Raise LLMNotAvailableError — no LLM is configured."""
raise LLMNotAvailableError(
"LLM provider is set to 'none'. This operation requires an LLM. "
"Set HINDSIGHT_API_LLM_PROVIDER to a real provider (e.g., openai, anthropic, gemini)."
)
async def cleanup(self) -> None:
"""No-op — nothing to clean up."""
pass
@@ -1,144 +0,0 @@
"""
Chunk storage for retain pipeline.
Handles storage of document chunks in the database.
"""
import hashlib
import logging
from dataclasses import dataclass
from ..memory_engine import fq_table
from .types import ChunkMetadata
logger = logging.getLogger(__name__)
def compute_chunk_hash(chunk_text: str) -> str:
"""Compute SHA256 hash of chunk text for delta comparison."""
return hashlib.sha256(chunk_text.encode()).hexdigest()
@dataclass
class ExistingChunk:
"""Represents a chunk already stored in the database."""
chunk_id: str
chunk_index: int
content_hash: str | None
async def load_existing_chunks(conn, bank_id: str, document_id: str) -> list[ExistingChunk]:
"""
Load existing chunk metadata for a document.
Returns list of ExistingChunk with chunk_id, chunk_index, and content_hash.
"""
rows = await conn.fetch(
f"""
SELECT chunk_id, chunk_index, content_hash
FROM {fq_table("chunks")}
WHERE document_id = $1 AND bank_id = $2
ORDER BY chunk_index
""",
document_id,
bank_id,
)
return [
ExistingChunk(
chunk_id=row["chunk_id"],
chunk_index=row["chunk_index"],
content_hash=row["content_hash"],
)
for row in rows
]
async def delete_chunks_by_ids(conn, chunk_ids: list[str]) -> None:
"""
Delete specific chunks by their IDs.
This cascades to memory_units (via FK with CASCADE delete)
and their links.
"""
if not chunk_ids:
return
await conn.execute(
f"DELETE FROM {fq_table('chunks')} WHERE chunk_id = ANY($1::text[])",
chunk_ids,
)
async def store_chunks_batch(conn, bank_id: str, document_id: str, chunks: list[ChunkMetadata]) -> dict[int, str]:
"""
Store document chunks in the database.
Args:
conn: Database connection
bank_id: Bank identifier
document_id: Document identifier
chunks: List of ChunkMetadata objects
Returns:
Dictionary mapping global chunk index to chunk_id
"""
if not chunks:
return {}
# Prepare chunk data for batch insert
chunk_ids = []
chunk_texts = []
chunk_indices = []
content_hashes = []
chunk_id_map = {}
for chunk in chunks:
chunk_id = f"{bank_id}_{document_id}_{chunk.chunk_index}"
chunk_ids.append(chunk_id)
chunk_texts.append(chunk.chunk_text)
chunk_indices.append(chunk.chunk_index)
content_hashes.append(compute_chunk_hash(chunk.chunk_text))
chunk_id_map[chunk.chunk_index] = chunk_id
# Batch upsert all chunks. ON CONFLICT makes this idempotent: re-submitting
# a retain under the same document_id (the pattern in vectorize-io/hindsight#977)
# may produce chunk_ids that already exist when upstream cascade-delete or
# delta-retain paths don't run (or race with a concurrent task). Overwriting
# is the correct behavior per the document_id grouping semantics — the caller
# intends this chunk to hold the latest content at that (document_id, index).
await conn.execute(
f"""
INSERT INTO {fq_table("chunks")} (chunk_id, document_id, bank_id, chunk_text, chunk_index, content_hash)
SELECT * FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::integer[], $6::text[])
ON CONFLICT (chunk_id) DO UPDATE SET
chunk_text = EXCLUDED.chunk_text,
chunk_index = EXCLUDED.chunk_index,
content_hash = EXCLUDED.content_hash
""",
chunk_ids,
[document_id] * len(chunk_texts),
[bank_id] * len(chunk_texts),
chunk_texts,
chunk_indices,
content_hashes,
)
return chunk_id_map
def map_facts_to_chunks(facts_chunk_indices: list[int], chunk_id_map: dict[int, str]) -> list[str | None]:
"""
Map fact chunk indices to chunk IDs.
Args:
facts_chunk_indices: List of chunk indices for each fact
chunk_id_map: Dictionary mapping chunk index to chunk_id
Returns:
List of chunk_ids (same length as facts_chunk_indices)
"""
chunk_ids = []
for chunk_idx in facts_chunk_indices:
chunk_id = chunk_id_map.get(chunk_idx)
chunk_ids.append(chunk_id)
return chunk_ids
@@ -1,194 +0,0 @@
"""
Entity labels models and helpers for retain pipeline.
Defines a controlled vocabulary of key:value classification labels
(e.g., 'pedagogy:scaffolding', 'interest:active') that are extracted
at retain time and stored as entities.
"""
from typing import Literal
from pydantic import BaseModel, Field, create_model
class LabelValue(BaseModel):
"""A single allowed value for a label group."""
value: str
description: str = ""
class LabelGroup(BaseModel):
"""A label group (dimension) with its type and allowed values."""
key: str
description: str = ""
type: Literal["value", "multi-values", "text"] = "value"
optional: bool = True
tag: bool = False
values: list[LabelValue] = []
class EntityLabelsConfig(BaseModel):
"""Entity labels configuration for a bank (controlled vocabulary)."""
attributes: list[LabelGroup] = []
def parse_entity_labels(raw: dict | list | None) -> EntityLabelsConfig | None:
"""
Parse raw entity labels config into EntityLabelsConfig.
Accepts:
- None → returns None
- list → list of attribute dicts (each may use legacy free_values/multi_value or new type field)
- dict → {attributes: [...]}
Legacy migration (backward-compat):
- free_values=True → type="text"
- multi_value=True → type="multi-values"
- neither / free_values=False → type="value"
Args:
raw: Raw entity labels config from bank config
Returns:
EntityLabelsConfig or None if raw is None/empty
"""
if raw is None:
return None
if isinstance(raw, list):
if not raw:
return None
attributes = [LabelGroup.model_validate(_migrate_label_group(a)) for a in raw]
return EntityLabelsConfig(attributes=attributes)
if isinstance(raw, dict):
attrs_raw = raw.get("attributes", [])
if not attrs_raw:
return None
attributes = [LabelGroup.model_validate(_migrate_label_group(a)) for a in attrs_raw]
return EntityLabelsConfig(attributes=attributes)
return None
def _migrate_label_group(raw: dict) -> dict:
"""Migrate legacy free_values/multi_value fields to the new type field."""
if not isinstance(raw, dict) or "type" in raw:
return raw
patched = dict(raw)
if patched.get("free_values"):
patched["type"] = "text"
elif patched.get("multi_value"):
patched["type"] = "multi-values"
else:
patched["type"] = "value"
# Remove legacy keys so Pydantic doesn't error on unknown fields
patched.pop("free_values", None)
patched.pop("multi_value", None)
return patched
def build_labels_model(labels_cfg: EntityLabelsConfig) -> type[BaseModel] | None:
"""
Build a dynamic Pydantic model for structured label extraction.
Each LabelGroup becomes a typed field based on its type:
- type="text" → str | None (always optional)
- type="value", optional=True → Literal["v1","v2"] | None
- type="value", optional=False → Literal["v1","v2"] (required)
- type="multi-values" → list[Literal["v1","v2"]]
Args:
labels_cfg: Parsed EntityLabelsConfig
Returns:
Dynamic Pydantic model class, or None if no groups defined
"""
fields: dict = {}
for group in labels_cfg.attributes:
if not group.key:
continue
description = group.description or group.key
if group.type == "text":
# Free-form: any string value accepted, always optional
fields[group.key] = (str | None, Field(default=None, description=description))
else:
# Enum-constrained: must have defined values
if not group.values:
continue
values = tuple(v.value for v in group.values if v.value)
if not values:
continue
# Literal[("v1", "v2")] is equivalent to Literal["v1", "v2"] in Python 3.11+
literal_type = Literal[values] # type: ignore[valid-type]
if group.type == "multi-values":
fields[group.key] = (
list[literal_type], # type: ignore[valid-type]
Field(default_factory=list, description=description),
)
elif group.optional:
fields[group.key] = (
literal_type | None, # type: ignore[valid-type]
Field(default=None, description=description),
)
else:
fields[group.key] = (
literal_type, # type: ignore[valid-type]
Field(description=description),
)
if not fields:
return None
return create_model("Labels", **fields)
def is_label_entity(text: str, labels_cfg: EntityLabelsConfig, labels_lookup: set[str]) -> bool:
"""
Return True if entity text belongs to any configured label group.
For enum groups: checks the pre-built lookup set.
For text groups: checks that the text starts with a known key prefix.
"""
if text.lower() in labels_lookup:
return True
for group in labels_cfg.attributes:
if group.type == "text" and group.key and text.lower().startswith(f"{group.key.lower()}:"):
return True
return False
def build_labels_lookup(labels_cfg: EntityLabelsConfig | list | None) -> set[str]:
"""
Build a set of valid 'key:value' label strings (lowercase) for fast lookup.
Accepts either EntityLabelsConfig or raw list/None for backwards compatibility.
Args:
labels_cfg: EntityLabelsConfig, raw list of attribute dicts, or None
Returns:
Set of lowercase 'key:value' strings
"""
if labels_cfg is None:
return set()
# Accept raw list/dict for backwards compatibility
if not isinstance(labels_cfg, EntityLabelsConfig):
parsed = parse_entity_labels(labels_cfg)
if parsed is None:
return set()
labels_cfg = parsed
valid = set()
for group in labels_cfg.attributes:
if group.type == "text":
continue # No fixed vocabulary — all values accepted in post-processing
for v in group.values:
if group.key and v.value:
valid.add(f"{group.key}:{v.value}".lower())
return valid
@@ -1,162 +0,0 @@
"""
Entity processing for retain pipeline.
Handles entity extraction, resolution, and link creation for stored facts.
"""
import logging
from . import link_utils
from .types import EntityLink, ProcessedFact
logger = logging.getLogger(__name__)
def _prepare_facts_for_entity_processing(
facts: list[ProcessedFact],
user_entities_per_content: dict[int, list[dict]] | None = None,
) -> tuple[list[str], list, list[list[dict]]]:
"""
Extract fact texts, dates, and merged entity lists from ProcessedFact objects.
Returns:
Tuple of (fact_texts, fact_dates, entities_per_fact)
"""
user_entities_per_content = user_entities_per_content or {}
fact_texts = [fact.fact_text for fact in facts]
fact_dates = [fact.occurred_start if fact.occurred_start is not None else fact.mentioned_at for fact in facts]
entities_per_fact = []
for fact in facts:
llm_entities = [{"text": entity.name, "type": "CONCEPT"} for entity in (fact.entities or [])]
user_entities = user_entities_per_content.get(fact.content_index, [])
seen_texts = {e["text"].lower() for e in llm_entities}
for user_entity in user_entities:
if user_entity["text"].lower() not in seen_texts:
llm_entities.append(
{
"text": user_entity["text"],
"type": user_entity.get("type", "CONCEPT"),
}
)
seen_texts.add(user_entity["text"].lower())
entities_per_fact.append(llm_entities)
return fact_texts, fact_dates, entities_per_fact
async def resolve_entities(
entity_resolver,
conn,
bank_id: str,
unit_ids: list[str],
facts: list[ProcessedFact],
log_buffer: list[str] = None,
user_entities_per_content: dict[int, list[dict]] = None,
entity_labels: list | None = None,
) -> tuple[list[str], list[tuple], dict[str, list[str]]]:
"""
Phase 1: Resolve entity names to canonical IDs (read-heavy).
Should be called on a SEPARATE connection OUTSIDE the main write transaction
to avoid holding the transaction open during expensive trigram scans.
Args:
entity_resolver: EntityResolver instance
conn: Database connection (separate from the main write transaction)
bank_id: Bank identifier
unit_ids: Placeholder unit IDs (used only for grouping)
facts: List of ProcessedFact objects
log_buffer: Optional buffer for detailed logging
user_entities_per_content: Dict mapping content_index to user-provided entities
entity_labels: Optional entity label taxonomy
Returns:
Tuple of (resolved_entity_ids, entity_to_unit, unit_to_entity_ids)
to pass to build_entity_links().
"""
if not unit_ids or not facts:
return [], [], {}
if len(unit_ids) != len(facts):
raise ValueError(f"Mismatch between unit_ids ({len(unit_ids)}) and facts ({len(facts)})")
fact_texts, fact_dates, entities_per_fact = _prepare_facts_for_entity_processing(facts, user_entities_per_content)
return await link_utils.resolve_entities_only(
entity_resolver,
conn,
bank_id,
unit_ids,
fact_texts,
"", # context (not used in current implementation)
fact_dates,
entities_per_fact,
log_buffer,
entity_labels=entity_labels,
)
async def build_entity_links(
entity_resolver,
conn,
bank_id: str,
unit_ids: list[str],
resolved_entity_ids: list[str],
entity_to_unit: list[tuple],
unit_to_entity_ids: dict[str, list[str]],
log_buffer: list[str] = None,
skip_unit_entities_insert: bool = False,
) -> list[EntityLink]:
"""
Build entity links for UI graph visualization.
Queries unit_entities to find shared entities between new and existing units,
then generates EntityLink objects. When called from Phase 3 (post-transaction),
set skip_unit_entities_insert=True since unit_entities were already inserted
in Phase 2.
Args:
entity_resolver: EntityResolver instance
conn: Database connection
bank_id: Bank identifier
unit_ids: Actual unit IDs (must already be inserted in the DB)
resolved_entity_ids: From resolve_entities()
entity_to_unit: From resolve_entities()
unit_to_entity_ids: From resolve_entities()
log_buffer: Optional buffer for detailed logging
skip_unit_entities_insert: Skip unit_entities INSERT (already done in Phase 2)
Returns:
List of EntityLink objects for batch insertion
"""
return await link_utils.build_entity_links_from_resolved(
entity_resolver,
conn,
bank_id,
unit_ids,
resolved_entity_ids,
entity_to_unit,
unit_to_entity_ids,
log_buffer,
skip_unit_entities_insert=skip_unit_entities_insert,
)
async def insert_entity_links_batch(conn, entity_links: list[EntityLink], bank_id: str) -> None:
"""
Insert entity links in batch.
Args:
conn: Database connection
entity_links: List of EntityLink objects
bank_id: Bank identifier (stored directly on memory_links for fast filtering)
"""
if not entity_links:
return
await link_utils.insert_entity_links_batch(conn, entity_links, bank_id)
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -1,67 +0,0 @@
"""
Graph retrieval strategies for memory recall.
This module provides an abstraction for graph-based memory retrieval,
allowing different algorithms to be swapped without changing the rest
of the recall pipeline.
"""
import logging
from abc import ABC, abstractmethod
from .tags import TagGroup, TagsMatch
from .types import GraphRetrievalTimings, RetrievalResult
logger = logging.getLogger(__name__)
class GraphRetriever(ABC):
"""
Abstract base class for graph-based memory retrieval.
Implementations traverse the memory graph (entity links, temporal links,
causal links) to find relevant facts that might not be found by
semantic or keyword search alone.
"""
@property
@abstractmethod
def name(self) -> str:
"""Return identifier for this retrieval strategy (e.g., 'link_expansion')."""
pass
@abstractmethod
async def retrieve(
self,
pool,
query_embedding_str: str,
bank_id: str,
fact_type: str,
budget: int,
query_text: str | None = None,
semantic_seeds: list[RetrievalResult] | None = None,
temporal_seeds: list[RetrievalResult] | None = None,
adjacency=None, # TypedAdjacency, optional pre-loaded graph
tags: list[str] | None = None, # Visibility scope tags for filtering
tags_match: TagsMatch = "any", # How to match tags: 'any' (OR) or 'all' (AND)
tag_groups: list[TagGroup] | None = None, # Compound boolean tag filter groups
) -> tuple[list[RetrievalResult], GraphRetrievalTimings | None]:
"""
Retrieve relevant facts via graph traversal.
Args:
pool: Database connection pool
query_embedding_str: Query embedding as string (for finding entry points)
bank_id: Memory bank identifier
fact_type: Fact type to filter ('world', 'experience', 'observation')
budget: Maximum number of nodes to explore/return
query_text: Original query text (optional, for some strategies)
semantic_seeds: Pre-computed semantic entry points (from semantic retrieval)
temporal_seeds: Pre-computed temporal entry points (from temporal retrieval)
adjacency: Pre-loaded typed adjacency graph (optional)
tags: Optional list of tags for visibility filtering (OR matching)
Returns:
Tuple of (List of RetrievalResult with activation scores, optional timing info)
"""
pass
@@ -1,553 +0,0 @@
"""
Link Expansion graph retrieval.
Expands from semantic/temporal seeds through three parallel, first-class signals
stored in memory_links:
1. Entity links — query-time self-join through unit_entities. Score = number of distinct
shared entities between the seed set and each candidate, computed via
COUNT(DISTINCT entity_id). Uses a LATERAL per-entity cap
(graph_per_entity_limit, default 200) to prevent high-fanout entities
from exploding the self-join intermediate rows.
2. Semantic links — precomputed kNN graph (each new fact linked to its top-5 most
similar existing facts at insert time, similarity >= 0.7). Checked
in both directions since the graph is not symmetric. Score = weight.
3. Causal links — explicit causal chains (causes/caused_by/enables/prevents).
Score = weight + 1.0 (boosted as highest-quality signal).
Entity expansion is bounded by graph_per_entity_limit (LATERAL cap per entity).
A timeout fallback (graph_expansion_timeout) drops entity expansion entirely if the
query still exceeds the budget.
For non-observation fact types the three expansions are issued as a single CTE query
(one roundtrip, one connection) with a `source` discriminator column so the Python
merge step can apply per-signal score transformations.
"""
import asyncio
import logging
import math
import time
from ...config import get_config
from ..db_utils import acquire_with_retry
from ..memory_engine import fq_table
from .graph_retrieval import GraphRetriever
from .tags import TagGroup, TagsMatch, filter_results_by_tag_groups, filter_results_by_tags
from .types import GraphRetrievalTimings, RetrievalResult
logger = logging.getLogger(__name__)
async def _find_semantic_seeds(
conn,
query_embedding_str: str,
bank_id: str,
fact_type: str,
limit: int = 20,
threshold: float = 0.3,
tags: list[str] | None = None,
tags_match: TagsMatch = "any",
tag_groups: list[TagGroup] | None = None,
) -> list[RetrievalResult]:
"""Find semantic seeds via embedding search."""
from .tags import build_tag_groups_where_clause, build_tags_where_clause_simple
tags_clause = build_tags_where_clause_simple(tags, 6, match=tags_match)
tag_groups_param_start = 6 + (1 if tags else 0)
groups_clause, groups_params, _ = build_tag_groups_where_clause(tag_groups, tag_groups_param_start)
params = [query_embedding_str, bank_id, fact_type, threshold, limit]
if tags:
params.append(tags)
params.extend(groups_params)
rows = await conn.fetch(
f"""
SELECT id, text, context, event_date, occurred_start, occurred_end,
mentioned_at, fact_type, document_id, chunk_id, tags, proof_count,
1 - (embedding <=> $1::vector) AS similarity
FROM {fq_table("memory_units")}
WHERE bank_id = $2
AND embedding IS NOT NULL
AND fact_type = $3
AND (1 - (embedding <=> $1::vector)) >= $4
{tags_clause}
{groups_clause}
ORDER BY embedding <=> $1::vector
LIMIT $5
""",
*params,
)
return [RetrievalResult.from_db_row(dict(r)) for r in rows]
class LinkExpansionRetriever(GraphRetriever):
"""
Graph retrieval via direct link expansion from seeds.
Runs three expansions through precomputed memory_links: entity co-occurrence,
semantic kNN, and causal chains, all bounded at retain time.
For non-observation fact types the three expansions are issued as a single CTE
query (one roundtrip, one connection slot) with a `source` discriminator column.
The Python merge step applies per-signal score transformations.
"""
def __init__(
self,
causal_weight_threshold: float = 0.3,
):
"""
Args:
causal_weight_threshold: Minimum weight for causal links to follow.
"""
self.causal_weight_threshold = causal_weight_threshold
@property
def name(self) -> str:
return "link_expansion"
async def retrieve(
self,
pool,
query_embedding_str: str,
bank_id: str,
fact_type: str,
budget: int,
query_text: str | None = None,
semantic_seeds: list[RetrievalResult] | None = None,
temporal_seeds: list[RetrievalResult] | None = None,
adjacency=None,
tags: list[str] | None = None,
tags_match: TagsMatch = "any",
tag_groups: list[TagGroup] | None = None,
) -> tuple[list[RetrievalResult], GraphRetrievalTimings | None]:
"""
Retrieve facts by expanding links from seeds.
Args:
pool: Database connection pool
query_embedding_str: Query embedding as string
bank_id: Memory bank ID
fact_type: Fact type to filter
budget: Maximum results to return
query_text: Original query text (unused)
semantic_seeds: Pre-computed semantic entry points
temporal_seeds: Pre-computed temporal entry points
adjacency: Unused, kept for interface compatibility
tags: Optional list of tags for visibility filtering
Returns:
Tuple of (results, timings)
"""
start_time = time.time()
timings = GraphRetrievalTimings(fact_type=fact_type)
async with acquire_with_retry(pool) as conn:
# Find seeds if not provided
if semantic_seeds:
all_seeds = list(semantic_seeds)
else:
seeds_start = time.time()
all_seeds = await _find_semantic_seeds(
conn,
query_embedding_str,
bank_id,
fact_type,
limit=20,
threshold=0.3,
tags=tags,
tags_match=tags_match,
tag_groups=tag_groups,
)
timings.seeds_time = time.time() - seeds_start
logger.debug(
f"[LinkExpansion] Found {len(all_seeds)} semantic seeds for fact_type={fact_type} "
f"(tags={tags}, tags_match={tags_match})"
)
if temporal_seeds:
all_seeds.extend(temporal_seeds)
if not all_seeds:
return [], timings
seed_ids = list({s.id for s in all_seeds})
timings.pattern_count = len(seed_ids)
query_start = time.time()
if fact_type == "observation":
entity_rows, semantic_rows, causal_rows = await self._expand_observations(conn, seed_ids, budget)
else:
entity_rows, semantic_rows, causal_rows = await self._expand_combined(conn, seed_ids, fact_type, budget)
timings.edge_load_time = time.time() - query_start
timings.db_queries = 1
timings.edge_count = len(entity_rows) + len(semantic_rows) + len(causal_rows)
# Merge results with additive intra-score: entity + semantic + causal ∈ [0, 3].
#
# Entity score: tanh(count × 0.5) maps shared-entity count to [0, 1]:
# 1 entity → 0.46, 2 → 0.76, 3 → 0.91, 4 → 0.96 (saturates naturally)
# Semantic score: similarity weight, already ∈ [0.7, 1.0].
# Causal score: link weight, already ∈ [0, 1].
#
# Facts appearing in multiple signals accumulate higher scores, rewarding
# convergent evidence. The outer RRF uses rank position from this sorted list.
entity_scores: dict[str, float] = {}
semantic_scores: dict[str, float] = {}
causal_scores: dict[str, float] = {}
row_map: dict[str, dict] = {}
for row in entity_rows:
fact_id = str(row["id"])
entity_scores[fact_id] = math.tanh(row["score"] * 0.5)
row_map[fact_id] = dict(row)
for row in semantic_rows:
fact_id = str(row["id"])
semantic_scores[fact_id] = max(semantic_scores.get(fact_id, 0.0), row["score"])
row_map.setdefault(fact_id, dict(row))
for row in causal_rows:
fact_id = str(row["id"])
causal_scores[fact_id] = max(causal_scores.get(fact_id, 0.0), row["score"])
row_map.setdefault(fact_id, dict(row))
all_ids = set(entity_scores) | set(semantic_scores) | set(causal_scores)
score_map = {
fid: entity_scores.get(fid, 0.0) + semantic_scores.get(fid, 0.0) + causal_scores.get(fid, 0.0)
for fid in all_ids
}
sorted_ids = sorted(score_map.keys(), key=lambda x: score_map[x], reverse=True)[:budget]
rows = [row_map[fact_id] for fact_id in sorted_ids]
results = []
for row in rows:
result = RetrievalResult.from_db_row(dict(row))
result.activation = row["score"]
results.append(result)
if tags:
results = filter_results_by_tags(results, tags, match=tags_match)
if tag_groups:
results = filter_results_by_tag_groups(results, tag_groups)
timings.result_count = len(results)
timings.traverse = time.time() - start_time
logger.debug(
f"LinkExpansion: {len(results)} results from {len(seed_ids)} seeds "
f"in {timings.traverse * 1000:.1f}ms (query: {timings.edge_load_time * 1000:.1f}ms)"
)
return results, timings
async def _expand_combined(
self,
conn,
seed_ids: list,
fact_type: str,
budget: int,
) -> tuple[list, list, list]:
"""
Single-roundtrip CTE query combining entity, semantic, and causal expansions.
Uses a `source` discriminator column so the caller can apply per-signal
score transformations. The three CTEs share one connection slot — important
for asyncpg which does not allow concurrent queries on the same connection.
Index coverage (requires migration d2e3f4a5b6c7):
entity: idx_memory_links_entity_covering (from_unit_id) INCLUDE (to_unit_id, entity_id)
WHERE link_type = 'entity' → index-only scan, no heap reads
semantic incoming:
idx_memory_links_to_type_weight (to_unit_id, link_type, weight DESC)
→ replaces costly BitmapAnd of two separate scans
"""
config = get_config()
ml = fq_table("memory_links")
mu = fq_table("memory_units")
ue = fq_table("unit_entities")
per_entity_limit = config.link_expansion_per_entity_limit
# Entity CTE with LATERAL fanout cap.
# Every seed entity (including high-frequency ones) is kept, but each
# entity's expansion is capped to per_entity_limit target units. The
# LATERAL subquery orders by unit_id DESC so the most recently inserted
# units are preferred (a recency proxy that is free — it rides the PK
# index with no extra sort).
entity_cte = f"""
seed_entities AS (
SELECT DISTINCT ue.entity_id
FROM {ue} ue
WHERE ue.unit_id = ANY($1::uuid[])
),
entity_expanded AS (
SELECT mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at,
mu.fact_type, mu.document_id, mu.chunk_id, mu.tags, mu.proof_count,
COUNT(DISTINCT se.entity_id)::float AS score,
'entity'::text AS source
FROM seed_entities se
CROSS JOIN LATERAL (
SELECT ue_target.unit_id
FROM {ue} ue_target
WHERE ue_target.entity_id = se.entity_id
AND ue_target.unit_id != ALL($1::uuid[])
ORDER BY ue_target.unit_id DESC
LIMIT {per_entity_limit}
) t
JOIN {mu} mu ON mu.id = t.unit_id
WHERE mu.fact_type = $2
GROUP BY mu.id
ORDER BY score DESC
LIMIT $3
)"""
semantic_causal_cte = f"""
semantic_expanded AS (
-- Semantic kNN: both outgoing (seeds → their kNN at insert time) and
-- incoming (facts inserted after seeds that found seeds as kNN).
-- Score = max similarity weight across both directions.
SELECT
id, text, context, event_date, occurred_start,
occurred_end, mentioned_at,
fact_type, document_id, chunk_id, tags, proof_count,
MAX(weight) AS score,
'semantic'::text AS source
FROM (
SELECT
mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at,
mu.fact_type, mu.document_id, mu.chunk_id, mu.tags, mu.proof_count,
ml.weight
FROM {ml} ml
JOIN {mu} mu ON mu.id = ml.to_unit_id
WHERE ml.from_unit_id = ANY($1::uuid[])
AND ml.link_type = 'semantic'
AND mu.fact_type = $2
AND mu.id != ALL($1::uuid[])
UNION ALL
SELECT
mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at,
mu.fact_type, mu.document_id, mu.chunk_id, mu.tags, mu.proof_count,
ml.weight
FROM {ml} ml
JOIN {mu} mu ON mu.id = ml.from_unit_id
WHERE ml.to_unit_id = ANY($1::uuid[])
AND ml.link_type = 'semantic'
AND mu.fact_type = $2
AND mu.id != ALL($1::uuid[])
) sem_raw
GROUP BY id, text, context, event_date, occurred_start,
occurred_end, mentioned_at,
fact_type, document_id, chunk_id, tags, proof_count
ORDER BY score DESC
LIMIT $3
),
causal_expanded AS (
-- Causal chains: explicit causes/enables/prevents links from seeds.
-- DISTINCT ON handles the case where a seed has multiple causal links
-- to the same target; best weight wins.
SELECT DISTINCT ON (mu.id)
mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at,
mu.fact_type, mu.document_id, mu.chunk_id, mu.tags, mu.proof_count,
ml.weight AS score,
'causal'::text AS source
FROM {ml} ml
JOIN {mu} mu ON ml.to_unit_id = mu.id
WHERE ml.from_unit_id = ANY($1::uuid[])
AND ml.link_type IN ('causes', 'caused_by', 'enables', 'prevents')
AND ml.weight >= $4
AND mu.fact_type = $2
ORDER BY mu.id, ml.weight DESC
LIMIT $3
)"""
full_query = f"""
WITH {entity_cte},
{semantic_causal_cte}
SELECT * FROM entity_expanded
UNION ALL
SELECT * FROM semantic_expanded
UNION ALL
SELECT * FROM causal_expanded
"""
params = [seed_ids, fact_type, budget, self.causal_weight_threshold]
try:
all_rows = await asyncio.wait_for(
conn.fetch(full_query, *params),
timeout=config.link_expansion_timeout,
)
except asyncio.TimeoutError:
logger.warning(
f"[LinkExpansion] Entity expansion timed out after {config.link_expansion_timeout}s "
f"for fact_type={fact_type}, falling back to semantic+causal only"
)
fallback_query = f"""
WITH {semantic_causal_cte}
SELECT * FROM semantic_expanded
UNION ALL
SELECT * FROM causal_expanded
"""
all_rows = await conn.fetch(fallback_query, *params)
entity_rows = [r for r in all_rows if r["source"] == "entity"]
semantic_rows = [r for r in all_rows if r["source"] == "semantic"]
causal_rows = [r for r in all_rows if r["source"] == "causal"]
return entity_rows, semantic_rows, causal_rows
async def _expand_observations(
self,
conn,
seed_ids: list,
budget: int,
) -> tuple[list, list, list]:
"""
Observation-specific expansion.
Observations don't have direct entity links in memory_links (they're created
by consolidation, not retain). Instead, traverse source_memory_ids → world
facts → entities → other world facts → their observations.
Semantic and causal expansions run as a second combined CTE query.
"""
source_ids_found: list = []
if logger.isEnabledFor(logging.DEBUG):
debug_rows = await conn.fetch(
f"""
SELECT id, source_memory_ids
FROM {fq_table("memory_units")}
WHERE id = ANY($1::uuid[])
""",
seed_ids,
)
for row in debug_rows:
if row["source_memory_ids"]:
source_ids_found.extend(row["source_memory_ids"])
logger.debug(
f"[LinkExpansion] observation graph: {len(seed_ids)} seeds, "
f"{len(source_ids_found)} source_memory_ids found"
)
config = get_config()
ue = fq_table("unit_entities")
per_entity_limit = config.link_expansion_per_entity_limit
connected_sources_cte = f"""
source_entities AS (
SELECT DISTINCT ue_seed.entity_id
FROM seed_sources ss
JOIN {ue} ue_seed ON ue_seed.unit_id = ss.source_id
),
connected_sources AS (
-- Find sources sharing entities with seed observation sources
-- via LATERAL-capped self-join (prevents hub entity fanout).
SELECT DISTINCT t.unit_id AS source_id
FROM source_entities se
CROSS JOIN LATERAL (
SELECT ue_target.unit_id
FROM {ue} ue_target
WHERE ue_target.entity_id = se.entity_id
ORDER BY ue_target.unit_id DESC
LIMIT {per_entity_limit}
) t
WHERE NOT EXISTS (
SELECT 1 FROM seed_sources ss WHERE ss.source_id = t.unit_id
)
)"""
entity_rows = await conn.fetch(
f"""
WITH seed_sources AS (
SELECT DISTINCT unnest(source_memory_ids) AS source_id
FROM {fq_table("memory_units")}
WHERE id = ANY($1::uuid[])
AND source_memory_ids IS NOT NULL
),
{connected_sources_cte},
connected_array AS (
SELECT array_agg(source_id) AS source_ids FROM connected_sources
)
SELECT
mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at,
mu.fact_type, mu.document_id, mu.chunk_id, mu.tags, mu.proof_count,
(SELECT COUNT(DISTINCT s) FROM unnest(mu.source_memory_ids) s WHERE s = ANY(ca.source_ids))::float AS score
FROM {fq_table("memory_units")} mu, connected_array ca
WHERE mu.fact_type = 'observation'
AND mu.id != ALL($1::uuid[])
AND ca.source_ids IS NOT NULL
AND mu.source_memory_ids && ca.source_ids
ORDER BY score DESC
LIMIT $2
""",
seed_ids,
budget,
)
logger.debug(f"[LinkExpansion] observation graph: found {len(entity_rows)} connected observations")
# Semantic + causal for observations in one query
ml = fq_table("memory_links")
mu = fq_table("memory_units")
sem_causal_rows = await conn.fetch(
f"""
WITH semantic_expanded AS (
SELECT
id, text, context, event_date, occurred_start,
occurred_end, mentioned_at,
fact_type, document_id, chunk_id, tags, proof_count,
MAX(weight) AS score,
'semantic'::text AS source
FROM (
SELECT mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at, mu.fact_type, mu.document_id,
mu.chunk_id, mu.tags, mu.proof_count, ml.weight
FROM {ml} ml JOIN {mu} mu ON mu.id = ml.to_unit_id
WHERE ml.from_unit_id = ANY($1::uuid[])
AND ml.link_type = 'semantic' AND mu.fact_type = 'observation'
AND mu.id != ALL($1::uuid[])
UNION ALL
SELECT mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at, mu.fact_type, mu.document_id,
mu.chunk_id, mu.tags, mu.proof_count, ml.weight
FROM {ml} ml JOIN {mu} mu ON mu.id = ml.from_unit_id
WHERE ml.to_unit_id = ANY($1::uuid[])
AND ml.link_type = 'semantic' AND mu.fact_type = 'observation'
AND mu.id != ALL($1::uuid[])
) sem_raw
GROUP BY id, text, context, event_date, occurred_start, occurred_end,
mentioned_at, fact_type, document_id, chunk_id, tags, proof_count
ORDER BY score DESC LIMIT $2
),
causal_expanded AS (
SELECT DISTINCT ON (mu.id)
mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
mu.occurred_end, mu.mentioned_at, mu.fact_type, mu.document_id,
mu.chunk_id, mu.tags, mu.proof_count, ml.weight AS score, 'causal'::text AS source
FROM {ml} ml JOIN {mu} mu ON ml.to_unit_id = mu.id
WHERE ml.from_unit_id = ANY($1::uuid[])
AND ml.link_type IN ('causes', 'caused_by', 'enables', 'prevents')
AND ml.weight >= $3 AND mu.fact_type = 'observation'
ORDER BY mu.id, ml.weight DESC LIMIT $2
)
SELECT * FROM semantic_expanded
UNION ALL
SELECT * FROM causal_expanded
""",
seed_ids,
budget,
self.causal_weight_threshold,
)
semantic_rows = [r for r in sem_causal_rows if r["source"] == "semantic"]
causal_rows = [r for r in sem_causal_rows if r["source"] == "causal"]
return entity_rows, semantic_rows, causal_rows
@@ -1,249 +0,0 @@
"""
Cross-encoder neural reranking for search results.
"""
import math
from datetime import datetime, timezone
from .types import MergedCandidate, ScoredResult
UTC = timezone.utc
# Multiplicative boost alphas for recency and temporal proximity.
# Each signal contributes at most ±(alpha/2) relative adjustment to the base CE score,
# so the max combined boost is (1 + alpha/2)^2 ≈ +21% and min is (1 - alpha/2)^2 ≈ -19%.
_RECENCY_ALPHA: float = 0.2
_TEMPORAL_ALPHA: float = 0.2
_PROOF_COUNT_ALPHA: float = 0.1 # Conservative: max ±5% for evidence strength
def apply_combined_scoring(
scored_results: list[ScoredResult],
now: datetime,
recency_alpha: float = _RECENCY_ALPHA,
temporal_alpha: float = _TEMPORAL_ALPHA,
proof_count_alpha: float = _PROOF_COUNT_ALPHA,
is_passthrough_reranker: bool = False,
) -> None:
"""Apply combined scoring to a list of ScoredResults in-place.
Uses the cross-encoder score as the primary relevance signal, with recency,
temporal proximity, and proof count applied as multiplicative boosts. This
ensures the influence of these secondary signals is always proportional to
the base relevance score, regardless of the cross-encoder model's score
calibration.
Formula::
recency_boost = 1 + recency_alpha * (recency - 0.5) # in [1-α/2, 1+α/2]
temporal_boost = 1 + temporal_alpha * (temporal - 0.5) # in [1-α/2, 1+α/2]
proof_count_boost = 1 + proof_count_alpha * (proof_norm - 0.5) # in [1-α/2, 1+α/2]
combined_score = CE_normalized * recency_boost * temporal_boost * proof_count_boost
proof_norm maps proof_count using a smooth logarithmic curve centered at 0.5,
clamped to [0, 1]:
proof_count=1 → 0.5 + 0 = 0.5 (neutral multiplier)
proof_count=150 → clamped to 1.0 (max +5% boost)
Temporal proximity is treated as neutral (0.5) when not set by temporal retrieval,
so temporal_boost collapses to 1.0 for non-temporal queries.
Proof count is treated as neutral (0.5) when not available (non-observation facts),
so proof_count_boost collapses to 1.0 for world/experience/opinion facts.
Args:
scored_results: Results from the cross-encoder reranker. Mutated in place.
now: Current UTC datetime for recency calculation.
recency_alpha: Max relative recency adjustment (default 0.2 → ±10%).
temporal_alpha: Max relative temporal adjustment (default 0.2 → ±10%).
proof_count_alpha: Max relative proof count adjustment (default 0.1 → ±5%).
"""
if now.tzinfo is None:
now = now.replace(tzinfo=UTC)
# When the configured cross-encoder is a passthrough (e.g.
# RRFPassthroughCrossEncoder used by slim deployments), every
# cross_encoder_score_normalized is identical and provides no relevance
# signal. In that case the multiplicative recency / temporal / proof_count
# boosts below become the *only* ranking signal — making the final order a
# pure recency sort regardless of how relevant a candidate actually is.
#
# Detect that case and seed cross_encoder_score_normalized from the RRF
# rank instead, so the boosts modulate a meaningful base score rather than
# replacing it. This is a no-op for real cross-encoders, which produce
# diverse scores.
# When the reranker is a passthrough (e.g. RRFPassthroughCrossEncoder used
# by slim deployments), every cross_encoder_score_normalized is identical
# and provides no relevance signal. The multiplicative recency / temporal /
# proof_count boosts below would then become the *only* ranking signal,
# making the final order a pure recency sort regardless of how relevant a
# candidate actually is.
#
# Seed cross_encoder_score_normalized from the RRF rank instead, so the
# boosts modulate a meaningful base score. Caller passes is_passthrough
# explicitly because "all scores identical" is too fragile a heuristic —
# a real reranker can also tie scores (especially in tests with synthetic
# data) and we'd corrupt legitimate single-result reranks.
if is_passthrough_reranker and scored_results:
n = len(scored_results)
sorted_by_rrf = sorted(
scored_results,
key=lambda s: getattr(getattr(s, "candidate", None), "rrf_score", 0.0),
reverse=True,
)
denom = max(1, n - 1)
for new_rank, sr in enumerate(sorted_by_rrf):
# Map rank → [0.1, 1.0] so the recency boost can still nudge
# ordering between adjacent candidates without overpowering RRF.
sr.cross_encoder_score_normalized = 1.0 - (0.9 * new_rank / denom)
for sr in scored_results:
# Recency: linear decay over 365 days → [0.1, 1.0]; neutral 0.5 if no date.
sr.recency = 0.5
if sr.retrieval.occurred_start:
occurred = sr.retrieval.occurred_start
if occurred.tzinfo is None:
occurred = occurred.replace(tzinfo=UTC)
days_ago = (now - occurred).total_seconds() / 86400
sr.recency = max(0.1, min(1.0, 1.0 - (days_ago / 365)))
# Temporal proximity: meaningful only for temporal queries; neutral otherwise.
sr.temporal = sr.retrieval.temporal_proximity if sr.retrieval.temporal_proximity is not None else 0.5
# Proof count: log-normalized evidence strength; neutral for non-observations.
proof_count = sr.retrieval.proof_count
if proof_count is not None and proof_count >= 1:
# Clamp to [0, 1] so extreme counts stay within documented ±5% range
proof_norm = min(1.0, max(0.0, 0.5 + (math.log(proof_count) / 10.0)))
else:
# Neutral baseline is precisely 0.5, ensuring neutral multiplier (1.0)
proof_norm = 0.5
# RRF: kept at 0.0 for trace continuity but excluded from scoring.
# RRF is batch-relative (min-max normalised) and redundant after reranking.
sr.rrf_normalized = 0.0
recency_boost = 1.0 + recency_alpha * (sr.recency - 0.5)
temporal_boost = 1.0 + temporal_alpha * (sr.temporal - 0.5)
proof_count_boost = 1.0 + proof_count_alpha * (proof_norm - 0.5)
sr.combined_score = sr.cross_encoder_score_normalized * recency_boost * temporal_boost * proof_count_boost
sr.weight = sr.combined_score
class CrossEncoderReranker:
"""
Neural reranking using a cross-encoder model.
Configured via environment variables (see cross_encoder.py).
Default local model is cross-encoder/ms-marco-MiniLM-L-6-v2.
"""
def __init__(self, cross_encoder=None):
"""
Initialize cross-encoder reranker.
Args:
cross_encoder: CrossEncoderModel instance. If None, creates one from
environment variables (defaults to local provider)
"""
if cross_encoder is None:
from hindsight_api.engine.cross_encoder import create_cross_encoder_from_env
cross_encoder = create_cross_encoder_from_env()
self.cross_encoder = cross_encoder
self._initialized = False
async def ensure_initialized(self):
"""Ensure the cross-encoder model is initialized (for lazy initialization)."""
if self._initialized:
return
import asyncio
cross_encoder = self.cross_encoder
# For local providers, run in thread pool to avoid blocking event loop
if cross_encoder.provider_name == "local":
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, lambda: asyncio.run(cross_encoder.initialize()))
else:
await cross_encoder.initialize()
self._initialized = True
async def rerank(self, query: str, candidates: list[MergedCandidate]) -> list[ScoredResult]:
"""
Rerank candidates using cross-encoder scores.
Args:
query: Search query
candidates: Merged candidates from RRF
Returns:
List of ScoredResult objects sorted by cross-encoder score
"""
if not candidates:
return []
# Prepare query-document pairs with date information
pairs = []
for candidate in candidates:
retrieval = candidate.retrieval
# Use text + context for better ranking
doc_text = retrieval.text
if retrieval.context:
doc_text = f"{retrieval.context}: {doc_text}"
# Add formatted date information for temporal awareness
if retrieval.occurred_start:
occurred_start = retrieval.occurred_start
# Format in two styles for better model understanding
# 1. ISO format: YYYY-MM-DD
date_iso = occurred_start.strftime("%Y-%m-%d")
# 2. Human-readable: "June 5, 2022"
date_readable = occurred_start.strftime("%B %d, %Y")
# Prepend date to document text
doc_text = f"[Date: {date_readable} ({date_iso})] {doc_text}"
pairs.append([query, doc_text])
# Get cross-encoder scores
scores = await self.cross_encoder.predict(pairs)
# Normalize scores using sigmoid to [0, 1] range
# Cross-encoder returns logits which can be negative
import math
import numpy as np
def sigmoid(x):
return 1 / (1 + np.exp(-x))
normalized_scores = [sigmoid(score) for score in scores]
# Create ScoredResult objects with cross-encoder scores
scored_results = []
for candidate, raw_score, norm_score in zip(candidates, scores, normalized_scores):
# Sanitize NaN scores (cross-encoder can return NaN for certain inputs).
# NaN propagates through all downstream scoring and Pydantic serializes
# NaN as JSON null, which breaks clients expecting numeric values.
raw = float(raw_score)
norm = float(norm_score)
if math.isnan(raw):
raw = 0.0
if math.isnan(norm):
norm = 0.0
scored_result = ScoredResult(
candidate=candidate,
cross_encoder_score=raw,
cross_encoder_score_normalized=norm,
weight=norm, # Initial weight is just cross-encoder score
)
scored_results.append(scored_result)
# Sort by cross-encoder score
scored_results.sort(key=lambda x: x.weight, reverse=True)
return scored_results
@@ -1,390 +0,0 @@
"""
Tags filtering utilities for retrieval.
Provides SQL building functions for filtering memories by tags.
Supports four matching modes via TagsMatch enum:
- "any": OR matching, includes untagged memories (default, backward compatible)
- "all": AND matching, includes untagged memories
- "any_strict": OR matching, excludes untagged memories
- "all_strict": AND matching, excludes untagged memories
OR matching (any/any_strict): Memory matches if ANY of its tags overlap with request tags
AND matching (all/all_strict): Memory matches if ALL request tags are present in its tags
"""
from __future__ import annotations
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field
TagsMatch = Literal["any", "all", "any_strict", "all_strict"]
def _parse_tags_match(match: TagsMatch) -> tuple[str, bool]:
"""
Parse TagsMatch into operator and include_untagged flag.
Returns:
Tuple of (operator, include_untagged)
- operator: "&&" for any/any_strict, "@>" for all/all_strict
- include_untagged: True for any/all, False for any_strict/all_strict
"""
if match == "any":
return "&&", True
elif match == "all":
return "@>", True
elif match == "any_strict":
return "&&", False
elif match == "all_strict":
return "@>", False
else:
# Default to "any" behavior
return "&&", True
def build_tags_where_clause(
tags: list[str] | None,
param_offset: int = 1,
table_alias: str = "",
match: TagsMatch = "any",
) -> tuple[str, list, int]:
"""
Build a SQL WHERE clause for filtering by tags.
Supports four matching modes:
- "any" (default): OR matching, includes untagged memories
- "all": AND matching, includes untagged memories
- "any_strict": OR matching, excludes untagged memories
- "all_strict": AND matching, excludes untagged memories
Args:
tags: List of tags to filter by. If None or empty, returns empty clause (no filtering).
param_offset: Starting parameter number for SQL placeholders (default 1).
table_alias: Optional table alias prefix (e.g., "mu." for "memory_units mu").
match: Matching mode. Defaults to "any".
Returns:
Tuple of (sql_clause, params, next_param_offset):
- sql_clause: SQL WHERE clause string
- params: List of parameter values to bind
- next_param_offset: Next available parameter number
Example:
>>> clause, params, next_offset = build_tags_where_clause(['user_a'], 3, 'mu.', 'any_strict')
>>> print(clause) # "AND mu.tags IS NOT NULL AND mu.tags != '{}' AND mu.tags && $3"
"""
if not tags:
return "", [], param_offset
column = f"{table_alias}tags" if table_alias else "tags"
operator, include_untagged = _parse_tags_match(match)
if include_untagged:
# Include untagged memories (NULL or empty array) OR matching tags
clause = f"AND ({column} IS NULL OR {column} = '{{}}' OR {column} {operator} ${param_offset})"
else:
# Strict: only memories with matching tags (exclude NULL and empty)
clause = f"AND {column} IS NOT NULL AND {column} != '{{}}' AND {column} {operator} ${param_offset}"
return clause, [tags], param_offset + 1
def build_tags_where_clause_simple(
tags: list[str] | None,
param_num: int,
table_alias: str = "",
match: TagsMatch = "any",
) -> str:
"""
Build a simple SQL WHERE clause for tags filtering.
This is a convenience version that returns just the clause string,
assuming the caller will add the tags array to their params list.
Args:
tags: List of tags to filter by. If None or empty, returns empty string.
param_num: Parameter number to use in the clause.
table_alias: Optional table alias prefix.
match: Matching mode. Defaults to "any".
Returns:
SQL clause string or empty string.
"""
if not tags:
return ""
column = f"{table_alias}tags" if table_alias else "tags"
operator, include_untagged = _parse_tags_match(match)
if include_untagged:
# Include untagged memories (NULL or empty array) OR matching tags
return f"AND ({column} IS NULL OR {column} = '{{}}' OR {column} {operator} ${param_num})"
else:
# Strict: only memories with matching tags (exclude NULL and empty)
return f"AND {column} IS NOT NULL AND {column} != '{{}}' AND {column} {operator} ${param_num}"
def filter_results_by_tags(
results: list,
tags: list[str] | None,
match: TagsMatch = "any",
) -> list:
"""
Filter retrieval results by tags in Python (for post-processing).
Used when SQL filtering isn't possible (e.g., graph traversal results).
Args:
results: List of RetrievalResult objects with a 'tags' attribute.
tags: List of tags to filter by. If None or empty, returns all results.
match: Matching mode. Defaults to "any".
Returns:
Filtered list of results.
"""
if not tags:
return results
_, include_untagged = _parse_tags_match(match)
is_any_match = match in ("any", "any_strict")
tags_set = set(tags)
filtered = []
for result in results:
result_tags = getattr(result, "tags", None)
# Check if untagged
is_untagged = result_tags is None or len(result_tags) == 0
if is_untagged:
if include_untagged:
filtered.append(result)
# else: skip untagged
else:
result_tags_set = set(result_tags)
if is_any_match:
# Any overlap
if result_tags_set & tags_set:
filtered.append(result)
else:
# All tags must be present
if tags_set <= result_tags_set:
filtered.append(result)
return filtered
# =============================================================================
# Compound tag group models (recursive boolean expressions)
# =============================================================================
class TagGroupLeaf(BaseModel):
"""A leaf tag filter: matches memories by tag list and match mode."""
tags: list[str]
match: TagsMatch = "any_strict"
class TagGroupAnd(BaseModel):
"""Compound AND group: all child filters must match."""
model_config = ConfigDict(populate_by_name=True)
filters: list[TagGroup] = Field(alias="and")
class TagGroupOr(BaseModel):
"""Compound OR group: at least one child filter must match."""
model_config = ConfigDict(populate_by_name=True)
filters: list[TagGroup] = Field(alias="or")
class TagGroupNot(BaseModel):
"""Compound NOT group: child filter must NOT match."""
model_config = ConfigDict(populate_by_name=True)
filter: TagGroup = Field(alias="not")
# TagGroup is a discriminated union; Pydantic will try left-to-right.
# TagGroupLeaf is identified by the presence of 'tags'.
# TagGroupAnd / TagGroupOr / TagGroupNot are compound (no 'tags' key).
TagGroup = Annotated[
TagGroupLeaf | TagGroupAnd | TagGroupOr | TagGroupNot,
Field(union_mode="left_to_right"),
]
# Rebuild forward-reference models so recursive TagGroup is resolved.
TagGroupAnd.model_rebuild()
TagGroupOr.model_rebuild()
TagGroupNot.model_rebuild()
# =============================================================================
# SQL builder for compound tag groups
# =============================================================================
def _build_group_clause(
group: TagGroup,
param_offset: int,
table_alias: str,
) -> tuple[str, list, int]:
"""
Recursively build an inner SQL clause (no leading AND/OR) for a single TagGroup.
Returns:
(inner_clause, params, next_param_offset)
"""
if isinstance(group, TagGroupLeaf):
column = f"{table_alias}tags" if table_alias else "tags"
operator, include_untagged = _parse_tags_match(group.match)
if include_untagged:
clause = f"({column} IS NULL OR {column} = '{{}}' OR {column} {operator} ${param_offset})"
else:
clause = f"({column} IS NOT NULL AND {column} != '{{}}' AND {column} {operator} ${param_offset})"
return clause, [group.tags], param_offset + 1
elif isinstance(group, TagGroupAnd):
parts = []
params: list = []
offset = param_offset
for child in group.filters:
child_clause, child_params, offset = _build_group_clause(child, offset, table_alias)
parts.append(child_clause)
params.extend(child_params)
inner = " AND ".join(parts)
return f"({inner})", params, offset
elif isinstance(group, TagGroupOr):
parts = []
params = []
offset = param_offset
for child in group.filters:
child_clause, child_params, offset = _build_group_clause(child, offset, table_alias)
parts.append(child_clause)
params.extend(child_params)
inner = " OR ".join(parts)
return f"({inner})", params, offset
elif isinstance(group, TagGroupNot):
child_clause, child_params, next_offset = _build_group_clause(group.filter, param_offset, table_alias)
return f"NOT {child_clause}", child_params, next_offset
else:
# Should never happen with proper Pydantic validation
return "", [], param_offset
def build_tag_groups_where_clause(
tag_groups: list[TagGroup] | None,
param_offset: int,
table_alias: str = "",
) -> tuple[str, list, int]:
"""
Build a SQL WHERE clause for compound tag group filtering.
Top-level groups are AND-ed together. Each group is a recursive boolean
expression (leaf, and, or, not).
Args:
tag_groups: List of TagGroup objects. If None or empty, returns empty clause.
param_offset: Starting parameter number for SQL placeholders.
table_alias: Optional table alias prefix (e.g., "mu." for "memory_units mu").
Returns:
Tuple of (sql_clause, params, next_param_offset):
- sql_clause: SQL WHERE clause string starting with "AND" (or empty string)
- params: List of parameter values to bind (one per leaf node)
- next_param_offset: Next available parameter number
Example:
>>> groups = [TagGroupLeaf(tags=["user:alice"], match="all_strict")]
>>> clause, params, next_offset = build_tag_groups_where_clause(groups, 3)
>>> print(clause) # "AND (tags IS NOT NULL AND tags != '{}' AND tags @> $3)"
"""
if not tag_groups:
return "", [], param_offset
all_params: list = []
all_clauses: list[str] = []
offset = param_offset
for group in tag_groups:
inner_clause, group_params, offset = _build_group_clause(group, offset, table_alias)
all_clauses.append(inner_clause)
all_params.extend(group_params)
combined = " AND ".join(all_clauses)
return f"AND {combined}", all_params, offset
# =============================================================================
# Python-side filter for compound tag groups (post-retrieval filtering)
# =============================================================================
def _match_group(result: object, group: TagGroup) -> bool:
"""
Recursively evaluate a TagGroup against a retrieval result.
Args:
result: Any object with a 'tags' attribute (list[str] or None).
group: The TagGroup to evaluate.
Returns:
True if the result matches the group, False otherwise.
"""
if isinstance(group, TagGroupLeaf):
result_tags = getattr(result, "tags", None)
is_untagged = result_tags is None or len(result_tags) == 0
_, include_untagged = _parse_tags_match(group.match)
is_any_match = group.match in ("any", "any_strict")
tags_set = set(group.tags)
if is_untagged:
return include_untagged
else:
result_tags_set = set(result_tags)
if is_any_match:
return bool(result_tags_set & tags_set)
else:
return tags_set <= result_tags_set
elif isinstance(group, TagGroupAnd):
return all(_match_group(result, child) for child in group.filters)
elif isinstance(group, TagGroupOr):
return any(_match_group(result, child) for child in group.filters)
elif isinstance(group, TagGroupNot):
return not _match_group(result, group.filter)
else:
return True
def filter_results_by_tag_groups(
results: list,
tag_groups: list[TagGroup] | None,
) -> list:
"""
Filter retrieval results by compound tag groups in Python (for post-processing).
Used when SQL filtering isn't possible (e.g., graph traversal results).
Top-level groups are AND-ed together.
Args:
results: List of RetrievalResult objects with a 'tags' attribute.
tag_groups: List of TagGroup objects. If None or empty, returns all results.
Returns:
Filtered list of results where ALL top-level groups match.
"""
if not tag_groups:
return results
return [r for r in results if all(_match_group(r, group) for group in tag_groups)]
@@ -1,79 +0,0 @@
"""File storage backends for uploaded files."""
from collections.abc import Callable
from .base import FileStorage
from .postgresql import PostgreSQLFileStorage
__all__ = ["FileStorage", "PostgreSQLFileStorage", "create_file_storage"]
def create_file_storage(
storage_type: str,
pool_getter: Callable | None = None,
schema: str | None = None,
schema_getter: Callable | None = None,
**kwargs,
) -> FileStorage:
"""
Create file storage backend based on configuration.
Args:
storage_type: "native" (PostgreSQL BYTEA) or "s3" (S3-compatible object storage)
pool_getter: Database pool getter (required for native)
schema: Static database schema (for native single-tenant)
schema_getter: Callable returning current schema at query time (for native multi-tenant)
**kwargs: Additional args passed to storage backend
Returns:
FileStorage instance
Raises:
ValueError: If storage_type is unknown or required args are missing
"""
if storage_type == "native":
if not pool_getter:
raise ValueError("pool_getter required for native (PostgreSQL) storage")
return PostgreSQLFileStorage(pool_getter=pool_getter, schema=schema, schema_getter=schema_getter)
elif storage_type == "s3":
from ...config import get_config
from .s3 import S3FileStorage
config = get_config()
bucket = config.file_storage_s3_bucket
if not bucket:
raise ValueError("HINDSIGHT_API_FILE_STORAGE_S3_BUCKET is required for S3 storage")
return S3FileStorage(
bucket=bucket,
region=config.file_storage_s3_region,
endpoint=config.file_storage_s3_endpoint,
access_key_id=config.file_storage_s3_access_key_id,
secret_access_key=config.file_storage_s3_secret_access_key,
)
elif storage_type == "gcs":
from ...config import get_config
from .gcs import GCSFileStorage
config = get_config()
bucket = config.file_storage_gcs_bucket
if not bucket:
raise ValueError("HINDSIGHT_API_FILE_STORAGE_GCS_BUCKET is required for GCS storage")
return GCSFileStorage(
bucket=bucket,
service_account_key=config.file_storage_gcs_service_account_key,
)
elif storage_type == "azure":
from ...config import get_config
from .azure import AzureFileStorage
config = get_config()
container = config.file_storage_azure_container
if not container:
raise ValueError("HINDSIGHT_API_FILE_STORAGE_AZURE_CONTAINER is required for Azure storage")
return AzureFileStorage(
container_name=container,
account_name=config.file_storage_azure_account_name,
account_key=config.file_storage_azure_account_key,
)
else:
raise ValueError(f"Unknown storage type: {storage_type}. Supported: 'native', 's3', 'gcs', 'azure'.")
@@ -1,62 +0,0 @@
"""Azure Blob Storage backend using obstore."""
import logging
from datetime import timedelta
import obstore as obs
from obstore.store import AzureStore
from .base import FileStorage
logger = logging.getLogger(__name__)
class AzureFileStorage(FileStorage):
"""
Azure Blob Storage backend.
Uses obstore (Rust-backed) for high-throughput async access to Azure Blob Storage.
Supports account key, SAS token, and default Azure credentials.
"""
def __init__(
self,
container_name: str,
account_name: str | None = None,
account_key: str | None = None,
):
kwargs: dict = {}
if account_name:
kwargs["account_name"] = account_name
if account_key:
kwargs["account_key"] = account_key
self._store = AzureStore(container_name, **kwargs)
logger.info(f"Initialized Azure file storage: container={container_name}, account={account_name}")
async def store(self, file_data: bytes, key: str, metadata: dict[str, str] | None = None) -> str:
await obs.put_async(self._store, key, file_data)
logger.debug(f"Stored file {key} ({len(file_data)} bytes) in Azure")
return key
async def retrieve(self, key: str) -> bytes:
try:
response = await obs.get_async(self._store, key)
return await response.bytes_async()
except Exception as e:
if "not found" in str(e).lower() or "BlobNotFound" in str(e):
raise FileNotFoundError(f"File not found: {key}") from e
raise
async def delete(self, key: str) -> None:
await obs.delete_async(self._store, key)
async def exists(self, key: str) -> bool:
try:
await obs.head_async(self._store, key)
return True
except Exception:
return False
async def get_download_url(self, key: str, expires_in: int = 3600) -> str:
return await obs.sign_async(self._store, "GET", key, timedelta(seconds=expires_in))
@@ -1,83 +0,0 @@
"""Abstract base class for file storage backends."""
from abc import ABC, abstractmethod
class FileStorage(ABC):
"""Abstract base for file storage backends."""
@abstractmethod
async def store(
self,
file_data: bytes,
key: str,
metadata: dict[str, str] | None = None,
) -> str:
"""
Store file and return storage key.
Args:
file_data: Raw file bytes
key: Storage key (e.g., "banks/{bank_id}/files/{file_id}.pdf")
metadata: Optional metadata to store with file
Returns:
Storage key that can be used to retrieve the file
"""
pass
@abstractmethod
async def retrieve(self, key: str) -> bytes:
"""
Retrieve file by storage key.
Args:
key: Storage key
Returns:
File data as bytes
Raises:
FileNotFoundError: If file does not exist
"""
pass
@abstractmethod
async def delete(self, key: str) -> None:
"""
Delete file by storage key.
Args:
key: Storage key
"""
pass
@abstractmethod
async def exists(self, key: str) -> bool:
"""
Check if file exists.
Args:
key: Storage key
Returns:
True if file exists, False otherwise
"""
pass
@abstractmethod
async def get_download_url(self, key: str, expires_in: int = 3600) -> str:
"""
Get a URL for downloading the file.
For PostgreSQL storage, this might be a relative API path.
For S3, this would be a pre-signed URL.
Args:
key: Storage key
expires_in: Expiration time in seconds (may be ignored for some backends)
Returns:
Download URL or path
"""
pass
@@ -1,105 +0,0 @@
"""Google Cloud Storage backend using obstore."""
import logging
import os
from datetime import datetime, timedelta, timezone
import obstore as obs
from obstore.store import GCSStore
from .base import FileStorage
logger = logging.getLogger(__name__)
def _make_google_auth_credential_provider():
"""Create a credential provider using google.auth (supports all credential types).
obstore's built-in credential parsing only supports service_account and
authorized_user JSON types. This provider uses the google-auth library
which additionally handles external_account (Workload Identity Federation),
impersonated credentials, and metadata-server credentials.
"""
import google.auth
import google.auth.transport.requests
credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
request = google.auth.transport.requests.Request()
def _provide():
credentials.refresh(request)
expiry = credentials.expiry
if expiry and expiry.tzinfo is None:
expiry = expiry.replace(tzinfo=timezone.utc)
return {"token": credentials.token, "expires_at": expiry}
return _provide
class GCSFileStorage(FileStorage):
"""
Google Cloud Storage backend.
Uses obstore (Rust-backed) for high-throughput async access to GCS.
Supports Application Default Credentials, service account keys, and explicit credentials.
"""
def __init__(
self,
bucket: str,
service_account_key: str | None = None,
):
kwargs: dict = {}
if service_account_key:
kwargs["service_account_key"] = service_account_key
else:
# Use google.auth credential provider for broad credential type support
# (service_account, authorized_user, external_account, metadata server, etc.)
try:
kwargs["credential_provider"] = _make_google_auth_credential_provider()
logger.info("Using google.auth credential provider for GCS")
except Exception as e:
logger.warning(
f"Failed to create google.auth credential provider, falling back to obstore defaults: {e}"
)
# Workaround for https://github.com/developmentseed/obstore/issues/605
# obstore's Rust layer doesn't support external_account credentials (Workload
# Identity Federation) and eagerly parses GOOGLE_APPLICATION_CREDENTIALS even
# when credential_provider is given. Per the obstore maintainer's guidance,
# remove env vars so the Rust code doesn't try to authenticate itself.
# google.auth (used by credential_provider above) has already loaded credentials.
gac = os.environ.pop("GOOGLE_APPLICATION_CREDENTIALS", None)
try:
self._store = GCSStore(bucket, **kwargs)
finally:
if gac is not None:
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = gac
logger.info(f"Initialized GCS file storage: bucket={bucket}")
async def store(self, file_data: bytes, key: str, metadata: dict[str, str] | None = None) -> str:
await obs.put_async(self._store, key, file_data)
logger.debug(f"Stored file {key} ({len(file_data)} bytes) in GCS")
return key
async def retrieve(self, key: str) -> bytes:
try:
response = await obs.get_async(self._store, key)
return await response.bytes_async()
except Exception as e:
if "not found" in str(e).lower():
raise FileNotFoundError(f"File not found: {key}") from e
raise
async def delete(self, key: str) -> None:
await obs.delete_async(self._store, key)
async def exists(self, key: str) -> bool:
try:
await obs.head_async(self._store, key)
return True
except Exception:
return False
async def get_download_url(self, key: str, expires_in: int = 3600) -> str:
return await obs.sign_async(self._store, "GET", key, timedelta(seconds=expires_in))
@@ -1,153 +0,0 @@
"""PostgreSQL BYTEA-based file storage (default, zero-config)."""
import logging
from collections.abc import Callable
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import asyncpg
from .base import FileStorage
logger = logging.getLogger(__name__)
def fq_table(table: str, schema: str | None = None) -> str:
"""Get fully-qualified table name with optional schema prefix."""
if schema:
return f'"{schema}".{table}'
return table
class PostgreSQLFileStorage(FileStorage):
"""
PostgreSQL BYTEA-based file storage.
Stores files directly in PostgreSQL using BYTEA columns.
This is the default storage backend - zero configuration required!
Pros:
- Works out of the box (no external dependencies)
- Transactional consistency with database
- Simple backups (included in pg_dump)
- Good performance for <10MB files
Cons:
- Database bloat for large/many files
- Not ideal for distributed deployments
- Higher cost than object storage at scale
For production/scale, consider S3FileStorage instead.
"""
def __init__(
self,
pool_getter: Callable[[], "asyncpg.Pool"],
schema: str | None = None,
schema_getter: Callable[[], str] | None = None,
):
"""
Initialize PostgreSQL file storage.
Args:
pool_getter: Function that returns asyncpg connection pool
schema: Static database schema (fallback for single-tenant / tests)
schema_getter: Callable returning current schema at query time (for multi-tenant)
"""
self._pool_getter = pool_getter
self._static_schema = schema
self._schema_getter = schema_getter
@property
def _schema(self) -> str | None:
"""Resolve schema dynamically per-request when schema_getter is provided."""
if self._schema_getter:
return self._schema_getter()
return self._static_schema
async def store(
self,
file_data: bytes,
key: str,
metadata: dict[str, str] | None = None,
) -> str:
"""Store file in PostgreSQL."""
pool = self._pool_getter()
async with pool.acquire() as conn:
await conn.execute(
f"""
INSERT INTO {fq_table("file_storage", self._schema)}
(storage_key, data)
VALUES ($1, $2)
ON CONFLICT (storage_key) DO UPDATE SET
data = EXCLUDED.data
""",
key,
file_data,
)
logger.debug(f"Stored file {key} ({len(file_data)} bytes) in PostgreSQL")
return key
async def retrieve(self, key: str) -> bytes:
"""Retrieve file from PostgreSQL."""
pool = self._pool_getter()
async with pool.acquire() as conn:
row = await conn.fetchrow(
f"""
SELECT data FROM {fq_table("file_storage", self._schema)}
WHERE storage_key = $1
""",
key,
)
if not row:
raise FileNotFoundError(f"File not found: {key}")
return bytes(row["data"])
async def delete(self, key: str) -> None:
"""Delete file from PostgreSQL."""
pool = self._pool_getter()
async with pool.acquire() as conn:
result = await conn.execute(
f"""
DELETE FROM {fq_table("file_storage", self._schema)}
WHERE storage_key = $1
""",
key,
)
# Check if anything was deleted
if result == "DELETE 0":
logger.warning(f"Attempted to delete non-existent file: {key}")
async def exists(self, key: str) -> bool:
"""Check if file exists in PostgreSQL."""
pool = self._pool_getter()
async with pool.acquire() as conn:
row = await conn.fetchrow(
f"""
SELECT 1 FROM {fq_table("file_storage", self._schema)}
WHERE storage_key = $1
""",
key,
)
return row is not None
async def get_download_url(self, key: str, expires_in: int = 3600) -> str:
"""
Get download URL for PostgreSQL-stored file.
Returns an API endpoint path (not a pre-signed URL since the file
is stored in the database). The expires_in parameter is ignored
for PostgreSQL storage.
"""
# Return API path for download endpoint
# (expires_in ignored for database storage - auth handled at API level)
return f"/v1/default/files/download/{key}"
@@ -1,71 +0,0 @@
"""S3 object storage backend using obstore."""
import logging
from datetime import timedelta
import obstore as obs
from obstore.store import S3Store
from .base import FileStorage
logger = logging.getLogger(__name__)
class S3FileStorage(FileStorage):
"""
S3-compatible object storage backend.
Uses obstore (Rust-backed) for high-throughput async access to
Amazon S3, MinIO, Cloudflare R2, and other S3-compliant APIs.
"""
def __init__(
self,
bucket: str,
region: str | None = None,
endpoint: str | None = None,
access_key_id: str | None = None,
secret_access_key: str | None = None,
):
kwargs: dict = {}
if region:
kwargs["region"] = region
if endpoint:
kwargs["endpoint"] = endpoint
# Allow plain HTTP for local S3-compatible services (MinIO, LocalStack, etc.)
if endpoint.startswith("http://"):
kwargs["allow_http"] = True
if access_key_id:
kwargs["access_key_id"] = access_key_id
if secret_access_key:
kwargs["secret_access_key"] = secret_access_key
self._store = S3Store(bucket, **kwargs)
logger.info(f"Initialized S3 file storage: bucket={bucket}, region={region}, endpoint={endpoint}")
async def store(self, file_data: bytes, key: str, metadata: dict[str, str] | None = None) -> str:
await obs.put_async(self._store, key, file_data)
logger.debug(f"Stored file {key} ({len(file_data)} bytes) in S3")
return key
async def retrieve(self, key: str) -> bytes:
try:
response = await obs.get_async(self._store, key)
return await response.bytes_async()
except Exception as e:
if "not found" in str(e).lower() or "NoSuchKey" in str(e):
raise FileNotFoundError(f"File not found: {key}") from e
raise
async def delete(self, key: str) -> None:
await obs.delete_async(self._store, key)
async def exists(self, key: str) -> bool:
try:
await obs.head_async(self._store, key)
return True
except Exception:
return False
async def get_download_url(self, key: str, expires_in: int = 3600) -> str:
return await obs.sign_async(self._store, "GET", key, timedelta(seconds=expires_in))
@@ -1,40 +0,0 @@
"""
Local MCP server entry point for use with Claude Code (HTTP transport).
This is a thin wrapper around the main hindsight-api server that pre-configures
sensible defaults for local use (embedded PostgreSQL via pg0, warning log level).
The full API runs on localhost:8888. Configure Claude Code's MCP settings:
claude mcp add --transport http hindsight http://localhost:8888/mcp/
Or pinned to a specific bank (single-bank mode):
claude mcp add --transport http hindsight http://localhost:8888/mcp/default/
Run with:
hindsight-local-mcp
Or with uvx:
uvx hindsight-api@latest hindsight-local-mcp
Environment variables:
HINDSIGHT_API_LLM_API_KEY: Required. API key for LLM provider.
HINDSIGHT_API_LLM_PROVIDER: Optional. LLM provider (default: "openai").
HINDSIGHT_API_LLM_MODEL: Optional. LLM model (default: "gpt-4o-mini").
HINDSIGHT_API_DATABASE_URL: Optional. Override database URL (default: pg0://hindsight-mcp).
"""
import os
def main() -> None:
"""Start the Hindsight API server with local defaults."""
# Set local defaults (only if not already configured by the user)
os.environ.setdefault("HINDSIGHT_API_DATABASE_URL", "pg0://hindsight-mcp")
from hindsight_api.main import main as api_main
api_main()
if __name__ == "__main__":
main()
File diff suppressed because it is too large Load Diff
@@ -1,13 +0,0 @@
"""Webhook system for Hindsight API event notifications."""
from .manager import WebhookManager
from .models import ConsolidationEventData, RetainEventData, WebhookConfig, WebhookEvent, WebhookEventType
__all__ = [
"WebhookManager",
"WebhookConfig",
"WebhookEvent",
"WebhookEventType",
"ConsolidationEventData",
"RetainEventData",
]
@@ -1,242 +0,0 @@
"""Webhook manager for delivering event notifications."""
import hashlib
import hmac
import json
import logging
import uuid
from datetime import datetime, timezone
from typing import TYPE_CHECKING
import asyncpg
from .models import WebhookConfig, WebhookEvent, WebhookHttpConfig
if TYPE_CHECKING:
from hindsight_api.extensions.tenant import TenantExtension
logger = logging.getLogger(__name__)
# Retry delay schedule in seconds: 5 retries after the first attempt.
# Fast early retries catch transient failures; later retries handle longer outages.
RETRY_DELAYS = [5, 300, 1800, 7200, 18000]
MAX_ATTEMPTS = len(RETRY_DELAYS) + 1 # first attempt + len(RETRY_DELAYS) retries
def _fq_table(table: str, schema: str | None = None) -> str:
"""Get fully-qualified table name with optional schema prefix."""
if schema:
return f'"{schema}".{table}'
return table
def _parse_http_config(value: str | dict | None) -> WebhookHttpConfig:
"""Parse http_config column value (JSONB returned as text or dict) into a model."""
if value is None:
return WebhookHttpConfig()
if isinstance(value, str):
return WebhookHttpConfig.model_validate_json(value)
return WebhookHttpConfig.model_validate(value)
class WebhookManager:
"""
Manages webhook registration and event firing.
Supports both global webhooks (configured via env vars) and per-bank
webhooks stored in the database. Deliveries are queued as async_operations
tasks (operation_type='webhook_delivery') and picked up by the worker poller.
"""
def __init__(
self,
pool: asyncpg.Pool,
global_webhooks: list[WebhookConfig],
tenant_extension: "TenantExtension | None" = None,
):
self._pool = pool
self._global_webhooks = global_webhooks
self._tenant_extension = tenant_extension
def _sign_payload(self, secret: str, payload_bytes: bytes) -> str:
"""Compute HMAC-SHA256 signature for a payload."""
return "sha256=" + hmac.new(secret.encode(), payload_bytes, hashlib.sha256).hexdigest()
async def fire_event(self, event: WebhookEvent, schema: str | None = None) -> None:
"""
Queue webhook deliveries for an event as async_operations tasks.
Loads per-bank and global webhooks, inserts pending webhook_delivery tasks for
any webhook whose event_types list matches the fired event type. The worker
poller picks these up and calls MemoryEngine._handle_webhook_delivery().
Args:
event: The event to deliver.
schema: Database schema (for multi-tenant). None = default schema.
"""
webhook_table = _fq_table("webhooks", schema)
ops_table = _fq_table("async_operations", schema)
now = datetime.now(timezone.utc)
payload_str = event.model_dump_json()
try:
# Load per-bank webhooks from DB (bank-specific + global NULL rows)
rows = await self._pool.fetch(
f"""
SELECT id, bank_id, url, secret, event_types, enabled, http_config::text
FROM {webhook_table}
WHERE (bank_id = $1 OR bank_id IS NULL) AND enabled = true
""",
event.bank_id,
)
db_webhooks = [
WebhookConfig(
id=str(row["id"]),
bank_id=row["bank_id"],
url=row["url"],
secret=row["secret"],
event_types=list(row["event_types"]) if row["event_types"] else [],
enabled=row["enabled"],
http_config=_parse_http_config(row["http_config"]),
)
for row in rows
]
# Merge with global webhooks from env config
all_webhooks = self._global_webhooks + db_webhooks
matched = 0
for webhook in all_webhooks:
if not webhook.enabled:
continue
if event.event.value not in webhook.event_types:
continue
operation_id = uuid.uuid4()
webhook_id = webhook.id if webhook.id else None
task_payload = json.dumps(
{
"type": "webhook_delivery",
"operation_id": str(operation_id),
"bank_id": event.bank_id,
"url": webhook.url,
"secret": webhook.secret,
"event_type": event.event.value,
"payload": payload_str,
"webhook_id": webhook_id,
"http_config": webhook.http_config.model_dump(),
}
)
await self._pool.execute(
f"""
INSERT INTO {ops_table}
(operation_id, bank_id, operation_type, status, task_payload, result_metadata, created_at, updated_at)
VALUES ($1, $2, 'webhook_delivery', 'pending', $3::jsonb, '{{}}'::jsonb, $4, $4)
""",
operation_id,
event.bank_id,
task_payload,
now,
)
matched += 1
logger.debug(f"Fired webhook event {event.event} for bank {event.bank_id}: {matched} delivery(ies) queued")
except Exception as e:
logger.error(f"Failed to queue webhook deliveries for event {event.event}: {e}")
async def fire_event_with_conn(
self, event: WebhookEvent, conn: asyncpg.Connection, schema: str | None = None
) -> None:
"""
Queue webhook deliveries within an existing database connection/transaction.
Identical to fire_event() but uses the provided connection instead of acquiring
one from the pool. Use this to atomically insert delivery tasks in the same
transaction as the primary operation (transactional outbox pattern).
Args:
event: The event to deliver.
conn: Existing asyncpg connection (may be inside an active transaction).
schema: Database schema (for multi-tenant). None = default schema.
"""
webhook_table = _fq_table("webhooks", schema)
ops_table = _fq_table("async_operations", schema)
now = datetime.now(timezone.utc)
payload_str = event.model_dump_json()
try:
rows = await conn.fetch(
f"""
SELECT id, bank_id, url, secret, event_types, enabled, http_config::text
FROM {webhook_table}
WHERE (bank_id = $1 OR bank_id IS NULL) AND enabled = true
""",
event.bank_id,
)
db_webhooks = [
WebhookConfig(
id=str(row["id"]),
bank_id=row["bank_id"],
url=row["url"],
secret=row["secret"],
event_types=list(row["event_types"]) if row["event_types"] else [],
enabled=row["enabled"],
http_config=_parse_http_config(row["http_config"]),
)
for row in rows
]
all_webhooks = self._global_webhooks + db_webhooks
matched = 0
for webhook in all_webhooks:
if not webhook.enabled:
continue
if event.event.value not in webhook.event_types:
continue
operation_id = uuid.uuid4()
webhook_id = webhook.id if webhook.id else None
task_payload = json.dumps(
{
"type": "webhook_delivery",
"operation_id": str(operation_id),
"bank_id": event.bank_id,
"url": webhook.url,
"secret": webhook.secret,
"event_type": event.event.value,
"payload": payload_str,
"webhook_id": webhook_id,
"http_config": webhook.http_config.model_dump(),
}
)
await conn.execute(
f"""
INSERT INTO {ops_table}
(operation_id, bank_id, operation_type, status, task_payload, result_metadata, created_at, updated_at)
VALUES ($1, $2, 'webhook_delivery', 'pending', $3::jsonb, '{{}}'::jsonb, $4, $4)
""",
operation_id,
event.bank_id,
task_payload,
now,
)
matched += 1
logger.debug(
f"Fired webhook event {event.event} for bank {event.bank_id}: {matched} delivery(ies) queued (in-transaction)"
)
except Exception as e:
logger.error(
f"Failed to queue webhook deliveries (in-transaction) for event {event.event}: {e}. "
"CRITICAL: The enclosing database transaction is now aborted and will roll back all changes."
)
raise
@@ -1,51 +0,0 @@
"""Pydantic models for the webhook system."""
from datetime import datetime
from enum import StrEnum
from pydantic import BaseModel, Field
class WebhookEventType(StrEnum):
CONSOLIDATION_COMPLETED = "consolidation.completed"
RETAIN_COMPLETED = "retain.completed"
class ConsolidationEventData(BaseModel):
observations_created: int | None = None
observations_updated: int | None = None
observations_deleted: int | None = None
error_message: str | None = None
class RetainEventData(BaseModel):
document_id: str | None = None
tags: list[str] | None = None
class WebhookEvent(BaseModel):
event: WebhookEventType
bank_id: str
operation_id: str
status: str # "completed" or "failed"
timestamp: datetime
data: ConsolidationEventData | RetainEventData
class WebhookHttpConfig(BaseModel):
"""HTTP delivery configuration for a webhook."""
method: str = Field(default="POST", description="HTTP method: GET or POST")
timeout_seconds: int = Field(default=30, description="HTTP request timeout in seconds")
headers: dict[str, str] = Field(default_factory=dict, description="Custom HTTP headers")
params: dict[str, str] = Field(default_factory=dict, description="Custom HTTP query parameters")
class WebhookConfig(BaseModel):
id: str
bank_id: str | None
url: str
secret: str | None
event_types: list[str]
enabled: bool
http_config: WebhookHttpConfig = Field(default_factory=WebhookHttpConfig)
@@ -1,9 +0,0 @@
from datetime import datetime
class RetryTaskAt(Exception):
"""Raise from a task handler to schedule a retry at a specific time."""
def __init__(self, retry_at: datetime, message: str = ""):
self.retry_at = retry_at
super().__init__(message)

Some files were not shown because too many files have changed in this diff Show More