fix(api)!: report contentless and mis-keyed publish payloads

AI_AGENT_INTEGRATION.md

@@ -241,7 +241,9 @@ Publish request fields:
 | `force_extraction` | No | `False` for normal background publishing. `True` for manual learn-now, tests, or final flushes where you intentionally want extraction to run immediately. | `false` |
 | `wait_for_response` | SDK/query option | `False` on interactive paths. `True` only when the caller is prepared to wait for extraction results. | `false` |
 
-Each interaction row should resemble Reflexio's `InteractionData` shape:
+Each interaction row must use Reflexio's `InteractionData` field names exactly.
+Unknown keys are dropped (and reported in the response's `warnings[]`), so a
+misspelling silently loses that value rather than being coerced:
 
 ```json
 {
@@ -497,8 +499,35 @@ Follow these rules for production agent integrations:
 - Keep hook latency bounded. Use short HTTP timeouts on interactive hooks.
 - Never let Reflexio exceptions fail the user's agent turn.
 - Buffer locally before network calls.
-- Advance publish watermarks only after success.
-- Retry failed publishes later.
+- Advance publish watermarks only after success — but **classify the failure
+  first**. Holding the watermark on a permanently-rejected batch wedges the
+  buffer: the same batch is re-sent forever and nothing is ever published.
+  Advancing on a transient failure loses data. Neither default is safe, so
+  branch on the status:
+
+  | Response | Meaning | Do |
+  |---|---|---|
+  | `5xx`, timeout, connection error | Transient | **Hold** the watermark; retry with backoff |
+  | `408`, `429` | Transient (slow down) | **Hold**; retry with backoff, honour `Retry-After` |
+  | `401`, `403` | Auth/config broken, not the payload | **Hold** and escalate — do not discard; the data is fine |
+  | `400`, `422` | The payload itself is invalid | **Advance** past it; retrying cannot succeed |
+
+  A `422` names the offending `interaction_data_list` index. Prefer
+  dead-lettering just that row and re-sending its valid siblings over
+  discarding the whole batch — the batch is rejected as a unit, so a blind
+  advance drops good interactions alongside the bad one.
+- Retry failed publishes later, using the same classification.
+- Build the wire payload with an **allowlist** of `InteractionData` fields, not
+  by passing your buffer record through minus a few keys. A denylist silently
+  ships your internal bookkeeping and rots every time you add a record key.
+- Every interaction must carry something — `content`, `shadow_content`,
+  `expert_content`, `interacted_image_url`, `image_encoding`, `tools_used`,
+  `citations`, `retrieved_learnings`, or a `user_action` other than `"none"`.
+  A wholly empty interaction is rejected with `422`; drop empty turns before
+  sending rather than letting one poison the batch.
+- Unknown fields are accepted but **dropped**, and named back to you in the
+  response's `warnings[]`. Check that list: a populated one means a field you
+  sent did not bind and that data was silently discarded.
 - Truncate large tool fields before publishing.
 - Do not run extraction synchronously on every turn.
 - Make retrieval query-aware; do not dump all memory into every prompt.

reflexio/client/client.py

@@ -464,6 +464,22 @@ def publish_interaction(
                 In ``wait_for_response=True`` mode the server waits for
                 extraction and the response includes ``request_id``,
                 storage routing, and real profile/playbook deltas.
+
+                ``warnings`` reports anything that was quietly altered:
+                unrecognised interaction fields that were dropped (a key you
+                sent did not bind and its value was discarded — check for a
+                typo), and interactions that carried nothing and were
+                skipped. Indices refer to the list as you passed it. The list
+                is bounded, so on a large batch treat it as a sample.
+
+        Raises:
+            ValueError: If ``session_id`` is missing or blank.
+            pydantic.ValidationError: Raised locally, before any HTTP call, if
+                an interaction carries nothing at all (no content, image,
+                tools, citations, learnings, or non-"none" user_action), if a
+                ``user_action`` has no ``user_action_description``, or if both
+                ``interacted_image_url`` and ``image_encoding`` are set. The
+                message names the offending ``interaction_data_list`` index.
         """
         if session_id is None or not session_id.strip():
             raise ValueError("session_id is required and cannot be empty")
@@ -490,6 +506,14 @@ def publish_interaction(
         result = self._publish_interaction_sync(
             request, wait_for_response=wait_for_response
         )
+        # Merge the warnings detected locally. Building InteractionData above
+        # already stripped the caller's unknown keys, so ``request.model_dump()``
+        # sends a clean payload and the server never sees them -- it cannot echo
+        # what it was never told. Without this, unrecognised fields are reported
+        # over raw HTTP but are invisible through the SDK, which is the primary
+        # integration path and the one this method's docstring promises.
+        if local_warnings := request.payload_warnings():
+            result.warnings = [*result.warnings, *local_warnings]
         self._cache.invalidate("get_profiles")
         self._cache.invalidate("get_agent_playbooks")
         return result

reflexio/integrations/openclaw/plugin/src/openclaw_smart/reflexio_adapter.py

@@ -17,6 +17,26 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+
+def _log_payload_warnings(result: object) -> None:
+    """Log anything the publish quietly altered. Never raises.
+
+    The publish has already succeeded by the time this runs, and the caller
+    advances its publish watermark on the adapter's return value -- so an
+    exception here would be misread as a publish failure and stall the buffer.
+    Total belt-and-braces on a diagnostic path is the right trade.
+    """
+    try:
+        warnings = getattr(result, "warnings", None) or []
+        if warnings:
+            _LOGGER.warning(
+                "publish_interaction altered the payload: %s",
+                "; ".join(str(warning) for warning in warnings),
+            )
+    except Exception:  # noqa: BLE001 - diagnostics must never break publishing
+        _LOGGER.debug("could not render publish warnings", exc_info=True)
+
+
 _ENV_URL = "REFLEXIO_URL"
 _DEFAULT_URL = "http://localhost:8071/"
 _SEARCH_MODE_HYBRID = "hybrid"  # reflexio.models.config_schema.SearchMode.HYBRID
@@ -80,7 +100,7 @@ def publish(
         if client is None:
             return False
         try:
-            client.publish_interaction(
+            result = client.publish_interaction(
                 user_id=project_id,
                 interactions=list(interactions),
                 agent_version=runtime.agent_version(),
@@ -89,11 +109,18 @@ def publish(
                 force_extraction=force_extraction,
                 skip_aggregation=skip_aggregation,
             )
-            return True
         except Exception as exc:  # noqa: BLE001
             _LOGGER.warning("publish_interaction failed: %s", exc)
             return False
 
+        # Diagnostics only, and deliberately OUTSIDE the try above: the publish
+        # has already succeeded, and the caller advances its watermark on our
+        # return value. Anything raised while merely reporting warnings would be
+        # caught as a publish failure, stalling the watermark and republishing
+        # the same batch on every later hook.
+        _log_payload_warnings(result)
+        return True
+
     def apply_extraction_defaults(self, *, window_size: int, stride_size: int) -> bool:
         """Push openclaw-smart's preferred extraction defaults to the reflexio server.
 

reflexio/integrations/openclaw/plugin/src/openclaw_smart/state.py

@@ -36,6 +36,28 @@
 
 _TOOL_DATA_FIELD_MAX_LEN = 256
 
+# Fields the server's ``InteractionData`` accepts. Declared literally rather
+# than imported so this module keeps no runtime dependency on ``reflexio``;
+# ``tests/test_state.py`` pins it against the real model so the two cannot
+# drift. Anything not in here is buffer-internal bookkeeping and must not
+# reach the wire.
+_INTERACTION_DATA_FIELDS = frozenset(
+    {
+        "created_at",
+        "role",
+        "content",
+        "shadow_content",
+        "expert_content",
+        "user_action",
+        "user_action_description",
+        "interacted_image_url",
+        "image_encoding",
+        "tools_used",
+        "citations",
+        "retrieved_learnings",
+    }
+)
+
 _VALID_CITATION_KINDS = frozenset(
     {"playbook", "profile", "user_playbook", "agent_playbook"}
 )
@@ -315,9 +337,13 @@ def unpublished_slice(
             pending_tools.append(tool_entry)
             continue
         if role in {"User", "Assistant"}:
-            turn = {
-                k: v for k, v in rec.items() if k not in {"role", "ts", "cited_items"}
-            }
+            # Allowlist, not denylist. This used to drop three known keys and
+            # pass everything else through, so buffer-internal bookkeeping
+            # (``user_id``, ``id``, ``kind``, ``title``, ...) rode along onto
+            # the wire. The server ignored it, but the denylist rots every time
+            # a hook adds a record key, and a stricter server turns that rot
+            # into a publish failure the adapter swallows silently.
+            turn = {k: v for k, v in rec.items() if k in _INTERACTION_DATA_FIELDS}
             turn["role"] = role
             if role == "Assistant":
                 citations = _to_wire_citations(rec.get("cited_items"))

reflexio/integrations/openclaw/plugin/tests/test_state.py

@@ -261,3 +261,45 @@ def test_to_wire_citations_preserves_explicit_playbook_source_kind() -> None:
         "user_playbook",
         "profile",
     ]
+
+
+class TestWirePayloadContract:
+    """The wire dict must contain only fields the server's model accepts.
+
+    ``unpublished_slice`` used to build the payload with a denylist (drop three
+    known keys, pass the rest through), so buffer bookkeeping such as
+    ``user_id`` rode along. The server ignored it -- until it didn't, at which
+    point the adapter swallowed the error and the publish watermark never
+    advanced, so the same batch retried forever and nothing was published.
+    """
+
+    def test_bookkeeping_keys_never_reach_the_wire(self):
+        _, turns = state.unpublished_slice(
+            [{"ts": 1, "role": "User", "content": "x", "user_id": "proj-a", "id": "r1"}]
+        )
+        assert turns, "expected one wire turn"
+        assert "user_id" not in turns[0]
+        assert "id" not in turns[0]
+        assert turns[0]["content"] == "x"
+
+    def test_allowlist_matches_the_server_model(self):
+        """Pin the literal field set against the real InteractionData."""
+        interaction_data = pytest.importorskip(
+            "reflexio.models.api_schema.domain.entities"
+        ).InteractionData
+        assert set(interaction_data.model_fields) == state._INTERACTION_DATA_FIELDS
+
+    def test_wire_turn_validates_against_the_server_model(self):
+        """Every dict the slicer emits must construct a real InteractionData."""
+        interaction_data = pytest.importorskip(
+            "reflexio.models.api_schema.domain.entities"
+        ).InteractionData
+        _, turns = state.unpublished_slice(
+            [
+                {"ts": 1, "role": "User", "content": "hi", "user_id": "p"},
+                {"ts": 2, "role": "Assistant", "content": "hello", "user_id": "p"},
+            ]
+        )
+        for turn in turns:
+            built = interaction_data(**turn)
+            assert built.unknown_field_names() == []