From a05c2f8ec1f03893d9805a39a30a15f2f061c2f0 Mon Sep 17 00:00:00 2001 From: Phil Date: Wed, 29 Apr 2026 13:18:00 -0600 Subject: [PATCH] fix(deriver): ignore blank observations before embedding (#615) * fix(deriver): ignore blank observations before embedding * Address PR review on observation normalization * Harden mock await arg access in tests * Unify blank observation filtering across tool paths * Move soft-delete query test back to fixture class --- src/crud/representation.py | 29 +++- src/utils/agent_tools.py | 20 ++- tests/crud/test_representation_manager.py | 185 ++++++++++++++++++++++ tests/utils/test_agent_tools.py | 91 +++++++++++ 4 files changed, 317 insertions(+), 8 deletions(-) diff --git a/src/crud/representation.py b/src/crud/representation.py index de97d0d3..9b689d06 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -26,6 +26,21 @@ from src.utils.representation import ( logger = logging.getLogger(__name__) +def _observation_text(obs: ExplicitObservation | DeductiveObservation) -> str: + """Return the canonical text payload for an explicit or deductive observation.""" + return obs.conclusion if isinstance(obs, DeductiveObservation) else obs.content + + +def _normalized_observation( + obs: ExplicitObservation | DeductiveObservation, +) -> ExplicitObservation | DeductiveObservation: + """Return an observation with its persisted/embed text normalized.""" + text = _observation_text(obs).strip() + if isinstance(obs, DeductiveObservation): + return obs.model_copy(update={"conclusion": text}) + return obs.model_copy(update={"content": text}) + + class RepresentationManager: """Unified manager for representation and document queries.""" @@ -67,15 +82,19 @@ class RepresentationManager: logger.debug("No observations to save") return new_documents - all_observations = representation.deductive + representation.explicit + all_observations = [ + _normalized_observation(obs) + for obs in representation.deductive + representation.explicit + if _observation_text(obs).strip() + ] + if not all_observations: + logger.debug("No non-empty observations to save") + return new_documents # Batch embed all observations batch_embed_start = time.perf_counter() - observation_texts = [ - obs.conclusion if isinstance(obs, DeductiveObservation) else obs.content - for obs in all_observations - ] + observation_texts = [_observation_text(obs) for obs in all_observations] try: embeddings = await embedding_client.simple_batch_embed(observation_texts) except ValueError as e: diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index d3bd93ab..0d95cf6d 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -33,6 +33,13 @@ logger = logging.getLogger(__name__) MAX_PEER_CARD_FACTS = 40 +def _normalized_observation_input( + obs: schemas.ObservationInput, +) -> schemas.ObservationInput: + """Return an observation input with content normalized for persistence/embedding.""" + return obs.model_copy(update={"content": obs.content.strip()}) + + def _base_observation_properties() -> dict[str, Any]: return { "content": { @@ -800,6 +807,13 @@ async def create_observations( logger.warning("create_observations called with empty list") return ObservationsCreatedResult(created_count=0, created_levels=[], failed=[]) + normalized_observations = [ + _normalized_observation_input(obs) for obs in observations if obs.content.strip() + ] + if not normalized_observations: + logger.info("No non-empty observations to create") + return ObservationsCreatedResult(created_count=0, created_levels=[], failed=[]) + # Phase 1: Ensure collection exists (short DB scope) async with tracked_db("create_observations.collection") as db: await crud.get_or_create_collection( @@ -810,12 +824,12 @@ async def create_observations( ) # Phase 2: Compute embeddings (no DB needed) - contents = [obs.content for obs in observations] + contents = [obs.content for obs in normalized_observations] embeddings_by_index: dict[int, list[float]] | None = None try: embeddings = await embedding_client.simple_batch_embed(contents) embeddings_by_index = dict( - zip(range(len(observations)), embeddings, strict=True) + zip(range(len(normalized_observations)), embeddings, strict=True) ) except Exception as e: logger.warning( @@ -826,7 +840,7 @@ async def create_observations( # Build document objects with pre-computed embeddings documents: list[schemas.DocumentCreate] = [] failed: list[ObservationFailure] = [] - for i, obs in enumerate(observations): + for i, obs in enumerate(normalized_observations): embedding: list[float] if embeddings_by_index is not None: embedding = embeddings_by_index[i] diff --git a/tests/crud/test_representation_manager.py b/tests/crud/test_representation_manager.py index e4c46853..dfd959d7 100644 --- a/tests/crud/test_representation_manager.py +++ b/tests/crud/test_representation_manager.py @@ -1,3 +1,8 @@ +from contextlib import asynccontextmanager +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + import pytest from nanoid import generate as generate_nanoid from sqlalchemy import func, update @@ -5,6 +10,25 @@ from sqlalchemy.ext.asyncio import AsyncSession from src import models from src.crud.representation import RepresentationManager +from src.utils.representation import ( + DeductiveObservation, + ExplicitObservation, + Representation, +) + + +@asynccontextmanager +async def _fake_tracked_db(_name: str): + yield object() + + +def _saved_observations(mock_save: AsyncMock): + call = mock_save.await_args + if "all_observations" in call.kwargs: + return call.kwargs["all_observations"] + if len(call.args) > 1: + return call.args[1] + raise AssertionError("missing all_observations in await args") class TestRepresentationManagerSoftDelete: @@ -134,3 +158,164 @@ class TestRepresentationManagerSoftDelete: result_ids = [doc.id for doc in results] assert doc_live.id in result_ids assert doc_deleted.id not in result_ids + + +class TestRepresentationManagerSave: + @pytest.mark.asyncio + async def test_save_representation_filters_blank_observations_before_embedding(self): + manager = RepresentationManager( + "workspace", + observer="observer", + observed="observed", + ) + representation = Representation( + explicit=[ + ExplicitObservation( + content=" ", + created_at=datetime.now(timezone.utc), + message_ids=[1], + session_name="session", + ), + ExplicitObservation( + content=" useful observation ", + created_at=datetime.now(timezone.utc), + message_ids=[1], + session_name="session", + ), + ] + ) + + with ( + patch("src.crud.representation.tracked_db", _fake_tracked_db), + patch( + "src.crud.representation.embedding_client.simple_batch_embed", + new=AsyncMock(return_value=[[0.1]]), + ) as mock_embed, + patch.object( + manager, + "_save_representation_internal", + new=AsyncMock(return_value=1), + ) as mock_save, + ): + saved = await manager.save_representation( + representation, + message_ids=[1], + session_name="session", + message_created_at=datetime.now(timezone.utc), + message_level_configuration=SimpleNamespace( + dream=SimpleNamespace(enabled=False) + ), + ) + + assert saved == 1 + mock_embed.assert_awaited_once_with(["useful observation"]) + saved_observations = _saved_observations(mock_save) + assert len(saved_observations) == 1 + assert saved_observations[0].content == "useful observation" + + @pytest.mark.asyncio + async def test_save_representation_filters_blank_deductive_observations(self): + manager = RepresentationManager( + "workspace", + observer="observer", + observed="observed", + ) + representation = Representation( + deductive=[ + DeductiveObservation( + conclusion=" ", + premises=["premise a"], + source_ids=["doc-a"], + created_at=datetime.now(timezone.utc), + message_ids=[1], + session_name="session", + ), + DeductiveObservation( + conclusion=" inferred conclusion ", + premises=["premise b"], + source_ids=["doc-b"], + created_at=datetime.now(timezone.utc), + message_ids=[1], + session_name="session", + ), + ] + ) + + with ( + patch("src.crud.representation.tracked_db", _fake_tracked_db), + patch( + "src.crud.representation.embedding_client.simple_batch_embed", + new=AsyncMock(return_value=[[0.2]]), + ) as mock_embed, + patch.object( + manager, + "_save_representation_internal", + new=AsyncMock(return_value=1), + ) as mock_save, + ): + saved = await manager.save_representation( + representation, + message_ids=[1], + session_name="session", + message_created_at=datetime.now(timezone.utc), + message_level_configuration=SimpleNamespace( + dream=SimpleNamespace(enabled=False) + ), + ) + + assert saved == 1 + mock_embed.assert_awaited_once_with(["inferred conclusion"]) + saved_observations = _saved_observations(mock_save) + assert len(saved_observations) == 1 + assert isinstance(saved_observations[0], DeductiveObservation) + assert saved_observations[0].conclusion == "inferred conclusion" + + @pytest.mark.asyncio + async def test_save_representation_skips_all_blank_observations(self): + manager = RepresentationManager( + "workspace", + observer="observer", + observed="observed", + ) + representation = Representation( + explicit=[ + ExplicitObservation( + content="", + created_at=datetime.now(timezone.utc), + message_ids=[1], + session_name="session", + ), + ExplicitObservation( + content="\n\t ", + created_at=datetime.now(timezone.utc), + message_ids=[1], + session_name="session", + ), + ] + ) + + with ( + patch("src.crud.representation.tracked_db", _fake_tracked_db), + patch( + "src.crud.representation.embedding_client.simple_batch_embed", + new=AsyncMock(), + ) as mock_embed, + patch.object( + manager, + "_save_representation_internal", + new=AsyncMock(), + ) as mock_save, + ): + saved = await manager.save_representation( + representation, + message_ids=[1], + session_name="session", + message_created_at=datetime.now(timezone.utc), + message_level_configuration=SimpleNamespace( + dream=SimpleNamespace(enabled=False) + ), + ) + + assert saved == 0 + mock_embed.assert_not_awaited() + mock_save.assert_not_awaited() diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index 4114cfe7..95406587 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -4,6 +4,7 @@ import asyncio from collections.abc import Callable from datetime import datetime, timedelta, timezone from typing import Any +from unittest.mock import AsyncMock import pytest from nanoid import generate as generate_nanoid @@ -377,6 +378,96 @@ class TestCreateObservations: assert len(created_documents) == 1 assert created_documents[0].content == "Embeds fine" + async def test_create_observations_filters_blank_content_before_embedding( + self, + tool_test_data: Any, + monkeypatch: pytest.MonkeyPatch, + ): + """Blank or whitespace-only observations are dropped before embedding/persistence.""" + workspace, peer1, peer2, session, _, _ = tool_test_data + created_documents: list[Any] = [] + + async def fake_batch_embed(texts: list[str]) -> list[list[float]]: + assert texts == ["trimmed observation"] + return [[0.4, 0.5, 0.6]] + + async def fake_create_documents( + _db: AsyncSession, + documents: list[Any], + workspace_name: str, + *, + observer: str, + observed: str, + deduplicate: bool = False, + ) -> list[Any]: + _ = (workspace_name, observer, observed, deduplicate) + created_documents.extend(documents) + return documents + + monkeypatch.setattr( + "src.utils.agent_tools.embedding_client.simple_batch_embed", + fake_batch_embed, + ) + monkeypatch.setattr( + "src.utils.agent_tools.crud.create_documents", fake_create_documents + ) + + result = await create_observations( + observations=[ + schemas.ObservationInput(content=" ", level="explicit"), + schemas.ObservationInput(content=" trimmed observation ", level="explicit"), + ], + observer=peer1.name, + observed=peer2.name, + session_name=session.name, + workspace_name=workspace.name, + message_ids=[], + message_created_at=str(datetime.now(timezone.utc)), + ) + + assert isinstance(result, ObservationsCreatedResult) + assert result.created_count == 1 + assert len(result.failed) == 0 + assert len(created_documents) == 1 + assert created_documents[0].content == "trimmed observation" + + async def test_create_observations_skips_all_blank_content( + self, + tool_test_data: Any, + monkeypatch: pytest.MonkeyPatch, + ): + """All-blank observations short-circuit without embedding or persistence.""" + workspace, peer1, peer2, session, _, _ = tool_test_data + batch_embed = AsyncMock() + create_documents = AsyncMock() + + monkeypatch.setattr( + "src.utils.agent_tools.embedding_client.simple_batch_embed", + batch_embed, + ) + monkeypatch.setattr( + "src.utils.agent_tools.crud.create_documents", create_documents + ) + + result = await create_observations( + observations=[ + schemas.ObservationInput(content=" ", level="explicit"), + schemas.ObservationInput(content="\n\t", level="explicit"), + ], + observer=peer1.name, + observed=peer2.name, + session_name=session.name, + workspace_name=workspace.name, + message_ids=[], + message_created_at=str(datetime.now(timezone.utc)), + ) + + assert isinstance(result, ObservationsCreatedResult) + assert result.created_count == 0 + assert len(result.failed) == 0 + batch_embed.assert_not_awaited() + create_documents.assert_not_awaited() + @pytest.mark.asyncio class TestDeleteObservations: