diff --git a/src/crud/document.py b/src/crud/document.py index f8560629..018f3411 100644 --- a/src/crud/document.py +++ b/src/crud/document.py @@ -176,7 +176,8 @@ async def query_documents_most_derived( limit: Maximum number of documents to return Returns: - Sequence of documents ordered by times_derived descending + Sequence of documents ordered by times_derived descending, + ties broken by created_at descending (most recent first) """ stmt = ( select(models.Document) @@ -186,7 +187,13 @@ async def query_documents_most_derived( models.Document.observed == observed, models.Document.deleted_at.is_(None), ) - .order_by(models.Document.times_derived.desc()) + .order_by( + models.Document.times_derived.desc(), + models.Document.created_at.desc(), + # created_at is the transaction timestamp, so documents created in + # the same batch share it -- id keeps the order deterministic. + models.Document.id, + ) .limit(limit) ) @@ -980,7 +987,13 @@ async def is_rejected_duplicate( If the document is not a duplicate, returns False. If the document is a duplicate AND the new document is superior, - deletes the existing document and returns False. + deletes the existing document and returns False. In this case + ``doc.times_derived`` is updated in place to carry the replaced + document's reinforcement count forward. + + If the document is a duplicate AND the existing document is superior, + increments the existing document's ``times_derived`` to record the + reinforcement, then returns True. """ # Step 1: Find potential duplicates using cosine similarity similar_docs = await query_documents( @@ -1014,12 +1027,20 @@ async def is_rejected_duplicate( logger.warning( f"[DUPLICATE DETECTION] Deleting existing in favor of new. new='{doc.content}', existing='{existing_doc.content}'." ) + # Carry the reinforcement count forward so replacing a duplicate counts as + # another derivation rather than resetting times_derived to 1. + doc.times_derived = max(doc.times_derived, existing_doc.times_derived + 1) # Soft-delete the existing document - reconciliation will clean up vectors and hard-delete existing_doc.deleted_at = datetime.datetime.now(datetime.timezone.utc) await db.flush() return False # Don't reject the new document - # Existing document has more information, reject the new one + # Existing document has more information, reject the new one but record the + # reinforcement: a semantic duplicate was derived again. Assign a SQL + # expression so the increment is atomic server-side -- concurrent workers + # reinforcing the same document must not lose updates. + existing_doc.times_derived = models.Document.times_derived + 1 + await db.flush() logger.warning( f"[DUPLICATE DETECTION] Rejecting new in favor of existing. new='{doc.content}', existing='{existing_doc.content}'." ) diff --git a/src/crud/representation.py b/src/crud/representation.py index 4ade79a4..558a5cce 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -444,7 +444,13 @@ class RepresentationManager: models.Document.observed == self.observed, models.Document.deleted_at.is_(None), ) - .order_by(models.Document.times_derived.desc()) + .order_by( + models.Document.times_derived.desc(), + models.Document.created_at.desc(), + # created_at is the transaction timestamp, so documents created + # in the same batch share it -- id keeps the order deterministic. + models.Document.id, + ) ) result = await db.execute(stmt) diff --git a/tests/crud/test_document.py b/tests/crud/test_document.py index f25b75d2..ccde3dac 100644 --- a/tests/crud/test_document.py +++ b/tests/crud/test_document.py @@ -6,6 +6,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src import crud, models, schemas +from src.crud.document import is_rejected_duplicate from src.exceptions import ResourceNotFoundException @@ -274,6 +275,196 @@ class TestDocumentCRUD: assert len(results) == 1 assert results[0].id == times_derived_map[2] + @pytest.mark.asyncio + async def test_most_derived_orders_by_recency_when_reinforcement_ties( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Regression: when times_derived ties, most-derived must fall back to + recency, not insertion order. Otherwise stale conclusions stick to the + front of the injected representation (the mid-Jan stickiness bug).""" + test_workspace, test_peer = sample_data + test_peer2, test_session, _ = await self._setup_test_data( + db_session, test_workspace, test_peer + ) + + base = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + # Three conclusions, all reinforced once -- the real-world steady state + # before the fix -- inserted oldest-first. + for i in range(3): + db_session.add( + models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + content=f"tie {i}", + session_name=test_session.name, + times_derived=1, + created_at=base + datetime.timedelta(days=i), + ) + ) + # A genuinely reinforced conclusion that is also the oldest of all. + db_session.add( + models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + content="hot", + session_name=test_session.name, + times_derived=5, + created_at=base - datetime.timedelta(days=10), + ) + ) + await db_session.flush() + + docs = await crud.query_documents_most_derived( + db_session, + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + limit=10, + ) + contents = [d.content for d in docs] + # Primary sort still wins: the actually-reinforced conclusion leads. + assert contents[0] == "hot" + # Ties break toward most-recent, not oldest-inserted. + assert contents[1:] == ["tie 2", "tie 1", "tie 0"] + + @pytest.mark.asyncio + async def test_duplicate_rejection_reinforces_existing( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Rejecting a new duplicate must bump the surviving doc's times_derived.""" + test_workspace, test_peer = sample_data + test_peer2, test_session, _ = await self._setup_test_data( + db_session, test_workspace, test_peer + ) + + await crud.create_documents( + db_session, + [ + schemas.DocumentCreate( + content="eri loves cats and dogs and birds and snakes", + embedding=[0.5] * 1536, + session_name=test_session.name, + times_derived=1, + metadata=schemas.DocumentMetadata( + message_ids=[1], + message_created_at="2026-01-01T00:00:00Z", + ), + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + # Fewer unique tokens -> existing wins -> new doc is rejected. + new_doc = schemas.DocumentCreate( + content="eri loves cats", + embedding=[0.5] * 1536, + session_name=test_session.name, + times_derived=1, + metadata=schemas.DocumentMetadata( + message_ids=[2], + message_created_at="2026-01-02T00:00:00Z", + ), + ) + rejected = await is_rejected_duplicate( + db_session, + new_doc, + test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + assert rejected is True + surviving = ( + await db_session.execute( + select(models.Document).where( + models.Document.workspace_name == test_workspace.name, + models.Document.observer == test_peer.name, + models.Document.observed == test_peer2.name, + models.Document.deleted_at.is_(None), + ) + ) + ).scalar_one() + assert surviving.times_derived == 2 + + @pytest.mark.asyncio + async def test_duplicate_replacement_carries_count_forward( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """When a new duplicate wins, it must inherit the replaced doc's count + 1 + rather than resetting reinforcement to 1.""" + test_workspace, test_peer = sample_data + test_peer2, test_session, _ = await self._setup_test_data( + db_session, test_workspace, test_peer + ) + + await crud.create_documents( + db_session, + [ + schemas.DocumentCreate( + content="eri loves cats", + embedding=[0.5] * 1536, + session_name=test_session.name, + times_derived=3, + metadata=schemas.DocumentMetadata( + message_ids=[1], + message_created_at="2026-01-01T00:00:00Z", + ), + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + # More information -> new wins -> existing is soft-deleted. + new_doc = schemas.DocumentCreate( + content="eri loves cats and dogs", + embedding=[0.5] * 1536, + session_name=test_session.name, + times_derived=1, + metadata=schemas.DocumentMetadata( + message_ids=[2], + message_created_at="2026-01-02T00:00:00Z", + ), + ) + rejected = await is_rejected_duplicate( + db_session, + new_doc, + test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + assert rejected is False + # Count carried forward onto the replacement (3 -> 4), not reset to 1. + assert new_doc.times_derived == 4 + live = ( + ( + await db_session.execute( + select(models.Document).where( + models.Document.workspace_name == test_workspace.name, + models.Document.observer == test_peer.name, + models.Document.observed == test_peer2.name, + models.Document.deleted_at.is_(None), + ) + ) + ) + .scalars() + .all() + ) + # Original is soft-deleted; replacement isn't inserted until create_documents runs. + assert len(live) == 0 + @pytest.mark.asyncio async def test_delete_document_success( self, diff --git a/tests/crud/test_representation_manager.py b/tests/crud/test_representation_manager.py index 141b61c7..7744e763 100644 --- a/tests/crud/test_representation_manager.py +++ b/tests/crud/test_representation_manager.py @@ -1,5 +1,5 @@ from contextlib import asynccontextmanager -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, patch import pytest @@ -180,6 +180,57 @@ class TestRepresentationManagerSoftDelete: assert doc_live.id in result_ids assert doc_deleted.id not in result_ids + @pytest.mark.asyncio + async def test_query_documents_most_derived_ties_break_by_recency( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Regression: when times_derived ties, the manager's most-derived query + must fall back to recency, not insertion order. Mirrors the equivalent + test on crud.query_documents_most_derived -- the query is duplicated in + both modules and must not drift.""" + test_workspace, test_peer = sample_data + test_peer2, test_session, _, manager = await self._setup( + db_session, test_workspace, test_peer + ) + + base = datetime(2026, 1, 1, tzinfo=timezone.utc) + # Three conclusions, all reinforced once, inserted oldest-first. + for i in range(3): + db_session.add( + models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + content=f"tie {i}", + session_name=test_session.name, + times_derived=1, + created_at=base + timedelta(days=i), + ) + ) + # A genuinely reinforced conclusion that is also the oldest of all. + db_session.add( + models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + content="hot", + session_name=test_session.name, + times_derived=5, + created_at=base - timedelta(days=10), + ) + ) + await db_session.flush() + + results = await manager._query_documents_most_derived(db_session, top_k=10) # pyright: ignore[reportPrivateUsage] + + contents = [doc.content for doc in results] + # Primary sort still wins: the actually-reinforced conclusion leads. + assert contents[0] == "hot" + # Ties break toward most-recent, not oldest-inserted. + assert contents[1:] == ["tie 2", "tie 1", "tie 0"] + class TestRepresentationManagerSave: @pytest.mark.asyncio