fix(knowledge): strengthen retry mechanisms for Celery sync tasks

Problem

The knowledge sync pipeline has insufficient retry mechanisms, causing silent data loss during transient failures (500 errors from Claude Code container, Graphiti/Pinecone API timeouts).

Evidence from batch classification of 347 articles (2026-02-21):

• 11 out of 58 articles failed classification with 500 Server Error from the Claude Code container
• ~49 out of 110 Graphiti sync tasks failed silently (110 queued → only 61 episodes created)
• Root cause: classification + Graphiti sync tasks competed for the same Claude Code container, causing overload

Current State

File: knowledge/tasks.py

| Task | Retries | Backoff | Issues |
|------|---------|---------|--------|
| classify_and_sync_to_graphiti (L245) | 2 @ 30s flat | None | Too few retries; flat delay doesn't help with sustained overload; fallback syncs ALL failed articles to Graphiti (defeats cost-saving purpose) |
| sync_document_to_graphiti (L130) | 3 @ 10s linear | delay * (retries+1) → 10/20/30s | Adequate for transient errors |
| update_document_in_index (L7) | 3 @ 5s | None | Only retries doc-not-found (L33-34). Pinecone API errors (L43-46) are caught but returned as {"status": "error"} with no retry |
| delete_knowledge_index_records (L49) | 0 | — | No bind=True, no max_retries. Fire-and-forget. Orphaned vectors if Pinecone is briefly down |
| delete_document_from_graphiti (L227) | 0 | — | No retry. Orphaned episodes if Graphiti is briefly down |

Required Changes

1. classify_and_sync_to_graphiti — increase retries + add exponential backoff

# Before
@shared_task(bind=True, max_retries=2, default_retry_delay=30)

# After
@shared_task(bind=True, max_retries=3, default_retry_delay=30)

And change retry countdown to exponential (L282-283):

# Before
raise self.retry(countdown=self.default_retry_delay)

# After
raise self.retry(countdown=self.default_retry_delay * (2 ** self.request.retries))
# → 30s, 60s, 120s

2. update_document_in_index — retry on Pinecone API errors

Currently L39-46:

try:
    result = PineconeRAGHandler().upsert_document(doc)
    ...
except Exception as e:
    error_msg = f"Error upserting document id={doc.id} to Pinecone: {e}"
    logger.error(error_msg, exc_info=True)
    return {"status": "error", ...}  # ← silently swallowed

Should retry:

except Exception as e:
    logger.error(f"Error upserting document id={doc.id} to Pinecone: {e}", exc_info=True)
    if self.request.retries < self.max_retries:
        raise self.retry(countdown=self.default_retry_delay * (self.request.retries + 1))
    return {"status": "error", "document_id": doc.id, "reason": str(e)}

3. delete_knowledge_index_records — add retries

# Before
@shared_task
def delete_knowledge_index_records(prefix_vector_id, total_chunks, namespace):

# After
@shared_task(bind=True, max_retries=3, default_retry_delay=10)
def delete_knowledge_index_records(self, prefix_vector_id, total_chunks, namespace):

Add retry in the except block (L77-80):

except Exception as e:
    logger.error(f"Error deleting vectors for {prefix_vector_id}: {e}", exc_info=True)
    if self.request.retries < self.max_retries:
        raise self.retry(countdown=self.default_retry_delay)
    return {"status": "error", "reason": str(e)}

4. delete_document_from_graphiti — add retries

# Before
@shared_task
def delete_document_from_graphiti(slug, group_id):

# After
@shared_task(bind=True, max_retries=2, default_retry_delay=10)
def delete_document_from_graphiti(self, slug, group_id):

Add retry:

deleted = _graphiti_delete_episodes_for_article(slug, group_id)
if deleted == 0:
    # Could be a transient failure — retry to be safe
    # (the helper function swallows errors and returns 0)
    if self.request.retries < self.max_retries:
        raise self.retry(countdown=self.default_retry_delay)
return {"status": "deleted", "slug": slug, "group_id": group_id, "deleted_count": deleted}

5. (Optional) Add rate limiting to prevent container overload

@shared_task(bind=True, max_retries=3, default_retry_delay=30, rate_limit='10/m')
def classify_and_sync_to_graphiti(self, document_id):

This prevents the scenario where bulk saves flood the classifier container. 10/min is conservative — adjust based on container capacity.

Impact

• Without fix: Transient API errors silently drop data — articles missing from Pinecone/Graphiti with no visibility
• With fix: Retries handle transient failures; only persistent failures result in error returns (which are logged)

Testing

• Unit tests: mock the API calls, verify retry behavior on 500 errors
• Integration: save a KnowledgeDoc while Graphiti/Pinecone is briefly unreachable, verify eventual sync

Notes

• delete_knowledge_index_records adding self parameter is a breaking change for any existing .si() calls — check knowledge/services.py L192 which uses .si(). Celery bind=True tasks receive self as the first arg automatically, so .si(prefix, chunks, ns) signature remains the same for callers.
• The rate_limit on classification is optional but recommended — it was the root cause of the 500 errors we observed.

@claude