diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index cebd9b94..65f8b667 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -26454,6 +26454,7 @@ def _conversation_read_output_schema(*, list_result: bool) -> Schema: "next_cursor": (str, type(None)), "coverage_complete": (bool, type(None)), "coverage": (dict, type(None)), + "ordering": (dict, type(None)), "warnings": (list, type(None)), "session_id": (str, type(None)), "chat_session_id": (str, type(None)), diff --git a/src/backend/services/conversation_management_service.py b/src/backend/services/conversation_management_service.py index 71a58693..42c8986f 100644 --- a/src/backend/services/conversation_management_service.py +++ b/src/backend/services/conversation_management_service.py @@ -773,6 +773,19 @@ def _append_visible(rows: Iterable[Mapping[str, Any]]) -> None: else: phase = "shared" position = None + if len(conversations) >= safe_limit: + # The owned stream is exhausted, but the shared stream has + # not been read. Preserve that transition for the next page. + next_cursor = encode_opaque_cursor( + purpose="conversation_list", + payload={ + "scope": cursor_scope, + "phase": "shared", + "position": None, + "coverage": _cursor_coverage(), + }, + ttl_seconds=_CONVERSATION_CURSOR_TTL_SECONDS, + ) if phase == "shared" and next_cursor is None and len(conversations) < safe_limit: remaining = max(1, safe_limit - len(conversations)) diff --git a/tests/backend/test_conversation_discovery_gateway.py b/tests/backend/test_conversation_discovery_gateway.py new file mode 100644 index 00000000..9f435aa2 --- /dev/null +++ b/tests/backend/test_conversation_discovery_gateway.py @@ -0,0 +1,174 @@ +"""Faithful discovery replay: real services, gateway and turn evidence; fixture storage.""" + +from __future__ import annotations + +import pytest + +from src.backend.integrations.internal_mcp import build_default_catalogue, catalogue +from src.backend.integrations.internal_mcp.gateway import ( + InternalMCPGateway, +) +from src.backend.integrations.internal_mcp.transport import InternalMCPTransport +from src.backend.security.access_control import override_current_actor +from src.backend.services import conversation_management_service as service +from src.backend.services.adaptive_turn_service import ( + _compact_evidence_envelope, + _resolve_turn_cursor_evidence_argument, +) +from src.backend.services.opaque_cursor_service import decode_opaque_cursor +from src.backend.services.turn_evidence_store import TrustedTurnScope, TurnEvidenceStore + + +@pytest.mark.parametrize("method", ["conversation_list", "conversation_search"]) +def test_gateway_filtered_pages_reach_owned_and_shared_candidates(monkeypatch, method): + scope = TrustedTurnScope( + user_concept_id="#V#alice", + organisation_concept_id="#V#org", + namespace="#V#alice@org", + ) + monkeypatch.setattr( + catalogue, "_register_dynamic_catalogue_methods", lambda **kwargs: None + ) + owned_positions = [] + + def owned_page(actor, **kwargs): + assert actor == scope.user_concept_id + assert kwargs["namespace"] == scope.namespace + position = kwargs["position"] + owned_positions.append(position) + assert position in (None, {"session_id": "named"}) + return { + "sessions": [ + { + "session_id": "named" if position is None else "unnamed", + "session_name": "Existing title" if position is None else None, + } + ], + "has_more": position is None, + "next_position": {"session_id": "named" if position is None else "unnamed"}, + } + + def invite_page(**kwargs): + assert kwargs["user_concept_id"] == scope.user_concept_id + assert kwargs["position"] is None + return { + "available": True, + "has_more": False, + "next_position": None, + "invites": [ + { + "session_id": "shared", + "invite_id": "invite-1", + "conversation_owner_user_id": "#V#owner", + "organisation_concept_id": "#V#org", + } + ], + } + + monkeypatch.setattr( + service.chat_history_service, + "get_chat_history_session_summaries_page", + owned_page, + ) + monkeypatch.setattr( + service, "list_accepted_invites_for_user_sessions", lambda **kwargs: [] + ) + monkeypatch.setattr(service, "list_accepted_invites_for_user_page", invite_page) + monkeypatch.setattr( + service, + "get_user_memberships", + lambda actor: {"memberships": [{"organisation_concept_id": "#V#org"}]}, + ) + monkeypatch.setattr( + service, + "apply_conversation_preferences", + lambda **kwargs: list(kwargs["conversations"]), + ) + monkeypatch.setattr( + service.chat_history_service, + "resolve_chat_history_namespace", + lambda owner: "#V#owner@org", + ) + monkeypatch.setattr( + service.chat_history_service, + "get_chat_history_session_summaries_for_owner_sessions", + lambda pairs, **kwargs: { + (owner, session): {"session_id": session, "session_name": None} + for owner, session in pairs + }, + ) + gateway = InternalMCPGateway( + catalogue=build_default_catalogue(), + transport=InternalMCPTransport(), + enabled=True, + ) + store = TurnEvidenceStore(scope, "discovery-turn") + arguments = { + "acting_user_concept_id": scope.user_concept_id, + "organisation_concept_id": scope.organisation_concept_id, + "namespace": scope.namespace, + "name_present": False, + } + arguments.update( + {"limit": 1} + if method == "conversation_list" + else {"query": "*", "page_size": 1} + ) + pages = [] + positions = [] + for index in range(3): + with override_current_actor( + scope.user_concept_id, scope.organisation_concept_id + ): + page = gateway.invoke(method, arguments).payload + assert page["success"] is True, page + assert page["ordering"]["tie_breaker"] == "session_id_or_invite_id" + pages.append(page) + envelope = store.record(method, f"call-{index}", page) + compact = _compact_evidence_envelope(envelope.to_mapping(), max_bytes=4000) + assert compact["source_diagnostics"]["coverage_complete"] == (index == 2) + assert compact["source_diagnostics"]["has_more"] == (index < 2) + if index < 2: + positions.append( + decode_opaque_cursor( + cursor=page["next_cursor"], purpose="conversation_list" + ) + ) + arguments, error = _resolve_turn_cursor_evidence_argument( + capability_name=method, + arguments={ + **{k: v for k, v in arguments.items() if k != "cursor"}, + "cursor_evidence_id": compact["evidence_id"], + }, + evidence_store=store, + scope=scope, + turn_id="discovery-turn", + ) + assert error is None + assert arguments["cursor"] == page["next_cursor"] + rows_key = "conversations" if method == "conversation_list" else "results" + assert [[row["session_id"] for row in page[rows_key]] for page in pages] == [ + [], + ["unnamed"], + ["shared"], + ] + assert pages[-1]["next_cursor"] is None + assert [(p["phase"], p["position"]) for p in positions] == [ + ("owned", {"session_id": "named"}), + ("shared", None), + ] + assert owned_positions == [None, {"session_id": "named"}] + + # A cursor cannot be continued in another actor's scope or a later turn. + with pytest.raises(service.ConversationManagementError, match="actor or filters"): + service.list_actor_conversations( + actor_user_id="#V#bob", cursor=pages[0]["next_cursor"] + ) + _, error = _resolve_turn_cursor_evidence_argument( + capability_name=method, + arguments={"cursor_evidence_id": envelope.evidence_id}, + evidence_store=store, + scope=scope, + turn_id="different-turn", + ) + assert error["error_code"] == "cursor_evidence_unavailable" diff --git a/tests/backend/test_conversation_management_service.py b/tests/backend/test_conversation_management_service.py index 9bf36d55..7e2e583e 100644 --- a/tests/backend/test_conversation_management_service.py +++ b/tests/backend/test_conversation_management_service.py @@ -532,6 +532,10 @@ def _accepted_for_sessions(**_kwargs): final = service.list_actor_conversations( actor_user_id="#V#alice", limit=1, cursor=first["next_cursor"] ) + assert final["next_cursor"] is not None + final = service.list_actor_conversations( + actor_user_id="#V#alice", limit=1, cursor=final["next_cursor"] + ) assert calls == {"owned": 2, "invite_lookup": 2} assert final["next_cursor"] is None