fix(retain): prevent duplicate memory units from delta upsert chunk issues

[nicoloboschi]

Summary

Fixes two bugs in the streaming retain pipeline that caused duplicate and stale memory units when documents were upserted multiple times with the same or different content.

Bug 1: Out-of-order chunk index assignment (root cause of delta always falling back)

The streaming retain pipeline extracts facts from chunks concurrently via asyncio.create_task. Chunks arrive at the database consumer in task-completion order (non-deterministic), but chunk_index was assigned based on arrival order:

# BUG: chunk_index based on arrival order, not original position
cm.chunk_index = global_chunk_offset + content_idx_in_batch

This caused chunk_id = {bank}_{doc}_{chunk_index} to be non-deterministic. On subsequent upserts, delta retain compared new content (deterministically chunked) against DB chunks stored at scrambled positions — every hash mismatched — so delta always fell back to full streaming re-processing, wasting LLM calls and risking duplicates through the recovery path.

Fix: Use the original global_idx (position in the pre-chunked content array) which is passed through the queue from the producer:

# FIX: deterministic chunk_id regardless of task completion order
cm.chunk_index = global_idx

Verified with test: test_repeated_upsert_chunks_not_scrambled — confirms chunks are stored at correct indices. test_delta_detects_unchanged_after_first_retain — confirms delta now correctly identifies all chunks as unchanged on second upsert (previously always fell back).

Bug 2: Concurrent upsert race condition

The streaming path splits into separate transactions:

1. TXN1: handle_document_tracking (cascade-delete old doc + insert new doc row)
2. LLM extraction (seconds of latency)
3. TXN2: store_chunks_batch + insert_facts_batch

Two concurrent retains for the same document could interleave:

Request A:  TXN1 (delete→nothing, insert doc) → LLM extraction → TXN2 (store chunks + units)
Request B:       TXN1 (delete A's doc row, but NO chunks to cascade yet!) → LLM → TXN2 (ON CONFLICT overwrites A's chunks + inserts own units)

Result: A's memory units (stale) + B's memory units both linked to same chunks = duplicates + data corruption.

Fix: PostgreSQL advisory lock per (bank_id, document_id) serializes concurrent retain operations:

• Keyed on hash(bank_id:document_id) — doesn't block unrelated documents
• Session-level lock held across the multi-transaction pipeline
• Second request waits, then sees correct state

Bug 3: Content hash mismatch in recovery detection (minor)

The recovery path computed new_content_hash from raw content, but the stored hash was computed after _sanitize_text(). If content contained control characters, recovery wouldn't trigger, causing unnecessary re-processing.

Fix: Apply same sanitization before hashing in recovery check.

Test plan

• test_repeated_upsert_chunks_not_scrambled — chunks stored at deterministic indices
• test_delta_detects_unchanged_after_first_retain — delta correctly detects unchanged content on 2nd/3rd upsert
• All 16 existing test_delta_retain.py tests pass
• All 50 test_retain.py tests pass
• All 8 HTTP API retain integration tests pass
• test_chunk_storage_upsert.py idempotency tests pass
• Lint passes