feat: publish retrieved learning links

plugin/src/claude_smart/publish.py

@@ -8,12 +8,14 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Literal
 
 from claude_smart import state
 from claude_smart.reflexio_adapter import Adapter
 
 PublishStatus = Literal["nothing", "ok", "failed"]
+_LOGGER = logging.getLogger(__name__)
 
 
 def publish_unpublished(
@@ -60,6 +62,20 @@ def publish_unpublished(
     _, interactions = state.unpublished_slice(records)
     if not interactions:
         return ("nothing", 0)
+    retrieved_state: dict[str, object] | None = None
+    try:
+        start_offset = state.retrieved_learning_watermark(records)
+        injected_entries, end_offset = state.read_injected_entries(
+            session_id, start_offset
+        )
+        retrieved_state = state.attach_retrieved_learnings(
+            records,
+            interactions,
+            injected_entries,
+            end_offset,
+        )
+    except Exception as exc:  # best-effort metadata must never block a publish
+        _LOGGER.warning("Could not attach retrieved learnings: %s", exc)
     client = adapter if adapter is not None else Adapter()
     ok = client.publish(
         session_id=session_id,
@@ -70,6 +86,9 @@ def publish_unpublished(
         skip_aggregation=skip_aggregation,
     )
     if ok:
-        state.append(session_id, {"published_up_to": len(records)})
+        watermark_record: dict[str, object] = {"published_up_to": len(records)}
+        if retrieved_state is not None:
+            watermark_record.update(retrieved_state)
+        state.append(session_id, watermark_record)
         return ("ok", len(interactions))
     return ("failed", len(interactions))

plugin/src/claude_smart/reflexio_adapter.py

@@ -121,9 +121,10 @@ def publish(
         if client is None:
             return False
         try:
+            interaction_list = list(interactions)
             kwargs = {
                 "user_id": project_id,
-                "interactions": list(interactions),
+                "interactions": interaction_list,
                 "agent_version": runtime.agent_version(),
                 "session_id": session_id,
                 "wait_for_response": False,
@@ -136,7 +137,24 @@ def publish(
                 _LOGGER.debug(
                     "publish_interaction client does not support override_learning_stall"
                 )
-            client.publish_interaction(**kwargs)
+            if _needs_raw_retrieved_learning_publish(interaction_list):
+                client._make_request(  # noqa: SLF001 - pinned-client compatibility
+                    "POST",
+                    "/api/publish_interaction",
+                    json={
+                        "user_id": project_id,
+                        "interaction_data_list": interaction_list,
+                        "agent_version": runtime.agent_version(),
+                        "session_id": session_id,
+                        "skip_aggregation": skip_aggregation,
+                        "force_extraction": force_extraction,
+                        "evaluation_only": False,
+                        "override_learning_stall": override_learning_stall,
+                    },
+                    params=None,
+                )
+            else:
+                client.publish_interaction(**kwargs)
             return True
         except Exception as exc:  # noqa: BLE001
             _LOGGER.warning("publish_interaction failed: %s", exc)
@@ -476,6 +494,27 @@ def _supports_keyword(callable_obj: Any, keyword: str) -> bool:
     return keyword in signature.parameters
 
 
+def _needs_raw_retrieved_learning_publish(
+    interactions: Sequence[dict[str, Any]],
+) -> bool:
+    """Return whether the pinned client would strip retrieved-learning links.
+
+    Reflexio 0.2.28 reconstructs dictionaries through an older Pydantic model
+    before HTTP, silently dropping ``retrieved_learnings``. Its authenticated
+    request helper can carry the complete body safely: old servers ignore the
+    field, while upgraded servers persist it.
+    """
+    if not any("retrieved_learnings" in item for item in interactions):
+        return False
+    try:
+        from reflexio.models.api_schema.service_schemas import (  # type: ignore[import-not-found]
+            InteractionData,
+        )
+    except ImportError:
+        return False
+    return "retrieved_learnings" not in InteractionData.model_fields
+
+
 def _filter_rejected_agent_playbooks(items: list[Any]) -> list[Any]:
     """Drop rejected shared skills defensively, even if an older backend ignores filters."""
     return [

plugin/src/claude_smart/state.py

@@ -9,6 +9,9 @@
   to the next assistant turn at ``Stop`` time
 - ``{"published_up_to": N}`` — high-water mark so Stop / SessionEnd don't
   re-publish rows already sent to reflexio
+- ``{"retrieved_folded_up_to": N, "retrieved_pending": [...]}`` — byte offset
+  read from the injection registry plus future-dated entries awaiting an
+  eligible Assistant turn
 
 The buffer exists for offline resilience: when reflexio is unreachable,
 Stop appends without publishing and the next successful hook drains.
@@ -19,6 +22,7 @@
 import json
 import logging
 import os
+from bisect import bisect_left
 from collections.abc import Iterable
 from pathlib import Path
 from typing import Any
@@ -38,6 +42,8 @@
 _VALID_CITATION_KINDS = frozenset(
     {"playbook", "profile", "user_playbook", "agent_playbook"}
 )
+_VALID_RETRIEVED_PLAYBOOK_KINDS = frozenset({"user_playbook", "agent_playbook"})
+_RETRIEVED_LEARNINGS_WIRE_CAP = 1000
 
 
 def _truncate_tool_data_field(value: Any) -> Any:
@@ -114,24 +120,197 @@ def read_injected(session_id: str) -> dict[str, dict[str, Any]]:
     (identical content produces the same hash-derived id, so the extra
     record only refreshes metadata).
     """
+    registry: dict[str, dict[str, Any]] = {}
+    entries, _ = read_injected_entries(session_id)
+    for entry in entries:
+        item_id = entry.get("id")
+        if isinstance(item_id, str) and item_id:
+            registry[item_id] = entry
+    return registry
+
+
+def read_injected_entries(
+    session_id: str, start_offset: int = 0
+) -> tuple[list[dict[str, Any]], int]:
+    """Return new injection-registry entries and the ending byte offset.
+
+    Unlike :func:`read_injected`, this reader intentionally preserves repeated
+    ids: each line represents a distinct injection event that must be folded
+    into exactly one published Assistant turn.
+
+    Args:
+        session_id: Host session identifier.
+        start_offset: Previously committed safe byte offset.
+
+    Returns:
+        Ordered decoded entries and the last completely consumed byte offset.
+    """
     path = injected_path(session_id)
     if not path.exists():
-        return {}
-    registry: dict[str, dict[str, Any]] = {}
+        return [], start_offset
+    entries: list[dict[str, Any]] = []
     with path.open("r", encoding="utf-8") as fh:
-        for line in fh:
-            line = line.strip()
+        if fcntl is not None:
+            try:
+                fcntl.flock(fh.fileno(), fcntl.LOCK_SH)
+            except OSError as exc:
+                _LOGGER.debug("shared flock failed on %s: %s", path, exc)
+        fh.seek(start_offset)
+        end_offset = start_offset
+        while True:
+            line_start = fh.tell()
+            line = fh.readline()
             if not line:
+                break
+            candidate = line.strip()
+            if not candidate:
+                end_offset = fh.tell()
                 continue
             try:
-                entry = json.loads(line)
+                entry = json.loads(candidate)
             except json.JSONDecodeError as exc:
+                if not line.endswith("\n"):
+                    end_offset = line_start
+                    break
                 _LOGGER.warning("Skipping malformed injected line in %s: %s", path, exc)
+                end_offset = fh.tell()
                 continue
-            item_id = entry.get("id")
-            if isinstance(item_id, str) and item_id:
-                registry[item_id] = entry
-    return registry
+            if isinstance(entry, dict):
+                entries.append(entry)
+            end_offset = fh.tell()
+    return entries, end_offset
+
+
+def retrieved_learning_watermark(records: list[dict[str, Any]]) -> int:
+    """Return the latest successfully published injection-file byte offset.
+
+    Args:
+        records: Raw persisted session-buffer records.
+
+    Returns:
+        Latest non-negative ``retrieved_folded_up_to`` value, or zero.
+    """
+    offset = 0
+    for record in records:
+        candidate = record.get("retrieved_folded_up_to")
+        if isinstance(candidate, int) and candidate >= 0:
+            offset = candidate
+    return offset
+
+
+def attach_retrieved_learnings(
+    records: list[dict[str, Any]],
+    interactions: list[dict[str, Any]],
+    injected_entries: list[dict[str, Any]],
+    end_offset: int,
+) -> dict[str, Any]:
+    """Fold injected entries into the first eligible Assistant interaction.
+
+    Args:
+        records: Raw persisted session-buffer records.
+        interactions: Unpublished wire interactions to enrich after validation.
+        injected_entries: Newly read injection-registry entries in file order.
+        end_offset: Safe byte offset reached in the injection registry.
+
+    Returns:
+        Registry offset and any not-yet-eligible entries. The caller persists
+        this state only after publish succeeds, making retries idempotent
+        without assuming registry timestamps are monotonic.
+    """
+    published = 0
+    pending: list[dict[str, Any]] = []
+    for rec in records:
+        if "published_up_to" in rec:
+            published = rec["published_up_to"]
+        if "retrieved_folded_up_to" in rec:
+            pending_value = rec.get("retrieved_pending", [])
+            pending = (
+                [item for item in pending_value if isinstance(item, dict)]
+                if isinstance(pending_value, list)
+                else []
+            )
+
+    assistant_timestamps: list[int] = []
+    for rec in records[published:]:
+        if rec.get("role") != "Assistant":
+            continue
+        ts = rec.get("ts")
+        assistant_timestamps.append(ts if isinstance(ts, int) else 0)
+    assistant_interaction_indexes = [
+        idx
+        for idx, interaction in enumerate(interactions)
+        if interaction.get("role") == "Assistant"
+    ]
+    assistant_count = min(len(assistant_timestamps), len(assistant_interaction_indexes))
+    staged_retrieved: dict[int, list[dict[str, str]]] = {}
+    seen_by_interaction: dict[int, set[tuple[str, str]]] = {}
+    for index in assistant_interaction_indexes:
+        existing = interactions[index].get("retrieved_learnings", [])
+        retrieved = [item.copy() for item in existing if isinstance(item, dict)]
+        staged_retrieved[index] = retrieved
+        seen_by_interaction[index] = {
+            (str(item.get("kind", "")), str(item.get("learning_id", "")))
+            for item in retrieved
+        }
+
+    skipped = 0
+    truncated = 0
+    attached_total = 0
+    remaining: list[dict[str, Any]] = []
+    for entry in [*pending, *injected_entries]:
+        entry_ts = entry.get("ts")
+        comparable_ts = entry_ts if isinstance(entry_ts, int) else 0
+        target = bisect_left(assistant_timestamps, comparable_ts, hi=assistant_count)
+        target = target if target < assistant_count else None
+        if target is None:
+            remaining.append(entry)
+            continue
+
+        real_id = entry.get("real_id")
+        kind = entry.get("kind")
+        if not isinstance(real_id, str) or not real_id:
+            skipped += 1
+            continue
+        if kind == "profile":
+            wire_kind = "profile"
+        elif (
+            kind == "playbook"
+            and entry.get("source_kind") in _VALID_RETRIEVED_PLAYBOOK_KINDS
+        ):
+            wire_kind = entry["source_kind"]
+        else:
+            skipped += 1
+            continue
+
+        interaction_index = assistant_interaction_indexes[target]
+        retrieved = staged_retrieved[interaction_index]
+        candidate = {"kind": wire_kind, "learning_id": real_id}
+        candidate_key = (wire_kind, real_id)
+        if candidate_key in seen_by_interaction[interaction_index]:
+            continue
+        if (
+            len(retrieved) >= _RETRIEVED_LEARNINGS_WIRE_CAP
+            or attached_total >= _RETRIEVED_LEARNINGS_WIRE_CAP
+        ):
+            truncated += 1
+            continue
+        retrieved.append(candidate)
+        seen_by_interaction[interaction_index].add(candidate_key)
+        attached_total += 1
+
+    for interaction_index, retrieved in staged_retrieved.items():
+        interactions[interaction_index]["retrieved_learnings"] = retrieved
+
+    if skipped:
+        _LOGGER.debug("Skipped %d unresolvable injected learning entries", skipped)
+    if truncated:
+        _LOGGER.warning(
+            "Dropped %d retrieved learning links at the publish wire cap", truncated
+        )
+    return {
+        "retrieved_folded_up_to": end_offset,
+        "retrieved_pending": remaining,
+    }
 
 
 def append(session_id: str, record: dict[str, Any]) -> None: