curo1305
6fed5ba531
Research, pattern mapping, and verification complete. Walking Skeleton mode active (MVP Phase 1). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
53 KiB
53 KiB
phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, user_setup, tags, must_haves
| phase | plan | type | wave | depends_on | files_modified | autonomous | requirements | user_setup | tags | must_haves | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 01-infrastructure-foundation | 05 | execute | 4 |
|
|
false |
|
|
|
Purpose: This plan closes the loop. After it ships, ROADMAP.md Phase 1 success criteria #1, #3, and #4 are all satisfied (criterion #2 was satisfied in Plan 03). Phase 1 ends with a usable single-user app whose entire internal architecture is the multi-user-ready PostgreSQL + MinIO + Celery stack.
Output: Updated backend/main.py, backend/api/documents.py, backend/api/topics.py, backend/services/classifier.py; new backend/celery_app.py + backend/tasks/document_tasks.py; cleaned backend/config.py; final test-suite cutover; deletion of data/ directory; and a passing end-to-end walking-skeleton verification checkpoint.
<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>
@CLAUDE.md @.planning/PROJECT.md @.planning/ROADMAP.md @.planning/STATE.md @.planning/phases/01-infrastructure-foundation/01-CONTEXT.md @.planning/phases/01-infrastructure-foundation/01-RESEARCH.md @.planning/phases/01-infrastructure-foundation/01-PATTERNS.md @.planning/phases/01-infrastructure-foundation/SKELETON.md @.planning/phases/01-infrastructure-foundation/01-04-SUMMARY.md @backend/services/storage.py @backend/db/models.py @backend/db/session.py @backend/deps/db.py @backend/storage/__init__.py After Plan 04, the async storage layer is in place. This plan wires consumers.Existing api/documents.py consumer points (must be ported to async + session injection):
storage.save_upload(content, file.filename, mime)→await storage.save_upload(session, content, file.filename, mime)storage.save_metadata(meta)→await storage.save_metadata(session, meta)storage.list_metadata(topic=topic)→await storage.list_metadata(session, topic=topic)storage.get_metadata(doc_id)→await storage.get_metadata(session, doc_id)storage.delete_document(doc_id)→await storage.delete_document(session, doc_id)await classifier.classify_document(saved["id"])→extract_and_classify.delay(saved["id"])(Celery task — STORE-08)
Existing api/topics.py consumer points:
storage.load_topics()→await storage.load_topics(session)storage.topic_doc_counts()→await storage.topic_doc_counts(session)storage.create_topic(...)→await storage.create_topic(session, ...)storage.update_topic(...)→await storage.update_topic(session, ...)storage.delete_topic(...)→await storage.delete_topic(session, ...)storage.get_metadata(...)→await storage.get_metadata(session, ...)(used by/suggest)
Existing services/classifier.py consumer points (called by both the soon-removed inline upload path and the new Celery task; module signature changes from async def classify_document(doc_id) accepting no session to async def classify_document(session, doc_id)):
- Used inside the Celery task wrapper via
asyncio.run(classify_document(session, doc_id))after manually opening a session
api/settings.py — KEEP AS-IS. The settings.json flat file lives until Phase 2 (D-03 settings deferred); the services/storage.load_settings() / save_settings() functions remain sync per Plan 04.
main.py lifespan contract (current → new):
# current
async def lifespan(app):
ensure_data_dirs()
yield
# new
async def lifespan(app):
# MinIO bucket auto-create
minio_client = Minio(settings.minio_endpoint, access_key=..., secret_key=..., secure=False)
if not await asyncio.to_thread(minio_client.bucket_exists, settings.minio_bucket):
await asyncio.to_thread(minio_client.make_bucket, settings.minio_bucket)
app.state.minio = minio_client
yield
await engine.dispose()
/health response contract (D-07):
{
"status": "ok",
"checks": {"postgres": "ok", "minio": "ok"}
}
Or "status": "degraded" if any check is not "ok".
Create `backend/tasks/__init__.py` as an empty file.
Create `backend/tasks/document_tasks.py`. Imports: `import asyncio`, `from celery_app import celery_app`, `from db.session import AsyncSessionLocal`, `from services import storage, extractor, classifier`, `from storage import get_storage_backend`. Define the task:
```python
@celery_app.task(name="tasks.document_tasks.extract_and_classify")
def extract_and_classify(document_id: str) -> dict:
return asyncio.run(_run(document_id))
async def _run(document_id: str) -> dict:
async with AsyncSessionLocal() as session:
meta = await storage.get_metadata(session, document_id)
if meta is None:
return {"document_id": document_id, "status": "not_found"}
# Fetch the bytes from MinIO so the extractor can read them
backend = get_storage_backend()
try:
obj_key = meta.get("object_key") or meta.get("path")
# The object_key shape is {user_id}/{doc_id}/{uuid4}{ext} — retrieve via storage_backend
# We don't have object_key on the metadata dict in v1 — read from DB directly:
from db.models import Document
import uuid as _uuid
doc = await session.get(Document, _uuid.UUID(document_id))
if doc is None or not doc.object_key:
return {"document_id": document_id, "status": "missing_object"}
file_bytes = await backend.get_object(doc.object_key)
text = extractor.extract_text_from_bytes(file_bytes, doc.content_type) if hasattr(extractor, "extract_text_from_bytes") else extractor.extract_text_bytes(file_bytes, doc.content_type)
meta["extracted_text"] = text
await storage.save_metadata(session, meta)
except Exception as e:
return {"document_id": document_id, "status": "extract_failed", "error": str(e)}
try:
topics = await classifier.classify_document(session, document_id)
return {"document_id": document_id, "status": "classified", "topics": topics}
except Exception as e:
# Non-fatal — preserve the existing convention from api/documents.py line 54-56
doc.status = "classification_failed"
await session.commit()
return {"document_id": document_id, "status": "classification_failed", "error": str(e)}
```
Note: If `services/extractor.py` only exposes `extract_text(path, mime)` (file-path-based), add a new helper `extract_text_from_bytes(file_bytes: bytes, mime: str)` to `services/extractor.py` that writes `file_bytes` to a `tempfile.NamedTemporaryFile(suffix=...)`, calls the existing `extract_text(tmp.name, mime)`, and unlinks the temp file. Do not modify any other behavior in `services/extractor.py`.
Update `backend/services/classifier.py`: change `async def classify_document(doc_id: str, topic_names: list[str] | None = None)` to `async def classify_document(session: AsyncSession, doc_id: str, topic_names: list[str] | None = None)`. Add `from sqlalchemy.ext.asyncio import AsyncSession` at the top. Replace `storage.get_metadata(doc_id)` → `await storage.get_metadata(session, doc_id)`. Replace `storage.load_settings()` → `storage.load_settings()` (unchanged — Phase 1 keeps the flat file; this is sync). Replace `storage.load_topics()` → `await storage.load_topics(session)` (note signature change — adapter call). Replace `storage.create_topic(name.strip())` → `await storage.create_topic(session, name.strip())`. Replace `storage.update_document_topics(doc_id, final_topics)` → `await storage.update_document_topics(session, doc_id, final_topics)`. Apply the same session-injection treatment to `suggest_topics_for_document(session, doc_id)`. Preserve `MAX_AI_CHARS = 8_000` and every other line verbatim.
Rewrite the lifespan function:
```python
@asynccontextmanager
async def lifespan(app: FastAPI):
# MinIO bucket initialization (RESEARCH.md Pattern 4)
minio_client = Minio(
settings.minio_endpoint,
access_key=settings.minio_access_key,
secret_key=settings.minio_secret_key,
secure=False,
)
exists = await asyncio.to_thread(minio_client.bucket_exists, settings.minio_bucket)
if not exists:
await asyncio.to_thread(minio_client.make_bucket, settings.minio_bucket)
app.state.minio = minio_client
yield
await engine.dispose()
```
Rewrite the `/health` handler (preserving `@app.get("/health")` and `async def`):
```python
@app.get("/health")
async def health(request: Request):
checks = {}
try:
async with AsyncSessionLocal() as session:
await session.execute(text("SELECT 1"))
checks["postgres"] = "ok"
except Exception as e:
checks["postgres"] = f"error: {type(e).__name__}: {e}"
try:
ok = await asyncio.to_thread(request.app.state.minio.bucket_exists, settings.minio_bucket)
checks["minio"] = "ok" if ok else "error: bucket missing"
except Exception as e:
checks["minio"] = f"error: {type(e).__name__}: {e}"
status = "ok" if all(v == "ok" for v in checks.values()) else "degraded"
return {"status": status, "checks": checks}
```
Keep the existing CORS middleware (`allow_origins=["*"]` for Phase 1 — Phase 2 locks down).
Rewrite `backend/api/documents.py`. Imports: replace `from services import storage, extractor, classifier` with `from sqlalchemy.ext.asyncio import AsyncSession`, `from deps.db import get_db`, `from services import storage, extractor`, `from tasks.document_tasks import extract_and_classify`, `from services import classifier` (only used by the `/classify` endpoint), and keep `from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Query, Depends` (add `Depends`). Preserve the router definition and `ALLOWED_MIME_TYPES` set.
For every route, append `session: AsyncSession = Depends(get_db)` to the signature. For each storage call, prepend `await` and pass `session` as the first arg:
- `upload_document`: read content, validate empty, generate filename + mime, call `await storage.save_upload(session, content, file.filename or "upload", mime)`, then `text = extractor.extract_text(saved["path"], mime)` — wait: `saved["path"]` is now the object_key, not a filesystem path. CHANGE the extraction step: pull bytes already in memory (`content`) and call the new `extractor.extract_text_from_bytes(content, mime)` helper introduced in Task 1. Build the `meta` dict exactly as before (preserve the keys), call `await storage.save_metadata(session, meta)`. If `auto_classify` is True, call `extract_and_classify.delay(saved["id"])` (NOTE: this re-fetches from MinIO inside the worker — acceptable for Phase 1; Phase 3 will pass the bytes through Redis directly if perf demands). Return `meta`. CHANGE: since classification is now async via Celery, the response no longer includes `topics` populated by the inline classifier call — set `meta["topics"] = []` and `meta["classified_at"] = None` and rely on the worker to update the DB row. Document this in a comment as the Phase 1 cutover behavior.
- `list_documents`: `docs = await storage.list_metadata(session, topic=topic)` then preserve the existing pagination math.
- `get_document`: `meta = await storage.get_metadata(session, doc_id)`; raise 404 if `None`.
- `delete_document`: `ok = await storage.delete_document(session, doc_id)`; raise 404 if `False`.
- `classify_document` (the `/classify` route): `meta = await storage.get_metadata(session, doc_id)`; raise 404 if None; preserve the inline `await classifier.classify_document(session, doc_id, topic_names)` call (this endpoint historically returned the topic list synchronously — keep that behavior; the upload-time path is the async one).
Rewrite `backend/api/topics.py` similarly: add `Depends` import, `from sqlalchemy.ext.asyncio import AsyncSession`, `from deps.db import get_db`. Add `session: AsyncSession = Depends(get_db)` to every route. Wrap every `storage.*` call with `await` and prepend `session`:
- `list_topics`: `topics = await storage.load_topics(session)`, `counts = await storage.topic_doc_counts(session)`.
- `create_topic`: `topic = await storage.create_topic(session, body.name, body.description, body.color)`.
- `update_topic`: `topic = await storage.update_topic(session, topic_id, name=body.name, description=body.description, color=body.color)`.
- `delete_topic`: `name = await storage.delete_topic(session, topic_id)`.
- `suggest_topics`: `meta = await storage.get_metadata(session, body.document_id)`; if None, 404; then `await classifier.suggest_topics_for_document(session, body.document_id)`.
Preserve every Pydantic model and HTTPException message verbatim.
Step 2 — prune `backend/config.py`. Read the current file (post-Plan 01). Remove:
- `DATA_DIR = Path(os.environ.get("DATA_DIR", "/app/data"))`
- `UPLOADS_DIR = DATA_DIR / "uploads"`
- `METADATA_DIR = DATA_DIR / "metadata"`
- `TOPICS_FILE = DATA_DIR / "topics.json"`
- the `def ensure_data_dirs():` function
- the `import json` at the top if no longer used (keep if `DEFAULT_SETTINGS` interpolates from JSON anywhere — currently no)
- the `import os` (no longer needed since `Settings` reads env via Pydantic)
Preserve:
- `from pathlib import Path` (used by `SETTINGS_FILE`)
- `DEFAULT_SYSTEM_PROMPT` (used by `services/classifier.py`)
- `DEFAULT_SETTINGS` (used by `services/storage.load_settings` fallback)
- `class Settings(BaseSettings):` with all Phase 1 fields (Plan 01 output)
- `settings = Settings()` module instance
Rebase `SETTINGS_FILE` as a derived path computed from `settings.data_dir`:
`SETTINGS_FILE = Path(settings.data_dir) / "settings.json"` — placed AFTER the `settings = Settings()` line. Add a comment: `# SETTINGS_FILE: still flat-file in Phase 1; migrates to users.ai_provider in Phase 2`.
Step 3 — prune `backend/tests/conftest.py`:
- DELETE the entire `isolated_data_dir` fixture (the autouse one that monkey-patches `config.DATA_DIR` etc.)
- DELETE the sync `client` fixture (`with TestClient(app) as c: yield c`)
- KEEP the `db_session`, `async_client`, `sample_txt`, `sample_pdf` fixtures introduced in Plan 02. Promote `async_client` so the previous behavior — fail gracefully if `deps.db` does not exist — is replaced with a hard dependency: remove the `try/except ImportError: pytest.skip(...)` wrapper inside the async fixtures because `deps.db.get_db` now exists.
- ADD a new `pytest_asyncio.fixture(scope="session")` named `live_services_available` that probes localhost:5432, localhost:9000, localhost:6379 via `socket.create_connection(..., timeout=1)`; if any probe fails, the fixture yields `False`; otherwise `True`. Update the `async_client` fixture (or add a new `live_async_client` fixture) to use an actual PostgreSQL + MinIO when `live_services_available` is True, falling back to the in-memory aiosqlite engine when False. Use `pytest.mark.skipif(not live_services_available, reason="docker compose not running")` on integration tests that need real MinIO; unit tests using only the in-memory DB do not need the skip marker. (Simpler approach acceptable: detect via env var `INTEGRATION=1`; if unset, skip integration tests.)
Step 4 — prune `backend/tests/test_documents.py`:
- DELETE the legacy sync tests: `test_upload_txt_no_classify`, `test_upload_pdf_no_classify`, `test_list_documents`, `test_list_documents_filter_by_topic`, `test_get_document`, `test_get_document_not_found`, `test_delete_document`, `test_delete_document_not_found`, `test_upload_empty_file`. (Plan 02 left them in place during the cutover; this plan completes the deletion.)
- On every `_async`-suffixed test added in Plan 02, REMOVE the `@pytest.mark.xfail(strict=False, reason="async storage layer implemented in plan 05")` marker.
- Update any test that previously referenced `import services.storage as st; st.update_document_topics(...)` to use the async ORM API via the `db_session` fixture: `from db.models import Document, DocumentTopic; from sqlalchemy import insert; ...`. For tests that need a topic-tagged document, build it via the API itself (call `POST /api/topics` then `PATCH /api/documents/.../classify`).
Step 5 — prune `backend/tests/test_health.py`:
- DELETE the legacy `def test_health(client):` (it used the sync TestClient fixture which is gone).
- REMOVE the `@pytest.mark.xfail` marker from `test_health_checks_postgres_and_minio`.
- If `live_services_available` is False, this test should be skipped via `pytest.mark.skipif(...)`.
Step 6 — run the full suite end-to-end against the in-memory engine: `cd backend && python3 -m pytest tests/ -v` should exit 0 with the storage tests PASSED, the alembic tests PASSED (or SKIPPED if no PostgreSQL available — the in-memory aiosqlite test path covers them), the health test SKIPPED or PASSED, and the async document tests PASSED or SKIPPED depending on `live_services_available`.
1. Ensure `.env` exists with all variables from `.env.example` filled in: `cp .env.example .env` (if not present) and replace each `changeme_*` placeholder with a value of your choice. The DATABASE_URL/DATABASE_MIGRATE_URL passwords MUST match the hardcoded passwords in `docker/postgres/initdb.d/01-init-users.sql` from Plan 01 (which itself was committed during Wave 1). The REDIS_URL password must match REDIS_PASSWORD.
2. Tear down any prior state: `docker compose down -v` (the `-v` deletes the postgres_data and minio_data named volumes so the init script will re-run).
3. Boot everything: `docker compose up --build -d`. Wait ~30 seconds.
4. Verify all services are healthy: `docker compose ps`. The `STATUS` column must show `Up (healthy)` for `postgres`, `minio`, `redis`, `backend`, AND `celery-worker`. If any is `unhealthy`, capture `docker compose logs <service>` and resolve before continuing.
5. Apply the migration against the live DB: `docker compose exec backend bash -lc "cd /app && alembic upgrade head"`. Must exit 0 with `Running upgrade -> 0001`.
6. Hit the health endpoint: `curl -s http://localhost:8000/health | python3 -m json.tool`. The response MUST be:
```
{
"status": "ok",
"checks": {
"postgres": "ok",
"minio": "ok"
}
}
```
7. Upload a real PDF or text file. Pick any small PDF (or use `printf 'Test document about invoices and contracts.' > /tmp/test.txt` first). Then:
```
curl -s -X POST http://localhost:8000/api/documents/upload \
-F "file=@/tmp/test.txt;type=text/plain" \
-F "auto_classify=true" | python3 -m json.tool
```
Confirm the response includes:
- `"id"` — a 36-character UUID string
- `"original_name": "test.txt"`
- `"size_bytes"` matching the file size
- `"topics": []` (classification is async — the Celery worker fills this in seconds later)
8. Confirm the document landed in PostgreSQL:
`docker compose exec postgres psql -U docuvault_app -d docuvault -c "SELECT id, filename, object_key, status FROM documents ORDER BY created_at DESC LIMIT 1;"`
— exactly one row; `object_key` starts with `null-user/` (D-03 sentinel from Plan 04); `status` is `pending` initially then `classified` or `classification_failed` after the worker runs.
9. Confirm the document landed in MinIO. The object key from step 8 will look like `null-user/<doc-uuid>/<random-uuid>.txt`. Either use the MinIO web console at `http://localhost:9001` (login with `MINIO_ROOT_USER` / `MINIO_ROOT_PASSWORD` from `.env`) and navigate to `docuvault` bucket → confirm the object exists with non-zero size — OR use `mc`:
`docker compose exec minio mc alias set local http://localhost:9000 $MINIO_ROOT_USER $MINIO_ROOT_PASSWORD` then `docker compose exec minio mc ls local/docuvault/null-user/`.
10. Confirm the Celery worker processed the task:
`docker compose logs celery-worker | tail -30`
— look for a `Task tasks.document_tasks.extract_and_classify[...] received` line followed by `succeeded` or a structured error. If the task succeeded, run:
`curl -s http://localhost:8000/api/documents | python3 -m json.tool`
— the response should show one item with `extracted_text` populated and possibly `topics` populated by the AI classifier (depending on AI provider config; if no `ANTHROPIC_API_KEY` / `OPENAI_API_KEY` is set, classification will fail gracefully and `status` will be `classification_failed` — that is acceptable for this walking-skeleton check; the storage layer worked.).
11. Delete the document:
`curl -s -X DELETE http://localhost:8000/api/documents/<id-from-step-7>` returns `{"success": true}`.
Then confirm the MinIO object is gone: `docker compose exec minio mc ls local/docuvault/null-user/<doc-uuid>/` returns empty or "Object does not exist".
12. Run the test suite against the live stack:
`docker compose exec -e INTEGRATION=1 backend bash -lc "cd /app && pytest tests/ -v"`
— every test PASSED, zero FAILED, zero XFAIL (skips for integration tests when INTEGRATION=0 are acceptable on a host-only run; when INTEGRATION=1 inside the container with live services, they must run and pass).
<threat_model>
Trust Boundaries
| Boundary | Description |
|---|---|
| Browser → FastAPI | HTTP/JSON (Phase 1: CORS * — Phase 2 locks down); multipart upload bytes traverse this boundary |
| FastAPI → Celery / Redis | Task payload is the document_id string only; no user input passed |
| FastAPI lifespan → MinIO | Bucket auto-create at startup; client persists on app.state.minio |
| Celery worker → MinIO + PostgreSQL | Worker re-fetches bytes from MinIO and reads/writes Document row |
STRIDE Threat Register
| Threat ID | Category | Component | Disposition | Mitigation Plan |
|---|---|---|---|---|
| T-01-05-01 | Spoofing | Unauthenticated upload endpoint accepting any client | accept | Phase 1 has no auth (D-03 — user_id nullable); upload accessible to anyone reaching localhost:8000. Phase 2 adds JWT + CSRF + rate-limit. Documented in SKELETON.md "Out of Scope". |
| T-01-05-02 | Tampering | MIME-type spoofing on upload | mitigate | The existing ALLOWED_MIME_TYPES set in api/documents.py is preserved verbatim. Phase 4 (DOC-02) adds magic-byte verification before download/preview. |
| T-01-05-03 | Information Disclosure | /health revealing internal error class names |
mitigate | /health error strings format as f"error: {type(e).__name__}: {e}" — exposes Python exception class name, which is acceptable for an internal/dev endpoint in Phase 1. Phase 2 will trim to "error" or "unhealthy" once the endpoint is reachable from the internet. Documented note in main.py. |
| T-01-05-04 | Tampering | Celery task receives untrusted document_id and might query arbitrary rows | mitigate | extract_and_classify only takes a document_id string from the upload path — never from a user query parameter. Task code does session.get(Document, uuid.UUID(document_id)) which raises ValueError for non-UUID input; no SQL injection vector. Document row lookup is single-row by primary key only. |
| T-01-05-05 | Denial of Service | Lifespan bucket-create on every reboot blocks startup | mitigate | if not bucket_exists: make_bucket is idempotent — fast on warm starts. If MinIO is unreachable at startup, lifespan raises and the FastAPI app fails to boot — this is intentional and surfaces the failure to Compose's depends_on: condition: service_healthy (which gated startup but cannot catch a post-startup MinIO crash). |
| T-01-05-06 | Information Disclosure | app.state.minio reused across handlers |
accept | The client holds connection state but no per-user credentials. All app handlers see the same app.state.minio — acceptable since Phase 1 has no per-user isolation. Phase 5 will introduce per-user StorageBackend instances for cloud backends. |
| T-01-05-SC | Tampering | npm/pip installs | N/A | No new package installs in this plan — all dependencies were added in Plan 01 and verified via RESEARCH.md Package Legitimacy Audit. |
| </threat_model> |
<success_criteria>
backend/main.pylifespan creates the MinIO bucket and disposes the engine;/healthreturns the postgres+minio shape per D-07.backend/api/documents.pyandbackend/api/topics.pyare entirely async-session-driven; upload-time classification is queued via Celery.delay().backend/celery_app.pyandbackend/tasks/document_tasks.pyare wired and discoverable.backend/services/classifier.pyaccepts anAsyncSession.backend/config.pyis pruned of legacy flat-file constants.backend/data/is deleted;tests/test_documents.pyis async-only;tests/conftest.pyno longer ships a sync TestClient fixture.docker compose upboots healthy; the walking skeleton end-to-end check from Task 4 passes. </success_criteria>