From 144ea2a07be2d7dedf62290f799e07ac36b4ea5d Mon Sep 17 00:00:00 2001 From: Aakash Kattelu Date: Wed, 12 Aug 2026 11:50:48 -0700 Subject: [PATCH] fix(deriver): surface failure when all observer saves fail When every observer's save_representation failed (e.g. embedding retries exhausted under a sustained 429), the deriver logged the error and returned normally, so the queue marked the work unit processed with zero documents saved. Collect per-observer errors and, after telemetry is emitted, raise RepresentationSaveError when no observer succeeded. Partial failures stay processed (saved observers must not be discarded) and are recorded via an additive failed_observer_count on RepresentationCompletedEvent. Refs #728 --- src/deriver/deriver.py | 23 +++- src/exceptions.py | 13 ++ src/telemetry/events/representation.py | 4 + tests/deriver/test_deriver_processing.py | 115 ++++++++++++++++++ .../test_representation_v2_fields.py | 2 + 5 files changed, 154 insertions(+), 3 deletions(-) diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 2ad7d14f..a139e4f5 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -7,6 +7,7 @@ from src import crud from src.config import ConfiguredModelSettings, settings from src.crud.representation import RepresentationManager from src.dependencies import tracked_db +from src.exceptions import RepresentationSaveError from src.llm import honcho_llm_call from src.llm.types import LLMTelemetryContext from src.models import Message @@ -196,6 +197,7 @@ async def process_representation_tasks_batch( agg_representation_result = crud.CreateDocumentsResult() successful_observer_count = 0 + save_errors: list[tuple[str, Exception]] = [] if observations.is_empty() or not message_ids: logger.warning( "Deriver generated zero observations for messages %s:%s in %s/%s!", @@ -236,10 +238,11 @@ async def process_representation_tasks_batch( representation_result.semantic_dup_replaced_count ) successful_observer_count += 1 - except Exception as e: - logger.error( - "Failed to save representation for observer %s: %s", observer, e + except Exception as e: # noqa: BLE001 + logger.exception( + "Failed to save representation for observer %s", observer ) + save_errors.append((observer, e)) # Log metrics overall_duration = (time.perf_counter() - overall_start) * 1000 @@ -337,5 +340,19 @@ async def process_representation_tasks_batch( 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, + failed_observer_count=len(save_errors), ) ) + + # If every observer's save failed, surface the failure to the queue manager so the + # work unit is marked errored instead of silently processed with zero documents saved + # (#728). Raised after telemetry so metrics still record the attempt. + if save_errors and successful_observer_count == 0: + details = "; ".join( + f"{observer}: {exc.__class__.__name__}: {exc}" + for observer, exc in save_errors + ) + raise RepresentationSaveError( + f"save_representation failed for all {len(save_errors)} observer(s): " + + details + ) from save_errors[0][1] diff --git a/src/exceptions.py b/src/exceptions.py index 129324b6..3df7815d 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -133,6 +133,19 @@ class VectorStoreError(HonchoException): detail = "Vector store operation failed" +@final +class RepresentationSaveError(HonchoException): + """Raised when every observer's representation save fails in a batch. + + Surfaced from the deriver to the queue manager so the work unit is marked + errored instead of silently processed with zero documents saved. The + underlying save exception is preserved as the cause via ``raise ... from``. + """ + + status_code = 500 + detail = "Representation save failed for all observers" + + class LLMError(Exception): """Exception raised when an LLM call fails. diff --git a/src/telemetry/events/representation.py b/src/telemetry/events/representation.py index ad6dfca4..c4291092 100644 --- a/src/telemetry/events/representation.py +++ b/src/telemetry/events/representation.py @@ -154,6 +154,10 @@ class RepresentationCompletedEvent(BaseEvent): default=0, description="Number of observers this representation was saved against", ) + failed_observer_count: int = Field( + default=0, + description="Number of observers whose save_representation failed (partial or total)", + ) def get_resource_id(self) -> str: """Resource ID includes workspace, session, and latest message for uniqueness.""" diff --git a/tests/deriver/test_deriver_processing.py b/tests/deriver/test_deriver_processing.py index 6785d159..0bc58d40 100644 --- a/tests/deriver/test_deriver_processing.py +++ b/tests/deriver/test_deriver_processing.py @@ -7,7 +7,9 @@ import pytest from src import crud, models from src.config import settings +from src.crud.representation import RepresentationManager from src.deriver.deriver import process_representation_tasks_batch +from src.exceptions import RepresentationSaveError from src.llm import HonchoLLMCallResponse from src.utils.representation import ( ExplicitObservationBase, @@ -70,6 +72,119 @@ class TestDeriverProcessing: assert kwargs["model_config"].stop_sequences == expected_config.stop_sequences assert "llm_settings" not in kwargs + async def test_all_observer_saves_failing_surfaces_failure(self): + """When every observer's save_representation fails, the batch must raise + instead of swallowing it. Without this the work unit is marked processed + with zero documents saved: silent memory loss (#728). + """ + message = Mock( + id=1, + public_id="msg_1", + 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="The user has a dog named Rover") + ] + ), + input_tokens=10, + output_tokens=5, + finish_reasons=["STOP"], + ) + + failing_save = AsyncMock(side_effect=RuntimeError("429 RESOURCE_EXHAUSTED")) + emitted: list[Any] = [] + with ( + patch( + "src.deriver.deriver.honcho_llm_call", + new_callable=AsyncMock, + return_value=mock_response, + ), + patch.object(RepresentationManager, "save_representation", failing_save), + patch("src.deriver.deriver.emit", side_effect=emitted.append), + pytest.raises(RepresentationSaveError, match="save_representation failed"), + ): + await process_representation_tasks_batch( + messages=[message], + message_level_configuration=configuration, + observers=["bob"], + observed="alice", + queue_item_message_ids=[1], + ) + + # Telemetry must fire *before* the raise so a total save failure is still + # visible to metrics. Guards against emit() being moved after the raise. + assert emitted, "expected telemetry to be emitted before the raised failure" + assert emitted[-1].observer_count == 0 + assert emitted[-1].failed_observer_count == 1 + + async def test_partial_observer_failure_is_processed_and_surfaced(self): + """When some observers save and one fails, the batch does NOT raise + (saved observers are kept) and the failure is visible via telemetry. + """ + message = Mock( + id=1, + public_id="msg_1", + 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="The user has a dog named Rover") + ] + ), + input_tokens=10, + output_tokens=5, + finish_reasons=["STOP"], + ) + + # bob succeeds, carol fails. + partial_save = AsyncMock( + side_effect=[ + crud.CreateDocumentsResult(), + RuntimeError("429 RESOURCE_EXHAUSTED"), + ] + ) + emitted: list[Any] = [] + with ( + patch( + "src.deriver.deriver.honcho_llm_call", + new_callable=AsyncMock, + return_value=mock_response, + ), + patch.object(RepresentationManager, "save_representation", partial_save), + 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 emitted, "expected a telemetry event to be emitted" + event = emitted[-1] + assert event.observer_count == 1 + assert event.failed_observer_count == 1 + async def test_process_representation_tasks_batch_passes_custom_instructions_into_prompt( self, ) -> None: diff --git a/tests/telemetry/test_representation_v2_fields.py b/tests/telemetry/test_representation_v2_fields.py index a41bed41..6f2251ea 100644 --- a/tests/telemetry/test_representation_v2_fields.py +++ b/tests/telemetry/test_representation_v2_fields.py @@ -51,6 +51,7 @@ class TestRepresentationV2AdditiveFields: assert event.exact_dup_existing_count == 0 assert event.semantic_dup_rejected_count == 0 assert event.semantic_dup_replaced_count == 0 + assert event.failed_observer_count == 0 def test_input_tokens_semantics_preserved(self): """The downstream metering key must remain 'queued-message tokens'. @@ -161,6 +162,7 @@ class TestRepresentationV2AdditiveFields: "exact_dup_existing_count", "semantic_dup_rejected_count", "semantic_dup_replaced_count", + "failed_observer_count", ): assert field in data, f"missing field: {field}" assert data["hit_batch_token_cap"] is True