RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #910 from plastic-labs/ulysspence/dev-1967-dedup-document-count 

feat(telemetry): counts documents deduped during representation, exact and semantically similar
This commit is contained in:
Ulysse Pence 2026-07-20 16:09:05 -04:00 committed by GitHub
parent 1684b84344 b7a00d5f6b
commit f133797164
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 378 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/crud/__init__.py
View File

@ -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",

52
src/crud/document.py
View File

@ -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(

21
src/crud/representation.py
View File

@ -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,

31
src/deriver/deriver.py
View File

@ -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,
)
)

25
src/telemetry/events/representation.py
View File

@ -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(

18
src/utils/agent_tools.py
View File

@ -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),

182
tests/crud/test_document.py
View File

@ -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(

17
tests/crud/test_representation_manager.py
View File

@ -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()

80
tests/deriver/test_deriver_processing.py
View File

@ -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."""

41
tests/telemetry/test_representation_v2_fields.py
View File

@ -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

12
tests/utils/test_agent_tools.py
View File

@ -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",