From e9ee5d4ba542d04b74b74115e7a3e890f742e9bd Mon Sep 17 00:00:00 2001 From: curo1305 Date: Thu, 4 Jun 2026 23:08:16 +0200 Subject: [PATCH 1/3] feat(07-04): Celery retry harness + _ClassificationError sentinel (D-09/D-10) - Add _ClassificationError sentinel raised by _run() for classification failures - Add _mark_classification_failed() async helper for final status writeback - Change decorator to bind=True, max_retries=3 - Outer sync task catches _ClassificationError and calls self.retry(countdown=[30,90,270]) - MaxRetriesExceededError caught in nested try/except to avoid re-raise from exc block - Promote test_retry_backoff and test_exhaustion_sets_failed_status from xfail to passing - Tests use push_request(retries=N) + patch.object(task, "retry") to verify countdown values - test_exhaustion_sets_failed_status uses assert_called_once_with (asyncio.run calling convention) --- backend/tasks/document_tasks.py | 91 ++++++++++++++++++++++----- backend/tests/test_document_tasks.py | 93 ++++++++++++++++++++++++---- 2 files changed, 158 insertions(+), 26 deletions(-) diff --git a/backend/tasks/document_tasks.py b/backend/tasks/document_tasks.py index 4d89c33..a86a2a1 100644 --- a/backend/tasks/document_tasks.py +++ b/backend/tasks/document_tasks.py @@ -12,17 +12,82 @@ Flow: 4. Extract text from bytes using services.extractor 5. Persist extracted_text back to the Document row 6. Call services.classifier.classify_document to assign topics - 7. Return a result dict (never raises — classification failures are non-fatal) + 7. Return a result dict; classification failures are retried with exponential backoff + +Celery retry harness (D-09, D-10 — Pitfall 3 from RESEARCH.md): + CRITICAL: self.retry() MUST be raised from the outer sync task, NOT inside + asyncio.run(). _ClassificationError is a sentinel raised by _run() to signal + a retryable classification failure. The outer extract_and_classify catches it + and calls self.retry(countdown=...) in the sync layer. """ import asyncio +from celery.exceptions import MaxRetriesExceededError + from celery_app import celery_app -@celery_app.task(name="tasks.document_tasks.extract_and_classify") -def extract_and_classify(document_id: str) -> dict: - """Synchronous Celery entry-point — delegates to async _run via asyncio.run.""" - return asyncio.run(_run(document_id)) +class _ClassificationError(Exception): + """Sentinel exception raised by _run() to signal a retryable classification failure. + + This exception escapes asyncio.run() and is caught by the outer sync task, + which then calls self.retry() in the Celery sync layer (not inside asyncio.run). + Non-classification failures (extract_failed, invalid_id) are NOT retried — + they return a dict directly from _run() without raising _ClassificationError. + """ + pass + + +async def _mark_classification_failed(document_id: str) -> None: + """Write doc.status = 'classification_failed' after all retries are exhausted. + + D-10: called by extract_and_classify's MaxRetriesExceededError handler via + asyncio.run(_mark_classification_failed(document_id)) so the final status + is written to DB regardless of Celery task context. + """ + import uuid as _uuid + + from db.session import AsyncSessionLocal + from db.models import Document + + async with AsyncSessionLocal() as session: + try: + doc_uuid = _uuid.UUID(document_id) + except ValueError: + return # Silently ignore invalid IDs — nothing to update + doc = await session.get(Document, doc_uuid) + if doc is not None: + doc.status = "classification_failed" + await session.commit() + + +@celery_app.task( + name="tasks.document_tasks.extract_and_classify", + bind=True, + max_retries=3, +) +def extract_and_classify(self, document_id: str) -> dict: + """Synchronous Celery entry-point — delegates to async _run via asyncio.run. + + Retry harness (D-09): classification failures are retried up to 3 times + with exponential backoff: 30s / 90s / 270s. + + Pitfall 3 guard: self.retry() is called HERE in the sync layer, never + inside asyncio.run(). _ClassificationError is the sentinel that escapes + asyncio.run() to trigger the retry. + """ + try: + return asyncio.run(_run(document_id)) + except _ClassificationError as exc: + # Exponential backoff: 30s on first retry, 90s on second, 270s on third + countdowns = [30, 90, 270] + countdown = countdowns[min(self.request.retries, 2)] + try: + raise self.retry(exc=exc, countdown=countdown) + except MaxRetriesExceededError: + # All retries exhausted — write final failure status to DB + asyncio.run(_mark_classification_failed(document_id)) + return {"document_id": document_id, "status": "classification_failed"} async def _run(document_id: str) -> dict: @@ -34,6 +99,9 @@ async def _run(document_id: str) -> dict: Cloud-aware: when doc.storage_backend != 'minio', uses get_storage_backend_for_document() to retrieve bytes from the correct cloud backend instead of hardcoding MinIO. + + Classification failures raise _ClassificationError (D-09 retryable sentinel). + Non-classification failures return a status dict (not retried). """ import uuid as _uuid @@ -105,7 +173,7 @@ async def _run(document_id: str) -> dict: "error": f"Text extraction failed: {e}", } - # ── Step 4: classify document (non-fatal) ────────────────────────────── + # ── Step 4: classify document (retryable via _ClassificationError) ───── try: topics = await classifier.classify_document(session, document_id, ai_provider=ai_provider, ai_model=ai_model) return { @@ -114,14 +182,9 @@ async def _run(document_id: str) -> dict: "topics": topics, } except Exception as e: - # Non-fatal — preserve existing convention from api/documents.py - doc.status = "classification_failed" - await session.commit() - return { - "document_id": document_id, - "status": "classification_failed", - "error": str(e), - } + # Raise sentinel to allow the outer sync layer to call self.retry() + # (D-09 Pitfall 3: self.retry() cannot be called inside asyncio.run) + raise _ClassificationError(str(e)) from e @celery_app.task(name="tasks.document_tasks.cleanup_abandoned_uploads") diff --git a/backend/tests/test_document_tasks.py b/backend/tests/test_document_tasks.py index 3b5d632..f90fa42 100644 --- a/backend/tests/test_document_tasks.py +++ b/backend/tests/test_document_tasks.py @@ -1,25 +1,94 @@ """ -Wave 0 xfail stubs for Phase 7 Celery document task tests. +Phase 7 Plan 04 — Celery retry harness tests. Covers: - D-09: Celery retry with 30s/90s/270s exponential backoff - D-10: After 3 retries doc.status = "classification_failed" -Each function is a placeholder to be promoted in Plan 07-04 once the -bind=True retry harness is implemented in document_tasks.py. +Test contract note (asyncio.run + AsyncMock): + Production code uses asyncio.run(_mark_classification_failed(document_id)). + asyncio.run() invokes the coroutine object returned by calling the factory: + _mark_classification_failed(document_id). This records the invocation as a + __call__ on the mock, NOT as an __await__. Therefore tests must assert via + assert_called_once_with(), NOT assert_awaited_once_with(). -Stub policy (STATE.md decision: xfail(strict=False) for Wave 0): - - Body is a single pytest.xfail() call — no assertion code. - - strict=False so unexpected passes (xpass) never break CI. +Celery bind=True calling convention: + For bind=True tasks, task.run() is the raw underlying function with self=task. + We push a fake request context via push_request(retries=N) to simulate retries, + then patch task.retry to capture the countdown argument rather than letting + Celery's called_directly path re-raise the original exception. """ import pytest +from unittest.mock import patch, AsyncMock +from celery.exceptions import Retry, MaxRetriesExceededError -@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-04") -async def test_retry_backoff(): - pytest.xfail("not implemented yet — Plan 07-04") +def test_retry_backoff(): + """extract_and_classify raises Retry with countdown 30/90/270 on retries 0/1/2. + + D-09: exponential backoff harness. The outer sync task must catch + _ClassificationError and call self.retry(exc=..., countdown=N) where N + comes from [30, 90, 270][min(self.request.retries, 2)]. + """ + from tasks.document_tasks import extract_and_classify, _ClassificationError + + doc_id = "11111111-1111-1111-1111-111111111111" + expected_countdowns = [30, 90, 270] + captured_countdowns = [] + + async def fake_run_raises(document_id): + raise _ClassificationError("fake classification error") + + def capture_retry(exc=None, countdown=None, **kwargs): + """Capture the countdown argument and raise Retry to simulate Celery behaviour.""" + captured_countdowns.append(countdown) + raise Retry(str(exc), exc) + + for retry_num in range(3): + extract_and_classify.push_request(retries=retry_num) + try: + with patch("tasks.document_tasks._run", side_effect=fake_run_raises), \ + patch.object(extract_and_classify, "retry", side_effect=capture_retry): + with pytest.raises(Retry): + extract_and_classify.run(doc_id) + finally: + extract_and_classify.pop_request() + + assert captured_countdowns == expected_countdowns, ( + f"Expected countdowns {expected_countdowns}, got {captured_countdowns}" + ) -@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-04") -async def test_exhaustion_sets_failed_status(): - pytest.xfail("not implemented yet — Plan 07-04") +def test_exhaustion_sets_failed_status(): + """When MaxRetriesExceededError is raised, _mark_classification_failed is called + and the task returns {'document_id': ..., 'status': 'classification_failed'}. + + D-10: final state writeback after exhausting all retries. + + Test contract: assert_called_once_with (not assert_awaited_once_with) because + asyncio.run(_mark_classification_failed(doc_id)) invokes the factory — that is + the __call__ path, not the __await__ path, on the AsyncMock. + """ + from tasks.document_tasks import extract_and_classify, _ClassificationError + + doc_id = "22222222-2222-2222-2222-222222222222" + mock_mark_failed = AsyncMock() + + async def fake_run_raises(document_id): + raise _ClassificationError("fake classification error") + + extract_and_classify.push_request(retries=3) + try: + with patch("tasks.document_tasks._run", side_effect=fake_run_raises), \ + patch("tasks.document_tasks._mark_classification_failed", mock_mark_failed), \ + patch.object(extract_and_classify, "retry", side_effect=MaxRetriesExceededError()): + result = extract_and_classify.run(doc_id) + finally: + extract_and_classify.pop_request() + + # _mark_classification_failed(document_id) was called (factory call, not awaited) + mock_mark_failed.assert_called_once_with(doc_id) + + # Return value reports final failure state + assert result["status"] == "classification_failed" + assert result["document_id"] == doc_id From 63cd707d5222f3067bbb9742a3ac44ca29bf494d Mon Sep 17 00:00:00 2001 From: curo1305 Date: Thu, 4 Jun 2026 23:10:49 +0200 Subject: [PATCH 2/3] feat(07-04): POST /classify re-queues Celery, removes synchronous classify (D-11) - Replace synchronous await classifier.classify_document() with extract_and_classify.delay() - Set doc.status='processing' and commit before dispatching Celery task - Return {'document_id': str, 'status': 'processing'} instead of assigned topics - Promote test_reclassify_requeues_celery from xfail stub to passing test - Add test_reclassify_cross_user_returns_404 for IDOR security invariant (T-07-10) --- backend/api/documents.py | 18 +++--- backend/tests/test_documents.py | 105 ++++++++++++++++++++++++++++++-- 2 files changed, 111 insertions(+), 12 deletions(-) diff --git a/backend/api/documents.py b/backend/api/documents.py index f38e563..18628b8 100644 --- a/backend/api/documents.py +++ b/backend/api/documents.py @@ -714,14 +714,17 @@ async def delete_document( async def classify_document( request: Request, doc_id: str, - body: dict = {}, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ): - """Reclassify a document's topics on demand. + """Re-queue a document for classification via Celery (D-11). + + Sets doc.status='processing', commits, dispatches extract_and_classify.delay(), + and returns {'document_id': str, 'status': 'processing'}. D-16: requires authenticated regular user. Asserts ownership — cross-user classify returns 404 (not 403) to avoid information leakage (T-03-11). + T-07-10: ownership enforced here; IDOR returns 404 per STATE.md policy. """ request.state.current_user = current_user try: @@ -733,13 +736,12 @@ async def classify_document( if doc is None or doc.user_id != current_user.id: raise HTTPException(404, "Document not found") - topic_names = body.get("topics") if body else None - try: - topics = await classifier.classify_document(session, doc_id, topic_names) - except Exception as e: - raise HTTPException(500, f"Classification failed: {e}") + doc.status = "processing" + await session.commit() - return {"topics": topics} + extract_and_classify.delay(str(doc.id)) + + return {"document_id": str(doc.id), "status": "processing"} # ── Range header parsing helper ─────────────────────────────────────────────── diff --git a/backend/tests/test_documents.py b/backend/tests/test_documents.py index 6a76fee..ae187c1 100644 --- a/backend/tests/test_documents.py +++ b/backend/tests/test_documents.py @@ -926,10 +926,107 @@ async def test_delete_cloud_remove_only(async_client, auth_user, db_session): # --------------------------------------------------------------------------- -# Phase 7 Wave 0 xfail stub — D-11 re-classify endpoint +# Phase 7 Plan 04 — D-11 re-classify endpoint (POST /{id}/classify → Celery) # --------------------------------------------------------------------------- -@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-04") -async def test_reclassify_requeues_celery(): - pytest.xfail("not implemented yet — Plan 07-04") +async def test_reclassify_requeues_celery(async_client, auth_user, db_session, monkeypatch): + """POST /api/documents/{id}/classify sets doc.status='processing', enqueues + Celery task via extract_and_classify.delay, and returns {'document_id': ..., + 'status': 'processing'}. + + D-11: reclassify endpoint triggers async processing instead of running + synchronously. The previous synchronous call to classifier.classify_document() + is replaced by extract_and_classify.delay(doc_id). + """ + import uuid as _uuid + from unittest.mock import MagicMock + from db.models import Document + + # Patch Celery delay to avoid real broker + mock_delay = MagicMock() + monkeypatch.setattr("tasks.document_tasks.extract_and_classify.delay", mock_delay) + + # Create a document owned by auth_user + doc_id = _uuid.uuid4() + doc = Document( + id=doc_id, + user_id=auth_user["user"].id, + filename="reclassify_test.txt", + content_type="text/plain", + size_bytes=100, + storage_backend="minio", + status="uploaded", + object_key=f"{auth_user['user'].id}/{doc_id}/{_uuid.uuid4()}.txt", + ) + db_session.add(doc) + await db_session.commit() + + resp = await async_client.post( + f"/api/documents/{doc_id}/classify", + headers=auth_user["headers"], + ) + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["status"] == "processing", f"Expected status=processing: {data}" + assert data["document_id"] == str(doc_id), f"Expected document_id={doc_id}: {data}" + + # Celery task was dispatched once with the doc's string ID + mock_delay.assert_called_once_with(str(doc_id)) + + # DB row was updated to processing + await db_session.refresh(doc) + assert doc.status == "processing", f"Expected doc.status=processing, got {doc.status}" + + +async def test_reclassify_cross_user_returns_404(async_client, auth_user, db_session): + """POST /api/documents/{other_user_doc_id}/classify by a different user returns 404. + + T-07-10 / D-16: cross-user reclassify attempt must return 404 (not 403) to + avoid information leakage. _get_owned_doc ownership check enforces this. + """ + import uuid as _uuid + from db.models import Document, User, Quota + from services.auth import hash_password, create_access_token + + # Create a document owned by auth_user + doc_id = _uuid.uuid4() + doc = Document( + id=doc_id, + user_id=auth_user["user"].id, + filename="other_user_doc.txt", + content_type="text/plain", + size_bytes=100, + storage_backend="minio", + status="uploaded", + object_key=f"{auth_user['user'].id}/{doc_id}/{_uuid.uuid4()}.txt", + ) + db_session.add(doc) + + # Create User B + user_b_id = _uuid.uuid4() + user_b = User( + id=user_b_id, + handle=f"classify_idor_{user_b_id.hex[:8]}", + email=f"classify_idor_{user_b_id.hex[:8]}@example.com", + password_hash=hash_password("Testpassword123!"), + role="user", + is_active=True, + password_must_change=False, + ) + quota_b = Quota(user_id=user_b_id, limit_bytes=104857600, used_bytes=0) + db_session.add(user_b) + db_session.add(quota_b) + await db_session.commit() + + token_b = create_access_token(str(user_b_id), "user") + headers_b = {"Authorization": f"Bearer {token_b}"} + + # User B attempts to reclassify User A's document — must get 404 + resp = await async_client.post( + f"/api/documents/{doc_id}/classify", + headers=headers_b, + ) + assert resp.status_code == 404, ( + f"Expected 404 for cross-user reclassify, got {resp.status_code}: {resp.text}" + ) From 0c1ae5284f0a15d9ca62d0f03d668c9b98e91c4c Mon Sep 17 00:00:00 2001 From: curo1305 Date: Thu, 4 Jun 2026 23:12:11 +0200 Subject: [PATCH 3/3] =?UTF-8?q?docs(07-04):=20add=20plan=20summary=20?= =?UTF-8?q?=E2=80=94=20Celery=20retry=20harness=20+=20POST=20/classify=20r?= =?UTF-8?q?e-queue=20complete?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- .../07-04-SUMMARY.md | 136 ++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 .planning/phases/07-redo-and-optimize-llm-integration/07-04-SUMMARY.md diff --git a/.planning/phases/07-redo-and-optimize-llm-integration/07-04-SUMMARY.md b/.planning/phases/07-redo-and-optimize-llm-integration/07-04-SUMMARY.md new file mode 100644 index 0000000..2bce352 --- /dev/null +++ b/.planning/phases/07-redo-and-optimize-llm-integration/07-04-SUMMARY.md @@ -0,0 +1,136 @@ +--- +phase: 07-redo-and-optimize-llm-integration +plan: 04 +subsystem: backend/tasks + backend/api +tags: [celery, retry, classification, async, tdd] +wave: 4 + +dependency_graph: + requires: + - 07-03 # load_provider_config + provider singletons + provides: + - D-09 # Celery exponential backoff (30/90/270s) + - D-10 # Final classification_failed status writeback + - D-11 # backend half: POST /classify re-queues Celery + affects: + - backend/tasks/document_tasks.py + - backend/api/documents.py + - backend/tests/test_document_tasks.py + - backend/tests/test_documents.py + +tech_stack: + added: [] + patterns: + - "_ClassificationError sentinel escapes asyncio.run() to trigger Celery retry in sync layer (Pitfall 3)" + - "MaxRetriesExceededError caught in nested try/except inside except _ClassificationError block" + - "push_request(retries=N) + patch.object(task, retry) pattern for testing bound Celery tasks" + - "assert_called_once_with (not assert_awaited_once_with) for asyncio.run(AsyncMock(...)) calling convention" + +key_files: + modified: + - backend/tasks/document_tasks.py + - backend/api/documents.py + - backend/tests/test_document_tasks.py + - backend/tests/test_documents.py + +decisions: + - "MaxRetriesExceededError caught in nested try/except inside except _ClassificationError — not as sibling except — because raise self.retry() raises MaxRetriesExceededError during exception handling, which would propagate past a sibling except block" + - "module-level extract_and_classify import retained in documents.py (already existed, already used in confirm and upload handlers) — no deferred import needed" + - "test_reclassify_cross_user_returns_404 added as new IDOR test for classify endpoint (plan said 'regression of existing IDOR test' but none existed for classify; added new test per security mandate)" + +metrics: + completed_date: "2026-06-04" + tasks_completed: 2 + tasks_total: 2 + files_modified: 4 +--- + +# Phase 07 Plan 04: Celery Retry Harness + POST /classify Re-queue Summary + +## One-liner + +Celery exponential-backoff retry harness (30s/90s/270s) via `_ClassificationError` sentinel + MaxRetriesExceededError nested catch, and `POST /classify` converted from synchronous classification to Celery re-queue. + +## Tasks Completed + +| Task | Name | Commit | Files | +|------|------|--------|-------| +| 1 | Celery retry harness + _ClassificationError + classification_failed writeback | e9ee5d4 | document_tasks.py, test_document_tasks.py | +| 2 | POST /api/documents/{id}/classify → re-queue Celery | 63cd707 | documents.py, test_documents.py | + +## What Was Built + +### Task 1: Celery retry harness (D-09 / D-10) + +`backend/tasks/document_tasks.py` was refactored to implement the self-healing retry loop: + +- `class _ClassificationError(Exception)` — sentinel raised by `_run()` when `classifier.classify_document()` fails; escapes `asyncio.run()` and is caught by the outer sync task +- `async def _mark_classification_failed(document_id)` — writes `doc.status = "classification_failed"` after all retries exhausted; called via `asyncio.run(...)` from the `MaxRetriesExceededError` handler +- Task decorator changed to `@celery_app.task(name=..., bind=True, max_retries=3)` +- Retry loop: `except _ClassificationError as exc` → `countdowns = [30, 90, 270]` → `raise self.retry(exc=exc, countdown=countdown)` +- `MaxRetriesExceededError` caught in a **nested** `try/except` inside the `_ClassificationError` handler (not as a sibling `except`) because `raise self.retry()` raises `MaxRetriesExceededError` during exception handling context + +**Critical architectural decision:** The nested catch was required. If `MaxRetriesExceededError` were a sibling `except` block after `except _ClassificationError`, Celery's `self.retry()` call raising `MaxRetriesExceededError` inside the `_ClassificationError` handler would propagate past it without being caught. + +### Task 2: POST /classify re-queue (D-11) + +`backend/api/documents.py` classify endpoint was refactored: + +- Removed: `await classifier.classify_document(session, doc_id, topic_names)` and the topics response +- Added: `doc.status = "processing"`, `await session.commit()`, `extract_and_classify.delay(str(doc.id))` +- Returns: `{"document_id": str(doc.id), "status": "processing"}` with HTTP 200 +- Ownership check retained inline (no `_get_owned_doc` helper existed; plan said to replicate if absent) +- `body: dict = {}` parameter removed (no longer needed since topics are written by the Celery task) + +## Test Coverage + +### Promoted from xfail: +- `test_document_tasks.py::test_retry_backoff` — verifies countdowns [30, 90, 270] for retries 0/1/2 +- `test_document_tasks.py::test_exhaustion_sets_failed_status` — verifies `_mark_classification_failed` called and return dict correct +- `test_documents.py::test_reclassify_requeues_celery` — verifies Celery delay called, doc.status=processing, response correct + +### New tests added: +- `test_documents.py::test_reclassify_cross_user_returns_404` — IDOR test for classify endpoint (T-07-10) + +### Test suite result: +- Before Plan 04: 366 passed, 12 xfailed, 1 pre-existing failure +- After Plan 04: 370 passed, 9 xfailed, 1 pre-existing failure (test_extract_docx missing module — unrelated) +- Net: +4 promoted tests, +1 new IDOR test + +## Deviations from Plan + +### Auto-fixed Issues + +**1. [Rule 1 - Bug] MaxRetriesExceededError requires nested try/except, not sibling** +- **Found during:** Task 1 implementation — tests confirmed the structure +- **Issue:** The plan described `except MaxRetriesExceededError` as a sibling handler after `except _ClassificationError`. However, `raise self.retry(exc=exc, countdown=countdown)` is called inside the `except _ClassificationError` block. When `self.retry()` raises `MaxRetriesExceededError`, Python is already unwinding the `_ClassificationError` handler — a sibling `except MaxRetriesExceededError` would not catch it. +- **Fix:** Used nested `try/except MaxRetriesExceededError` inside the `except _ClassificationError` block +- **Files modified:** `backend/tasks/document_tasks.py` +- **Commit:** e9ee5d4 + +**2. [Rule 2 - Missing Critical Functionality] Added IDOR test for classify endpoint** +- **Found during:** Task 2 — plan referenced "regression of existing IDOR test" but no such test existed for POST /classify +- **Issue:** T-07-10 (cross-user reclassify → 404) had no test coverage +- **Fix:** Added `test_reclassify_cross_user_returns_404` +- **Files modified:** `backend/tests/test_documents.py` +- **Commit:** 63cd707 + +### No architectural changes required. + +## Threat Surface Scan + +No new network endpoints or trust boundaries introduced. The `POST /classify` endpoint already existed; this plan changed its implementation from synchronous to async-queued. The IDOR invariant (T-07-10) is enforced by the ownership check `doc.user_id != current_user.id → 404` — same pattern as all other document endpoints. + +No new entries needed in threat model. + +## Known Stubs + +None — all classification path changes write real state to the DB via the Celery task. + +## Self-Check: PASSED + +- `backend/tasks/document_tasks.py` contains `class _ClassificationError`, `bind=True`, `max_retries=3`, `async def _mark_classification_failed`, `countdowns = [30, 90, 270]`, `raise self.retry(exc=` +- `backend/api/documents.py` POST classify contains `extract_and_classify.delay(` and `doc.status = "processing"`, does NOT contain `await classifier.classify_document(` +- Commits e9ee5d4 and 63cd707 exist in git log +- `pytest tests/test_document_tasks.py tests/test_documents.py::test_reclassify_requeues_celery` exits 0 +- Full suite: 370 passed, 1 pre-existing failure (test_extract_docx — unrelated)