From 9b001b61be212e324d5533a60ca48dd9558a3674 Mon Sep 17 00:00:00 2001 From: xiaoan17 Date: Thu, 29 Oct 2026 14:09:13 +0800 Subject: [PATCH] fix(chat): keep the provider reply projection in the composition root PR #7542 added a core capability import of loopx.extensions.lark.private_conversations, so tests/architecture/test_control_plane_import_boundaries.py:: test_lark_projection_sinks_are_owned_by_the_extension_layer fails on main. A core capability must not own an extension projection sink. Move the projection to the HTTP composition root that already joins provider history onto the selected page, and exercise it there. The core projection now carries no provider history at all. Joining the provider history in the composition root grows loopx/chat_server.py from 2020 to 2026 lines, so the reviewed module metric ceiling for that module is refreshed in the same change, as the maintainability ratchet requires. The growth is the composition call itself; no other module metric moved. Signed-off-by: xiaoan17 --- loopx/canary/module_metric_baseline.json | 2 +- .../capabilities/manager_context/roundtrip.py | 4 +- loopx/chat_server.py | 14 +++-- tests/test_manager_context_roundtrip.py | 57 ++++++++++++++++--- 4 files changed, 63 insertions(+), 14 deletions(-) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index 5b498482ac..a87d31f47d 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/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 diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index c0361c60eb..3e599f6f63 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/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 diff --git a/loopx/chat_server.py b/loopx/chat_server.py index d625be9824..9b73bc13da 100644 --- a/loopx/chat_server.py +++ b/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: diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 65ac1df2e8..195b00e28f 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/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"]}