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

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

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.
 

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

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:

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