feat(librarian): signal degraded search coverage

- parse source_counts into HybridRAGResponse and additively parse the
  shared-contract source_status/degraded fields when present (absence
  tolerated, so deploy order between tatlock and library-desk never
  matters)
- hybrid_search appends a one-line coverage note when a leg reported
  'failed' (or degraded is set), falling back to inferring silent legs
  from source_counts on older library-desk versions

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/agents/librarian/test_hybrid_contract.py

@@ -17,7 +17,7 @@ import httpx
 import pytest
 
 from src.agents.librarian.client import HybridRAGResponse, LibraryDeskClient
-from src.agents.librarian.tools import SOURCE_ICONS, hybrid_search
+from src.agents.librarian.tools import SOURCE_ICONS, _coverage_note, hybrid_search
 
 FIXTURE_PATH = Path(__file__).parent / "fixtures" / "hybrid_query_recorded.json"
 
@@ -150,6 +150,58 @@ class TestHybridRAGContract:
         assert config["enable_documents"] is False
         assert config["enable_volatile"] is False
 
+    @pytest.mark.asyncio
+    async def test_source_counts_and_timing_parsed(
+        self, client_with_recorded_response
+    ):
+        """source_counts and timing map into the response model."""
+        response = await client_with_recorded_response.hybrid_search(
+            "home server infrastructure", user="testuser"
+        )
+
+        assert response.source_counts == {"graph": 3, "web": 2}
+        assert response.timing.get("total_ms", 0) > 0
+
+    @pytest.mark.asyncio
+    async def test_source_status_absent_is_tolerated(
+        self, client_with_recorded_response
+    ):
+        """Recorded response predates source_status/degraded - defaults apply."""
+        response = await client_with_recorded_response.hybrid_search(
+            "home server infrastructure", user="testuser"
+        )
+
+        assert response.source_status == {}
+        assert response.degraded is False
+
+    @pytest.mark.asyncio
+    async def test_source_status_parsed_when_present(self, recorded_response):
+        """Additive source_status/degraded fields parse when the service sends them."""
+        enriched = dict(recorded_response)
+        enriched["source_status"] = {
+            "vector": "ok",
+            "graph": "ok",
+            "web": "failed",
+            "volatile": "disabled",
+            "documents": "ok",
+        }
+        enriched["degraded"] = True
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = enriched
+        mock_response.raise_for_status = MagicMock()
+        mock_httpx = AsyncMock(spec=httpx.AsyncClient)
+        mock_httpx.post.return_value = mock_response
+
+        client = LibraryDeskClient(base_url="http://test:8089", api_key="test-key")
+        client._client = mock_httpx
+
+        response = await client.hybrid_search("home server infrastructure", user="u")
+
+        assert response.degraded is True
+        assert response.source_status["web"] == "failed"
+        assert response.source_status["volatile"] == "disabled"
+
     @pytest.mark.asyncio
     async def test_tool_renders_no_unknown_results(self, client_with_recorded_response, monkeypatch):
         """The hybrid_search tool renders real sources and non-zero scores."""
@@ -173,3 +225,93 @@ class TestHybridRAGContract:
         assert "unknown" not in output
         assert "score: 0.00" not in output
         assert "•" not in output, "every source value should map to an icon"
+
+
+@pytest.mark.unit
+class TestCoverageNote:
+    """Coverage note makes degraded searches visible to model and user."""
+
+    def _response(self, **kwargs) -> HybridRAGResponse:
+        return HybridRAGResponse(**kwargs)
+
+    def test_no_note_when_all_legs_report(self):
+        response = self._response(
+            source_counts={
+                "vector": 2,
+                "graph": 1,
+                "web": 2,
+                "documents": 1,
+                "volatile": 1,
+            },
+        )
+        note = _coverage_note(
+            response, include_web=True, include_documents=True, include_volatile=True
+        )
+        assert note == ""
+
+    def test_note_when_leg_missing_from_counts(self):
+        response = self._response(source_counts={"graph": 3, "web": 2})
+        note = _coverage_note(
+            response, include_web=True, include_documents=True, include_volatile=True
+        )
+        assert "Coverage note" in note
+        assert "vector" in note
+        assert "documents" in note
+        assert "volatile" in note
+
+    def test_disabled_legs_are_not_reported_missing(self):
+        response = self._response(source_counts={"vector": 2, "graph": 1})
+        note = _coverage_note(
+            response,
+            include_web=False,
+            include_documents=False,
+            include_volatile=False,
+        )
+        assert note == ""
+
+    def test_note_prefers_source_status_failures(self):
+        response = self._response(
+            source_counts={"vector": 2, "graph": 1},
+            source_status={
+                "vector": "ok",
+                "graph": "ok",
+                "web": "failed",
+                "volatile": "disabled",
+                "documents": "ok",
+            },
+            degraded=True,
+        )
+        note = _coverage_note(
+            response, include_web=True, include_documents=True, include_volatile=True
+        )
+        assert "failed" in note
+        assert "web" in note
+        # disabled legs are not reported as failures
+        assert "volatile" not in note
+
+    def test_no_note_when_status_all_ok(self):
+        response = self._response(
+            source_counts={"graph": 1},
+            source_status={
+                "vector": "ok",
+                "graph": "ok",
+                "web": "ok",
+                "volatile": "ok",
+                "documents": "ok",
+            },
+            degraded=False,
+        )
+        note = _coverage_note(
+            response, include_web=True, include_documents=True, include_volatile=True
+        )
+        assert note == ""
+
+    def test_degraded_without_named_failures(self):
+        response = self._response(
+            source_status={"vector": "ok", "graph": "ok"},
+            degraded=True,
+        )
+        note = _coverage_note(
+            response, include_web=True, include_documents=True, include_volatile=True
+        )
+        assert "partial" in note