chore: merge executor worktree (worktree-agent-a707cbd4c3780710b)

.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)

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
+    doc.status = "processing"
-    try:
+    await session.commit()
-        topics = await classifier.classify_document(session, doc_id, topic_names)
-    except Exception as e:
-        raise HTTPException(500, f"Classification failed: {e}")
 
-    return {"topics": topics}
+    extract_and_classify.delay(str(doc.id))
+
+    return {"document_id": str(doc.id), "status": "processing"}
 
 
 # ── Range header parsing helper ───────────────────────────────────────────────

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")
+class _ClassificationError(Exception):
-def extract_and_classify(document_id: str) -> dict:
+    """Sentinel exception raised by _run() to signal a retryable classification failure.
-    """Synchronous Celery entry-point — delegates to async _run via asyncio.run."""
+
+    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
+            # Raise sentinel to allow the outer sync layer to call self.retry()
-            doc.status = "classification_failed"
+            # (D-09 Pitfall 3: self.retry() cannot be called inside asyncio.run)
-            await session.commit()
+            raise _ClassificationError(str(e)) from e
-            return {
-                "document_id": document_id,
-                "status": "classification_failed",
-                "error": str(e),
-            }
 
 
 @celery_app.task(name="tasks.document_tasks.cleanup_abandoned_uploads")

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
+Test contract note (asyncio.run + AsyncMock):
-bind=True retry harness is implemented in document_tasks.py.
+  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):
+Celery bind=True calling convention:
-  - Body is a single pytest.xfail() call — no assertion code.
+  For bind=True tasks, task.run() is the raw underlying function with self=task.
-  - strict=False so unexpected passes (xpass) never break CI.
+  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")
+def test_retry_backoff():
-async def test_retry_backoff():
+    """extract_and_classify raises Retry with countdown 30/90/270 on retries 0/1/2.
-    pytest.xfail("not implemented yet — Plan 07-04")
+
+    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")
+def test_exhaustion_sets_failed_status():
-async def test_exhaustion_sets_failed_status():
+    """When MaxRetriesExceededError is raised, _mark_classification_failed is called
-    pytest.xfail("not implemented yet — Plan 07-04")
+    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

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(async_client, auth_user, db_session, monkeypatch):
-async def test_reclassify_requeues_celery():
+    """POST /api/documents/{id}/classify sets doc.status='processing', enqueues
-    pytest.xfail("not implemented yet — Plan 07-04")
+    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}"
+    )