fix(chat): keep the provider reply projection in the composition root

loopx/canary/module_metric_baseline.json

@@ -41,7 +41,7 @@
     "loopx/chat_server.py": {
       "any_count": 25,
       "dict_any_count": 0,
-      "lines": 2020
+      "lines": 2026
     },
     "loopx/chat_store.py": {
       "lines": 2066

loopx/capabilities/manager_context/roundtrip.py

@@ -307,8 +307,8 @@ def project_chat_session_snapshot(root, store, session_id, *, registry,
     # or re-reading every off-page Turn for each page.
     snapshot["messages"] = project_collaboration(store, root, session_id, snapshot["messages"],
         registry=registry, context_messages=context_messages)
-    from ...extensions.lark.private_conversations import project_private_reply_history
-    snapshot["messages"] = project_private_reply_history(store, store.load_session(session_id) or {}, snapshot["messages"])
+    # Provider-owned history joins the selected page at the composition root.
+    # Core capabilities never import an extension projection sink.
     if page is not None:
         snapshot["message_page"] = page["message_page"]
     return snapshot

loopx/chat_server.py

@@ -59,7 +59,7 @@
 from .capabilities.native_chat.model_settings import normalize_agent_reasoning_effort, normalize_agent_model_id
 from .extensions.lark.private_conversation_api import PrivateConversationRequestMixin, PRIVATE_CONVERSATIONS_PATH
 from .extensions.lark.conversation_identity import observe_lark_conversation_identity
-from .extensions.lark.private_conversations import LarkPrivateConversations
+from .extensions.lark.private_conversations import LarkPrivateConversations, project_private_reply_history
 from .chat_loopx_mode import DelegationWakeService, handle_loopx_request
 from .capabilities.manager_context.roundtrip import (
     ReturnService,
@@ -933,10 +933,16 @@ def _session_snapshot(self, session_id: str) -> None:
                 raise ValueError("message_limit must be between 1 and 200")
             self.server.runtime_controller.recover_exited_turn(session_id)
             snapshot = cast(Callable[..., dict[str, object]], project_chat_session_snapshot)
-            self._send_json(snapshot(
-                self.server.runtime_controller.coordination_runtime_root, self.server.chat_store, session_id,
+            payload = snapshot(
+                self.server.runtime_controller.coordination_runtime_root,
+                self.server.chat_store, session_id,
                 registry=self.server.registry_path, message_limit=message_limit,
-                message_cursor=query.get("message_cursor", [None])[0]))
+                message_cursor=query.get("message_cursor", [None])[0])
+            # The HTTP composition root joins provider history onto the selected page.
+            payload["messages"] = project_private_reply_history(
+                self.server.chat_store, self.server.chat_store.load_session(session_id) or {},
+                payload["messages"])
+            self._send_json(payload)
         except KeyError:
             self._send_error("chat session was not found", status=404)
         except ValueError as exc:

tests/test_manager_context_roundtrip.py

@@ -72,11 +72,54 @@ def frozen_private_reply(tmp_path):
     return store, session, turn, request, native, record, source_text, parent
 
 
-def test_frozen_private_reply_projects_original_text_without_mutating_model_input(frozen_private_reply, tmp_path):
+@pytest.fixture
+def private_snapshot(frozen_private_reply, tmp_path):
+    """Exercise provider history at its shipped HTTP composition boundary."""
+    import http.client
+    from types import SimpleNamespace
+    from urllib.parse import urlencode
+
+    from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
+
+    store, session, *_ = frozen_private_reply
+    server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
+    server.chat_store, server.verbose = store, False
+    server.registry_path, server.runtime_root = tmp_path / "absent.json", tmp_path
+    server.runtime_controller = SimpleNamespace(
+        recover_exited_turn=lambda _: None, close=lambda: None,
+        # Chat keeps its own storage when no registry names another root.
+        coordination_runtime_root=store.root.parent,
+    )
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+
+    def read(**query):
+        connection = http.client.HTTPConnection(*server.server_address, timeout=15)
+        try:
+            connection.request("GET", f"/api/chat/sessions/{session['session_id']}?{urlencode(query)}")
+            response = connection.getresponse()
+            payload = json.loads(response.read())
+            assert response.status == 200, payload
+            return payload
+        finally:
+            connection.close()
+
+    try:
+        yield read
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
+def test_frozen_private_reply_projects_original_text_without_mutating_model_input(frozen_private_reply, tmp_path, private_snapshot):
     store, session, turn, _, _, _, source_text, parent = frozen_private_reply
     before = {str(p): p.read_bytes() for p in store.root.rglob("*") if p.is_file()}
-    snapshot = project_chat_session_snapshot(tmp_path, store, session["session_id"], registry=tmp_path / "absent.json",
+    # The core projection carries no provider history; the composition root joins it.
+    generic = project_chat_session_snapshot(tmp_path, store, session["session_id"], registry=tmp_path / "absent.json",
                                           message_limit=1)
+    assert "reply_context" not in generic["messages"][0]
+    snapshot = private_snapshot(message_limit=1)
     message = snapshot["messages"][0]
     assert message["text"] == turn["message"]
     assert message["reply_context"]["source_text"] == source_text
@@ -86,7 +129,7 @@ def test_frozen_private_reply_projects_original_text_without_mutating_model_inpu
     assert {str(p): p.read_bytes() for p in store.root.rglob("*") if p.is_file()} == before
 
 
-def test_reply_history_keeps_failed_and_recorded_recovery_context(frozen_private_reply, tmp_path):
+def test_reply_history_keeps_failed_and_recorded_recovery_context(frozen_private_reply, tmp_path, private_snapshot):
     store, session, original, request, native, record, source_text, _ = frozen_private_reply
     client = f"external-{request}-recovery-21-quota-1"
     store.update_turn(session["session_id"], original["turn_id"], status="failed", error_code="synthetic_upstream_failure")
@@ -96,19 +139,19 @@ def test_reply_history_keeps_failed_and_recorded_recovery_context(frozen_private
     record["turn_id"] = recovery["turn_id"]
     for directory, row in (("external-requests", native), ("lark-private-deliveries", record)):
         _write(store.root / directory / f"{request}.json", row)
-    snapshot = project_chat_session_snapshot(tmp_path, store, session["session_id"], registry=tmp_path / "absent.json")
+    snapshot = private_snapshot()
     assert [m["reply_context"]["source_text"] for m in snapshot["messages"]] == [source_text, source_text]
     # A plausible-looking suffix without its original native receipt is not a source.
     store.update_turn(session["session_id"], recovery["turn_id"], status="completed")
     store.release_active_turn(session["session_id"], recovery["turn_id"])
     store.create_turn(session["session_id"], client_turn_id=f"external-{request}-recovery-999", message=original["message"], origin="lark")
-    snapshot = project_chat_session_snapshot(tmp_path, store, session["session_id"], registry=tmp_path / "absent.json")
+    snapshot = private_snapshot()
     assert "reply_context" not in snapshot["messages"][-1]
 
 
 @pytest.mark.parametrize("drift", ["session", "turn", "binding", "source", "native_message", "frozen_message",
                                   "parent", "thread", "order", "timestamp", "sender", "event_shape", "missing", "user_spoof", "foreign_origin"])
-def test_private_reply_view_requires_exact_frozen_source(frozen_private_reply, tmp_path, drift):
+def test_private_reply_view_requires_exact_frozen_source(frozen_private_reply, tmp_path, drift, private_snapshot):
     store, session, turn, request, native, record, _, _ = frozen_private_reply
     path = store.root / "lark-private-deliveries" / f"{request}.json"
     if drift in {"session", "turn", "binding"}:
@@ -139,7 +182,7 @@ def test_private_reply_view_requires_exact_frozen_source(frozen_private_reply, t
         path.unlink()
     if drift == "foreign_origin":
         store.append_message(session["session_id"], role="user", text=turn["message"], turn_id=turn["turn_id"], origin="peer")
-    snapshot = project_chat_session_snapshot(tmp_path, store, session["session_id"], registry=tmp_path / "absent.json")
+    snapshot = private_snapshot()
     message = snapshot["messages"][-1]
     assert "reply_context" not in message
     assert message["text"] in {turn["message"], record["reply_context_message"]}