diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 58bcd9f0..b34e4d17 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -5,6 +5,7 @@ from .collection import ( ) from .deriver import get_deriver_status, get_queue_status from .document import ( + CreateDocumentsResult, create_documents, create_observations, delete_document, @@ -83,6 +84,7 @@ __all__ = [ "get_deriver_status", "get_queue_status", # Document + "CreateDocumentsResult", "create_documents", "create_observations", "fetch_documents_by_ids", diff --git a/src/crud/document.py b/src/crud/document.py index ed381340..e46f311a 100644 --- a/src/crud/document.py +++ b/src/crud/document.py @@ -1,5 +1,7 @@ import datetime from collections.abc import Sequence +from dataclasses import dataclass, field +from enum import Enum from logging import getLogger from typing import Any, cast @@ -441,6 +443,15 @@ def _normalize_content(content: str) -> str: return content.strip().lower() +@dataclass +class CreateDocumentsResult: + created_documents: list[schemas.DocumentCreate] = field(default_factory=list) + exact_dup_in_batch_count: int = 0 + exact_dup_existing_count: int = 0 + semantic_dup_rejected_count: int = 0 + semantic_dup_replaced_count: int = 0 + + async def create_documents( db: AsyncSession, documents: list[schemas.DocumentCreate], @@ -449,7 +460,7 @@ async def create_documents( observer: str, observed: str, deduplicate: bool = False, -) -> list[schemas.DocumentCreate]: +) -> CreateDocumentsResult: """ Create multiple documents with optional duplicate detection. @@ -517,6 +528,10 @@ async def create_documents( # duplicates within a single inference call collapse to one document. seen_in_batch: set[str] = set() + exact_dup_existing_count = 0 + exact_dup_in_batch_count = 0 + semantic_dup_rejected_count = 0 + semantic_dup_replaced_count = 0 for doc in documents: try: normalized_content = _normalize_content(doc.content) @@ -524,6 +539,7 @@ async def create_documents( # Exact-match dedup, always on: # 1) collapse exact duplicates within this batch (drop silently). if normalized_content in seen_in_batch: + exact_dup_in_batch_count += 1 continue seen_in_batch.add(normalized_content) @@ -542,16 +558,22 @@ async def create_documents( doc.times_derived, ) await db.flush() + exact_dup_existing_count += 1 continue # for each document, if deduplicate is True, perform a process # that checks against existing documents and either rejects this document # as a duplicate OR deletes an existing document that is a duplicate. if deduplicate: - is_duplicate = await is_rejected_duplicate( + duplicate_result = await is_rejected_duplicate( db, doc, workspace_name, observer=observer, observed=observed ) - if is_duplicate: + if duplicate_result is SemanticRejectionResult.REPLACED_EXISTING: + # Existing doc was soft-deleted in favor of this one; the + # new doc still gets inserted below. + semantic_dup_replaced_count += 1 + elif duplicate_result is SemanticRejectionResult.REJECTED: + semantic_dup_rejected_count += 1 continue metadata_dict = doc.metadata.model_dump(exclude_none=True) @@ -703,7 +725,13 @@ async def create_documents( "Failed to create documents due to integrity constraint violation" ) from e - return accepted_documents + return CreateDocumentsResult( + created_documents=accepted_documents, + exact_dup_existing_count=exact_dup_existing_count, + exact_dup_in_batch_count=exact_dup_in_batch_count, + semantic_dup_rejected_count=semantic_dup_rejected_count, + semantic_dup_replaced_count=semantic_dup_replaced_count, + ) async def delete_document( @@ -1053,6 +1081,12 @@ async def create_observations( return honcho_documents +class SemanticRejectionResult(Enum): + NOT_DUPLICATE = 0 + REPLACED_EXISTING = 1 + REJECTED = 2 + + async def is_rejected_duplicate( db: AsyncSession, doc: schemas.DocumentCreate, @@ -1060,7 +1094,7 @@ async def is_rejected_duplicate( *, observer: str, observed: str, -) -> bool: +) -> SemanticRejectionResult: """ Check if a document is a duplicate of an existing document. @@ -1094,7 +1128,7 @@ async def is_rejected_duplicate( ) if not similar_docs: - return False + return SemanticRejectionResult.NOT_DUPLICATE existing_doc = similar_docs[0] @@ -1121,7 +1155,9 @@ async def is_rejected_duplicate( # 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 + return ( + SemanticRejectionResult.REPLACED_EXISTING + ) # Don't reject the new document # Existing document has more information, reject the new one but record the # reinforcement: a semantic duplicate was derived again. greatest(...) keeps @@ -1138,7 +1174,7 @@ async def is_rejected_duplicate( doc.content, existing_doc.content, ) - return True + return SemanticRejectionResult.REJECTED async def cleanup_soft_deleted_documents( diff --git a/src/crud/representation.py b/src/crud/representation.py index 558a5cce..3061c2d9 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -64,7 +64,7 @@ class RepresentationManager: session_name: str, message_created_at: datetime.datetime, message_level_configuration: ResolvedConfiguration, - ) -> int: + ) -> crud.CreateDocumentsResult: """ Save Representation objects to the collection as a set of documents. @@ -75,14 +75,15 @@ class RepresentationManager: message_created_at: Timestamp when the message was created Returns: - The number of *new documents saved* + The result of document creation, including saved documents and + deduplication counts. """ - new_documents = 0 + empty_result = crud.CreateDocumentsResult() if not representation.deductive and not representation.explicit: logger.debug("No observations to save") - return new_documents + return empty_result all_observations = [ _normalized_observation(obs) @@ -91,7 +92,7 @@ class RepresentationManager: ] if not all_observations: logger.debug("No non-empty observations to save") - return new_documents + return empty_result # Batch embed all observations batch_embed_start = time.perf_counter() @@ -123,7 +124,7 @@ class RepresentationManager: # Batch create document objects create_document_start = time.perf_counter() async with tracked_db("representation_manager.save_representation") as db: - new_documents = await self._save_representation_internal( + new_documents_result = await self._save_representation_internal( db, all_observations, embeddings, @@ -141,7 +142,7 @@ class RepresentationManager: "ms", ) - return new_documents + return new_documents_result async def _save_representation_internal( self, @@ -152,7 +153,7 @@ class RepresentationManager: session_name: str, message_created_at: datetime.datetime, message_level_configuration: ResolvedConfiguration, - ) -> int: + ) -> crud.CreateDocumentsResult: # get_or_create_collection already handles IntegrityError with rollback and a retry collection = await crud.get_or_create_collection( db, @@ -191,7 +192,7 @@ class RepresentationManager: ) # Use bulk creation with optional duplicate detection - accepted_documents = await crud.create_documents( + accepted_documents_result = await crud.create_documents( db, documents_to_create, self.workspace_name, @@ -206,7 +207,7 @@ class RepresentationManager: except Exception as e: logger.warning(f"Failed to check dream scheduling: {e}") - return len(accepted_documents) + return accepted_documents_result async def get_working_representation( self, diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 3e68d084..2ad7d14f 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -194,6 +194,7 @@ async def process_representation_tasks_batch( latest_message.created_at, ) + agg_representation_result = crud.CreateDocumentsResult() successful_observer_count = 0 if observations.is_empty() or not message_ids: logger.warning( @@ -213,12 +214,26 @@ async def process_representation_tasks_batch( ) try: - await representation_manager.save_representation( - observations, - message_ids, - latest_message.session_name, - latest_message.created_at, - message_level_configuration, + representation_result = ( + await representation_manager.save_representation( + observations, + message_ids, + latest_message.session_name, + latest_message.created_at, + message_level_configuration, + ) + ) + agg_representation_result.exact_dup_existing_count += ( + representation_result.exact_dup_existing_count + ) + agg_representation_result.exact_dup_in_batch_count += ( + representation_result.exact_dup_in_batch_count + ) + agg_representation_result.semantic_dup_rejected_count += ( + representation_result.semantic_dup_rejected_count + ) + agg_representation_result.semantic_dup_replaced_count += ( + representation_result.semantic_dup_replaced_count ) successful_observer_count += 1 except Exception as e: @@ -318,5 +333,9 @@ async def process_representation_tasks_batch( hit_batch_token_cap=hit_batch_token_cap, hit_input_token_cap=response.hit_input_token_cap, observer_count=successful_observer_count, + exact_dup_existing_count=agg_representation_result.exact_dup_existing_count, + exact_dup_in_batch_count=agg_representation_result.exact_dup_in_batch_count, + semantic_dup_rejected_count=agg_representation_result.semantic_dup_rejected_count, + semantic_dup_replaced_count=agg_representation_result.semantic_dup_replaced_count, ) ) diff --git a/src/telemetry/events/representation.py b/src/telemetry/events/representation.py index 92db6806..ad6dfca4 100644 --- a/src/telemetry/events/representation.py +++ b/src/telemetry/events/representation.py @@ -100,6 +100,31 @@ class RepresentationCompletedEvent(BaseEvent): default=0, description="Estimated tokens for the system/scaffold portion of the prompt", ) + exact_dup_in_batch_count: int = Field( + default=0, + description="Number of documents produced in this representation that had the same normalized content", + ) + exact_dup_existing_count: int = Field( + default=0, + description=( + "Number of documents previously written that had a representation that had the same normalized " + "content as a document in this representation" + ), + ) + semantic_dup_rejected_count: int = Field( + default=0, + description=( + "Number of documents in this representation rejected because their cosine-similarity was high " + "for an existing document but were worse than the corresponding existing document" + ), + ) + semantic_dup_replaced_count: int = Field( + default=0, + description=( + "Number of documents in this representation that replaced existing documents because their " + "cosine-similarity was high and they were better than the corresponding existing document" + ), + ) # Cap configuration + hit flags () batch_max_tokens: int = Field( diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index ca9db2dd..b0cf3051 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -987,14 +987,16 @@ async def create_observations( accepted: list[schemas.DocumentCreate] = [] if documents: async with tracked_db("create_observations.save") as db: - accepted = await crud.create_documents( - db, - documents=documents, - workspace_name=workspace_name, - observer=observer, - observed=observed, - deduplicate=True, - ) + accepted = ( + await crud.create_documents( + db, + documents=documents, + workspace_name=workspace_name, + observer=observer, + observed=observed, + deduplicate=True, + ) + ).created_documents logger.info( "Created %d observations in %s/%s/%s", len(accepted), diff --git a/tests/crud/test_document.py b/tests/crud/test_document.py index 1258b522..65768c83 100644 --- a/tests/crud/test_document.py +++ b/tests/crud/test_document.py @@ -6,7 +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.crud.document import SemanticRejectionResult, is_rejected_duplicate from src.exceptions import ResourceNotFoundException @@ -381,7 +381,7 @@ class TestDocumentCRUD: observed=test_peer2.name, ) - assert rejected is True + assert rejected is SemanticRejectionResult.REJECTED surviving = ( await db_session.execute( select(models.Document).where( @@ -445,7 +445,7 @@ class TestDocumentCRUD: observed=test_peer2.name, ) - assert rejected is False + assert rejected is SemanticRejectionResult.REPLACED_EXISTING # Count carried forward onto the replacement (3 -> 4), not reset to 1. assert new_doc.times_derived == 4 live = ( @@ -509,7 +509,7 @@ class TestDocumentCRUD: ), ] - accepted = await crud.create_documents( + result = await crud.create_documents( db_session, documents=doc_schemas, workspace_name=test_workspace.name, @@ -517,8 +517,13 @@ class TestDocumentCRUD: observed=test_peer2.name, deduplicate=False, ) + accepted = result.created_documents assert len(accepted) == 1 + assert result.exact_dup_in_batch_count == 2 + assert result.exact_dup_existing_count == 0 + assert result.semantic_dup_rejected_count == 0 + assert result.semantic_dup_replaced_count == 0 live = ( ( await db_session.execute( @@ -571,7 +576,7 @@ class TestDocumentCRUD: ) # Case/whitespace variant of the existing content -> exact match. - accepted = await crud.create_documents( + result = await crud.create_documents( db_session, [ schemas.DocumentCreate( @@ -590,8 +595,13 @@ class TestDocumentCRUD: observed=test_peer2.name, deduplicate=False, ) + accepted = result.created_documents assert len(accepted) == 0 + assert result.exact_dup_existing_count == 1 + assert result.exact_dup_in_batch_count == 0 + assert result.semantic_dup_rejected_count == 0 + assert result.semantic_dup_replaced_count == 0 surviving = ( ( await db_session.execute( @@ -663,25 +673,27 @@ class TestDocumentCRUD: # Incoming exact match claims more accumulated reinforcement (5) than # existing + 1 (3) -> incoming wins. - accepted = await crud.create_documents( - db_session, - [ - schemas.DocumentCreate( - content="user likes coffee ", - embedding=[0.9] * 1536, - session_name=test_session.name, - times_derived=5, - metadata=schemas.DocumentMetadata( - message_ids=[2], - message_created_at="2026-01-02T00:00:00Z", - ), - ) - ], - workspace_name=test_workspace.name, - observer=test_peer.name, - observed=test_peer2.name, - deduplicate=False, - ) + accepted = ( + await crud.create_documents( + db_session, + [ + schemas.DocumentCreate( + content="user likes coffee ", + embedding=[0.9] * 1536, + session_name=test_session.name, + times_derived=5, + metadata=schemas.DocumentMetadata( + message_ids=[2], + message_created_at="2026-01-02T00:00:00Z", + ), + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + deduplicate=False, + ) + ).created_documents assert len(accepted) == 0 live = await _live() assert len(live) == 1 @@ -689,24 +701,26 @@ class TestDocumentCRUD: # A normal re-derivation (times_derived defaults to 1) now bumps by one: # greatest(existing + 1, 1) -> existing + 1. - accepted = await crud.create_documents( - db_session, - [ - schemas.DocumentCreate( - content="USER LIKES COFFEE", - embedding=[0.4] * 1536, - session_name=test_session.name, - metadata=schemas.DocumentMetadata( - message_ids=[3], - message_created_at="2026-01-03T00:00:00Z", - ), - ) - ], - workspace_name=test_workspace.name, - observer=test_peer.name, - observed=test_peer2.name, - deduplicate=False, - ) + accepted = ( + await crud.create_documents( + db_session, + [ + schemas.DocumentCreate( + content="USER LIKES COFFEE", + embedding=[0.4] * 1536, + session_name=test_session.name, + metadata=schemas.DocumentMetadata( + message_ids=[3], + message_created_at="2026-01-03T00:00:00Z", + ), + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + deduplicate=False, + ) + ).created_documents assert len(accepted) == 0 live = await _live() assert len(live) == 1 @@ -746,7 +760,7 @@ class TestDocumentCRUD: ) db_session.autoflush = False - accepted = await crud.create_documents( + result = await crud.create_documents( db_session, [ schemas.DocumentCreate( @@ -775,9 +789,14 @@ class TestDocumentCRUD: observed=test_peer2.name, deduplicate=True, ) + accepted = result.created_documents assert len(accepted) == 1 assert accepted[0].content == "User likes coffee and tea" + assert result.exact_dup_existing_count == 1 + assert result.semantic_dup_replaced_count == 1 + assert result.exact_dup_in_batch_count == 0 + assert result.semantic_dup_rejected_count == 0 surviving = ( ( @@ -797,6 +816,65 @@ class TestDocumentCRUD: assert surviving[0].content == "User likes coffee and tea" assert surviving[0].times_derived == 3 + @pytest.mark.asyncio + async def test_semantic_dedup_rejected_counts( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """A semantically-similar doc with less information than the existing one + is rejected, and the rejection is counted on the result.""" + 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. + result = await crud.create_documents( + db_session, + [ + 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", + ), + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + deduplicate=True, + ) + + assert len(result.created_documents) == 0 + assert result.semantic_dup_rejected_count == 1 + assert result.exact_dup_in_batch_count == 0 + assert result.exact_dup_existing_count == 0 + assert result.semantic_dup_replaced_count == 0 + @pytest.mark.asyncio async def test_delete_document_success( self, @@ -902,15 +980,17 @@ class TestDocumentCRUD: ] # Create documents - count = await crud.create_documents( - db_session, - documents=doc_schemas, - workspace_name=test_workspace.name, - observer=test_peer.name, - observed=test_peer2.name, - ) + created_documents = ( + await crud.create_documents( + db_session, + documents=doc_schemas, + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + ).created_documents - assert len(count) == 2 + assert len(created_documents) == 2 # Verify documents were created stmt = select(models.Document).where( diff --git a/tests/crud/test_representation_manager.py b/tests/crud/test_representation_manager.py index 7744e763..f4be22dd 100644 --- a/tests/crud/test_representation_manager.py +++ b/tests/crud/test_representation_manager.py @@ -1,6 +1,6 @@ from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from nanoid import generate as generate_nanoid @@ -8,6 +8,7 @@ from sqlalchemy import func, update from sqlalchemy.ext.asyncio import AsyncSession from src import models +from src.crud.document import CreateDocumentsResult from src.crud.representation import RepresentationManager from src.schemas.configuration import ( ResolvedConfiguration, @@ -268,7 +269,9 @@ class TestRepresentationManagerSave: patch.object( manager, "_save_representation_internal", - new=AsyncMock(return_value=1), + new=AsyncMock( + return_value=CreateDocumentsResult(created_documents=[MagicMock()]) + ), ) as mock_save, ): saved = await manager.save_representation( @@ -279,7 +282,7 @@ class TestRepresentationManagerSave: message_level_configuration=_resolved_config(), ) - assert saved == 1 + assert len(saved.created_documents) == 1 mock_embed.assert_awaited_once_with(["useful observation"]) saved_observations = _saved_observations(mock_save) assert len(saved_observations) == 1 @@ -322,7 +325,9 @@ class TestRepresentationManagerSave: patch.object( manager, "_save_representation_internal", - new=AsyncMock(return_value=1), + new=AsyncMock( + return_value=CreateDocumentsResult(created_documents=[MagicMock()]) + ), ) as mock_save, ): saved = await manager.save_representation( @@ -333,7 +338,7 @@ class TestRepresentationManagerSave: message_level_configuration=_resolved_config(), ) - assert saved == 1 + assert len(saved.created_documents) == 1 mock_embed.assert_awaited_once_with(["inferred conclusion"]) saved_observations = _saved_observations(mock_save) assert len(saved_observations) == 1 @@ -384,6 +389,6 @@ class TestRepresentationManagerSave: message_level_configuration=_resolved_config(), ) - assert saved == 0 + assert len(saved.created_documents) == 0 mock_embed.assert_not_awaited() mock_save.assert_not_awaited() diff --git a/tests/deriver/test_deriver_processing.py b/tests/deriver/test_deriver_processing.py index a2058bf9..6785d159 100644 --- a/tests/deriver/test_deriver_processing.py +++ b/tests/deriver/test_deriver_processing.py @@ -5,11 +5,15 @@ from unittest.mock import AsyncMock, Mock, patch import pytest -from src import models +from src import crud, models from src.config import settings from src.deriver.deriver import process_representation_tasks_batch from src.llm import HonchoLLMCallResponse -from src.utils.representation import PromptRepresentation, Representation +from src.utils.representation import ( + ExplicitObservationBase, + PromptRepresentation, + Representation, +) from src.utils.work_unit import construct_work_unit_key, parse_work_unit_key @@ -316,6 +320,78 @@ class TestDeriverProcessing: for record in caplog.records ) + async def test_emits_dedup_counts_summed_across_observers(self) -> None: + """RepresentationCompletedEvent dedup counts must be the sum across all + observer collections, not the last observer's result.""" + message = Mock( + id=1, + public_id="msg_dedup", + session_name="session-1", + workspace_name="workspace-1", + peer_name="alice", + content="hello", + token_count=5, + created_at=datetime.now(timezone.utc), + ) + configuration = Mock() + configuration.reasoning.enabled = True + + mock_response = HonchoLLMCallResponse( + content=PromptRepresentation( + explicit=[ExplicitObservationBase(content="alice says hello")] + ), + input_tokens=10, + output_tokens=5, + finish_reasons=["STOP"], + ) + + manager = Mock() + manager.save_representation = AsyncMock( + side_effect=[ + crud.CreateDocumentsResult( + exact_dup_in_batch_count=1, + exact_dup_existing_count=2, + semantic_dup_rejected_count=3, + semantic_dup_replaced_count=4, + ), + crud.CreateDocumentsResult( + exact_dup_in_batch_count=10, + exact_dup_existing_count=20, + semantic_dup_rejected_count=30, + semantic_dup_replaced_count=40, + ), + ] + ) + emitted: list[Any] = [] + + with ( + patch( + "src.deriver.deriver.honcho_llm_call", + new_callable=AsyncMock, + return_value=mock_response, + ), + patch( + "src.deriver.deriver.RepresentationManager", + return_value=manager, + ), + patch("src.deriver.deriver.emit", side_effect=emitted.append), + ): + await process_representation_tasks_batch( + messages=[message], + message_level_configuration=configuration, + observers=["bob", "carol"], + observed="alice", + queue_item_message_ids=[1], + ) + + assert len(emitted) == 1 + event = emitted[0] + assert event.observer_count == 2 + assert event.exact_dup_in_batch_count == 11 + assert event.exact_dup_existing_count == 22 + assert event.semantic_dup_rejected_count == 33 + assert event.semantic_dup_replaced_count == 44 + class TestBackwardsCompatibility: """Test backwards compatibility for queue items created before the deduplication change.""" diff --git a/tests/telemetry/test_representation_v2_fields.py b/tests/telemetry/test_representation_v2_fields.py index 669da96b..a41bed41 100644 --- a/tests/telemetry/test_representation_v2_fields.py +++ b/tests/telemetry/test_representation_v2_fields.py @@ -2,8 +2,6 @@ """tests for RepresentationCompletedEvent additive fields + truncation. Targets: -- Schema stays at v2 (additive, no bump). Existing `input_tokens` semantics - unchanged. - fields are defaultable (no breakage for callers that ignore them) and round-trip through Pydantic serialization. - `HonchoLLMCallResponse.hit_input_token_cap` defaults to False but can be @@ -17,10 +15,6 @@ from src.telemetry.events.representation import RepresentationCompletedEvent class TestRepresentationV2AdditiveFields: - def test_schema_stays_at_v2(self): - """is additive — schema_version must NOT bump to 3.""" - assert RepresentationCompletedEvent.schema_version() == 2 - def test_new_fields_are_optional(self): """Existing callers must keep working without supplying any new fields. All new fields default.""" @@ -53,6 +47,10 @@ class TestRepresentationV2AdditiveFields: assert event.hit_batch_token_cap is False assert event.hit_input_token_cap is False assert event.observer_count == 0 + assert event.exact_dup_in_batch_count == 0 + assert event.exact_dup_existing_count == 0 + assert event.semantic_dup_rejected_count == 0 + assert event.semantic_dup_replaced_count == 0 def test_input_tokens_semantics_preserved(self): """The downstream metering key must remain 'queued-message tokens'. @@ -159,10 +157,41 @@ class TestRepresentationV2AdditiveFields: "hit_batch_token_cap", "hit_input_token_cap", "observer_count", + "exact_dup_in_batch_count", + "exact_dup_existing_count", + "semantic_dup_rejected_count", + "semantic_dup_replaced_count", ): assert field in data, f"missing field: {field}" assert data["hit_batch_token_cap"] is True + def test_dedup_count_fields_round_trip(self): + event = RepresentationCompletedEvent( + workspace_name="ws", + session_name="s", + observed="user", + queue_items_processed=1, + earliest_message_id="m1", + latest_message_id="m1", + message_count=1, + explicit_conclusion_count=0, + context_preparation_ms=10.0, + llm_call_ms=100.0, + total_duration_ms=110.0, + input_tokens=100, + total_input_tokens=200, + output_tokens=50, + exact_dup_in_batch_count=2, + exact_dup_existing_count=3, + semantic_dup_rejected_count=4, + semantic_dup_replaced_count=5, + ) + data = event.model_dump(mode="json") + assert data["exact_dup_in_batch_count"] == 2 + assert data["exact_dup_existing_count"] == 3 + assert data["semantic_dup_rejected_count"] == 4 + assert data["semantic_dup_replaced_count"] == 5 + class TestHitInputTokenCapFlag: """`HonchoLLMCallResponse.hit_input_token_cap` is the bridge between the diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index 12037bf7..3f2dad6b 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -318,10 +318,10 @@ class TestCreateObservations: observer: str, observed: str, deduplicate: bool = False, - ) -> list[Any]: + ) -> crud.CreateDocumentsResult: _ = (workspace_name, observer, observed, deduplicate) created_documents.extend(documents) - return documents + return crud.CreateDocumentsResult(created_documents=documents) monkeypatch.setattr( "src.utils.agent_tools.embedding_client.simple_batch_embed", @@ -379,10 +379,10 @@ class TestCreateObservations: observer: str, observed: str, deduplicate: bool = False, - ) -> list[Any]: + ) -> crud.CreateDocumentsResult: _ = (workspace_name, observer, observed, deduplicate) created_documents.extend(documents) - return documents + return crud.CreateDocumentsResult(created_documents=documents) monkeypatch.setattr( "src.utils.agent_tools.embedding_client.simple_batch_embed", @@ -438,10 +438,10 @@ class TestCreateObservations: observer: str, observed: str, deduplicate: bool = False, - ) -> list[Any]: + ) -> crud.CreateDocumentsResult: _ = (workspace_name, observer, observed, deduplicate) created_documents.extend(documents) - return documents + return crud.CreateDocumentsResult(created_documents=documents) monkeypatch.setattr( "src.utils.agent_tools.embedding_client.simple_batch_embed",