Compare commits
5 Commits
e9caeaa4d8
...
e2322a2909
| Author | SHA1 | Date | |
|---|---|---|---|
| e2322a2909 | |||
| 8c062ae461 | |||
| 3d3631c620 | |||
| 88b2db0c3d | |||
| 0ca698acfa |
5
.claude/agents/Explore.md
Normal file
5
.claude/agents/Explore.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
name: Explore
|
||||||
|
description: Fast, read-only codebase search
|
||||||
|
model: sonnet
|
||||||
|
---
|
||||||
186
.claude/hooks/python_quality.py
Normal file
186
.claude/hooks/python_quality.py
Normal file
@@ -0,0 +1,186 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Run Python quality tools for Claude Code Write/Edit hooks.
|
||||||
|
|
||||||
|
The hook is intentionally non-blocking: it formats/fixes what Ruff can fix safely,
|
||||||
|
then reports remaining Ruff/ty diagnostics back to Claude as additional context.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import hashlib
|
||||||
|
import json
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
PYTHON_SUFFIXES = {".py", ".pyi"}
|
||||||
|
SKIP_PARTS = {".git", ".venv", "__pycache__"}
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
payload = _read_payload()
|
||||||
|
file_path = _extract_file_path(payload)
|
||||||
|
if file_path is None:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
repo_root = _repo_root()
|
||||||
|
path = _resolve_path(file_path, repo_root)
|
||||||
|
|
||||||
|
if not _should_check(path):
|
||||||
|
return 0
|
||||||
|
|
||||||
|
relative_path = _display_path(path, repo_root)
|
||||||
|
before = _sha256(path)
|
||||||
|
|
||||||
|
fix = _run(["uv", "run", "ruff", "check", "--fix", str(path)], repo_root)
|
||||||
|
fmt = _run(["uv", "run", "ruff", "format", str(path)], repo_root)
|
||||||
|
lint = _run(["uv", "run", "ruff", "check", str(path)], repo_root)
|
||||||
|
|
||||||
|
# `--error-on-warning` makes ty warnings visible to the hook while this
|
||||||
|
# script still exits 0, so warnings are reported but do not stop Claude.
|
||||||
|
types = _run(["uv", "run", "ty", "check", "--error-on-warning", str(path)], repo_root)
|
||||||
|
|
||||||
|
after = _sha256(path)
|
||||||
|
changed = before != after
|
||||||
|
|
||||||
|
message = _build_message(
|
||||||
|
relative_path=relative_path,
|
||||||
|
changed=changed,
|
||||||
|
fix=fix,
|
||||||
|
fmt=fmt,
|
||||||
|
lint=lint,
|
||||||
|
types=types,
|
||||||
|
)
|
||||||
|
if message:
|
||||||
|
print(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"suppressOutput": True,
|
||||||
|
"hookSpecificOutput": {
|
||||||
|
"hookEventName": "PostToolUse",
|
||||||
|
"additionalContext": message,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
def _read_payload() -> dict[str, object]:
|
||||||
|
try:
|
||||||
|
raw = sys.stdin.read()
|
||||||
|
return json.loads(raw) if raw.strip() else {}
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
|
||||||
|
def _extract_file_path(payload: dict[str, object]) -> str | None:
|
||||||
|
response = payload.get("tool_response")
|
||||||
|
if isinstance(response, dict):
|
||||||
|
for key in ("filePath", "file_path"):
|
||||||
|
value = response.get(key)
|
||||||
|
if isinstance(value, str) and value:
|
||||||
|
return value
|
||||||
|
|
||||||
|
tool_input = payload.get("tool_input")
|
||||||
|
if isinstance(tool_input, dict):
|
||||||
|
value = tool_input.get("file_path")
|
||||||
|
if isinstance(value, str) and value:
|
||||||
|
return value
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _repo_root() -> Path:
|
||||||
|
result = subprocess.run(
|
||||||
|
["git", "rev-parse", "--show-toplevel"],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
if result.returncode == 0:
|
||||||
|
return Path(result.stdout.strip()).resolve()
|
||||||
|
return Path.cwd().resolve()
|
||||||
|
|
||||||
|
|
||||||
|
def _resolve_path(file_path: str, repo_root: Path) -> Path:
|
||||||
|
path = Path(file_path).expanduser()
|
||||||
|
if not path.is_absolute():
|
||||||
|
path = repo_root / path
|
||||||
|
return path.resolve()
|
||||||
|
|
||||||
|
|
||||||
|
def _should_check(path: Path) -> bool:
|
||||||
|
return path.is_file() and path.suffix in PYTHON_SUFFIXES and SKIP_PARTS.isdisjoint(path.parts)
|
||||||
|
|
||||||
|
|
||||||
|
def _display_path(path: Path, repo_root: Path) -> str:
|
||||||
|
try:
|
||||||
|
return str(path.relative_to(repo_root))
|
||||||
|
except ValueError:
|
||||||
|
return str(path)
|
||||||
|
|
||||||
|
|
||||||
|
def _sha256(path: Path) -> str:
|
||||||
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||||
|
|
||||||
|
|
||||||
|
def _run(command: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
|
||||||
|
return subprocess.run(
|
||||||
|
command,
|
||||||
|
cwd=cwd,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _combined_output(result: subprocess.CompletedProcess[str]) -> str:
|
||||||
|
return "\n".join(part.strip() for part in (result.stdout, result.stderr) if part.strip())
|
||||||
|
|
||||||
|
|
||||||
|
def _command_label(result: subprocess.CompletedProcess[str]) -> str:
|
||||||
|
return " ".join(result.args) if isinstance(result.args, list) else str(result.args)
|
||||||
|
|
||||||
|
|
||||||
|
def _build_message(
|
||||||
|
*,
|
||||||
|
relative_path: str,
|
||||||
|
changed: bool,
|
||||||
|
fix: subprocess.CompletedProcess[str],
|
||||||
|
fmt: subprocess.CompletedProcess[str],
|
||||||
|
lint: subprocess.CompletedProcess[str],
|
||||||
|
types: subprocess.CompletedProcess[str],
|
||||||
|
) -> str:
|
||||||
|
lines: list[str] = []
|
||||||
|
|
||||||
|
if changed:
|
||||||
|
lines.append(
|
||||||
|
f"Python quality hook updated `{relative_path}` with Ruff safe fixes/formatting."
|
||||||
|
)
|
||||||
|
lines.append("Read the file before making another manual edit to avoid stale text.")
|
||||||
|
|
||||||
|
for result in (fix, fmt, lint, types):
|
||||||
|
if result.returncode == 0:
|
||||||
|
continue
|
||||||
|
|
||||||
|
output = _combined_output(result)
|
||||||
|
if not output:
|
||||||
|
output = f"Command exited with status {result.returncode}."
|
||||||
|
|
||||||
|
lines.extend(
|
||||||
|
[
|
||||||
|
"",
|
||||||
|
f"Remaining diagnostics from `{_command_label(result)}`:",
|
||||||
|
"```text",
|
||||||
|
output[-6000:],
|
||||||
|
"```",
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
return "\n".join(lines).strip()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
20
.claude/settings.json
Normal file
20
.claude/settings.json
Normal file
@@ -0,0 +1,20 @@
|
|||||||
|
{
|
||||||
|
"enabledPlugins": {
|
||||||
|
"astral@astral-sh": true
|
||||||
|
},
|
||||||
|
"hooks": {
|
||||||
|
"PostToolUse": [
|
||||||
|
{
|
||||||
|
"matcher": "Write|Edit",
|
||||||
|
"hooks": [
|
||||||
|
{
|
||||||
|
"type": "command",
|
||||||
|
"command": "repo=$(git rev-parse --show-toplevel 2>/dev/null || pwd); python3 \"$repo/.claude/hooks/python_quality.py\"",
|
||||||
|
"timeout": 120,
|
||||||
|
"statusMessage": "Running Ruff and ty on Python edit"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
0
.env.example
Normal file
0
.env.example
Normal file
77
.env.langfuse.example
Normal file
77
.env.langfuse.example
Normal file
@@ -0,0 +1,77 @@
|
|||||||
|
# Langfuse local development environment.
|
||||||
|
#
|
||||||
|
# Usage:
|
||||||
|
# cp .env.langfuse.example .env.langfuse
|
||||||
|
# # replace all CHANGE_ME values before starting
|
||||||
|
# docker compose --env-file .env.langfuse -f docker-compose.langfuse.yml up -d
|
||||||
|
#
|
||||||
|
# Generate secrets with commands such as:
|
||||||
|
# openssl rand -base64 32
|
||||||
|
# openssl rand -hex 16
|
||||||
|
# openssl rand -hex 32
|
||||||
|
#
|
||||||
|
# Do not commit .env.langfuse.
|
||||||
|
|
||||||
|
# Public URL used by Langfuse itself. For local browser access this should match
|
||||||
|
# http://localhost:3000. For a VM, set this to your HTTPS domain.
|
||||||
|
LANGFUSE_NEXTAUTH_URL=http://localhost:3000
|
||||||
|
|
||||||
|
# Host ports. Internal container ports stay unchanged and are used for
|
||||||
|
# container-to-container communication.
|
||||||
|
LANGFUSE_WEB_PORT=3000
|
||||||
|
LANGFUSE_WORKER_PORT=3030
|
||||||
|
LANGFUSE_POSTGRES_HOST_PORT=15432
|
||||||
|
LANGFUSE_REDIS_HOST_PORT=16379
|
||||||
|
LANGFUSE_CLICKHOUSE_HTTP_PORT=18123
|
||||||
|
LANGFUSE_CLICKHOUSE_NATIVE_PORT=19000
|
||||||
|
LANGFUSE_MINIO_API_PORT=9090
|
||||||
|
LANGFUSE_MINIO_CONSOLE_PORT=9091
|
||||||
|
|
||||||
|
# Core Langfuse secrets. Replace these values.
|
||||||
|
LANGFUSE_NEXTAUTH_SECRET=CHANGE_ME_generate_with_openssl_rand_base64_32
|
||||||
|
LANGFUSE_SALT=CHANGE_ME_generate_with_openssl_rand_hex_16
|
||||||
|
# Must be exactly 64 hex characters. Generate with: openssl rand -hex 32
|
||||||
|
LANGFUSE_ENCRYPTION_KEY=CHANGE_ME_generate_with_openssl_rand_hex_32
|
||||||
|
|
||||||
|
# Langfuse internal Postgres. This is not the chatbot application's Postgres.
|
||||||
|
LANGFUSE_POSTGRES_VERSION=17
|
||||||
|
LANGFUSE_POSTGRES_USER=postgres
|
||||||
|
LANGFUSE_POSTGRES_PASSWORD=CHANGE_ME_langfuse_postgres_password
|
||||||
|
LANGFUSE_POSTGRES_DB=postgres
|
||||||
|
|
||||||
|
# Langfuse internal ClickHouse.
|
||||||
|
LANGFUSE_CLICKHOUSE_USER=clickhouse
|
||||||
|
LANGFUSE_CLICKHOUSE_PASSWORD=CHANGE_ME_langfuse_clickhouse_password
|
||||||
|
LANGFUSE_CLICKHOUSE_CLUSTER_ENABLED=false
|
||||||
|
|
||||||
|
# Langfuse internal Redis.
|
||||||
|
LANGFUSE_REDIS_PORT=6379
|
||||||
|
LANGFUSE_REDIS_AUTH=CHANGE_ME_langfuse_redis_password
|
||||||
|
|
||||||
|
# Langfuse local MinIO/object storage.
|
||||||
|
LANGFUSE_MINIO_ROOT_USER=minio
|
||||||
|
LANGFUSE_MINIO_ROOT_PASSWORD=CHANGE_ME_langfuse_minio_password
|
||||||
|
LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID=minio
|
||||||
|
LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY=CHANGE_ME_langfuse_minio_password
|
||||||
|
LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID=minio
|
||||||
|
LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY=CHANGE_ME_langfuse_minio_password
|
||||||
|
LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT_PUBLIC=http://localhost:9090
|
||||||
|
LANGFUSE_S3_BATCH_EXPORT_ACCESS_KEY_ID=minio
|
||||||
|
LANGFUSE_S3_BATCH_EXPORT_SECRET_ACCESS_KEY=CHANGE_ME_langfuse_minio_password
|
||||||
|
LANGFUSE_S3_BATCH_EXPORT_EXTERNAL_ENDPOINT=http://localhost:9090
|
||||||
|
|
||||||
|
# Development defaults.
|
||||||
|
LANGFUSE_TELEMETRY_ENABLED=false
|
||||||
|
LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES=false
|
||||||
|
|
||||||
|
# Optional headless initialization. Leave blank to create the organization,
|
||||||
|
# project, user, and API keys in the UI after startup.
|
||||||
|
LANGFUSE_INIT_ORG_ID=
|
||||||
|
LANGFUSE_INIT_ORG_NAME=
|
||||||
|
LANGFUSE_INIT_PROJECT_ID=
|
||||||
|
LANGFUSE_INIT_PROJECT_NAME=
|
||||||
|
LANGFUSE_INIT_PROJECT_PUBLIC_KEY=
|
||||||
|
LANGFUSE_INIT_PROJECT_SECRET_KEY=
|
||||||
|
LANGFUSE_INIT_USER_EMAIL=
|
||||||
|
LANGFUSE_INIT_USER_NAME=
|
||||||
|
LANGFUSE_INIT_USER_PASSWORD=
|
||||||
13
.gitignore
vendored
13
.gitignore
vendored
@@ -9,5 +9,16 @@ wheels/
|
|||||||
# Virtual environments
|
# Virtual environments
|
||||||
.venv
|
.venv
|
||||||
|
|
||||||
# Enviroment variables
|
# Test and coverage artifacts
|
||||||
|
.pytest_cache/
|
||||||
|
.coverage
|
||||||
|
coverage.xml
|
||||||
|
htmlcov/
|
||||||
|
|
||||||
|
# Environment variables
|
||||||
.env
|
.env
|
||||||
|
.env.langfuse
|
||||||
|
infra/langfuse/.env
|
||||||
|
|
||||||
|
# Claude Code local overrides
|
||||||
|
.claude/settings.local.json
|
||||||
|
|||||||
51
README.md
51
README.md
@@ -0,0 +1,51 @@
|
|||||||
|
# Talie chatbot service
|
||||||
|
|
||||||
|
Architecture decisions live in [`docs/adr`](docs/adr). The first implementation
|
||||||
|
milestone is documented in the [ingestion vertical-slice plan](docs/plans/001-ingestion-vertical-slice.md).
|
||||||
|
|
||||||
|
## Local Langfuse
|
||||||
|
|
||||||
|
This repo includes a root-level development Compose file for Langfuse:
|
||||||
|
|
||||||
|
- [`docker-compose.langfuse.yml`](docker-compose.langfuse.yml)
|
||||||
|
- [`.env.langfuse.example`](.env.langfuse.example)
|
||||||
|
|
||||||
|
Start Langfuse locally:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
cp .env.langfuse.example .env.langfuse
|
||||||
|
# edit .env.langfuse and replace CHANGE_ME values
|
||||||
|
|
||||||
|
docker compose --env-file .env.langfuse -f docker-compose.langfuse.yml up -d
|
||||||
|
```
|
||||||
|
|
||||||
|
Open:
|
||||||
|
|
||||||
|
```text
|
||||||
|
http://localhost:3000
|
||||||
|
```
|
||||||
|
|
||||||
|
If the chatbot app runs on your host machine, configure it with:
|
||||||
|
|
||||||
|
```env
|
||||||
|
LANGFUSE_HOST=http://localhost:3000
|
||||||
|
```
|
||||||
|
|
||||||
|
If the chatbot app later runs inside the same Compose project/network as
|
||||||
|
Langfuse, configure it with:
|
||||||
|
|
||||||
|
```env
|
||||||
|
LANGFUSE_HOST=http://langfuse-web:3000
|
||||||
|
```
|
||||||
|
|
||||||
|
A future app stack can be launched together with Langfuse using multiple Compose
|
||||||
|
files:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
docker compose \
|
||||||
|
-f docker-compose.yml \
|
||||||
|
-f docker-compose.langfuse.yml \
|
||||||
|
--env-file .env \
|
||||||
|
--env-file .env.langfuse \
|
||||||
|
up -d
|
||||||
|
```
|
||||||
|
|||||||
@@ -58,7 +58,11 @@ Conventions:
|
|||||||
|
|
||||||
- Use SQLAlchemy async sessions in FastAPI (`AsyncSession`) and one session per
|
- Use SQLAlchemy async sessions in FastAPI (`AsyncSession`) and one session per
|
||||||
request/job unit of work.
|
request/job unit of work.
|
||||||
- Use Alembic for all DDL. FastAPI startup opens connections and checks
|
- Use Alembic for all DDL. Bootstrap the migration environment once with
|
||||||
|
`uv run alembic init -t async alembic`, then commit the generated
|
||||||
|
`alembic.ini`, `alembic/env.py`, and `alembic/versions/` directory. Retain the
|
||||||
|
async Alembic template and configure it to load the application's database URL
|
||||||
|
and SQLAlchemy model metadata. FastAPI startup opens connections and checks
|
||||||
readiness; it does not create or alter tables.
|
readiness; it does not create or alter tables.
|
||||||
- Prefer UUID primary keys generated by the application. Avoid integer IDs that
|
- Prefer UUID primary keys generated by the application. Avoid integer IDs that
|
||||||
leak tenant size and make distributed workers harder to compose.
|
leak tenant size and make distributed workers harder to compose.
|
||||||
|
|||||||
@@ -0,0 +1,270 @@
|
|||||||
|
# 0012. Application resource lifetime and dependency ownership
|
||||||
|
|
||||||
|
## Status
|
||||||
|
|
||||||
|
Proposed
|
||||||
|
|
||||||
|
## Context
|
||||||
|
|
||||||
|
ADR-0008 defines the FastAPI REST boundary and says the app lifespan owns
|
||||||
|
long-lived resources. ADR-0009 chooses SQLAlchemy 2.x, Alembic migrations, and
|
||||||
|
async sessions for the Postgres layer. We need a project-wide rule for how those
|
||||||
|
resources, and future resources with similar lifetimes, are created, shared, and
|
||||||
|
closed.
|
||||||
|
|
||||||
|
The service will need objects with different lifetimes:
|
||||||
|
|
||||||
|
- application-lifetime objects such as SQLAlchemy engines, session factories,
|
||||||
|
Qdrant clients, LangGraph checkpointers/stores, compiled graphs, HTTP clients,
|
||||||
|
observability clients, model clients, Redis clients, or other connection pools;
|
||||||
|
- request/job-lifetime objects such as SQLAlchemy sessions, transactions,
|
||||||
|
authentication context, tenant context, request ids, and unit-of-work state;
|
||||||
|
- operation-lifetime objects such as temporary files, one-off streams, locks, or
|
||||||
|
short-lived connections checked out for a single block.
|
||||||
|
|
||||||
|
Using module-level mutable resource instances for everything would make startup,
|
||||||
|
shutdown, testing, concurrency, and transaction boundaries hard to reason about.
|
||||||
|
The most dangerous example is a global SQLAlchemy `Session`: sessions track
|
||||||
|
identity state and transaction state, and are not safe to share across concurrent
|
||||||
|
requests.
|
||||||
|
|
||||||
|
At the same time, creating expensive clients or connection pools for every
|
||||||
|
request would waste resources and lose pooling benefits. We need explicit
|
||||||
|
ownership: the component that creates a resource is responsible for closing it.
|
||||||
|
|
||||||
|
## Decision
|
||||||
|
|
||||||
|
### Use FastAPI lifespan for application-lifetime resources
|
||||||
|
|
||||||
|
Create long-lived shared resources in the FastAPI lifespan function and close
|
||||||
|
them after the lifespan `yield`.
|
||||||
|
|
||||||
|
Application-lifetime resources include:
|
||||||
|
|
||||||
|
- SQLAlchemy async `Engine`;
|
||||||
|
- SQLAlchemy `async_sessionmaker` bound to that engine;
|
||||||
|
- Qdrant client;
|
||||||
|
- LangGraph checkpointer/store and compiled graph;
|
||||||
|
- shared `httpx.AsyncClient` instances;
|
||||||
|
- Redis or cache clients;
|
||||||
|
- Langfuse/observability clients;
|
||||||
|
- embedding, reranking, LLM, or other expensive model clients.
|
||||||
|
|
||||||
|
The lifespan function is the owner of these resources. It is responsible for
|
||||||
|
calling cleanup methods such as `dispose()`, `aclose()`, `close()`, or equivalent
|
||||||
|
shutdown hooks.
|
||||||
|
|
||||||
|
Indicative shape:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from collections.abc import AsyncIterator
|
||||||
|
from contextlib import asynccontextmanager
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
from fastapi import FastAPI
|
||||||
|
|
||||||
|
|
||||||
|
@asynccontextmanager
|
||||||
|
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||||
|
app.state.database = Database(settings.database_url)
|
||||||
|
app.state.http_client = httpx.AsyncClient()
|
||||||
|
|
||||||
|
try:
|
||||||
|
yield
|
||||||
|
finally:
|
||||||
|
await app.state.http_client.aclose()
|
||||||
|
await app.state.database.close()
|
||||||
|
|
||||||
|
|
||||||
|
app = FastAPI(lifespan=lifespan)
|
||||||
|
```
|
||||||
|
|
||||||
|
Avoid opening external resource clients as import-time side effects. Module-level
|
||||||
|
constants, settings objects, pure functions, type aliases, and stateless helpers
|
||||||
|
are acceptable. Mutable network/database clients should be constructed by the
|
||||||
|
application owner, not by importing a module.
|
||||||
|
|
||||||
|
### Treat SQLAlchemy engine and session as different lifetimes
|
||||||
|
|
||||||
|
A SQLAlchemy `Engine` is an application-lifetime pool manager, not a single
|
||||||
|
request's transaction. It may keep database connections open in a pool and reuse
|
||||||
|
them across requests.
|
||||||
|
|
||||||
|
A SQLAlchemy `AsyncSession` is a request/job-lifetime unit of work. It tracks ORM
|
||||||
|
identity state and transaction state and may check out a database connection from
|
||||||
|
the engine when needed.
|
||||||
|
|
||||||
|
Therefore:
|
||||||
|
|
||||||
|
- create one engine/session factory per FastAPI process during lifespan;
|
||||||
|
- create one `AsyncSession` per request or background-job unit of work;
|
||||||
|
- never share one `AsyncSession` globally across concurrent requests;
|
||||||
|
- never run DDL such as `create_all()` at FastAPI startup; use Alembic as decided
|
||||||
|
in ADR-0009.
|
||||||
|
|
||||||
|
### Use dependencies for request-lifetime resources
|
||||||
|
|
||||||
|
Use FastAPI dependencies, usually `yield` dependencies, for objects that should
|
||||||
|
exist for one request and then be cleaned up.
|
||||||
|
|
||||||
|
Indicative database session dependency:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from collections.abc import AsyncIterator
|
||||||
|
|
||||||
|
from fastapi import Request
|
||||||
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
|
|
||||||
|
async def get_db_session(request: Request) -> AsyncIterator[AsyncSession]:
|
||||||
|
database: Database = request.app.state.database
|
||||||
|
|
||||||
|
async with database.session_factory() as session:
|
||||||
|
try:
|
||||||
|
yield session
|
||||||
|
except Exception:
|
||||||
|
await session.rollback()
|
||||||
|
raise
|
||||||
|
```
|
||||||
|
|
||||||
|
The dependency owns the session lifetime. Route handlers, services, and
|
||||||
|
repositories must not close a session they did not create.
|
||||||
|
|
||||||
|
Authentication and tenant context follow the same rule: resolve them through
|
||||||
|
request dependencies, then pass the trusted context into code that needs it.
|
||||||
|
|
||||||
|
### Keep transaction boundaries explicit
|
||||||
|
|
||||||
|
Routes or application-service functions own transaction boundaries. Repository
|
||||||
|
and CRUD functions receive a session and perform database work, but they should
|
||||||
|
not secretly commit, rollback, or close the session unless their contract
|
||||||
|
explicitly says they own a complete unit of work.
|
||||||
|
|
||||||
|
Preferred shape:
|
||||||
|
|
||||||
|
```python
|
||||||
|
async def create_user(session: AsyncSession, data: CreateUserRequest) -> User:
|
||||||
|
user = User(email=data.email)
|
||||||
|
session.add(user)
|
||||||
|
return user
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/users")
|
||||||
|
async def create_user_endpoint(
|
||||||
|
body: CreateUserRequest,
|
||||||
|
session: SessionDep,
|
||||||
|
) -> UserResponse:
|
||||||
|
user = await create_user(session, body)
|
||||||
|
await session.commit()
|
||||||
|
return UserResponse.model_validate(user)
|
||||||
|
```
|
||||||
|
|
||||||
|
Avoid hidden global access:
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Do not do this.
|
||||||
|
session = SessionLocal()
|
||||||
|
|
||||||
|
async def create_user(data: CreateUserRequest) -> User:
|
||||||
|
session.add(User(email=data.email))
|
||||||
|
await session.commit()
|
||||||
|
```
|
||||||
|
|
||||||
|
This keeps unit-of-work ownership visible, makes multi-step transactions easier,
|
||||||
|
and lets tests pass in their own sessions.
|
||||||
|
|
||||||
|
### Use context managers for operation-lifetime resources
|
||||||
|
|
||||||
|
Objects needed only inside one function or block should use `with` or
|
||||||
|
`async with` rather than app state or module globals.
|
||||||
|
|
||||||
|
Examples include file handles, temporary streams, one-off SQLAlchemy connections,
|
||||||
|
locks, and response streams.
|
||||||
|
|
||||||
|
```python
|
||||||
|
async with http_client.stream("GET", url) as response:
|
||||||
|
...
|
||||||
|
```
|
||||||
|
|
||||||
|
or:
|
||||||
|
|
||||||
|
```python
|
||||||
|
async with engine.connect() as connection:
|
||||||
|
...
|
||||||
|
```
|
||||||
|
|
||||||
|
The context manager owns and releases the resource at block exit.
|
||||||
|
|
||||||
|
### Pass dependencies explicitly through service boundaries
|
||||||
|
|
||||||
|
Routers adapt HTTP requests into typed application calls. Services and
|
||||||
|
repositories should receive the resources they use as explicit parameters:
|
||||||
|
|
||||||
|
```python
|
||||||
|
async def get_point(
|
||||||
|
session: AsyncSession,
|
||||||
|
qdrant: QdrantClient,
|
||||||
|
auth: AuthContext,
|
||||||
|
point_id: str,
|
||||||
|
) -> PointResponse:
|
||||||
|
...
|
||||||
|
```
|
||||||
|
|
||||||
|
Do not have lower layers import mutable resource singletons. Explicit parameters
|
||||||
|
make ownership clear and allow FastAPI dependency overrides, test fixtures, and
|
||||||
|
background workers to supply equivalent resources.
|
||||||
|
|
||||||
|
### Account for worker process multiplicity
|
||||||
|
|
||||||
|
FastAPI lifespan runs once per worker process. If the service runs with multiple
|
||||||
|
Uvicorn/Gunicorn workers, each worker has its own app instance, engine, and
|
||||||
|
connection pool. Database pool sizes and external client limits must be chosen
|
||||||
|
with worker count in mind.
|
||||||
|
|
||||||
|
## Consequences
|
||||||
|
|
||||||
|
### Positive
|
||||||
|
- Resource ownership and cleanup become explicit: lifespan closes app resources,
|
||||||
|
dependencies close request resources, and context managers close block-scoped
|
||||||
|
resources.
|
||||||
|
- SQLAlchemy sessions are not shared across concurrent requests, avoiding unsafe
|
||||||
|
identity-map and transaction-state reuse.
|
||||||
|
- Connection pooling is preserved because expensive clients and engines are
|
||||||
|
created once per process instead of once per request.
|
||||||
|
- Transaction boundaries are easier to reason about because routes or
|
||||||
|
application services decide when to commit or rollback.
|
||||||
|
- Services and repositories are easier to test because sessions, clients, and
|
||||||
|
auth contexts can be injected directly.
|
||||||
|
- The same policy can be reused for future Redis, Qdrant, Langfuse, HTTP, model,
|
||||||
|
and worker resources.
|
||||||
|
|
||||||
|
### Negative
|
||||||
|
- This adds boilerplate: lifespan setup, app state typing/conventions,
|
||||||
|
dependencies, and explicit parameters must be maintained.
|
||||||
|
- Developers must distinguish engine/session, application/request, and
|
||||||
|
owner/borrower lifetimes instead of importing a convenient global.
|
||||||
|
- App state access needs discipline and typing helpers so resources do not become
|
||||||
|
an unstructured service locator.
|
||||||
|
- Multi-worker deployments require explicit pool sizing because each process owns
|
||||||
|
its own pools and clients.
|
||||||
|
|
||||||
|
## Alternatives Considered
|
||||||
|
|
||||||
|
- **Module-level resource singletons**: rejected as the default for mutable
|
||||||
|
external resources. Import-time construction makes startup order, tests,
|
||||||
|
reloads, and shutdown cleanup harder to control. Module-level constants and
|
||||||
|
stateless helpers remain acceptable.
|
||||||
|
- **One global SQLAlchemy session**: rejected. A session is a unit of work with
|
||||||
|
identity and transaction state, not an application-wide pool. Sharing it across
|
||||||
|
requests is unsafe and makes rollback/error handling ambiguous.
|
||||||
|
- **Create every client/session inside each CRUD function**: rejected. It hides
|
||||||
|
ownership and transaction boundaries, prevents multi-step units of work, and
|
||||||
|
wastes pooling benefits for expensive clients.
|
||||||
|
- **Automatically commit every successful request in the session dependency**:
|
||||||
|
deferred as a default. It reduces route boilerplate, but it can make write
|
||||||
|
boundaries too implicit and does not fit every read/write or multi-transaction
|
||||||
|
flow. Routes or application services should own commits unless a future ADR
|
||||||
|
adopts a unit-of-work abstraction.
|
||||||
|
- **Use a generic global service locator**: rejected for now. It can reduce
|
||||||
|
parameter lists, but it obscures dependencies and ownership. FastAPI
|
||||||
|
dependencies plus explicit function parameters are clearer at this stage.
|
||||||
182
docs/adr/0013-s3-compatible-object-storage-with-minio.md
Normal file
182
docs/adr/0013-s3-compatible-object-storage-with-minio.md
Normal file
@@ -0,0 +1,182 @@
|
|||||||
|
# 0013. S3-compatible object storage with MinIO
|
||||||
|
|
||||||
|
## Status
|
||||||
|
|
||||||
|
Proposed
|
||||||
|
|
||||||
|
## Context
|
||||||
|
|
||||||
|
ADR-0008 defines source-file upload endpoints and returns a job-shaped ingestion
|
||||||
|
response. ADR-0009 models `source_files` and leaves room for a `storage_uri`
|
||||||
|
where original files are retained. The service now needs a clear place to store
|
||||||
|
uploaded file bytes and any derived ingestion artifacts without putting large
|
||||||
|
binary/text payloads in Postgres or broker messages.
|
||||||
|
|
||||||
|
The storage decision has to satisfy several constraints:
|
||||||
|
|
||||||
|
- Uploaded files may be larger than is appropriate for Postgres rows.
|
||||||
|
- Ingestion workers need to fetch the exact file bytes after the HTTP request has
|
||||||
|
completed.
|
||||||
|
- Postgres should remain the source of truth for tenant metadata, file metadata,
|
||||||
|
ingestion job state, audit records, and retention decisions.
|
||||||
|
- Broker messages should carry durable identifiers, not raw file content.
|
||||||
|
- Local/self-hosted deployment should stay practical and avoid coupling this app
|
||||||
|
to Langfuse's internal object-storage services.
|
||||||
|
|
||||||
|
## Decision
|
||||||
|
|
||||||
|
Use **MinIO** as the application's S3-compatible object store for uploaded source
|
||||||
|
files and derived ingestion blobs.
|
||||||
|
|
||||||
|
MinIO stores bytes. Postgres stores metadata and authoritative state.
|
||||||
|
|
||||||
|
### Store source-file bytes in MinIO
|
||||||
|
|
||||||
|
When `POST /v1/files` receives an upload, the service stores the raw file in an
|
||||||
|
application-owned MinIO bucket before or within the same application flow that
|
||||||
|
creates the `source_files` and `ingestion_jobs` rows.
|
||||||
|
|
||||||
|
`source_files` keeps the durable pointer and metadata, including fields already
|
||||||
|
sketched in ADR-0009:
|
||||||
|
|
||||||
|
- `tenant_id`;
|
||||||
|
- `domain`;
|
||||||
|
- `source_filename`;
|
||||||
|
- `source_type`;
|
||||||
|
- `content_sha256`;
|
||||||
|
- `byte_size`;
|
||||||
|
- `storage_uri` or equivalent bucket/key fields;
|
||||||
|
- lifecycle status and audit timestamps.
|
||||||
|
|
||||||
|
Object keys are internal identifiers, not trusted user filenames. Prefer a shape
|
||||||
|
like:
|
||||||
|
|
||||||
|
```text
|
||||||
|
tenants/{tenant_id}/source-files/{file_id}/original
|
||||||
|
```
|
||||||
|
|
||||||
|
or, when content-addressing is useful:
|
||||||
|
|
||||||
|
```text
|
||||||
|
tenants/{tenant_id}/source-files/{file_id}/{content_sha256}
|
||||||
|
```
|
||||||
|
|
||||||
|
The original filename is stored as metadata in Postgres, not used as the object
|
||||||
|
key authority.
|
||||||
|
|
||||||
|
### Keep MinIO separate from authoritative state
|
||||||
|
|
||||||
|
MinIO is not the source of truth for:
|
||||||
|
|
||||||
|
- tenant identity or authorization;
|
||||||
|
- file ownership;
|
||||||
|
- ingestion job status;
|
||||||
|
- ingestion progress events;
|
||||||
|
- audit logs;
|
||||||
|
- LangGraph checkpoints or Store memory;
|
||||||
|
- Qdrant point state;
|
||||||
|
- broker delivery state.
|
||||||
|
|
||||||
|
Workers always load `source_files` and `ingestion_jobs` from Postgres by id, then
|
||||||
|
use the stored object pointer to fetch bytes from MinIO.
|
||||||
|
|
||||||
|
### Use MinIO for retained blobs, not queue payloads
|
||||||
|
|
||||||
|
Broker messages reference ids such as `file_id` and `ingestion_job_id`; they do
|
||||||
|
not contain raw file bytes, extracted text, chunks, embeddings, or large parser
|
||||||
|
outputs.
|
||||||
|
|
||||||
|
MinIO may also store derived blobs when retention policy allows it, for example:
|
||||||
|
|
||||||
|
- converted `.doc` -> `.docx` outputs from ADR-0004;
|
||||||
|
- extracted text snapshots used for debugging failed ingestion;
|
||||||
|
- parser diagnostics or quarantined uploads;
|
||||||
|
- exported artifacts generated by maintenance jobs.
|
||||||
|
|
||||||
|
Those derived objects still need Postgres metadata if they are user-visible,
|
||||||
|
auditable, or subject to retention/erasure policy.
|
||||||
|
|
||||||
|
### Use application-owned buckets and credentials
|
||||||
|
|
||||||
|
The app must not casually reuse Langfuse's internal MinIO bucket, credentials, or
|
||||||
|
lifecycle. Langfuse object storage belongs to the Langfuse stack from ADR-0010;
|
||||||
|
this service needs its own application bucket(s) and credentials.
|
||||||
|
|
||||||
|
A local deployment may run MinIO in the same Docker environment, but with a
|
||||||
|
separate bucket such as:
|
||||||
|
|
||||||
|
```text
|
||||||
|
chatbot-source-files
|
||||||
|
```
|
||||||
|
|
||||||
|
Production may use MinIO or another S3-compatible service behind the same object
|
||||||
|
storage interface, provided it preserves tenant isolation, encryption, backup,
|
||||||
|
and retention requirements.
|
||||||
|
|
||||||
|
### Manage the client as an application-lifetime resource
|
||||||
|
|
||||||
|
The object-storage client is an application-lifetime external client under
|
||||||
|
ADR-0012. It is created by the FastAPI lifespan or worker process startup and
|
||||||
|
closed on shutdown when the client library requires explicit cleanup.
|
||||||
|
|
||||||
|
Lower-level ingestion services receive an object-storage client or interface
|
||||||
|
explicitly. They do not import a module-level mutable MinIO singleton.
|
||||||
|
|
||||||
|
For async code, object-storage operations must not block the event loop. Use an
|
||||||
|
async-capable S3 client, or isolate synchronous SDK calls in the appropriate
|
||||||
|
threadpool boundary.
|
||||||
|
|
||||||
|
### Enforce privacy, retention, and integrity rules
|
||||||
|
|
||||||
|
- Compute and store `content_sha256` for uploaded bytes.
|
||||||
|
- Enforce allowed content types/extensions and maximum file size before retaining
|
||||||
|
files, as required by ADR-0008.
|
||||||
|
- Prefer private buckets; generate short-lived presigned URLs only for explicit
|
||||||
|
internal workflows that need them.
|
||||||
|
- Apply server-side encryption and backups according to deployment policy.
|
||||||
|
- Deletion and tenant-erasure workflows must delete both Postgres metadata and
|
||||||
|
MinIO objects, while preserving audit records according to tenant/legal
|
||||||
|
retention requirements.
|
||||||
|
|
||||||
|
## Consequences
|
||||||
|
|
||||||
|
### Positive
|
||||||
|
- Large uploaded files and derived blobs stay out of Postgres rows and broker
|
||||||
|
payloads.
|
||||||
|
- Ingestion can run in separate workers after the HTTP request finishes because
|
||||||
|
workers can fetch the retained object by durable id.
|
||||||
|
- The service keeps S3-compatible portability while using MinIO for local and
|
||||||
|
self-hosted deployments.
|
||||||
|
- `source_files.storage_uri` from ADR-0009 now has a concrete storage backend.
|
||||||
|
- Separating app MinIO from Langfuse MinIO avoids accidental coupling to
|
||||||
|
Langfuse's internal lifecycle and credentials.
|
||||||
|
|
||||||
|
### Negative
|
||||||
|
- Adds another external dependency to configure, secure, back up, and monitor.
|
||||||
|
- Retention and tenant erasure now have to coordinate Postgres rows, Qdrant
|
||||||
|
points, and MinIO objects.
|
||||||
|
- Upload flows need cleanup handling for partial failures, such as object upload
|
||||||
|
succeeding but the Postgres transaction failing.
|
||||||
|
- Async FastAPI handlers need care if the chosen S3/MinIO client is synchronous.
|
||||||
|
- Object keys and bucket policies become part of the security boundary; mistakes
|
||||||
|
can expose cross-tenant files.
|
||||||
|
|
||||||
|
## Alternatives Considered
|
||||||
|
|
||||||
|
- **Store uploaded files directly in Postgres**: rejected. It simplifies
|
||||||
|
transactional consistency, but bloats the relational database, complicates
|
||||||
|
backups, and mixes large binary/blob storage with metadata and audit queries.
|
||||||
|
- **Store only extracted chunks in Qdrant and discard originals immediately**:
|
||||||
|
rejected as the default. It makes failed-ingestion debugging, reprocessing,
|
||||||
|
parser upgrades, and retention review harder. Tenants may still configure
|
||||||
|
shorter retention later.
|
||||||
|
- **Use the broker message as file transport**: rejected. Brokers should carry
|
||||||
|
small work identifiers and metadata; raw files, extracted text, chunks, and
|
||||||
|
embeddings belong in object storage or the appropriate durable store.
|
||||||
|
- **Reuse Langfuse's internal MinIO service/bucket**: rejected. Langfuse object
|
||||||
|
storage is part of the observability stack, not this application's file store.
|
||||||
|
Sharing it would couple credentials, retention, backups, and operational
|
||||||
|
lifecycle across unrelated systems.
|
||||||
|
- **Bind directly to a cloud-only S3 provider**: rejected for now. The project is
|
||||||
|
self-hosted/local-development oriented, and MinIO gives an S3-compatible API
|
||||||
|
while keeping deployment portable.
|
||||||
332
docs/adr/0014-durable-job-dispatch-with-rabbitmq.md
Normal file
332
docs/adr/0014-durable-job-dispatch-with-rabbitmq.md
Normal file
@@ -0,0 +1,332 @@
|
|||||||
|
# 0014. Durable job dispatch with RabbitMQ
|
||||||
|
|
||||||
|
## Status
|
||||||
|
|
||||||
|
Proposed
|
||||||
|
|
||||||
|
## Context
|
||||||
|
|
||||||
|
ADR-0008 makes file ingestion job-shaped: `POST /v1/files` returns `202 Accepted`
|
||||||
|
with `file_id`, `ingestion_job_id`, and a queued/running status. ADR-0009 makes
|
||||||
|
Postgres the durable source of truth for `source_files`, `ingestion_jobs`, and
|
||||||
|
`ingestion_job_events`. ADR-0013 stores uploaded file bytes in MinIO so ingestion
|
||||||
|
workers can fetch them after the request completes.
|
||||||
|
|
||||||
|
What is still missing is the delivery mechanism that tells workers which durable
|
||||||
|
jobs are ready to process.
|
||||||
|
|
||||||
|
The broker decision has to preserve existing boundaries:
|
||||||
|
|
||||||
|
- Postgres remains authoritative for job status, progress, audit, tenancy, and
|
||||||
|
retention-sensitive state.
|
||||||
|
- MinIO stores file bytes and derived blobs, not queue state.
|
||||||
|
- Broker messages carry ids and correlation metadata, not raw files or chunks.
|
||||||
|
- Chat runs are not queued by this service: ADR-0007/0008 allow one in-flight run
|
||||||
|
per `thread_id` and return `409 Conflict` for concurrent runs.
|
||||||
|
- Application-lifetime clients are created and closed by lifespan/startup owners
|
||||||
|
under ADR-0012, not import-time globals.
|
||||||
|
|
||||||
|
## Decision
|
||||||
|
|
||||||
|
Use **RabbitMQ** (via the **aio-pika** async SDK) as the durable pub/sub and
|
||||||
|
job-dispatch broker for asynchronous ingestion and maintenance work.
|
||||||
|
|
||||||
|
RabbitMQ's work-queue model gives ingestion dispatch what it needs directly:
|
||||||
|
durable exchanges/queues, manual acknowledgement, redelivery on nack/crash, and
|
||||||
|
per-queue dead-lettering for poison messages, all while Postgres remains the
|
||||||
|
source of truth for application state.
|
||||||
|
|
||||||
|
### Start with ingestion jobs
|
||||||
|
|
||||||
|
The first broker-backed workflow is source-file ingestion:
|
||||||
|
|
||||||
|
```text
|
||||||
|
POST /v1/files
|
||||||
|
-> authenticate and resolve tenant
|
||||||
|
-> store source bytes in MinIO (ADR-0013)
|
||||||
|
-> create/update source_files row
|
||||||
|
-> create ingestion_jobs row with status='queued'
|
||||||
|
-> create an unpublished outbox_events row
|
||||||
|
-> commit one transaction
|
||||||
|
-> return 202 Accepted
|
||||||
|
|
||||||
|
outbox publisher
|
||||||
|
-> publish the RabbitMQ message containing durable ids
|
||||||
|
-> mark the outbox event published
|
||||||
|
```
|
||||||
|
|
||||||
|
A dedicated ingestion worker process consumes the message:
|
||||||
|
|
||||||
|
```text
|
||||||
|
RabbitMQ message
|
||||||
|
-> load ingestion_jobs/source_files from Postgres
|
||||||
|
-> bind logging context from durable ids
|
||||||
|
-> fetch source object from MinIO
|
||||||
|
-> parse, chunk, embed, and mutate Qdrant points
|
||||||
|
-> append ingestion_job_events
|
||||||
|
-> mark ingestion_jobs succeeded/failed/cancelled
|
||||||
|
-> ack or nack/reject the message
|
||||||
|
```
|
||||||
|
|
||||||
|
### Worker-owned ingestion chunk CRUD
|
||||||
|
|
||||||
|
For file ingestion and re-ingestion, generated chunk/point CRUD is performed by
|
||||||
|
the ingestion worker, not by the FastAPI publisher/request path.
|
||||||
|
|
||||||
|
The HTTP request path only owns the lightweight durable handoff:
|
||||||
|
|
||||||
|
- authenticate the caller and derive tenant context;
|
||||||
|
- store uploaded bytes in MinIO;
|
||||||
|
- create or update `source_files`;
|
||||||
|
- create `ingestion_jobs(status='queued')`;
|
||||||
|
- create the corresponding unpublished `outbox_events` row;
|
||||||
|
- commit the Postgres transaction and return `202 Accepted`.
|
||||||
|
|
||||||
|
The separate outbox-publisher process owns publication of the RabbitMQ message
|
||||||
|
containing durable ids and records its outcome on the outbox event.
|
||||||
|
|
||||||
|
The ingestion worker owns the expensive and retryable side effects:
|
||||||
|
|
||||||
|
- parsing source files;
|
||||||
|
- chunking;
|
||||||
|
- embedding and reranking-vector generation when applicable;
|
||||||
|
- creating, updating, reordering, and soft-deleting generated Qdrant points;
|
||||||
|
- appending `ingestion_job_events`;
|
||||||
|
- updating ingestion job status, counters, and failure summaries.
|
||||||
|
|
||||||
|
Shared chunk/point mutation logic should live in a service/repository layer that
|
||||||
|
can be reused by both direct API routes and ingestion workers. Direct
|
||||||
|
`/v1/points` CRUD remains a synchronous API responsibility unless the operation
|
||||||
|
is explicitly bulk or job-shaped. Ingestion-generated mutations are asynchronous
|
||||||
|
worker responsibility because they must be safe to retry and resume by
|
||||||
|
`ingestion_job_id`.
|
||||||
|
|
||||||
|
### Keep messages small and versioned
|
||||||
|
|
||||||
|
Broker payloads carry identifiers and correlation metadata only. A typical
|
||||||
|
message is:
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"type": "ingestion.job.created",
|
||||||
|
"version": 1,
|
||||||
|
"tenant_id": "...",
|
||||||
|
"file_id": "...",
|
||||||
|
"ingestion_job_id": "...",
|
||||||
|
"request_id": "...",
|
||||||
|
"api_key_id": "..."
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
Do not put raw file bytes, extracted text, chunks, embeddings, raw prompts, model
|
||||||
|
outputs, or secrets in RabbitMQ messages.
|
||||||
|
|
||||||
|
Workers must treat message metadata as routing/correlation input, not as the only
|
||||||
|
authority. Before doing tenant-scoped work, workers reload the durable job and
|
||||||
|
file rows from Postgres and verify the ids are consistent.
|
||||||
|
|
||||||
|
### Use explicit routing keys, a topic exchange, and durable queues
|
||||||
|
|
||||||
|
Use stable, dot-separated routing keys. Initial routing keys:
|
||||||
|
|
||||||
|
```text
|
||||||
|
ingestion.job.created
|
||||||
|
ingestion.job.retry_requested
|
||||||
|
maintenance.retention.requested
|
||||||
|
maintenance.erasure.requested
|
||||||
|
```
|
||||||
|
|
||||||
|
Create an application-owned durable topic exchange for dispatching work
|
||||||
|
messages, and durable (quorum-type) queues bound to it by pattern, for example:
|
||||||
|
|
||||||
|
```text
|
||||||
|
exchange: chatbot.jobs (topic, durable)
|
||||||
|
|
||||||
|
queue: ingestion.jobs <- binding pattern ingestion.job.*
|
||||||
|
queue: maintenance.jobs <- binding pattern maintenance.*
|
||||||
|
```
|
||||||
|
|
||||||
|
Multiple worker processes share a queue as competing consumers, so work is
|
||||||
|
distributed without each process independently receiving every message. Set
|
||||||
|
consumer prefetch (QoS) explicitly rather than relying on the client default, so
|
||||||
|
one slow consumer cannot starve the others or accumulate unbounded unacked
|
||||||
|
messages.
|
||||||
|
|
||||||
|
Ack messages only after the worker has persisted the resulting job state and
|
||||||
|
progress events to Postgres. Redelivery must be safe: ingestion processors should
|
||||||
|
be idempotent by `ingestion_job_id` and Qdrant point ids/upsert semantics.
|
||||||
|
|
||||||
|
### Run subscribers as worker processes, not FastAPI side effects
|
||||||
|
|
||||||
|
The outbox-publisher process owns the RabbitMQ publisher connection/channel and
|
||||||
|
publishes committed outbox events. FastAPI routes create durable outbox intent
|
||||||
|
only; they do not publish ingestion notifications directly.
|
||||||
|
|
||||||
|
Subscribers should run in separate worker process entrypoints. Do not start a
|
||||||
|
consumer inside every FastAPI web worker by default: with multiple
|
||||||
|
Uvicorn/Gunicorn workers, each process has its own lifespan and could create
|
||||||
|
surprising duplicate consumers or resource pressure.
|
||||||
|
|
||||||
|
Worker processes create their own application-lifetime resources at startup:
|
||||||
|
|
||||||
|
- SQLAlchemy engine/sessionmaker;
|
||||||
|
- MinIO/S3 object-storage client;
|
||||||
|
- Qdrant client;
|
||||||
|
- RabbitMQ connection/channel (aio-pika, via `aio_pika.connect_robust` for
|
||||||
|
automatic reconnection);
|
||||||
|
- observability/logging clients;
|
||||||
|
- ingestion model clients.
|
||||||
|
|
||||||
|
Each message gets its own request/job-lifetime SQLAlchemy `AsyncSession` and
|
||||||
|
explicit transaction boundaries, following ADR-0012.
|
||||||
|
|
||||||
|
### Use a transactional outbox for job dispatch
|
||||||
|
|
||||||
|
Use a transactional outbox to atomically record the ingestion job and the intent
|
||||||
|
to dispatch it. The HTTP publisher does not publish directly to RabbitMQ after
|
||||||
|
committing the ingestion-job transaction.
|
||||||
|
|
||||||
|
Instead, the same Postgres transaction creates or updates the durable job state
|
||||||
|
and inserts an unpublished outbox event:
|
||||||
|
|
||||||
|
```text
|
||||||
|
same DB transaction:
|
||||||
|
insert/update source_files
|
||||||
|
insert ingestion_jobs(status='queued')
|
||||||
|
insert outbox_events(type='ingestion.job.created', payload, published_at=NULL)
|
||||||
|
commit
|
||||||
|
```
|
||||||
|
|
||||||
|
A separate outbox publisher process reads unpublished events, publishes the
|
||||||
|
versioned message to RabbitMQ, and records the publication outcome:
|
||||||
|
|
||||||
|
```text
|
||||||
|
outbox publisher:
|
||||||
|
claim unpublished outbox event
|
||||||
|
publish to RabbitMQ (persistent message, publisher confirms enabled)
|
||||||
|
mark outbox event published once the broker confirms receipt
|
||||||
|
```
|
||||||
|
|
||||||
|
This removes the failure window where an `ingestion_jobs` row commits but the API
|
||||||
|
process crashes before recording that it must be dispatched. If the transaction
|
||||||
|
commits, the job and its durable dispatch intent both exist; the outbox publisher
|
||||||
|
can resume publication after a restart.
|
||||||
|
|
||||||
|
An outbox event must have a stable UUID event id. Include that id in the
|
||||||
|
RabbitMQ message so consumers and operators can correlate it back to the outbox
|
||||||
|
record. RabbitMQ has no broker-native publish-deduplication mechanism comparable
|
||||||
|
to some other brokers: publisher confirms guarantee the broker accepted the
|
||||||
|
message, but the publisher can still crash after the broker confirms and before
|
||||||
|
`published_at` is recorded, causing a duplicate publish on retry. There is no
|
||||||
|
broker-side safety net for that case here — **workers must be idempotent by
|
||||||
|
`ingestion_job_id`**, and Qdrant mutations must use deterministic point
|
||||||
|
ids/upsert semantics, as the sole guarantee against duplicate processing.
|
||||||
|
|
||||||
|
Outbox events are delivery records, not a replacement for `ingestion_jobs` or
|
||||||
|
`ingestion_job_events`. Postgres continues to own application job state and
|
||||||
|
progress. A stale-job scanner remains a repair/operational check for jobs or
|
||||||
|
outbox events that have not advanced as expected, not the primary dispatch path.
|
||||||
|
|
||||||
|
Add the `outbox_events` schema through an Alembic migration. Do not create or
|
||||||
|
alter the table at FastAPI startup, as required by ADR-0009.
|
||||||
|
|
||||||
|
### Configure retries, dead letters, and observability
|
||||||
|
|
||||||
|
Declare queues with an explicit dead-letter exchange (`x-dead-letter-exchange`)
|
||||||
|
pointing at a durable `chatbot.jobs.dlx` fanout exchange bound to a durable
|
||||||
|
dead-letter queue. A message lands there when a worker rejects/nacks it with
|
||||||
|
`requeue=False`. Workers should nack with `requeue=False` after a bounded local
|
||||||
|
retry count for a given delivery, rather than looping redelivery indefinitely.
|
||||||
|
Poison messages or repeatedly failing jobs should end as durable Postgres
|
||||||
|
failures with an `ingestion_job_events` error entry, and the dead-letter queue
|
||||||
|
gives operators a place to inspect the raw message for messages that failed
|
||||||
|
before any Postgres state could be written.
|
||||||
|
|
||||||
|
At worker ingress, bind structured logging context from durable identifiers:
|
||||||
|
|
||||||
|
- `tenant_id`;
|
||||||
|
- `request_id`;
|
||||||
|
- `api_key_id`;
|
||||||
|
- `file_id`;
|
||||||
|
- `ingestion_job_id`;
|
||||||
|
- RabbitMQ exchange/routing key/delivery tag when available.
|
||||||
|
|
||||||
|
Use stable event names such as:
|
||||||
|
|
||||||
|
- `ingestion.job.dispatched`;
|
||||||
|
- `ingestion.job.received`;
|
||||||
|
- `ingestion.job.completed`;
|
||||||
|
- `ingestion.job.failed`;
|
||||||
|
- `broker.publish.failed`;
|
||||||
|
- `broker.message.redelivered`.
|
||||||
|
|
||||||
|
### Do not queue chat runs through RabbitMQ yet
|
||||||
|
|
||||||
|
ADR-0007 and ADR-0008 intentionally reject queueing concurrent chat runs per
|
||||||
|
thread: one run may be in flight, and concurrent runs return `409 Conflict`.
|
||||||
|
This ADR does not change that decision.
|
||||||
|
|
||||||
|
A future ADR may introduce broker-backed work for chat-adjacent tasks such as
|
||||||
|
summarization, retention, erasure, or offline evaluation, but synchronous chat run
|
||||||
|
execution remains owned by the FastAPI/LangGraph boundary unless superseded.
|
||||||
|
|
||||||
|
## Consequences
|
||||||
|
|
||||||
|
### Positive
|
||||||
|
- File ingestion can move out of HTTP request handling into scalable worker
|
||||||
|
processes without changing the REST contract from ADR-0008.
|
||||||
|
- RabbitMQ gives durable delivery, manual ack/redelivery, per-queue
|
||||||
|
dead-lettering, and competing-consumer work distribution without adopting
|
||||||
|
Kafka-level operational complexity.
|
||||||
|
- Postgres remains the auditable source of truth for jobs, progress, tenant
|
||||||
|
ownership, failure state, and the durable intent to dispatch each job.
|
||||||
|
- The transactional outbox prevents a committed job from losing its initial
|
||||||
|
dispatch intent when the API process fails before a direct broker publish.
|
||||||
|
- MinIO, Postgres, Qdrant, and the broker each have distinct responsibilities:
|
||||||
|
bytes, metadata/state, vector search, and work delivery.
|
||||||
|
- Separate worker processes avoid surprising subscriptions in every FastAPI web
|
||||||
|
worker and make GPU/model-heavy ingestion resources easier to size.
|
||||||
|
|
||||||
|
### Negative
|
||||||
|
- Adds another external service to run, secure, monitor, and back up where
|
||||||
|
queue/exchange durability is required.
|
||||||
|
- Developers must reason about redelivery and idempotency; handlers may run more
|
||||||
|
than once for the same `ingestion_job_id`.
|
||||||
|
- Adds an `outbox_events` table, Alembic migration, outbox publisher process, and
|
||||||
|
monitoring for unpublished or stuck events.
|
||||||
|
- RabbitMQ has no broker-native publish deduplication. Unlike brokers that offer
|
||||||
|
one, duplicate publication after a publisher crash between broker confirmation
|
||||||
|
and outbox-event marking is not mitigated by the broker at all; idempotent
|
||||||
|
consumers are the only protection.
|
||||||
|
- RabbitMQ exchange, queue, binding, and dead-letter-exchange configuration must
|
||||||
|
be managed explicitly per environment.
|
||||||
|
- The app now needs web, ingestion-worker, and outbox-publisher deployment/runbook
|
||||||
|
conventions.
|
||||||
|
|
||||||
|
## Alternatives Considered
|
||||||
|
|
||||||
|
- **NATS JetStream**: rejected. JetStream keeps the broker footprint lighter and
|
||||||
|
offers built-in publish deduplication, but the project wants RabbitMQ's more
|
||||||
|
mature work-queue/routing model (topic exchanges, per-queue DLX, prefetch-based
|
||||||
|
fair dispatch) and the aio-pika SDK for asyncio integration.
|
||||||
|
- **Redis Streams**: rejected for this decision. Redis Streams would work for an
|
||||||
|
MVP and has simple local deployment, but RabbitMQ is a cleaner dedicated broker
|
||||||
|
for durable pub/sub and worker coordination without also mixing cache
|
||||||
|
responsibilities into the same service.
|
||||||
|
- **Kafka**: rejected. Kafka is an excellent durable event log at high scale, but
|
||||||
|
its operational overhead is unnecessary for this service's initial ingestion
|
||||||
|
and maintenance jobs.
|
||||||
|
- **FastAPI `BackgroundTasks` or in-process asyncio tasks**: rejected as the
|
||||||
|
durable design. They are simple, but work can be lost on process restart and
|
||||||
|
they do not coordinate multiple workers cleanly.
|
||||||
|
- **Post-commit direct RabbitMQ publish plus a repair scanner**: rejected as the
|
||||||
|
primary dispatch design. It is simpler, but a crash after the job transaction
|
||||||
|
commits and before the publish is recorded can strand a queued job until the
|
||||||
|
scanner finds it. The transactional outbox persists the dispatch intent in the
|
||||||
|
same transaction as the job.
|
||||||
|
- **Database polling only**: rejected as the main dispatch mechanism. Polling
|
||||||
|
Postgres for queued jobs is a useful repair path, but relying on polling alone
|
||||||
|
adds latency and unnecessary database load once a broker is available.
|
||||||
|
- **Use the broker as the source of truth for job state**: rejected. Broker state
|
||||||
|
is delivery state. Application job state, progress, audit, and tenant ownership
|
||||||
|
remain in Postgres.
|
||||||
277
docs/adr/0015-modular-monolith-package-architecture.md
Normal file
277
docs/adr/0015-modular-monolith-package-architecture.md
Normal file
@@ -0,0 +1,277 @@
|
|||||||
|
# 0015. Modular monolith package architecture
|
||||||
|
|
||||||
|
## Status
|
||||||
|
|
||||||
|
Proposed
|
||||||
|
|
||||||
|
## Context
|
||||||
|
|
||||||
|
The repository currently contains only a small FastAPI-oriented scaffold under
|
||||||
|
`src/`, while the ADRs define several substantial capabilities and external
|
||||||
|
boundaries: FastAPI HTTP endpoints, Postgres and Alembic, MinIO object storage,
|
||||||
|
RabbitMQ with a transactional outbox, Qdrant point/retrieval operations,
|
||||||
|
LangGraph conversational execution, and separate worker processes.
|
||||||
|
|
||||||
|
Without an explicit package structure, implementation can drift toward route
|
||||||
|
handlers that call SDK clients directly, workers that duplicate HTTP logic, and
|
||||||
|
generic catch-all directories such as `utils`, `services`, or `clients`. That
|
||||||
|
would make tenant isolation, transactions, retries, resource ownership, and tests
|
||||||
|
harder to apply consistently.
|
||||||
|
|
||||||
|
ADR-0012 requires application-lifetime resource ownership and explicit dependency
|
||||||
|
passing. ADR-0014 requires workers to own ingestion-generated Chunk/Point CRUD,
|
||||||
|
while direct point routes remain synchronous API work. The repository structure
|
||||||
|
must support both entrypoints reusing the same application behavior without
|
||||||
|
coupling worker code to FastAPI routes or LangGraph nodes to SDK details.
|
||||||
|
|
||||||
|
## Decision
|
||||||
|
|
||||||
|
Use a **modular monolith with explicit infrastructure adapters**. The application
|
||||||
|
uses one deployable codebase with separate web, worker, and outbox-publisher
|
||||||
|
process entrypoints. Business/application behavior is grouped by capability;
|
||||||
|
external systems are isolated behind infrastructure adapters.
|
||||||
|
|
||||||
|
Use `src/` as the explicit Python package root.
|
||||||
|
|
||||||
|
### Package layout
|
||||||
|
|
||||||
|
Create packages as they become necessary, following this structure:
|
||||||
|
|
||||||
|
```text
|
||||||
|
src/
|
||||||
|
├── __init__.py
|
||||||
|
├── main.py
|
||||||
|
├── config.py
|
||||||
|
├── bootstrap/
|
||||||
|
│ ├── lifespan.py
|
||||||
|
│ └── dependencies.py
|
||||||
|
├── api/
|
||||||
|
│ ├── dependencies/
|
||||||
|
│ ├── routers/
|
||||||
|
│ ├── schemas/
|
||||||
|
│ └── router.py
|
||||||
|
├── application/
|
||||||
|
│ ├── files/
|
||||||
|
│ ├── ingestion/
|
||||||
|
│ ├── points/
|
||||||
|
│ ├── retrieval/
|
||||||
|
│ ├── threads/
|
||||||
|
│ └── ports/
|
||||||
|
├── agent/
|
||||||
|
│ ├── graph.py
|
||||||
|
│ ├── state.py
|
||||||
|
│ ├── nodes/
|
||||||
|
│ ├── prompts/
|
||||||
|
│ ├── tools/
|
||||||
|
│ └── persistence.py
|
||||||
|
├── infrastructure/
|
||||||
|
│ ├── postgres/
|
||||||
|
│ │ ├── models/
|
||||||
|
│ │ ├── repositories/
|
||||||
|
│ │ ├── database.py
|
||||||
|
│ │ └── outbox.py
|
||||||
|
│ ├── qdrant/
|
||||||
|
│ ├── minio/
|
||||||
|
│ ├── rabbitmq/
|
||||||
|
│ ├── langgraph/
|
||||||
|
│ └── observability/
|
||||||
|
├── messaging/
|
||||||
|
│ ├── events.py
|
||||||
|
│ ├── subjects.py
|
||||||
|
│ └── outbox_publisher.py
|
||||||
|
└── workers/
|
||||||
|
├── ingestion.py
|
||||||
|
└── maintenance.py
|
||||||
|
```
|
||||||
|
|
||||||
|
Keep Alembic configuration and migration revisions at the repository root:
|
||||||
|
|
||||||
|
```text
|
||||||
|
alembic.ini
|
||||||
|
alembic/
|
||||||
|
├── env.py
|
||||||
|
└── versions/
|
||||||
|
```
|
||||||
|
|
||||||
|
Keep tests outside the application package and organize them by testing boundary:
|
||||||
|
|
||||||
|
```text
|
||||||
|
tests/
|
||||||
|
├── unit/
|
||||||
|
│ ├── application/
|
||||||
|
│ └── agent/
|
||||||
|
├── integration/
|
||||||
|
│ ├── postgres/
|
||||||
|
│ ├── minio/
|
||||||
|
│ ├── rabbitmq/
|
||||||
|
│ └── qdrant/
|
||||||
|
└── e2e/
|
||||||
|
```
|
||||||
|
|
||||||
|
### Dependency direction
|
||||||
|
|
||||||
|
Entry adapters call application services; application services depend on typed
|
||||||
|
ports/contracts; infrastructure packages implement those ports.
|
||||||
|
|
||||||
|
```text
|
||||||
|
FastAPI routes / RabbitMQ workers / LangGraph nodes
|
||||||
|
-> application services
|
||||||
|
-> application ports
|
||||||
|
-> infrastructure adapters
|
||||||
|
```
|
||||||
|
|
||||||
|
The reverse direction is prohibited:
|
||||||
|
|
||||||
|
- infrastructure adapters do not import FastAPI routers, workers, or graph nodes;
|
||||||
|
- workers do not call FastAPI route functions;
|
||||||
|
- LangGraph nodes do not call API route functions or construct mutable SDK clients;
|
||||||
|
- application services do not import concrete MinIO, RabbitMQ, Qdrant, or
|
||||||
|
SQLAlchemy client construction code;
|
||||||
|
- routes do not call raw Qdrant, MinIO, or RabbitMQ SDK methods directly.
|
||||||
|
|
||||||
|
Use ports selectively for external side effects and persistence boundaries; do not
|
||||||
|
add interfaces around pure local functions merely to satisfy a pattern.
|
||||||
|
|
||||||
|
### API package
|
||||||
|
|
||||||
|
`api/` is the HTTP adapter only:
|
||||||
|
|
||||||
|
- `routers/` map HTTP operations to application-service calls;
|
||||||
|
- `dependencies/` resolve request-lifetime objects such as `AuthContext` and
|
||||||
|
`AsyncSession`;
|
||||||
|
- `schemas/` contains public Pydantic request/response/error models;
|
||||||
|
- `router.py` composes versioned route groups.
|
||||||
|
|
||||||
|
API schemas are separate from SQLAlchemy ORM models and RabbitMQ message schemas.
|
||||||
|
Routes perform HTTP validation and response mapping, but not parsing, embedding,
|
||||||
|
Qdrant mutations, or transaction-independent business workflows.
|
||||||
|
|
||||||
|
### Application package
|
||||||
|
|
||||||
|
`application/` contains reusable use-case behavior. It has no FastAPI request
|
||||||
|
objects, RabbitMQ consumer loops, or SDK client construction.
|
||||||
|
|
||||||
|
- `files/` creates source-file records, validates lifecycle actions, and returns
|
||||||
|
file/job status.
|
||||||
|
- `ingestion/` performs parse/chunk/embed/index orchestration after the worker
|
||||||
|
receives a job.
|
||||||
|
- `points/` applies tenant-aware direct Point CRUD rules and shared generated-point
|
||||||
|
mutation behavior.
|
||||||
|
- `retrieval/` owns retrieval use cases used by the graph; it does not expose raw
|
||||||
|
Qdrant SDK details.
|
||||||
|
- `threads/` coordinates run-level application behavior without taking ownership
|
||||||
|
of conversation/session records reserved for the main backend and LangGraph.
|
||||||
|
- `ports/` defines narrow contracts for external side effects, including object
|
||||||
|
storage, message publishing, point storage, and repositories where useful.
|
||||||
|
|
||||||
|
Both FastAPI routes and worker consumers call these services. This prevents a
|
||||||
|
second, inconsistent ingestion implementation from growing inside `workers/`.
|
||||||
|
|
||||||
|
### Agent package
|
||||||
|
|
||||||
|
All LangGraph-specific application graph code lives in `agent/`:
|
||||||
|
|
||||||
|
- `graph.py` builds and compiles the graph from explicitly passed dependencies;
|
||||||
|
- `state.py` defines graph state and graph-facing result types;
|
||||||
|
- `nodes/` contains focused graph-node behavior such as triage, retrieval,
|
||||||
|
generation, verification, and memory extraction;
|
||||||
|
- `prompts/` holds prompt identifiers/templates or prompt access helpers;
|
||||||
|
- `tools/` contains graph tool definitions;
|
||||||
|
- `persistence.py` contains graph-facing persistence configuration/types.
|
||||||
|
|
||||||
|
Concrete `AsyncPostgresSaver` and `AsyncPostgresStore` setup belongs in
|
||||||
|
`infrastructure/langgraph/`, then is passed into the graph factory during
|
||||||
|
bootstrap. Graph nodes call application services, particularly
|
||||||
|
`application/retrieval/`, instead of embedding Qdrant query logic.
|
||||||
|
|
||||||
|
### Infrastructure package
|
||||||
|
|
||||||
|
`infrastructure/` contains concrete integrations and resource setup.
|
||||||
|
|
||||||
|
- `postgres/` owns SQLAlchemy engine/sessionmaker setup, ORM models, repository
|
||||||
|
implementations, and transactional outbox persistence.
|
||||||
|
- `qdrant/` owns Qdrant client lifecycle, collection/bootstrap helpers, low-level
|
||||||
|
point operations, and hybrid retrieval adapter mechanics.
|
||||||
|
- `minio/` implements object-storage operations against MinIO/S3-compatible APIs.
|
||||||
|
- `rabbitmq/` owns the RabbitMQ connection/channel lifecycle plus low-level
|
||||||
|
publish and consumer adapters (aio-pika).
|
||||||
|
- `langgraph/` configures the concrete Postgres-backed LangGraph persistence
|
||||||
|
adapters.
|
||||||
|
- `observability/` configures structlog and Langfuse integrations.
|
||||||
|
|
||||||
|
Infrastructure code receives configuration and is created by a process owner; it
|
||||||
|
must not create mutable external clients at import time.
|
||||||
|
|
||||||
|
### Messaging and workers
|
||||||
|
|
||||||
|
`messaging/` contains versioned event schemas, stable routing-key names, and the
|
||||||
|
outbox-publisher orchestration. The outbox publisher coordinates Postgres outbox
|
||||||
|
records with the RabbitMQ adapter; it does not become a second source of job
|
||||||
|
state.
|
||||||
|
|
||||||
|
`workers/` contains thin process entrypoints and consumer loops. A worker creates
|
||||||
|
application-lifetime dependencies, consumes a durable RabbitMQ message, binds
|
||||||
|
job logging context, and invokes the corresponding application service. It does
|
||||||
|
not hold parsing/chunking/Qdrant business logic itself.
|
||||||
|
|
||||||
|
For example:
|
||||||
|
|
||||||
|
```text
|
||||||
|
workers/ingestion.py
|
||||||
|
-> application/ingestion/processor.py
|
||||||
|
-> application/points/service.py
|
||||||
|
-> infrastructure/qdrant/points.py
|
||||||
|
```
|
||||||
|
|
||||||
|
### Bootstrap and resource ownership
|
||||||
|
|
||||||
|
`bootstrap/` composes configuration and concrete infrastructure adapters for each
|
||||||
|
process entrypoint. FastAPI lifespan owns web-process resources; worker and outbox
|
||||||
|
publisher startup own their corresponding resources. This implements ADR-0012
|
||||||
|
without turning `app.state` or module globals into an untyped service locator.
|
||||||
|
|
||||||
|
## Consequences
|
||||||
|
|
||||||
|
### Positive
|
||||||
|
|
||||||
|
- The repository has clear homes for LangGraph, Qdrant, MinIO, RabbitMQ, Postgres,
|
||||||
|
API, worker, and outbox-publisher code before implementation grows.
|
||||||
|
- HTTP routes, worker consumers, and LangGraph nodes reuse application services
|
||||||
|
while remaining separate transport/execution adapters.
|
||||||
|
- SDK-specific details are isolated, making integration tests and test doubles
|
||||||
|
practical without hiding all code behind unnecessary abstractions.
|
||||||
|
- The layout directly supports ADR-0012's explicit resource ownership and
|
||||||
|
ADR-0014's worker-owned ingestion Chunk/Point CRUD.
|
||||||
|
- A single codebase remains simple to deploy while allowing web, worker, and
|
||||||
|
outbox-publisher processes to scale independently.
|
||||||
|
|
||||||
|
### Negative
|
||||||
|
|
||||||
|
- The initial directory structure is more elaborate than a route-plus-models
|
||||||
|
FastAPI starter application.
|
||||||
|
- Developers must maintain dependency direction rather than importing a concrete
|
||||||
|
client wherever it is convenient.
|
||||||
|
- Some capabilities span several packages by design, for example an upload route,
|
||||||
|
application service, MinIO adapter, outbox repository, and publisher process.
|
||||||
|
- Ports/contracts should remain narrow; excessive abstraction would add ceremony
|
||||||
|
without improving testability or substitutability.
|
||||||
|
|
||||||
|
## Alternatives Considered
|
||||||
|
|
||||||
|
- **Technology-first packages only**: rejected. Directories such as `db`,
|
||||||
|
`qdrant`, `rabbitmq`, and `langgraph` are immediately discoverable, but feature
|
||||||
|
workflows become scattered across every integration package and encourage
|
||||||
|
transport adapters to own business behavior.
|
||||||
|
- **Feature-first packages only**: rejected. Keeping all file, point, and thread
|
||||||
|
code together is attractive, but it obscures ownership of shared external
|
||||||
|
clients and risks duplicating infrastructure integration logic across features.
|
||||||
|
- **Full clean architecture with interfaces for every class/function**: rejected.
|
||||||
|
The application needs clear external boundaries, not abstraction around pure
|
||||||
|
helper functions. Ports are reserved for persistence and external side effects.
|
||||||
|
- **Microservices for ingestion, retrieval, and chat from the start**: rejected.
|
||||||
|
The project needs independent web/worker processes, but a single modular
|
||||||
|
codebase avoids premature network boundaries, deployment complexity, and
|
||||||
|
distributed transaction concerns.
|
||||||
|
- **Put LangGraph under `api/` or workers under `ingestion/` only**: rejected.
|
||||||
|
LangGraph and RabbitMQ workers are independent execution adapters; putting one
|
||||||
|
under another would invert dependencies and make reuse/testing harder.
|
||||||
215
docs/adr/0016-testing-strategy-and-quality-gates.md
Normal file
215
docs/adr/0016-testing-strategy-and-quality-gates.md
Normal file
@@ -0,0 +1,215 @@
|
|||||||
|
# 0016. Testing strategy and quality gates
|
||||||
|
|
||||||
|
## Status
|
||||||
|
|
||||||
|
Proposed
|
||||||
|
|
||||||
|
## Context
|
||||||
|
|
||||||
|
The project has ADRs for tenant-scoped ingestion, explicit resource ownership,
|
||||||
|
MinIO object storage, transactional outbox dispatch, RabbitMQ workers,
|
||||||
|
Qdrant indexing, and a modular monolith. It has no test runner, test fixtures,
|
||||||
|
or executable test suite yet.
|
||||||
|
|
||||||
|
The first CSV ingestion slice has correctness properties that cannot be left to
|
||||||
|
manual testing: Alembic is the only schema-management path; tenant identity is
|
||||||
|
trusted server-side context; a file, job, and outbox event commit atomically;
|
||||||
|
workers are safe under at-least-once delivery; and generated Qdrant points are
|
||||||
|
idempotent and tenant-filtered. ADR-0015 already reserves a test layout by
|
||||||
|
boundary, while ADR-0012 requires explicit dependencies and resource lifetimes
|
||||||
|
that should make tests practical without import-time client patching.
|
||||||
|
|
||||||
|
Tests need to give fast feedback during implementation without replacing
|
||||||
|
integration coverage with mocks or making routine development depend on Docker,
|
||||||
|
provider credentials, live models, or Langfuse availability.
|
||||||
|
|
||||||
|
ADR-0014's transactional-outbox decision controls ingestion dispatch. The upload
|
||||||
|
path records durable dispatch intent; a separate outbox publisher publishes it.
|
||||||
|
|
||||||
|
## Decision
|
||||||
|
|
||||||
|
### Use pytest with explicit async support
|
||||||
|
|
||||||
|
Use pytest as the test runner. Add `pytest`, `pytest-asyncio`, `httpx`,
|
||||||
|
`asgi-lifespan`, `testcontainers`, `pytest-timeout`, and `pytest-cov` as
|
||||||
|
development dependencies.
|
||||||
|
|
||||||
|
Configure `pytest-asyncio` in strict mode. Async tests and fixtures must be
|
||||||
|
explicit. FastAPI tests use `httpx.AsyncClient`, `ASGITransport`, and
|
||||||
|
`LifespanManager` so they exercise application startup and shutdown according to
|
||||||
|
ADR-0012. `FastAPI.TestClient` is not the default project test client.
|
||||||
|
|
||||||
|
Register these markers:
|
||||||
|
|
||||||
|
- one primary boundary marker per test: `unit`, `integration`, or `e2e`;
|
||||||
|
- `postgres`, `minio`, `rabbitmq`, or `qdrant` for the real service used by an
|
||||||
|
integration test;
|
||||||
|
- `slow` only where a test materially exceeds the normal integration feedback
|
||||||
|
target;
|
||||||
|
- `live_provider` for an opt-in, credential-gated external-provider smoke test.
|
||||||
|
|
||||||
|
Name tests `test_<unit>_<scenario>_<outcome>` and use Arrange–Act–Assert.
|
||||||
|
|
||||||
|
### Test by architectural boundary
|
||||||
|
|
||||||
|
Use the test layout reserved by ADR-0015, with shared support for fixtures and
|
||||||
|
assertions:
|
||||||
|
|
||||||
|
```text
|
||||||
|
tests/
|
||||||
|
├── conftest.py
|
||||||
|
├── fakes.py
|
||||||
|
├── support/
|
||||||
|
│ ├── factories.py
|
||||||
|
│ └── assertions.py
|
||||||
|
├── unit/
|
||||||
|
│ ├── application/
|
||||||
|
│ └── agent/
|
||||||
|
├── integration/
|
||||||
|
│ ├── postgres/
|
||||||
|
│ ├── minio/
|
||||||
|
│ ├── rabbitmq/
|
||||||
|
│ └── qdrant/
|
||||||
|
└── e2e/
|
||||||
|
```
|
||||||
|
|
||||||
|
- **Unit tests** are the default development feedback loop. They cover pure
|
||||||
|
application policy, validation, lifecycle transitions, tenant propagation,
|
||||||
|
deterministic IDs, CSV chunking, event construction, and LangGraph
|
||||||
|
control-flow policy.
|
||||||
|
- **Integration tests** validate a production infrastructure adapter against its
|
||||||
|
real backing service and its deployment-relevant behavior.
|
||||||
|
- **End-to-end tests** prove a small vertical-slice acceptance contract through
|
||||||
|
the real application composition. They do not replace faster unit or adapter
|
||||||
|
tests.
|
||||||
|
|
||||||
|
Hand-written fakes and spies implement narrow application-owned ports, not
|
||||||
|
MinIO, RabbitMQ, Qdrant, or model SDK-shaped interfaces. Scripted model, embedder,
|
||||||
|
retrieval, clock, and UUID fakes make normal test runs deterministic.
|
||||||
|
|
||||||
|
### Apply pragmatic TDD
|
||||||
|
|
||||||
|
For application behavior, HTTP contracts, database migrations, reliability
|
||||||
|
rules, and defects, first write a focused failing test that describes the
|
||||||
|
observable requirement. Make the smallest change that passes it, then refactor
|
||||||
|
while the relevant suite is green.
|
||||||
|
|
||||||
|
TDD applies to behavior and regressions, not as an artificial ritual for pure
|
||||||
|
refactors or configuration-only changes with no observable behavior change.
|
||||||
|
Those changes must preserve and extend existing relevant coverage as needed.
|
||||||
|
|
||||||
|
Before introducing a concrete adapter, write tests for the consuming
|
||||||
|
application port. Before each Alembic migration, write the empty-database
|
||||||
|
migration test or extend the existing migration test. Add the corresponding
|
||||||
|
real-adapter integration test before declaring that boundary complete.
|
||||||
|
|
||||||
|
### Use disposable real infrastructure in integration tests
|
||||||
|
|
||||||
|
Use Testcontainers as the standard automated integration-test resource mechanism
|
||||||
|
for Postgres, MinIO, RabbitMQ, and Qdrant.
|
||||||
|
|
||||||
|
- Tests never connect to a developer's local services or Langfuse-owned storage
|
||||||
|
and credentials.
|
||||||
|
- Start containers at suite or session scope, then isolate data per test with
|
||||||
|
unique data, object-key prefixes, exchange/queue/binding names, and collection
|
||||||
|
names.
|
||||||
|
- Disable parallel integration execution until fixture isolation and cleanup are
|
||||||
|
proven worker-safe.
|
||||||
|
- Fixtures expose typed settings or connection values. Tests create production
|
||||||
|
adapters through their normal constructors.
|
||||||
|
|
||||||
|
Docker Compose remains the mechanism for manual local validation and a later,
|
||||||
|
serialized operational smoke test where web, outbox-publisher, and worker run as
|
||||||
|
independent processes. It is not the default pytest fixture mechanism.
|
||||||
|
|
||||||
|
### Treat invariants as reusable contracts
|
||||||
|
|
||||||
|
Test the following requirements at the applicable application, adapter, and E2E
|
||||||
|
boundaries:
|
||||||
|
|
||||||
|
- Tenant identity comes from server-side authenticated context. Request payloads,
|
||||||
|
query parameters, object metadata, and broker messages cannot override it.
|
||||||
|
- Cross-tenant access does not disclose tenant-owned data. Public routes normally
|
||||||
|
return `404` for inaccessible resources.
|
||||||
|
- Alembic creates the schema from an empty database. Tests never use
|
||||||
|
`Base.metadata.create_all()`, and FastAPI startup performs readiness checks only,
|
||||||
|
never DDL.
|
||||||
|
- The upload transaction records `source_files`, a queued `ingestion_jobs` row,
|
||||||
|
and an unpublished `outbox_events` row atomically. The HTTP route does not
|
||||||
|
directly publish the ingestion event.
|
||||||
|
- Broker messages contain durable identifiers and correlation metadata only. The
|
||||||
|
worker reloads job and source-file records from Postgres before tenant-scoped
|
||||||
|
side effects.
|
||||||
|
- Worker acknowledgement follows durable progress or terminal-state persistence.
|
||||||
|
Duplicate publication and redelivery do not regress terminal jobs, inflate
|
||||||
|
counters, or create duplicate logical chunks.
|
||||||
|
- MinIO keys are server-derived internal paths. Qdrant reads and mutations use a
|
||||||
|
server-derived tenant filter, deterministic point IDs, and upsert semantics.
|
||||||
|
|
||||||
|
### Keep correctness tests separate from model-quality evaluation
|
||||||
|
|
||||||
|
Pytest verifies deterministic application behavior: graph policy, schemas,
|
||||||
|
redaction, correlation metadata, retry budgets, and graceful observability
|
||||||
|
failure handling. Normal pytest runs never call a paid or live model provider.
|
||||||
|
|
||||||
|
Langfuse datasets and experiments evaluate prompts, model behavior, retrieval,
|
||||||
|
citations, and response quality. They are promotion evidence, not a replacement
|
||||||
|
for application correctness tests. Live-provider smoke tests, if introduced,
|
||||||
|
are opt-in, credential-gated, rate-limited, and excluded from ordinary local and
|
||||||
|
pull-request runs.
|
||||||
|
|
||||||
|
### Establish phased quality gates
|
||||||
|
|
||||||
|
Initially require Ruff format checking, Ruff linting, Ty type checking, and unit
|
||||||
|
tests. Require Docker-capable integration tests as their adapters are
|
||||||
|
implemented. Run the separate-process Compose E2E smoke as a serialized
|
||||||
|
pre-release or scheduled gate until it is reliable enough for every pull request.
|
||||||
|
|
||||||
|
Collect coverage reports but do not set a percentage threshold before the first
|
||||||
|
vertical slice has meaningful implementation. Later introduce a scoped,
|
||||||
|
ratcheting threshold rather than encouraging low-value coverage.
|
||||||
|
|
||||||
|
## Consequences
|
||||||
|
|
||||||
|
### Positive
|
||||||
|
|
||||||
|
- Unit tests provide fast, deterministic TDD feedback for core application
|
||||||
|
behavior.
|
||||||
|
- Real-service tests cover the behaviors least safe to simulate: Alembic
|
||||||
|
migrations, object storage, RabbitMQ acknowledgements/redelivery, and Qdrant
|
||||||
|
filtering/upserts.
|
||||||
|
- Explicit fakes reinforce the dependency direction and resource ownership rules
|
||||||
|
from ADR-0012 and ADR-0015.
|
||||||
|
- The ingestion path has concrete tenant-isolation and reliability contracts,
|
||||||
|
rather than only a happy-path demonstration.
|
||||||
|
- Live model quality can improve through Langfuse experiments without making
|
||||||
|
application tests nondeterministic or expensive.
|
||||||
|
|
||||||
|
### Negative
|
||||||
|
|
||||||
|
- Docker is required for integration and E2E suites.
|
||||||
|
- Testcontainers add setup time and require careful fixture cleanup.
|
||||||
|
- Maintaining real-service coverage and test data isolation adds engineering
|
||||||
|
effort.
|
||||||
|
- E2E tests do not prove semantic quality of LLM responses; that remains an
|
||||||
|
evaluation responsibility.
|
||||||
|
|
||||||
|
## Alternatives Considered
|
||||||
|
|
||||||
|
- **Mock all external SDKs**: rejected. Mocks cannot prove migrations, real
|
||||||
|
RabbitMQ acknowledgement/redelivery behavior, MinIO semantics, or Qdrant
|
||||||
|
tenant filtering.
|
||||||
|
- **Use full-stack Compose tests only**: rejected. They are slow and opaque for
|
||||||
|
the default development loop and make failures difficult to localize.
|
||||||
|
- **Run all integration containers on every pytest invocation**: rejected. Test
|
||||||
|
boundaries should be selected deliberately for fast local feedback.
|
||||||
|
- **Use live model providers in routine tests**: rejected because they are
|
||||||
|
nondeterministic, costly, slow, credential-dependent, and hard to assert.
|
||||||
|
- **Use Langfuse as the primary regression-test runner**: rejected. Langfuse is
|
||||||
|
the quality/evaluation plane; pytest remains the deterministic application test
|
||||||
|
framework.
|
||||||
|
- **Create schemas with `Base.metadata.create_all()` in fixtures**: rejected. It
|
||||||
|
bypasses the production Alembic migration path.
|
||||||
|
- **Require strict test-first work for every non-behavioral refactor**: rejected.
|
||||||
|
TDD should protect observable behavior and regressions, not add ceremony where
|
||||||
|
no behavior changes.
|
||||||
287
docs/plans/001-ingestion-vertical-slice.md
Normal file
287
docs/plans/001-ingestion-vertical-slice.md
Normal file
@@ -0,0 +1,287 @@
|
|||||||
|
# 001. Ingestion vertical-slice implementation plan
|
||||||
|
|
||||||
|
## Purpose
|
||||||
|
|
||||||
|
This plan turns the accepted architectural direction in the ADRs into the first
|
||||||
|
working product slice: a tenant-scoped CSV upload is stored in MinIO, represented
|
||||||
|
by durable Postgres records, dispatched through RabbitMQ using a
|
||||||
|
transactional outbox, processed by a separate worker, and indexed as Qdrant
|
||||||
|
points.
|
||||||
|
|
||||||
|
This is an implementation plan, not an Architecture Decision Record. ADRs explain
|
||||||
|
why major technologies and boundaries were chosen; this document defines the
|
||||||
|
order, scope, and verification criteria for implementing them.
|
||||||
|
|
||||||
|
## Architecture baseline
|
||||||
|
|
||||||
|
The first vertical slice uses these responsibilities:
|
||||||
|
|
||||||
|
| System | Responsibility |
|
||||||
|
|---|---|
|
||||||
|
| FastAPI | HTTP boundary, validation, authentication, tenant derivation, and job creation. |
|
||||||
|
| Postgres | Tenant/auth data, source-file metadata, ingestion job state/progress, audit, and transactional outbox events. |
|
||||||
|
| MinIO | Private source-file bytes and retained derived ingestion blobs. |
|
||||||
|
| RabbitMQ | Durable delivery of ingestion and maintenance work. |
|
||||||
|
| Ingestion worker | Parsing, chunking, embedding, ingestion-generated Chunk/Point CRUD, and job status updates. |
|
||||||
|
| Qdrant | Tenant-filtered generated chunks and their vectors/payloads. |
|
||||||
|
|
||||||
|
The controlling ADRs are:
|
||||||
|
|
||||||
|
- [ADR-0008](../adr/0008-rest-api-and-fastapi-boundary.md): FastAPI REST boundary
|
||||||
|
and job-shaped file ingestion contract.
|
||||||
|
- [ADR-0009](../adr/0009-postgres-sqlalchemy-alembic-schema.md): Postgres source
|
||||||
|
files, jobs, audit, and migration conventions.
|
||||||
|
- [ADR-0012](../adr/0012-application-resource-lifetime-and-dependency-ownership.md):
|
||||||
|
resource lifetime, dependency injection, and explicit transaction ownership.
|
||||||
|
- [ADR-0013](../adr/0013-s3-compatible-object-storage-with-minio.md): MinIO object
|
||||||
|
storage boundary.
|
||||||
|
- [ADR-0014](../adr/0014-durable-job-dispatch-with-rabbitmq.md): RabbitMQ,
|
||||||
|
transactional outbox, separate workers, and worker-owned ingestion Chunk/Point
|
||||||
|
CRUD.
|
||||||
|
|
||||||
|
The cited ADRs are currently proposed. Treat them as the implementation baseline
|
||||||
|
only after the project owner accepts them; code should not silently diverge from
|
||||||
|
them.
|
||||||
|
|
||||||
|
## First release scope
|
||||||
|
|
||||||
|
### In scope
|
||||||
|
|
||||||
|
- `POST /v1/files` for authenticated tenant-scoped **CSV** upload.
|
||||||
|
- File validation, size limits, content hashing, and streaming upload to MinIO.
|
||||||
|
- Alembic-managed Postgres schema for the minimal tenant/auth, source file,
|
||||||
|
ingestion job, job event, and outbox records needed by this slice.
|
||||||
|
- Transactional outbox publication of `ingestion.job.created` to RabbitMQ.
|
||||||
|
- A separate ingestion worker process with a durable RabbitMQ consumer.
|
||||||
|
- CSV parsing and deterministic chunk creation.
|
||||||
|
- Tenant-filtered Qdrant point upserts using deterministic point identifiers.
|
||||||
|
- Job status/progress persistence and `GET /v1/files/{file_id}` status lookup.
|
||||||
|
- Structured correlation logging at HTTP, outbox, and worker ingress.
|
||||||
|
- Automated tests for the critical state transitions, redelivery, and tenant
|
||||||
|
boundaries.
|
||||||
|
|
||||||
|
### Explicitly out of scope
|
||||||
|
|
||||||
|
- XLSX, DOCX, and legacy DOC ingestion.
|
||||||
|
- The conversational LangGraph API and SSE streaming.
|
||||||
|
- Final reranker selection, GPU deployment, or unresolved model licensing from
|
||||||
|
ADR-0005.
|
||||||
|
- Direct `/v1/points` CRUD endpoints beyond the reusable service layer required by
|
||||||
|
ingestion.
|
||||||
|
- Full tenant erasure and hard-deletion workflow.
|
||||||
|
- A public download API or presigned object URLs.
|
||||||
|
- Exactly-once end-to-end processing. The worker must instead be safe under
|
||||||
|
at-least-once delivery.
|
||||||
|
|
||||||
|
## Required invariants
|
||||||
|
|
||||||
|
The implementation must preserve these rules from the ADRs:
|
||||||
|
|
||||||
|
1. `tenant_id` is derived from trusted authentication context; it is never
|
||||||
|
accepted from the upload body, query parameters, MinIO metadata, or a broker
|
||||||
|
message as authority.
|
||||||
|
2. MinIO stores bytes; Postgres stores metadata, lifecycle state, job progress,
|
||||||
|
audit records, and dispatch intent.
|
||||||
|
3. Broker messages contain stable IDs and correlation metadata only. They never
|
||||||
|
contain file bytes, extracted text, chunks, embeddings, secrets, raw prompts,
|
||||||
|
or raw model output.
|
||||||
|
4. The worker reloads the job and source-file records from Postgres before doing
|
||||||
|
tenant-scoped work.
|
||||||
|
5. The worker, not the HTTP publisher, performs parsing, chunking, embedding, and
|
||||||
|
generated Qdrant Chunk/Point CRUD.
|
||||||
|
6. Qdrant reads and mutations are tenant-filtered. Ingestion-generated point IDs
|
||||||
|
are deterministic so retrying a job does not create duplicate logical chunks.
|
||||||
|
7. The worker acknowledges a RabbitMQ message only after it has durably persisted
|
||||||
|
the applicable Postgres progress/final state.
|
||||||
|
8. Application clients are built at FastAPI or worker-process startup and closed
|
||||||
|
at shutdown. No mutable clients are opened as import-time globals.
|
||||||
|
|
||||||
|
## Required decisions before implementing the affected phase
|
||||||
|
|
||||||
|
The first implementation should use these defaults unless a later ADR changes
|
||||||
|
them:
|
||||||
|
|
||||||
|
### Source-file idempotency and replacement
|
||||||
|
|
||||||
|
- Use `(tenant_id, domain, content_sha256)` to recognize identical uploads.
|
||||||
|
- An identical active upload should return the existing source-file/job reference
|
||||||
|
rather than create a duplicate ingestion.
|
||||||
|
- A changed upload creates a new ingestion job. Existing active Qdrant points are
|
||||||
|
replaced only after the new job completes successfully, so a failed re-ingestion
|
||||||
|
does not remove a working index.
|
||||||
|
- Preserve the original filename in Postgres metadata. MinIO object keys remain
|
||||||
|
internal ID-based paths.
|
||||||
|
|
||||||
|
This policy should be made explicit in ADR-0009 before implementing re-ingestion
|
||||||
|
rather than becoming an accidental repository behavior.
|
||||||
|
|
||||||
|
### File deletion
|
||||||
|
|
||||||
|
For the first release, file deletion should be soft and job-shaped:
|
||||||
|
|
||||||
|
1. mark the source file as deletion requested/soft deleted in Postgres;
|
||||||
|
2. write a maintenance outbox event;
|
||||||
|
3. have a worker soft-delete the related Qdrant points;
|
||||||
|
4. retain the MinIO object until an explicit retention or hard-erasure workflow.
|
||||||
|
|
||||||
|
Hard deletion requires a later retention/erasure implementation covering MinIO,
|
||||||
|
Qdrant, and the relevant Postgres data.
|
||||||
|
|
||||||
|
### Broker operations
|
||||||
|
|
||||||
|
Before deploying an environment, define and document:
|
||||||
|
|
||||||
|
- RabbitMQ exchange, queue, binding, and dead-letter-exchange configuration;
|
||||||
|
- queue name, prefetch, manual-ack policy, and DLX retry/backoff;
|
||||||
|
- worker concurrency and resource limits;
|
||||||
|
- outbox polling/publish interval and stuck-event alert threshold;
|
||||||
|
- how failed jobs are inspected, retried, and cancelled.
|
||||||
|
|
||||||
|
These are deployment/runbook settings, not new ADRs unless they change the
|
||||||
|
reliability guarantee or system boundary.
|
||||||
|
|
||||||
|
## Build order
|
||||||
|
|
||||||
|
### Phase 1: Foundation and local dependencies
|
||||||
|
|
||||||
|
1. Add typed configuration in `src/config.py` for Postgres, MinIO, RabbitMQ,
|
||||||
|
Qdrant, application limits, and logging.
|
||||||
|
2. Populate `.env.example` with non-secret local-development configuration.
|
||||||
|
3. Add application Docker Compose services for Postgres, MinIO, RabbitMQ, and
|
||||||
|
Qdrant. Keep application MinIO buckets/credentials separate from Langfuse
|
||||||
|
infrastructure.
|
||||||
|
4. Add direct Python dependencies and lock them with `uv`:
|
||||||
|
SQLAlchemy async/Postgres driver, MinIO/S3 client, aio-pika, Qdrant client,
|
||||||
|
and structured logging dependencies chosen by ADR-0011.
|
||||||
|
5. Add the pytest foundation from ADR-0016: async test configuration, boundary
|
||||||
|
markers, and support for dependency-injected fakes. Add a lifespan smoke test
|
||||||
|
before creating external clients.
|
||||||
|
6. Create FastAPI lifespan setup and typed dependency helpers without creating
|
||||||
|
schema at startup.
|
||||||
|
|
||||||
|
**Exit criteria:** local infrastructure starts; readiness checks can report each
|
||||||
|
required dependency; clients are opened/closed by process owners; fast unit tests
|
||||||
|
run without Docker or live providers.
|
||||||
|
|
||||||
|
### Phase 2: Database, migrations, and domain contracts
|
||||||
|
|
||||||
|
1. Define SQLAlchemy models and Alembic migrations for the minimum required
|
||||||
|
tables: `tenants`, `api_keys`, `source_files`, `ingestion_jobs`,
|
||||||
|
`ingestion_job_events`, and `outbox_events`.
|
||||||
|
2. Define Pydantic request/response/message schemas, including a versioned
|
||||||
|
`ingestion.job.created` message containing `event_id`, `tenant_id`, `file_id`,
|
||||||
|
`ingestion_job_id`, `request_id`, and `api_key_id`.
|
||||||
|
3. Implement explicit repositories/services with a request/job-lifetime
|
||||||
|
`AsyncSession`; routes/services own commit/rollback boundaries as specified by
|
||||||
|
ADR-0012.
|
||||||
|
4. Write the empty-database migration test before each schema revision. Create
|
||||||
|
Testcontainers-based Postgres fixtures for tenants, hashed API keys, database
|
||||||
|
sessions, and migrations. Do not use `create_all()` in test fixtures.
|
||||||
|
|
||||||
|
**Exit criteria:** migrations create the schema from an empty database; application
|
||||||
|
startup performs no DDL; schema and repository tests verify tenant-scoped
|
||||||
|
reads/writes and valid job transitions.
|
||||||
|
|
||||||
|
### Phase 3: MinIO upload and durable job creation
|
||||||
|
|
||||||
|
1. Implement API-key authentication and `AuthContext` tenant derivation.
|
||||||
|
2. Implement `POST /v1/files` for CSV only, including streaming-size controls,
|
||||||
|
file-type validation, SHA-256 calculation, and a private MinIO upload using
|
||||||
|
an internal object key.
|
||||||
|
3. In one Postgres transaction, persist `source_files`, create
|
||||||
|
`ingestion_jobs(status='queued')`, and write the corresponding unpublished
|
||||||
|
`outbox_events` row.
|
||||||
|
4. Return `202 Accepted` with `file_id`, `ingestion_job_id`, and `queued` status.
|
||||||
|
5. Implement `GET /v1/files/{file_id}` with tenant filtering and a public status
|
||||||
|
response that does not expose raw storage credentials or internal artifacts.
|
||||||
|
6. Add cleanup/compensation handling for a MinIO upload that succeeds while the
|
||||||
|
database transaction fails.
|
||||||
|
7. Add unit/API tests for trusted tenant derivation, CSV validation, idempotency,
|
||||||
|
`202 Accepted`, and tenant-scoped status. Add MinIO adapter integration tests
|
||||||
|
for server-derived private object paths and compensation behavior.
|
||||||
|
|
||||||
|
**Exit criteria:** an authenticated CSV upload creates a private object, a queued
|
||||||
|
job, and an unpublished outbox event; a tenant cannot retrieve another tenant's
|
||||||
|
file status; the HTTP path does not publish directly to RabbitMQ.
|
||||||
|
|
||||||
|
### Phase 4: RabbitMQ and outbox publisher
|
||||||
|
|
||||||
|
1. Provision/verify the application-owned RabbitMQ topic exchange, queue, and
|
||||||
|
binding configuration through deployment/bootstrap code rather than route
|
||||||
|
startup side effects.
|
||||||
|
2. Implement an outbox-publisher process that safely claims unpublished events,
|
||||||
|
publishes them with the stable event id using publisher confirms, and records
|
||||||
|
success/failure attempts.
|
||||||
|
3. RabbitMQ has no broker-native message de-duplication; retain idempotency in
|
||||||
|
all consumers as the sole guard against duplicate processing.
|
||||||
|
4. Add monitoring/logging for publish attempts, unpublished-event age, and
|
||||||
|
repeated failures.
|
||||||
|
5. Add unit tests for event claiming and retryable failures, then Testcontainers
|
||||||
|
RabbitMQ integration tests for durable publication, restart/retry, duplicate
|
||||||
|
publication, and message metadata.
|
||||||
|
|
||||||
|
**Exit criteria:** an outbox event becomes a durable RabbitMQ message after a
|
||||||
|
publisher restart; retrying publication cannot create duplicate application work.
|
||||||
|
|
||||||
|
### Phase 5: Ingestion worker and Qdrant Chunk/Point CRUD
|
||||||
|
|
||||||
|
1. Create a separate worker entrypoint with its own application-lifetime database,
|
||||||
|
MinIO, RabbitMQ, Qdrant, model, and logging clients.
|
||||||
|
2. Consume `ingestion.job.created`; reload and validate Postgres records before
|
||||||
|
fetching the MinIO object.
|
||||||
|
3. Transition the job from `queued` to `running` conditionally, append progress
|
||||||
|
events, parse CSV, create deterministic chunks, and upsert tenant-scoped
|
||||||
|
Qdrant points.
|
||||||
|
4. Mark the job `succeeded` with counters or `failed` with a safe error summary;
|
||||||
|
acknowledge only after final/progress state is persisted.
|
||||||
|
5. Make a repeated delivery of the same `ingestion_job_id` safe: no duplicate
|
||||||
|
logical chunks, no incorrect counters, and no transition from a terminal state
|
||||||
|
back to `running`.
|
||||||
|
6. Add unit tests for deterministic CSV chunks, point IDs, and terminal job
|
||||||
|
transitions. Add Testcontainers Qdrant and RabbitMQ integration tests for
|
||||||
|
tenant-filtered upserts, acknowledgement after durable state, redelivery, and
|
||||||
|
parser/Qdrant failure handling.
|
||||||
|
|
||||||
|
**Exit criteria:** a successful upload reaches `succeeded`, and its points are
|
||||||
|
retrievable only under the owning tenant's Qdrant filter. Forced worker failure
|
||||||
|
and message redelivery produce a correct final job state.
|
||||||
|
|
||||||
|
### Phase 6: Operations, integration tests, and documentation
|
||||||
|
|
||||||
|
1. Add an operator runbook covering local startup, migrations, MinIO bucket setup,
|
||||||
|
RabbitMQ exchange/queue setup, web/worker/outbox commands, and job replay.
|
||||||
|
2. Add a serialized Compose-based operational smoke test where the web process,
|
||||||
|
outbox publisher, and worker run independently for upload through indexed
|
||||||
|
points. Testcontainers remains the standard pytest mechanism for individual
|
||||||
|
adapter integration tests.
|
||||||
|
3. Add end-to-end tests for duplicate upload, duplicate RabbitMQ delivery, outbox
|
||||||
|
publisher crash/restart, worker crash/restart, tenant isolation, and failed
|
||||||
|
parser/Qdrant behavior.
|
||||||
|
4. Add health/readiness checks that distinguish process health from dependency
|
||||||
|
readiness.
|
||||||
|
5. Update the README with local-start instructions and links to ADRs, this plan,
|
||||||
|
and the operations runbook.
|
||||||
|
|
||||||
|
**Exit criteria:** a new developer can start the stack, apply migrations, upload a
|
||||||
|
CSV, observe the job through completion, and understand how to investigate or
|
||||||
|
retry a failure.
|
||||||
|
|
||||||
|
## Definition of done for the vertical slice
|
||||||
|
|
||||||
|
The first slice is done when the following path works in local Compose and is
|
||||||
|
covered by automated tests:
|
||||||
|
|
||||||
|
```text
|
||||||
|
POST /v1/files (authenticated CSV upload)
|
||||||
|
-> raw bytes stored privately in MinIO
|
||||||
|
-> source file, queued job, and outbox event committed in Postgres
|
||||||
|
-> outbox publisher writes a durable RabbitMQ message
|
||||||
|
-> ingestion worker processes and indexes deterministic Qdrant points
|
||||||
|
-> Postgres records progress and terminal job status
|
||||||
|
-> GET /v1/files/{file_id} reports that status within the owning tenant only
|
||||||
|
```
|
||||||
|
|
||||||
|
The next implementation work after this slice is direct Point CRUD, retrieval,
|
||||||
|
and then the LangGraph conversational flow. Do not couple those later milestones
|
||||||
|
to the initial ingestion path unless they are needed to preserve one of the
|
||||||
|
invariants above.
|
||||||
@@ -5,14 +5,41 @@ description = "Add your description here"
|
|||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.13"
|
requires-python = ">=3.13"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
"aio-pika>=9.5.0",
|
||||||
|
"alembic>=1.19.1",
|
||||||
"fastapi[standard]==0.141.1",
|
"fastapi[standard]==0.141.1",
|
||||||
"pydantic-settings>=2.14.2",
|
"langgraph>=1.2.10",
|
||||||
|
"pydantic-settings>=2.15.0",
|
||||||
|
"sqlalchemy>=2.0.51",
|
||||||
]
|
]
|
||||||
|
|
||||||
[dependency-groups]
|
[dependency-groups]
|
||||||
dev = [
|
dev = [
|
||||||
"ruff>=0.16.1",
|
"asgi-lifespan>=2.1.0",
|
||||||
"ty>=0.0.65",
|
"httpx>=0.28.1",
|
||||||
|
"pytest>=8.3.5",
|
||||||
|
"pytest-asyncio>=0.25.3",
|
||||||
|
"pytest-cov>=6.0.0",
|
||||||
|
"pytest-timeout>=2.3.1",
|
||||||
|
"ruff>=0.16.2",
|
||||||
|
"testcontainers>=4.9.2",
|
||||||
|
"ty>=0.0.69",
|
||||||
|
]
|
||||||
|
|
||||||
|
[tool.pytest.ini_options]
|
||||||
|
testpaths = ["tests"]
|
||||||
|
asyncio_mode = "strict"
|
||||||
|
timeout = 10
|
||||||
|
markers = [
|
||||||
|
"unit: fast tests with no external services",
|
||||||
|
"integration: tests against a real disposable service",
|
||||||
|
"e2e: end-to-end vertical-slice tests",
|
||||||
|
"postgres: integration test using Postgres",
|
||||||
|
"minio: integration test using MinIO",
|
||||||
|
"rabbitmq: integration test using RabbitMQ",
|
||||||
|
"qdrant: integration test using Qdrant",
|
||||||
|
"slow: test exceeds normal integration feedback time",
|
||||||
|
"live_provider: opt-in test that calls an external provider",
|
||||||
]
|
]
|
||||||
|
|
||||||
[tool.ruff]
|
[tool.ruff]
|
||||||
@@ -45,3 +72,17 @@ indent-style = "space"
|
|||||||
|
|
||||||
[tool.ty.environment]
|
[tool.ty.environment]
|
||||||
python-version = "3.13"
|
python-version = "3.13"
|
||||||
|
|
||||||
|
[tool.ty.rules]
|
||||||
|
# High-value, currently ignore/warn by
|
||||||
|
possibly-unresolved-reference = "error"
|
||||||
|
missing-type-argument = "warn"
|
||||||
|
possibly-missing-attribute = "warn"
|
||||||
|
|
||||||
|
# Nice-to-have discipline:
|
||||||
|
unused-ignore-comment = "warn"
|
||||||
|
missing-override-decorator = "warn"
|
||||||
|
|
||||||
|
# Turn OFF or downgrade if noisy on your codebase:
|
||||||
|
possibly-missing-import = "ignore"
|
||||||
|
division-by-zero = "ignore"
|
||||||
|
|||||||
5
src/config.py
Normal file
5
src/config.py
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
from pydantic_settings import BaseSettings
|
||||||
|
|
||||||
|
|
||||||
|
class Settings(BaseSettings):
|
||||||
|
pass
|
||||||
0
src/main.py
Normal file
0
src/main.py
Normal file
Reference in New Issue
Block a user