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

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
This commit is contained in:
Phil 2026-04-29 13:18:00 -06:00 committed by GitHub
parent 94ade07c12
commit a05c2f8ec1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 317 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/crud/representation.py
View File

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

20
src/utils/agent_tools.py
View File

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

185
tests/crud/test_representation_manager.py
View File

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

91
tests/utils/test_agent_tools.py
View File

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