[nicoloboschi]

Fixes #6337.

Problem

Two retains of the same document with the same content, running at once (a client retrying a slow retain, or a task run again while its first run is still writing), stored the facts of a multi-batch document twice.

Each batch checked "is the document still mine?" by comparing the document's content hash with its own. Two runs with the same content carry the same hash, so the check passed for both, and both wrote every batch after the first. Chunks were deduplicated by their fixed ids, facts were not. Recall, consolidation and everything else reading the bank saw the duplicates.

The same weak check guarded the two other routes the issue names: a second retain that starts after the first has committed a batch (the streaming recovery path for a sliced document, and the delta path for one that fits a single batch).

Fix

Under the per-document row lock, a batch now drops the chunks already stored with the same text, together with their facts. A chunk commits in the same transaction as its facts, so "chunk stored" means "its facts are stored". Whichever run reaches a chunk first writes it, so the document ends complete once the run that last cleared it finishes. The delta write gets the same guard.

This is the approach the reporter proposed. Their first idea, aborting a run when it finds an overlap, leaves gaps when both runs abort, so it is not used.

Details:

• Phase-1 entity resolution and causal relations point at facts by position, so the kept facts are renumbered and both are rebuilt (_drop_committed_chunks).
• The entity mention counts phase 1 queued for dropped facts are un-queued (forget_pending_mentions), so mentions are not counted twice either.
• The guard reads only this batch's chunk rows (new optional chunk_indices filter on load_existing_chunks, = ANY($3::integer[]), which the Oracle rewrite turns into IN (...)).
• Store-owned banks write a retain in one atomic call and need no guard.

What changes

Only same-content overlaps. Runs with different content still hand over as before (the later one wins, the other stops). Every batch except a fresh first one does one more small read: its own chunks' hashes.

Tests

• test_concurrent_retains_of_the_same_content_store_each_fact_once: the reporter's test (kept as their commit). Failed 3/3 before, passes after.
• test_concurrent_same_content_retains_out_of_order_store_each_chunk_once (12 seeds): chunks finish extracting in random order, so the runs batch different chunks. 12/12 failed before, pass after, and the document always ends complete.
• test_same_content_retain_overtaken_between_load_and_write_stores_each_chunk_once (delta and recovery variants): pauses the runs at exact points so B reads the half-written document and A writes the rest before B does; asserts B really reached the guard. Both failed before.
• test_drop_committed_chunks.py: unit tests for the renumbering (entities, ANN links, causal links, mention counts).
• System test test_92b_concurrent_same_retain.py: two identical retains at once through the real server; the exact set of facts, one per chunk. Failed 2/2 before.

Locally: 1589 tests across the 60 retain-related files pass; ruff, ty and lint.sh are clean.

Thanks to @skybleak333 for the precise report, the failing test, and the analysis of why the obvious fix is wrong.