diff --git a/plugin/src/claude_smart/reflexio_adapter.py b/plugin/src/claude_smart/reflexio_adapter.py index 1378836..7c0bc15 100644 --- a/plugin/src/claude_smart/reflexio_adapter.py +++ b/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: diff --git a/tests/test_adapter.py b/tests/test_adapter.py index f76f9a6..491b6bb 100644 --- a/tests/test_adapter.py +++ b/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 diff --git a/tests/test_publish.py b/tests/test_publish.py index ecbeb14..b5e10d0 100644 --- a/tests/test_publish.py +++ b/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