feat(adapter): surface the publish warnings the server returns

plugin/src/claude_smart/reflexio_adapter.py

@@ -125,6 +125,47 @@ def publish(
         client = self._get_client()
         if client is None:
             return PublishResult(False)
+        result, response = self._attempt_publish(
+            client,
+            session_id=session_id,
+            project_id=project_id,
+            request_id=request_id,
+            interactions=interactions,
+            force_extraction=force_extraction,
+            override_learning_stall=override_learning_stall,
+            skip_aggregation=skip_aggregation,
+        )
+        if result.ok:
+            # Deliberately outside `_attempt_publish` and every try inside it.
+            # The publish has already been accepted, and `publish_unpublished`
+            # advances the buffer watermark only on a truthy result — so a raise
+            # while reading diagnostics would report a *successful* publish as
+            # failed and re-send the same batch on every later hook. The nested
+            # guard covers the logging handler too, not just the extraction.
+            try:
+                for warning in _publish_warnings(response):
+                    _LOGGER.warning("reflexio dropped part of the payload: %s", warning)
+            except Exception as exc:  # noqa: BLE001 — diagnostics must never fail a publish.
+                _LOGGER.debug("could not read publish warnings: %s", exc)
+        return result
+
+    def _attempt_publish(
+        self,
+        client: Any,
+        *,
+        session_id: str,
+        project_id: str,
+        request_id: str | None,
+        interactions: Sequence[dict[str, Any]],
+        force_extraction: bool,
+        override_learning_stall: bool,
+        skip_aggregation: bool,
+    ) -> tuple[PublishResult, Any]:
+        """Run the publish and return ``(result, raw response)``.
+
+        Split out of ``publish`` so the warning read has somewhere to live that
+        is outside this method's ``except`` — see the caller.
+        """
         try:
             interaction_list = list(interactions)
             raw_request = getattr(client, "_make_request", None)
@@ -142,13 +183,13 @@ def publish(
                     "source": _SOURCE,
                 }
                 try:
-                    raw_request(
+                    raw_response = raw_request(
                         "POST",
                         "/api/publish_interaction",
                         json=payload,
                         params=None,
                     )
-                    return PublishResult(True, request_id)
+                    return PublishResult(True, request_id), raw_response
                 except Exception as exc:  # noqa: BLE001
                     if not _needs_raw_retrieved_learning_publish(interaction_list):
                         raise
@@ -163,13 +204,13 @@ def publish(
                             interaction_list
                         ),
                     }
-                    raw_request(
+                    fallback_response = raw_request(
                         "POST",
                         "/api/publish_interaction",
                         json=fallback_payload,
                         params=None,
                     )
-                    return PublishResult(True, request_id)
+                    return PublishResult(True, request_id), fallback_response
             if _needs_raw_retrieved_learning_publish(interaction_list):
                 _LOGGER.warning(
                     "Stable raw publishing is unavailable; publishing "
@@ -190,11 +231,11 @@ def publish(
             response = client.publish_interaction(**kwargs)
             response_request_id = getattr(response, "request_id", None)
             if isinstance(response_request_id, str) and response_request_id:
-                return PublishResult(True, response_request_id)
-            return PublishResult(True)
+                return PublishResult(True, response_request_id), response
+            return PublishResult(True), response
         except Exception as exc:  # noqa: BLE001
             _LOGGER.warning("publish_interaction failed: %s", exc)
-            return PublishResult(False)
+            return PublishResult(False), None
 
     def apply_extraction_defaults(self, *, window_size: int, stride_size: int) -> bool:
         """Push claude-smart's preferred extraction defaults to the reflexio server.
@@ -494,6 +535,27 @@ def _record_read_error(self, operation: str, exc: Exception) -> None:
         _LOGGER.debug("%s failed: %s", operation, exc)
 
 
+def _publish_warnings(response: Any) -> list[str]:
+    """Pull ``warnings`` off a publish response, tolerating any shape.
+
+    Defensive rather than total: ``getattr`` swallows only ``AttributeError``,
+    a mapping can override ``get``, and ``str`` runs a caller-supplied
+    ``__str__``. The caller wraps this so those cannot fail an accepted
+    publish. ``_extract_items`` is not reused because its ``list(value)``
+    raises on a non-iterable.
+
+    Both publish paths land here: the raw ``_make_request`` path returns a
+    parsed JSON dict, the client path a response object.
+    """
+    if isinstance(response, dict):
+        value = response.get("warnings")
+    else:
+        value = getattr(response, "warnings", None)
+    if not isinstance(value, (list, tuple)):
+        return []
+    return [str(item) for item in value]
+
+
 def _extract_items(response: Any, field: str) -> list[Any]:
     """Pull a list field from a reflexio response object or dict, tolerating shape drift."""
     if response is None:

tests/test_adapter.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+import logging
 import sys
 import threading
 import time
 from types import SimpleNamespace
 from typing import Any
 
+import pytest
+
 from claude_smart import reflexio_adapter
 
 
@@ -1009,3 +1012,128 @@ def test_mark_stall_notified_calls_through(monkeypatch):
     monkeypatch.setattr(adapter, "_get_client", lambda: stub)
     adapter.mark_stall_notified()
     assert stub.notified_called is True
+
+
+_ADAPTER_LOGGER = "claude_smart.reflexio_adapter"
+
+
+class _WarningClient(_FakeClient):
+    """Fake whose publish paths echo a server ``warnings`` payload."""
+
+    def __init__(self, *, response: Any = None, raw_response: Any = None) -> None:
+        super().__init__()
+        self._response = response
+        self._raw_response = raw_response
+
+    def publish_interaction(self, **kwargs):
+        self.published_kwargs = kwargs
+        return self._response
+
+    def _make_request(self, method, path, **kwargs):
+        self.raw_request = {"method": method, "path": path, **kwargs}
+        return self._raw_response
+
+
+class _WarningsPropertyRaises:
+    """``getattr(obj, "warnings", None)`` absorbs only ``AttributeError``."""
+
+    @property
+    def warnings(self) -> list[str]:
+        raise RuntimeError("warnings unavailable")
+
+
+class _MappingGetRaises(dict):
+    """``isinstance(response, dict)`` holds, but the override is what runs."""
+
+    def get(self, *_args, **_kwargs):
+        raise KeyError("boom")
+
+
+class _Unprintable:
+    def __str__(self) -> str:
+        raise ValueError("cannot render")
+
+
+class TestPublishWarnings:
+    """The server reports fields it could not bind; the hook log must show them.
+
+    Silent field-dropping is the defect this channel exists for: a publish of
+    50 mis-keyed interactions returned 200 and stored 50 empty rows. This
+    plugin's raw ``_make_request`` path matters most — it posts the payload
+    directly, so unknown keys really do reach the server rather than being
+    stripped client-side first.
+    """
+
+    @staticmethod
+    def _adapter_records(caplog):
+        return [r for r in caplog.records if r.name == _ADAPTER_LOGGER]
+
+    def test_client_path_logs_server_warnings(self, caplog) -> None:
+        client = _WarningClient(
+            response=SimpleNamespace(
+                warnings=["interaction_data_list[0]: ignored unrecognised field(s) Content"],
+                request_id="r1",
+            )
+        )
+        with caplog.at_level(logging.WARNING, logger=_ADAPTER_LOGGER):
+            result = _adapter_with(client).publish(
+                session_id="s1",
+                project_id="p1",
+                interactions=[{"role": "User", "content": "hi"}],
+            )
+        assert result.ok is True
+        assert "unrecognised field(s) Content" in caplog.text
+
+    def test_raw_request_path_logs_server_warnings(self, caplog) -> None:
+        """The pinned-request_id path returns parsed JSON, not a model."""
+        client = _WarningClient(raw_response={"warnings": ["dropped foo"]})
+        with caplog.at_level(logging.WARNING, logger=_ADAPTER_LOGGER):
+            result = _adapter_with(client).publish(
+                session_id="s1",
+                project_id="p1",
+                request_id="request-1",
+                interactions=[{"role": "User", "content": "hi"}],
+            )
+        assert result.ok is True
+        assert result.request_id == "request-1"
+        assert "dropped foo" in caplog.text
+
+    def test_quiet_when_there_is_nothing_to_report(self, caplog) -> None:
+        client = _WarningClient(response=SimpleNamespace(warnings=[]))
+        with caplog.at_level(logging.WARNING, logger=_ADAPTER_LOGGER):
+            result = _adapter_with(client).publish(
+                session_id="s1",
+                project_id="p1",
+                interactions=[{"role": "User", "content": "hi"}],
+            )
+        assert result.ok is True
+        assert self._adapter_records(caplog) == []
+
+    @pytest.mark.parametrize(
+        "response",
+        [
+            None,
+            SimpleNamespace(),
+            {"warnings": None},
+            {"warnings": 5},
+            _WarningsPropertyRaises(),
+            _MappingGetRaises(),
+            {"warnings": [_Unprintable()]},
+        ],
+    )
+    def test_hostile_response_shapes_still_report_success(self, response) -> None:
+        """A publish the server ACCEPTED must never be reported as failed.
+
+        ``publish_unpublished`` advances the buffer watermark only on a truthy
+        result, so raising while reading diagnostics would re-send an accepted
+        batch on every subsequent hook — duplicates forever, caused purely by
+        the code meant to improve observability. The last three shapes are the
+        ones that actually escape the helper's isinstance guard.
+        """
+        client = _WarningClient(response=response, raw_response=response)
+        result = _adapter_with(client).publish(
+            session_id="s1",
+            project_id="p1",
+            interactions=[{"role": "User", "content": "hi"}],
+        )
+        assert result.ok is True

tests/test_publish.py

@@ -378,3 +378,47 @@ def test_publish_keeps_first_links_at_request_cap(session_dir) -> None:
     assert len(retrieved) == 1000
     assert retrieved[0]["learning_id"] == "0"
     assert retrieved[-1]["learning_id"] == "999"
+
+
+class _HostileWarningsClient:
+    """A server response whose ``warnings`` blows up on access.
+
+    Not contrived: a computed pydantic field or a lazy client wrapper that
+    re-reads the socket on attribute access has exactly this shape, and
+    ``getattr(obj, "warnings", None)`` absorbs only ``AttributeError``.
+    """
+
+    @property
+    def warnings(self) -> list[str]:
+        raise RuntimeError("warnings unavailable")
+
+    def publish_interaction(self, **_kwargs: Any) -> Any:
+        return self
+
+    def _make_request(self, _method: str, _path: str, **_kwargs: Any) -> Any:
+        return self
+
+
+def test_watermark_advances_even_if_reading_warnings_blows_up(session_dir) -> None:
+    """A publish the server accepted must be marked published.
+
+    This is the whole reason the adapter reads warnings outside the try that
+    guards the publish call. If a diagnostic read could surface as a failed
+    publish, the watermark would never advance and the next hook would re-send
+    the same accepted batch — duplicates forever, caused by the observability
+    code. Goes through the real Adapter rather than a fake so the guard itself
+    is under test.
+    """
+    _append_assistant("s1", 10)
+    adapter = Adapter()
+    adapter._client = _HostileWarningsClient()  # bypass lazy construction
+
+    assert publish.publish_unpublished(
+        session_id="s1",
+        project_id="project",
+        force_extraction=False,
+        skip_aggregation=False,
+        adapter=adapter,
+    ) == ("ok", 1)
+
+    assert state.read_all("s1")[-1]["published_up_to"] == 1