diff --git a/api/.importlinter b/api/.importlinter index f1fb5726f27..e010fe8a0e9 100644 --- a/api/.importlinter +++ b/api/.importlinter @@ -238,6 +238,29 @@ forbidden_modules = sqlalchemy werkzeug +[importlinter:contract:installed-app-message-boundary] +name = Installed app message service is framework and persistence neutral +type = forbidden +source_modules = + services.installed_app_message_service +forbidden_modules = + controllers + extensions + flask + models + repositories + sqlalchemy + werkzeug + +[importlinter:contract:installed-app-message-persistence-boundary] +name = Installed app message persistence does not call the legacy message service +type = forbidden +source_modules = + repositories.installed_app_message_repository +forbidden_modules = + services.message_service +allow_indirect_imports = True + [importlinter:contract:conversation-persistence-boundary] name = Conversation persistence does not call legacy conversation or workspace services type = forbidden diff --git a/api/controllers/console/explore/error.py b/api/controllers/console/explore/error.py index e0489463cdc..7314d6737fa 100644 --- a/api/controllers/console/explore/error.py +++ b/api/controllers/console/explore/error.py @@ -59,6 +59,24 @@ class ConversationNameRequiredHTTPError(InstalledAppHTTPError): code = 400 +class MessageNotFoundHTTPError(InstalledAppHTTPError): + error_code = "message_not_found" + description = "The message was not found for this account and app." + code = 404 + + +class MessageCursorNotFoundHTTPError(InstalledAppHTTPError): + error_code = "message_cursor_not_found" + description = "The message cursor is not in this conversation. Refresh the list and try again." + code = 404 + + +class MessageFeedbackRatingRequiredHTTPError(InstalledAppHTTPError): + error_code = "message_feedback_rating_required" + description = "A rating is required when there is no existing feedback to remove." + code = 400 + + class NotCompletionAppError(BaseHTTPException): error_code = "not_completion_app" description = "Not Completion App" diff --git a/api/controllers/console/explore/message.py b/api/controllers/console/explore/message.py index 9e38c44c2c4..f0c955ff1c6 100644 --- a/api/controllers/console/explore/message.py +++ b/api/controllers/console/explore/message.py @@ -1,14 +1,19 @@ +"""HTTP admission and response contracts for installed-app messages.""" + import logging +from collections.abc import Callable +from functools import wraps from typing import Literal from uuid import UUID -from flask import request -from pydantic import BaseModel, TypeAdapter -from sqlalchemy.orm import Session -from werkzeug.exceptions import InternalServerError, NotFound +from flask import Response +from flask_restx import Resource +from pydantic import BaseModel +from werkzeug.exceptions import HTTPException, InternalServerError, Unauthorized from controllers.common.controller_schemas import MessageFeedbackPayload, MessageListQuery from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models +from controllers.console import console_ns from controllers.console.app.error import ( AppMoreLikeThisDisabledError, AppUnavailableError, @@ -17,39 +22,39 @@ from controllers.console.app.error import ( ProviderNotInitializeError, ProviderQuotaExceededError, ) -from controllers.console.app.wraps import with_session from controllers.console.explore.error import ( AppSuggestedQuestionsAfterAnswerDisabledError, + ConversationNotFoundHTTPError, + InstalledAppNotFoundHTTPError, + MessageCursorNotFoundHTTPError, + MessageFeedbackRatingRequiredHTTPError, + MessageNotFoundHTTPError, NotChatAppError, NotCompletionAppError, ) -from controllers.console.explore.wraps import InstalledAppResource -from controllers.console.wraps import model_validate, with_current_user -from core.app.entities.app_invoke_entities import InvokeFrom +from controllers.console.explore.installed_app_admission import get_installed_app +from controllers.console.flask_admission import console_account_admission +from controllers.console.wraps import model_validate from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError -from extensions.ext_database import db -from fields.conversation_fields import MessageResponseSource, ResultResponse -from fields.message_fields import ( - ExploreMessageInfiniteScrollPagination, - ExploreMessageListItem, - SuggestedQuestionsResponse, -) +from extensions.ext_application_services import application_services +from fields.conversation_fields import ResultResponse +from fields.message_fields import ExploreMessageInfiniteScrollPagination, SuggestedQuestionsResponse from graphon.model_runtime.errors.invoke import InvokeError from libs import helper -from models import Account -from models.enums import FeedbackRating -from models.model import AppMode, InstalledApp -from services.app_generate_service import AppGenerateService +from machinery.context import RequestContext +from services.account_errors import AccountNotFoundError +from services.app_definition_query_service import AppDefinitionUnavailableError from services.errors.app import MoreLikeThisDisabledError +from services.errors.base import BaseServiceError from services.errors.conversation import ConversationNotExistsError from services.errors.message import ( FirstMessageNotExistsError, MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError, ) -from services.message_service import MessageService - -from .. import console_ns +from services.installed_app_access_service import InstalledAppNotFoundError, InstalledAppRef +from services.installed_app_generation_service import InstalledAppNotCompletionError +from services.installed_app_message_service import FeedbackRatingRequiredError, MessageNotChatAppError logger = logging.getLogger(__name__) @@ -67,176 +72,161 @@ register_response_schema_models( ) +def _message_errors[**P, R](view: Callable[P, R]) -> Callable[P, R]: + @wraps(view) + def decorated(*args: P.args, **kwargs: P.kwargs) -> R: + try: + return view(*args, **kwargs) + except InstalledAppNotFoundError as error: + raise InstalledAppNotFoundHTTPError() from error + except AppDefinitionUnavailableError as error: + raise AppUnavailableError() from error + except AccountNotFoundError as error: + raise Unauthorized("Account no longer exists.") from error + except MessageNotChatAppError as error: + raise NotChatAppError() from error + except InstalledAppNotCompletionError as error: + raise NotCompletionAppError() from error + except MessageNotExistsError as error: + raise MessageNotFoundHTTPError() from error + except FirstMessageNotExistsError as error: + raise MessageCursorNotFoundHTTPError() from error + except ConversationNotExistsError as error: + raise ConversationNotFoundHTTPError() from error + except FeedbackRatingRequiredError as error: + raise MessageFeedbackRatingRequiredHTTPError() from error + except MoreLikeThisDisabledError as error: + raise AppMoreLikeThisDisabledError() from error + except SuggestedQuestionsAfterAnswerDisabledError as error: + raise AppSuggestedQuestionsAfterAnswerDisabledError() from error + except ProviderTokenNotInitError as error: + raise ProviderNotInitializeError(error.description) from error + except QuotaExceededError as error: + raise ProviderQuotaExceededError() from error + except ModelCurrentlyNotSupportError as error: + raise ProviderModelCurrentlyNotSupportError() from error + except InvokeError as error: + raise CompletionRequestError(error.description) from error + except (HTTPException, ValueError): + raise + except Exception as error: + logger.exception("Installed-app message operation failed") + raise InternalServerError() from error + + return decorated + + @console_ns.route( "/installed-apps//messages", endpoint="installed_app_messages", ) -class MessageListApi(InstalledAppResource): +class MessageListApi(Resource): @console_ns.doc(params=query_params_from_model(MessageListQuery)) @console_ns.response(200, "Success", console_ns.models[ExploreMessageInfiniteScrollPagination.__name__]) - @with_current_user - def get(self, current_user: Account, installed_app: InstalledApp): - session = db.session() - app_model = installed_app.app_with_session(session=session) - if app_model is None: - raise AppUnavailableError() - - app_mode = AppMode.value_of(app_model.mode) - if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}: - raise NotChatAppError() - args = MessageListQuery.model_validate(request.args.to_dict()) - - try: - pagination = MessageService.pagination_by_first_id( - app_model, - current_user, - args.conversation_id, - args.first_id or None, - args.limit, - session=session, - ) - adapter = TypeAdapter(ExploreMessageListItem) - items = [ - adapter.validate_python(MessageResponseSource(message, session=session), from_attributes=True) - for message in pagination.data - ] - return ExploreMessageInfiniteScrollPagination( - limit=pagination.limit, - has_more=pagination.has_more, - data=items, - ).model_dump(mode="json") - except ConversationNotExistsError: - raise NotFound("Conversation Not Exists.") - except FirstMessageNotExistsError: - raise NotFound("First Message Not Exists.") + @console_account_admission() + @get_installed_app + @model_validate(MessageListQuery) + @_message_errors + def get( + self, query: MessageListQuery, request_context: RequestContext, installed_app: InstalledAppRef + ) -> dict[str, object]: + page = application_services().installed_app_messages.get_page( + installed_app=installed_app, + account_id=request_context.account_id, + conversation_id=query.conversation_id, + first_id=query.first_id or None, + limit=query.limit, + ) + return helper.dump_response(ExploreMessageInfiniteScrollPagination, page) @console_ns.route( "/installed-apps//messages//feedbacks", endpoint="installed_app_message_feedback", ) -class MessageFeedbackApi(InstalledAppResource): +class MessageFeedbackApi(Resource): @console_ns.expect(console_ns.models[MessageFeedbackPayload.__name__]) @console_ns.response(200, "Feedback submitted successfully", console_ns.models[ResultResponse.__name__]) - @with_current_user + @console_account_admission() + @get_installed_app @model_validate(MessageFeedbackPayload) + @_message_errors def post( - self, req_data: MessageFeedbackPayload, current_user: Account, installed_app: InstalledApp, message_id: UUID - ): - app_model = installed_app.app_with_session(session=db.session()) - if app_model is None: - raise AppUnavailableError() - - message_id_str = str(message_id) - - try: - MessageService.create_feedback( - app_model=app_model, - message_id=message_id_str, - user=current_user, - rating=FeedbackRating(req_data.rating) if req_data.rating else None, - content=req_data.content, - session=db.session(), - ) - except MessageNotExistsError: - raise NotFound("Message Not Exists.") - - return ResultResponse(result="success").model_dump(mode="json") + self, + payload: MessageFeedbackPayload, + request_context: RequestContext, + installed_app: InstalledAppRef, + message_id: UUID, + ) -> dict[str, object]: + application_services().installed_app_messages.set_feedback( + installed_app=installed_app, + account_id=request_context.account_id, + message_id=str(message_id), + rating=payload.rating, + content=payload.content, + ) + return helper.dump_response(ResultResponse, {"result": "success"}) @console_ns.route( "/installed-apps//messages//more-like-this", endpoint="installed_app_more_like_this", ) -class MessageMoreLikeThisApi(InstalledAppResource): +class MessageMoreLikeThisApi(Resource): @console_ns.doc(params=query_params_from_model(MoreLikeThisQuery)) @console_ns.response(200, "Success") - @with_current_user - @with_session - def get(self, session: Session, current_user: Account, installed_app: InstalledApp, message_id: UUID): - app_model = installed_app.app_with_session(session=session) - if app_model is None: - raise AppUnavailableError() - if app_model.mode != "completion": - raise NotCompletionAppError() - - message_id_str = str(message_id) - - args = MoreLikeThisQuery.model_validate(request.args.to_dict()) - - streaming = args.response_mode == "streaming" - - try: - response = AppGenerateService.generate_more_like_this( - session=session, - app_model=app_model, - user=current_user, - message_id=message_id_str, - invoke_from=InvokeFrom.EXPLORE, - streaming=streaming, - ) - # response-contract:ignore compact_generate_response - return helper.compact_generate_response(response) - except MessageNotExistsError: - raise NotFound("Message Not Exists.") - except MoreLikeThisDisabledError: - raise AppMoreLikeThisDisabledError() - except ProviderTokenNotInitError as ex: - raise ProviderNotInitializeError(ex.description) - except QuotaExceededError: - raise ProviderQuotaExceededError() - except ModelCurrentlyNotSupportError: - raise ProviderModelCurrentlyNotSupportError() - except InvokeError as e: - raise CompletionRequestError(e.description) - except ValueError as e: - raise e - except Exception: - logger.exception("internal server error.") - raise InternalServerError() + @console_account_admission() + @get_installed_app + @model_validate(MoreLikeThisQuery) + @_message_errors + def get( + self, + query: MoreLikeThisQuery, + request_context: RequestContext, + installed_app: InstalledAppRef, + message_id: UUID, + ) -> Response: + response = application_services().installed_app_generation.generate_more_like_this( + installed_app=installed_app, + account_id=request_context.account_id, + message_id=str(message_id), + streaming=query.response_mode == "streaming", + ) + # response-contract:ignore compact_generate_response + return helper.compact_generate_response(response) @console_ns.route( "/installed-apps//messages//suggested-questions", endpoint="installed_app_suggested_question", ) -class MessageSuggestedQuestionApi(InstalledAppResource): +class MessageSuggestedQuestionApi(Resource): @console_ns.response(200, "Success", console_ns.models[SuggestedQuestionsResponse.__name__]) - @with_current_user - def get(self, current_user: Account, installed_app: InstalledApp, message_id: UUID): - app_model = installed_app.app_with_session(session=db.session()) - if app_model is None: - raise AppUnavailableError() - app_mode = AppMode.value_of(app_model.mode) - if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}: - raise NotChatAppError() - - message_id_str = str(message_id) - + @console_account_admission() + @get_installed_app + @_message_errors + def get( + self, request_context: RequestContext, installed_app: InstalledAppRef, message_id: UUID + ) -> dict[str, object]: try: - questions = MessageService.get_suggested_questions_after_answer( - app_model=app_model, - user=current_user, - message_id=message_id_str, - invoke_from=InvokeFrom.EXPLORE, - session=db.session(), + questions = application_services().installed_app_messages.get_suggested_questions( + installed_app=installed_app, + account_id=request_context.account_id, + message_id=str(message_id), ) - except MessageNotExistsError: - raise NotFound("Message not found") - except ConversationNotExistsError: - raise NotFound("Conversation not found") - except SuggestedQuestionsAfterAnswerDisabledError: - raise AppSuggestedQuestionsAfterAnswerDisabledError() - except ProviderTokenNotInitError as ex: - raise ProviderNotInitializeError(ex.description) - except QuotaExceededError: - raise ProviderQuotaExceededError() - except ModelCurrentlyNotSupportError: - raise ProviderModelCurrentlyNotSupportError() - except InvokeError as e: - raise CompletionRequestError(e.description) - except Exception: - logger.exception("internal server error.") - raise InternalServerError() - - return SuggestedQuestionsResponse(data=questions).model_dump(mode="json") + except ( + BaseServiceError, + AppDefinitionUnavailableError, + ProviderTokenNotInitError, + QuotaExceededError, + ModelCurrentlyNotSupportError, + InvokeError, + ): + raise + except ValueError as error: + # Legacy model/history configuration failures are server failures, + # not invalid request parameters. Replace this when that runtime + # exposes typed configuration errors. + logger.exception("Suggested-question runtime failed for message %s", message_id) + raise InternalServerError() from error + return helper.dump_response(SuggestedQuestionsResponse, {"data": questions}) diff --git a/api/extensions/ext_application_services.py b/api/extensions/ext_application_services.py index e4fab175f77..ea0d90f1d3b 100644 --- a/api/extensions/ext_application_services.py +++ b/api/extensions/ext_application_services.py @@ -57,6 +57,7 @@ from repositories.file_grant_repository import FileGrantRepository from repositories.human_input_file_upload_repository import SQLAlchemyHumanInputFileUploadRepository from repositories.installation_state_repository import InstallationStateRepository from repositories.installed_app_conversation_repository import SQLAlchemyInstalledAppConversationRepository +from repositories.installed_app_message_repository import SQLAlchemyInstalledAppMessageRepository from repositories.installed_app_repository import SQLAlchemyInstalledAppRepository from repositories.message_file_preview_repository import MessageFilePreviewQueryRepository from repositories.oauth_access_token_repository import SQLAlchemyOAuthAccessTokenRepository @@ -187,6 +188,8 @@ from services.installed_app_access_service import InstalledAppAccessService from services.installed_app_conversation_service import InstalledAppConversationService from services.installed_app_generation_adapters import AppGenerateServiceRuntime as InstalledAppGenerateServiceRuntime from services.installed_app_generation_service import InstalledAppGenerationService +from services.installed_app_message_adapters import InstalledAppMessageRuntime, emit_installed_app_feedback +from services.installed_app_message_service import InstalledAppMessageService from services.installed_app_service import InstalledAppService from services.knowledge.api_key_service import DatasetApiKeyService from services.message_file_preview_service import MessageFilePreviewService @@ -390,6 +393,7 @@ class ApplicationServices: installed_app_access: InstalledAppAccessService installed_app_conversations: InstalledAppConversationService installed_app_generation: InstalledAppGenerationService + installed_app_messages: InstalledAppMessageService installed_apps: InstalledAppService notifications: NotificationService step_by_step_tour: StepByStepTourService @@ -592,6 +596,7 @@ def build_application_services( get_user_permissions=webapp_access.batch_get_user_permissions, ) app_preview_repository = AppPreviewQueryRepository(session_factory=database_client) + installed_app_message_runtime = InstalledAppMessageRuntime(session_factory=database_client) feature_gateway = FeatureServiceGateway() accounts = SQLAlchemyAccountRepository(session_factory=database_client) integrations = SQLAlchemyAccountIntegrationRepository(session_factory=database_client) @@ -830,6 +835,12 @@ def build_application_services( usage=installed_apps, runtime=InstalledAppGenerateServiceRuntime(session_factory=database_client), ), + installed_app_messages=InstalledAppMessageService( + messages=SQLAlchemyInstalledAppMessageRepository(session_factory=database_client), + get_extra_contents=installed_app_message_runtime.get_extra_contents, + suggested_questions=installed_app_message_runtime.get_suggested_questions, + emit_feedback=emit_installed_app_feedback, + ), installed_apps=InstalledAppService( installed_apps=installed_apps, get_workspace_role=workspace_query_repository.get_account_role, diff --git a/api/repositories/installed_app_message_repository.py b/api/repositories/installed_app_message_repository.py new file mode 100644 index 00000000000..30dd6148de2 --- /dev/null +++ b/api/repositories/installed_app_message_repository.py @@ -0,0 +1,215 @@ +"""Persist and materialize messages within an admitted installation.""" + +from typing import cast, override + +from pydantic import JsonValue +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from models.enums import ConversationFromSource, FeedbackFromSource, FeedbackRating +from models.model import App, Conversation, InstalledApp, Message, MessageAgentThought, MessageFeedback +from services.errors.conversation import ConversationNotExistsError +from services.errors.message import FirstMessageNotExistsError, MessageNotExistsError +from services.installed_app_access_service import InstalledAppNotFoundError, InstalledAppRef +from services.installed_app_message_service import ( + FeedbackRatingRequiredError, + InstalledAppMessageStore, + MessageAgentThoughtRecord, + MessageFeedbackEvent, + MessageFeedbackRecord, + MessageFileRecord, + MessageInputValue, + MessagePage, + MessageRating, + MessageRecord, +) + + +class SQLAlchemyInstalledAppMessageRepository(InstalledAppMessageStore): + def __init__(self, *, session_factory: sessionmaker[Session]) -> None: + self._session_factory: sessionmaker[Session] = session_factory + + @override + def get_page( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + conversation_id: str, + first_id: str | None, + limit: int, + ) -> MessagePage: + with self._session_factory() as session: + app = self._get_app(session=session, installed_app=installed_app) + if not conversation_id: + return MessagePage(limit=limit, has_more=False, data=()) + conversation = session.scalar( + select(Conversation).where( + Conversation.id == conversation_id, + Conversation.app_id == app.id, + Conversation.from_source == ConversationFromSource.CONSOLE, + Conversation.from_account_id == account_id, + Conversation.from_end_user_id.is_(None), + Conversation.is_deleted.is_(False), + ) + ) + if conversation is None: + raise ConversationNotExistsError( + f"Conversation {conversation_id} does not belong to this account and app." + ) + statement = select(Message).where(Message.conversation_id == conversation.id) + if first_id: + first_message = session.scalar(statement.where(Message.id == first_id)) + if first_message is None: + raise FirstMessageNotExistsError( + f"The first_id cursor {first_id} does not belong to conversation {conversation_id}." + ) + statement = statement.where( + Message.created_at < first_message.created_at, Message.id != first_message.id + ) + messages = list(session.scalars(statement.order_by(Message.created_at.desc()).limit(limit + 1)).all()) + has_more = len(messages) > limit + # Preserve ascending presentation of each newest-first page and the + # existing strict timestamp cursor, including tied timestamps. + messages = messages[:limit] + messages.reverse() + return MessagePage( + limit=limit, + has_more=has_more, + data=tuple(self._to_record(message=message, session=session) for message in messages), + ) + + @override + def set_feedback( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + message_id: str, + rating: MessageRating | None, + content: str | None, + ) -> MessageFeedbackEvent | None: + with self._session_factory.begin() as session: + app = self._get_app(session=session, installed_app=installed_app) + message = session.scalar( + select(Message).where( + Message.id == message_id, + Message.app_id == app.id, + Message.from_source == ConversationFromSource.CONSOLE, + Message.from_account_id == account_id, + Message.from_end_user_id.is_(None), + ) + ) + if message is None: + raise MessageNotExistsError(f"Message {message_id} does not belong to this account and app.") + feedback = message.admin_feedback_with_session(session=session) + if rating is None: + if feedback is None: + raise FeedbackRatingRequiredError( + f"Message {message_id} has no admin feedback to remove; a rating is required." + ) + session.delete(feedback) + return None + if feedback is None: + session.add( + MessageFeedback( + app_id=app.id, + conversation_id=message.conversation_id, + message_id=message.id, + rating=FeedbackRating(rating), + content=content, + from_source=FeedbackFromSource.ADMIN, + from_end_user_id=None, + from_account_id=account_id, + ) + ) + else: + feedback.rating = FeedbackRating(rating) + feedback.content = content + return MessageFeedbackEvent( + tenant_id=app.tenant_id, + app_id=app.id, + conversation_id=message.conversation_id, + message_id=message.id, + account_id=account_id, + rating=rating, + content=content, + ) + + @staticmethod + def _get_app(*, session: Session, installed_app: InstalledAppRef) -> App: + app = session.scalar( + select(App) + .join(InstalledApp, InstalledApp.app_id == App.id) + .where( + InstalledApp.id == installed_app.id, + InstalledApp.tenant_id == installed_app.tenant_id, + InstalledApp.app_id == installed_app.app_id, + ) + ) + if app is None: + raise InstalledAppNotFoundError(f"Installed app {installed_app.id} no longer exists") + return app + + @staticmethod + def _to_record(*, message: Message, session: Session) -> MessageRecord: + # Retain model input/file restoration (including its file I/O and legacy + # commit) until that boundary is migrated; do not recreate the file factory. + inputs = cast(dict[str, MessageInputValue], message.inputs_with_session(session=session)) + feedback = message.user_feedback_with_session(session=session) + thoughts = [ + SQLAlchemyInstalledAppMessageRepository._thought_record(thought) + for thought in message.agent_thoughts_with_session(session=session) + ] + files = [ + MessageFileRecord( + id=file["id"], + filename=file["filename"], + type=file["type"], + url=file.get("url"), + mime_type=file.get("mime_type"), + size=file.get("size"), + transfer_method=file["transfer_method"], + belongs_to=file.get("belongs_to"), + upload_file_id=file.get("upload_file_id"), + ) + for file in message.message_files_with_session(session=session) + ] + return MessageRecord( + id=message.id, + conversation_id=message.conversation_id, + parent_message_id=message.parent_message_id, + inputs=inputs, + query=message.query, + answer=message.re_sign_file_url_answer, + feedback=MessageFeedbackRecord(rating=feedback.rating.value) if feedback is not None else None, + retriever_resources=cast(list[dict[str, JsonValue]] | None, message.retriever_resources), + created_at=message.created_at, + agent_thoughts=thoughts, + message_files=files, + message_tokens=message.message_tokens, + answer_tokens=message.answer_tokens, + provider_response_latency=message.provider_response_latency, + total_price=message.total_price, + currency=message.currency, + status=message.status.value, + error=message.error, + metadata=cast(JsonValue, message.message_metadata_dict), + ) + + @staticmethod + def _thought_record(thought: MessageAgentThought) -> MessageAgentThoughtRecord: + return MessageAgentThoughtRecord( + id=thought.id, + message_id=thought.message_id, + message_chain_id=thought.message_chain_id, + position=thought.position, + thought=thought.thought, + answer=thought.answer, + tool=thought.tool, + tool_labels=cast(JsonValue, thought.tool_labels), + tool_input=thought.tool_input, + created_at=thought.created_at, + observation=thought.observation, + files=cast(list[str], thought.files), + ) diff --git a/api/services/installed_app_generation_adapters.py b/api/services/installed_app_generation_adapters.py index f19405b17d8..0b2751fd6e7 100644 --- a/api/services/installed_app_generation_adapters.py +++ b/api/services/installed_app_generation_adapters.py @@ -1,22 +1,59 @@ """Bridge pure installed app requests to the existing generation runtime.""" import logging -from collections.abc import Mapping -from typing import override +from collections.abc import Callable, Generator, Mapping +from typing import cast, override from sqlalchemy.orm import Session, sessionmaker +from core.app.apps.completion.app_generator import CompletionAppGenerator from core.app.entities.app_invoke_entities import InvokeFrom from models import Account, App from services.account_errors import AccountNotFoundError from services.app_definition_query_service import AppDefinitionUnavailableError from services.app_generate_service import AppGenerateService from services.conversation_service import ConversationService -from services.installed_app_generation_service import GenerationResponse, InstalledAppGenerationRuntime +from services.installed_app_generation_service import ( + GenerationResponse, + GenerationStream, + InstalledAppGenerationRuntime, +) logger = logging.getLogger(__name__) +class _MoreLikeThisEventStream: + def __init__(self, source: Generator[Mapping[str, object] | str, None, None]) -> None: + self._source: Generator[Mapping[str, object] | str, None, None] = source + self._events: GenerationStream = cast(GenerationStream, CompletionAppGenerator.convert_to_event_stream(source)) + self._closed: bool = False + + def __iter__(self) -> "_MoreLikeThisEventStream": + return self + + def __next__(self) -> str: + if self._closed: + raise StopIteration + try: + return next(self._events) + except BaseException: + try: + self.close() + except BaseException: + logger.exception("Failed to close the more-like-this response after an error") + raise + + def close(self) -> None: + if self._closed: + return + self._closed = True + try: + self._events.close() + finally: + # Closing an unstarted conversion generator does not reach its source. + self._source.close() + + class AppGenerateServiceRuntime(InstalledAppGenerationRuntime): def __init__(self, *, session_factory: sessionmaker[Session]) -> None: self._session_factory: sessionmaker[Session] = session_factory @@ -30,6 +67,52 @@ class AppGenerateServiceRuntime(InstalledAppGenerationRuntime): args: Mapping[str, object], streaming: bool, ) -> GenerationResponse: + conversation_id = args.get("conversation_id") + app, account = self._load_generation_context( + app_id=app_id, + account_id=account_id, + conversation_id=conversation_id if isinstance(conversation_id, str) else None, + ) + return self._run_generation( + lambda session: AppGenerateService.generate( + session=session, + app_model=app, + user=account, + args=args, + invoke_from=InvokeFrom.EXPLORE, + streaming=streaming, + ) + ) + + @override + def generate_more_like_this( + self, + *, + app_id: str, + account_id: str, + message_id: str, + streaming: bool, + ) -> GenerationResponse: + app, account = self._load_generation_context(app_id=app_id, account_id=account_id) + + def generate(session: Session) -> GenerationResponse: + response = AppGenerateService.generate_more_like_this( + session=session, + app_model=app, + user=account, + message_id=message_id, + invoke_from=InvokeFrom.EXPLORE, + streaming=streaming, + ) + if isinstance(response, Mapping): + return response + return _MoreLikeThisEventStream(response) + + return self._run_generation(generate) + + def _load_generation_context( + self, *, app_id: str, account_id: str, conversation_id: str | None = None + ) -> tuple[App, Account]: with self._session_factory(expire_on_commit=False) as read_session: app = read_session.get(App, app_id) if app is None: @@ -40,28 +123,22 @@ class AppGenerateServiceRuntime(InstalledAppGenerationRuntime): # Resolve the validated chat conversation before a streaming generator # is created, preserving the legacy eager 404 and ownership checks. - conversation_id = args.get("conversation_id") - if isinstance(conversation_id, str) and conversation_id: + if conversation_id: ConversationService.get_conversation( app_model=app, conversation_id=conversation_id, user=account, session=read_session, ) + return app, account + def _run_generation(self, generate: Callable[[Session], GenerationResponse]) -> GenerationResponse: # Release actor/app reads before the runtime's quota and rate-limit checks. # Legacy generators still own their internal database and external I/O. response: GenerationResponse | None = None try: with self._session_factory(expire_on_commit=False) as session: - generated_response: GenerationResponse = AppGenerateService.generate( - session=session, - app_model=app, - user=account, - args=args, - invoke_from=InvokeFrom.EXPLORE, - streaming=streaming, - ) + generated_response = generate(session) response = generated_response session.commit() except BaseException: diff --git a/api/services/installed_app_generation_service.py b/api/services/installed_app_generation_service.py index 708e777ee9c..8d7bb01aa2c 100644 --- a/api/services/installed_app_generation_service.py +++ b/api/services/installed_app_generation_service.py @@ -45,6 +45,15 @@ class InstalledAppGenerationRuntime(Protocol): streaming: bool, ) -> GenerationResponse: ... + def generate_more_like_this( + self, + *, + app_id: str, + account_id: str, + message_id: str, + streaming: bool, + ) -> GenerationResponse: ... + class InstalledAppGenerationService: def __init__( @@ -77,6 +86,25 @@ class InstalledAppGenerationService: return self._generate(installed_app=installed_app, account_id=account_id, args=args, streaming=True) + def generate_more_like_this( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + message_id: str, + streaming: bool, + ) -> GenerationResponse: + if installed_app.app_mode != "completion": + raise InstalledAppNotCompletionError(f"App {installed_app.app_id} is not a completion app") + + # Regenerating an earlier message does not record installation usage. + return self._runtime.generate_more_like_this( + app_id=installed_app.app_id, + account_id=account_id, + message_id=message_id, + streaming=streaming, + ) + def generate_workflow( self, *, installed_app: InstalledAppRef, account_id: str, args: Mapping[str, object] ) -> GenerationResponse: diff --git a/api/services/installed_app_message_adapters.py b/api/services/installed_app_message_adapters.py new file mode 100644 index 00000000000..cd64b7dea3a --- /dev/null +++ b/api/services/installed_app_message_adapters.py @@ -0,0 +1,88 @@ +"""Boundaries to the existing message runtime and execution extra contents.""" + +import logging +from collections.abc import Mapping, Sequence + +from flask import current_app +from pydantic import JsonValue +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from core.app.entities.app_invoke_entities import InvokeFrom +from extensions.ext_database import db +from models import Account, App, InstalledApp +from repositories.sqlalchemy_execution_extra_content_repository import SQLAlchemyExecutionExtraContentRepository +from services.account_errors import AccountNotFoundError +from services.installed_app_access_service import InstalledAppNotFoundError, InstalledAppRef +from services.installed_app_message_service import MessageFeedbackEvent +from services.message_service import MessageService + +logger = logging.getLogger(__name__) + + +class InstalledAppMessageRuntime: + def __init__(self, *, session_factory: sessionmaker[Session]) -> None: + self._session_factory: sessionmaker[Session] = session_factory + self._extra_contents: SQLAlchemyExecutionExtraContentRepository = SQLAlchemyExecutionExtraContentRepository( + session_maker=session_factory + ) + + def get_extra_contents(self, *, message_ids: Sequence[str]) -> Mapping[str, list[dict[str, JsonValue]]]: + contents = self._extra_contents.get_by_message_ids(message_ids) + return { + message_id: [content.model_dump(mode="json", exclude_none=True) for content in items] + for message_id, items in zip(message_ids, contents, strict=True) + } + + def get_suggested_questions(self, *, installed_app: InstalledAppRef, account_id: str, message_id: str) -> list[str]: + with self._session_factory(expire_on_commit=False) as session: + app = session.scalar( + select(App) + .join(InstalledApp, InstalledApp.app_id == App.id) + .where( + InstalledApp.id == installed_app.id, + InstalledApp.tenant_id == installed_app.tenant_id, + InstalledApp.app_id == installed_app.app_id, + ) + ) + if app is None: + raise InstalledAppNotFoundError(f"Installed app {installed_app.id} no longer exists") + account = session.get(Account, account_id) + if account is None: + raise AccountNotFoundError(f"Account {account_id} no longer exists") + + # TODO: Migrate the legacy history, model configuration, and provider + # lookups together. They still use db.session internally. A separate app + # context releases that scoped session without touching the request's. + with current_app.app_context(): + return MessageService.get_suggested_questions_after_answer( + app_model=app, + user=account, + message_id=message_id, + invoke_from=InvokeFrom.EXPLORE, + session=db.session(), + ) + + +def emit_installed_app_feedback(*, feedback: MessageFeedbackEvent) -> None: + # Feedback has already committed. Telemetry failure must not reject it. + try: + from core.telemetry import FeedbackCreatedEvent, TelemetryContext, emit + + emit( + FeedbackCreatedEvent( + context=TelemetryContext(tenant_id=feedback.tenant_id), + payload={ + "message_id": feedback.message_id, + "app_id": feedback.app_id, + "conversation_id": feedback.conversation_id, + "from_end_user_id": None, + "from_account_id": feedback.account_id, + "rating": feedback.rating, + "from_source": "admin", + "content": feedback.content, + }, + ) + ) + except Exception: + logger.warning("Failed to emit feedback telemetry for message %s", feedback.message_id, exc_info=True) diff --git a/api/services/installed_app_message_service.py b/api/services/installed_app_message_service.py new file mode 100644 index 00000000000..e6d4aaa0727 --- /dev/null +++ b/api/services/installed_app_message_service.py @@ -0,0 +1,201 @@ +"""Message reads and feedback for an admitted installation.""" + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field, replace +from datetime import datetime +from decimal import Decimal +from typing import Literal, Protocol + +from pydantic import JsonValue + +from graphon.file import File +from services.installed_app_access_service import InstalledAppRef + +type MessageRating = Literal["like", "dislike"] +type MessageInputValue = JsonValue | File | list[File] + + +class MessageNotChatAppError(Exception): + pass + + +class FeedbackRatingRequiredError(Exception): + pass + + +@dataclass(frozen=True, slots=True) +class MessageFeedbackRecord: + rating: str | None + + +@dataclass(frozen=True, slots=True) +class MessageFileRecord: + id: str + filename: str | None + type: str + url: str | None + mime_type: str | None + size: int | None + transfer_method: str + belongs_to: str | None + upload_file_id: str | None + + +@dataclass(frozen=True, slots=True) +class MessageAgentThoughtRecord: + id: str + message_id: str + message_chain_id: str | None + position: int + thought: str | None + answer: str | None + tool: str | None + tool_labels: JsonValue + tool_input: str | None + created_at: datetime | None + observation: str | None + files: list[str] + + +@dataclass(frozen=True, slots=True) +class MessageRecord: + id: str + conversation_id: str + parent_message_id: str | None + inputs: dict[str, MessageInputValue] + query: str + answer: str + feedback: MessageFeedbackRecord | None + retriever_resources: list[dict[str, JsonValue]] | None + created_at: datetime | None + agent_thoughts: list[MessageAgentThoughtRecord] + message_files: list[MessageFileRecord] + message_tokens: int + answer_tokens: int + provider_response_latency: float + total_price: Decimal | None + currency: str | None + status: str + error: str | None + metadata: JsonValue + extra_contents: list[dict[str, JsonValue]] = field(default_factory=list) + + +@dataclass(frozen=True, slots=True) +class MessagePage: + limit: int + has_more: bool + data: tuple[MessageRecord, ...] + + +@dataclass(frozen=True, slots=True) +class MessageFeedbackEvent: + tenant_id: str + app_id: str + conversation_id: str + message_id: str + account_id: str + rating: str + content: str | None + + +class InstalledAppMessageStore(Protocol): + def get_page( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + conversation_id: str, + first_id: str | None, + limit: int, + ) -> MessagePage: ... + + def set_feedback( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + message_id: str, + rating: MessageRating | None, + content: str | None, + ) -> MessageFeedbackEvent | None: ... + + +class MessageExtraContentsQuery(Protocol): + def __call__(self, *, message_ids: Sequence[str]) -> Mapping[str, list[dict[str, JsonValue]]]: ... + + +class SuggestedQuestionGenerator(Protocol): + def __call__(self, *, installed_app: InstalledAppRef, account_id: str, message_id: str) -> list[str]: ... + + +class MessageFeedbackEmitter(Protocol): + def __call__(self, *, feedback: MessageFeedbackEvent) -> None: ... + + +class InstalledAppMessageService: + def __init__( + self, + *, + messages: InstalledAppMessageStore, + get_extra_contents: MessageExtraContentsQuery, + suggested_questions: SuggestedQuestionGenerator, + emit_feedback: MessageFeedbackEmitter, + ) -> None: + self._messages: InstalledAppMessageStore = messages + self._get_extra_contents: MessageExtraContentsQuery = get_extra_contents + self._suggested_questions: SuggestedQuestionGenerator = suggested_questions + self._emit_feedback: MessageFeedbackEmitter = emit_feedback + + def get_page( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + conversation_id: str, + first_id: str | None, + limit: int, + ) -> MessagePage: + self._require_chat_app(installed_app) + page = self._messages.get_page( + installed_app=installed_app, + account_id=account_id, + conversation_id=conversation_id, + first_id=first_id, + limit=limit, + ) + if not page.data: + return page + contents = self._get_extra_contents(message_ids=[message.id for message in page.data]) + return replace( + page, + data=tuple(replace(message, extra_contents=contents.get(message.id, [])) for message in page.data), + ) + + def set_feedback( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + message_id: str, + rating: MessageRating | None, + content: str | None, + ) -> None: + feedback = self._messages.set_feedback( + installed_app=installed_app, + account_id=account_id, + message_id=message_id, + rating=rating, + content=content, + ) + if feedback is not None: + self._emit_feedback(feedback=feedback) + + def get_suggested_questions(self, *, installed_app: InstalledAppRef, account_id: str, message_id: str) -> list[str]: + self._require_chat_app(installed_app) + return self._suggested_questions(installed_app=installed_app, account_id=account_id, message_id=message_id) + + @staticmethod + def _require_chat_app(installed_app: InstalledAppRef) -> None: + if installed_app.app_mode not in {"chat", "agent-chat", "advanced-chat"}: + raise MessageNotChatAppError(f"Installed app {installed_app.id} is not a chat app.") diff --git a/api/tests/unit_tests/controllers/console/explore/test_installed_app_completion.py b/api/tests/unit_tests/controllers/console/explore/test_installed_app_completion.py index bd5d5783144..af0b802fa99 100644 --- a/api/tests/unit_tests/controllers/console/explore/test_installed_app_completion.py +++ b/api/tests/unit_tests/controllers/console/explore/test_installed_app_completion.py @@ -61,6 +61,11 @@ class _Runtime: raise self.error return self.response + def generate_more_like_this( + self, *, app_id: str, account_id: str, message_id: str, streaming: bool + ) -> GenerationResponse: + pytest.fail(f"Unexpected more-like-this call: {app_id=}, {account_id=}, {message_id=}, {streaming=}") + @dataclass(frozen=True) class _Services: diff --git a/api/tests/unit_tests/controllers/console/explore/test_message.py b/api/tests/unit_tests/controllers/console/explore/test_message.py index 0f8e37bc5ff..828dcb15a05 100644 --- a/api/tests/unit_tests/controllers/console/explore/test_message.py +++ b/api/tests/unit_tests/controllers/console/explore/test_message.py @@ -1,561 +1,588 @@ +"""Explore message HTTP contracts through real admission, services and SQLite.""" + +import json +from collections.abc import Generator, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime, timedelta from decimal import Decimal -from unittest.mock import MagicMock, patch +from typing import Literal +from uuid import uuid4 import pytest -from flask import Flask -from sqlalchemy.orm import Session, scoped_session, sessionmaker -from werkzeug.exceptions import InternalServerError, NotFound +from pydantic import JsonValue +from sqlalchemy import Connection, event, select +from sqlalchemy.orm import Session, SessionTransaction, sessionmaker +from werkzeug.test import TestResponse import controllers.console.explore.message as module -from controllers.console.app.error import ( - AppMoreLikeThisDisabledError, - CompletionRequestError, - ProviderModelCurrentlyNotSupportError, - ProviderNotInitializeError, - ProviderQuotaExceededError, -) -from controllers.console.explore.error import ( - AppSuggestedQuestionsAfterAnswerDisabledError, - NotChatAppError, - NotCompletionAppError, -) -from core.errors.error import ( - ModelCurrentlyNotSupportError, - ProviderTokenNotInitError, - QuotaExceededError, -) +from core.app.entities.app_invoke_entities import InvokeFrom +from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError from graphon.model_runtime.errors.invoke import InvokeError -from models import Account -from models.enums import ConversationFromSource -from models.model import App, AppMode, InstalledApp, Message +from models import App, AppMode, InstalledApp +from models.enums import ConversationFromSource, ConversationStatus, FeedbackFromSource, FeedbackRating +from models.model import Conversation, Message, MessageFeedback +from repositories.installed_app_message_repository import SQLAlchemyInstalledAppMessageRepository +from repositories.installed_app_repository import SQLAlchemyInstalledAppRepository +from services.errors.app import MoreLikeThisDisabledError from services.errors.conversation import ConversationNotExistsError -from services.errors.message import ( - FirstMessageNotExistsError, - MessageNotExistsError, - SuggestedQuestionsAfterAnswerDisabledError, +from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError +from services.installed_app_access_service import InstalledAppRef +from services.installed_app_generation_service import GenerationResponse, InstalledAppGenerationService +from services.installed_app_message_service import InstalledAppMessageService, MessageFeedbackEvent +from tests.unit_tests.controllers.console.explore.test_installed_app_admission import ( + _assert_json_response, + _Harness, + harness, ) +__all__ = ["harness"] -def unwrap(func): - bound_self = getattr(func, "__self__", None) - while hasattr(func, "__wrapped__"): - func = func.__wrapped__ - if bound_self is not None: - return func.__get__(bound_self, bound_self.__class__) - return func +_CREATED_AT = datetime(2024, 1, 1) +type _Operation = Literal["list", "feedback", "more-like-this", "suggested-questions"] +_OPERATIONS: tuple[_Operation, ...] = ("list", "feedback", "more-like-this", "suggested-questions") +_BLOCKING: dict[str, object] = {"answer": "你好", "metadata": {}, "usage": None} -def make_message(*, app_id: str): +@dataclass(frozen=True) +class _Services: + installed_app_messages: InstalledAppMessageService + installed_app_generation: InstalledAppGenerationService + + +@dataclass +class _Messages: + harness: _Harness + factory: sessionmaker[Session] + sessions: list[Session] = field(default_factory=list) + extras: dict[str, list[dict[str, JsonValue]]] = field(default_factory=dict) + extra_calls: list[list[str]] = field(default_factory=list) + feedback_events: list[MessageFeedbackEvent] = field(default_factory=list) + questions: list[str] = field(default_factory=lambda: ["Next question?", "还有吗？"]) + question_calls: list[tuple[InstalledAppRef, str, str]] = field(default_factory=list) + generation_calls: list[tuple[str, str, str, bool]] = field(default_factory=list) + generation_response: GenerationResponse = field(default_factory=lambda: dict(_BLOCKING)) + external_error: Exception | None = None + + def assert_sessions_closed(self) -> None: + assert all(not session.in_transaction() and not session.identity_map for session in self.sessions) + + def get_extra_contents(self, *, message_ids: Sequence[str]) -> Mapping[str, list[dict[str, JsonValue]]]: + self.assert_sessions_closed() + self.extra_calls.append(list(message_ids)) + return self.extras + + def emit_feedback(self, *, feedback: MessageFeedbackEvent) -> None: + self.assert_sessions_closed() + with self.factory() as session: + persisted = session.scalar(select(MessageFeedback).where(MessageFeedback.message_id == feedback.message_id)) + assert persisted is not None + assert persisted.rating.value == feedback.rating + assert persisted.content == feedback.content + self.feedback_events.append(feedback) + + def suggested_questions(self, *, installed_app: InstalledAppRef, account_id: str, message_id: str) -> list[str]: + self.assert_sessions_closed() + self.question_calls.append((installed_app, account_id, message_id)) + if self.external_error is not None: + raise self.external_error + return self.questions + + def generate( + self, *, app_id: str, account_id: str, args: Mapping[str, object], streaming: bool + ) -> GenerationResponse: + pytest.fail(f"Unexpected normal generation: {app_id=}, {account_id=}, {args=}, {streaming=}") + + def generate_more_like_this( + self, *, app_id: str, account_id: str, message_id: str, streaming: bool + ) -> GenerationResponse: + self.assert_sessions_closed() + self.assert_unused() + self.generation_calls.append((app_id, account_id, message_id, streaming)) + if self.external_error is not None: + raise self.external_error + return self.generation_response + + def assert_unused(self) -> None: + with self.factory() as session: + installation = session.get(InstalledApp, self.harness.installed_app.id) + assert installation is not None + assert installation.last_used_at is None + + def request( + self, + operation: _Operation, + *, + conversation_id: str = "", + message_id: str | None = None, + installed_app_id: str | None = None, + query: str | None = None, + body: dict[str, object] | None = None, + ) -> TestResponse: + url = f"/installed-apps/{installed_app_id or self.harness.installed_app.id}/messages" + if operation == "list": + url += f"?conversation_id={conversation_id}" + if query: + url += f"&{query}" + else: + suffix = "feedbacks" if operation == "feedback" else operation + url += f"/{message_id or uuid4()}/{suffix}" + if operation == "more-like-this": + url += f"?{query if query is not None else 'response_mode=blocking'}" + return self.harness.app.test_client().open(url, method="POST" if operation == "feedback" else "GET", json=body) + + +@pytest.fixture +def messages( + harness: _Harness, monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessionmaker[Session] +) -> _Messages: + factory = sessionmaker(bind=sqlite_session_factory.kw["bind"], expire_on_commit=False) + state = _Messages(harness=harness, factory=factory) + _set_mode(state, AppMode.CHAT) + + @event.listens_for(factory, "after_begin") + def track_session(session: Session, _transaction: SessionTransaction, _connection: Connection) -> None: + state.sessions.append(session) + + services = _Services( + installed_app_messages=InstalledAppMessageService( + messages=SQLAlchemyInstalledAppMessageRepository(session_factory=factory), + get_extra_contents=state.get_extra_contents, + suggested_questions=state.suggested_questions, + emit_feedback=state.emit_feedback, + ), + installed_app_generation=InstalledAppGenerationService( + usage=SQLAlchemyInstalledAppRepository(session_factory=factory), runtime=state + ), + ) + monkeypatch.setattr(module, "application_services", lambda: services) + for resource, suffix in ( + (module.MessageListApi, ""), + (module.MessageFeedbackApi, "//feedbacks"), + (module.MessageMoreLikeThisApi, "//more-like-this"), + (module.MessageSuggestedQuestionApi, "//suggested-questions"), + ): + harness.api.add_resource(resource, f"/installed-apps//messages{suffix}") + return state + + +def _set_mode(state: _Messages, mode: AppMode) -> None: + with state.factory.begin() as session: + app = session.get(App, state.harness.target_app.id) + assert app is not None + app.mode = mode + + +def _conversation( + state: _Messages, + *, + app_id: str | None = None, + account_id: str | None = None, + source: ConversationFromSource = ConversationFromSource.CONSOLE, + end_user_id: str | None = None, + is_deleted: bool = False, +) -> Conversation: + conversation = Conversation( + app_id=app_id or state.harness.target_app.id, + mode=AppMode.CHAT, + name="Conversation", + status=ConversationStatus.NORMAL, + _inputs={}, + from_source=source, + from_account_id=account_id or state.harness.account.id, + from_end_user_id=end_user_id, + invoke_from=InvokeFrom.EXPLORE, + is_deleted=is_deleted, + ) + with state.factory.begin() as session: + session.add(conversation) + return conversation + + +def _message( + state: _Messages, + conversation: Conversation, + *, + age: int = 0, + app_id: str | None = None, + account_id: str | None = None, + source: ConversationFromSource = ConversationFromSource.CONSOLE, + end_user_id: str | None = None, +) -> Message: message = Message( - id="m1", - app_id=app_id, - conversation_id="11111111-1111-1111-1111-111111111111", - query="hello", - message={"role": "user", "content": "hello"}, - answer="", - message_tokens=0, + app_id=app_id or conversation.app_id, + conversation_id=conversation.id, + _inputs={"zero": 0, "disabled": False, "null": None, "empty": []}, + query="Hello", + message={}, + answer="你好", + message_tokens=2, + answer_tokens=3, message_unit_price=Decimal(0), - answer_tokens=0, answer_unit_price=Decimal(0), + total_price=Decimal(0), provider_response_latency=0, currency="USD", - from_source=ConversationFromSource.API, - app_mode=AppMode.CHAT, + status="normal", + message_metadata=json.dumps({"retriever_resources": [], "zero": 0}), + from_source=source, + from_account_id=account_id or state.harness.account.id, + from_end_user_id=end_user_id, + created_at=_CREATED_AT - timedelta(minutes=age), ) - message._inputs = {} - message.status = "normal" + with state.factory.begin() as session: + session.add(message) return message -def make_installed_app(session: Session, mode: str | None = None) -> InstalledApp: - app_model = App( - tenant_id="owner-tenant", - name="Explore App", - mode=mode or AppMode.CHAT, - enable_site=True, - enable_api=False, - ) - session.add(app_model) - session.flush() - installed_app = InstalledApp( - tenant_id="viewer-tenant", - app_id=app_model.id, - app_owner_tenant_id=app_model.tenant_id, - position=0, - is_pinned=False, - last_used_at=None, - ) - session.add(installed_app) - session.commit() - return installed_app +def _error(response: TestResponse, *, status: int, code: str, message: str | None = None) -> None: + assert response.status_code == status + body = response.get_json() + assert body["code"] == code + assert body["status"] == status + assert isinstance(body["message"], str) + assert body["message"] + if message is not None: + assert body["message"] == message + assert response.headers["Content-Type"] == "application/json" + assert int(response.headers["Content-Length"]) == len(response.data) -class _UsesSQLiteSession: - sqlite_session: Session - account: Account - - @pytest.fixture(autouse=True) - def _bind_database( - self, - sqlite_session: Session, - sqlite_session_factory: sessionmaker[Session], - monkeypatch: pytest.MonkeyPatch, - ): - self.sqlite_session = sqlite_session - self.account = Account(name="User", email="user@example.com") - self.account.id = "account-1" - session_proxy = scoped_session(sqlite_session_factory) - monkeypatch.setattr(module.db, "session", session_proxy) - yield - session_proxy.remove() - - -class TestMessageListApi(_UsesSQLiteSession): - def test_get_success(self, app: Flask): - api = module.MessageListApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - pagination = MagicMock( - limit=20, - has_more=False, - data=[make_message(app_id=installed_app.app_id), make_message(app_id=installed_app.app_id)], +def test_list_preserves_response_and_fetches_older_messages_in_display_order(messages: _Messages) -> None: + conversation = _conversation(messages) + oldest = _message(messages, conversation, age=2) + middle = _message(messages, conversation, age=1) + newest = _message(messages, conversation) + messages.extras[middle.id] = [{"type": "human_input", "workflow_run_id": "workflow-1", "submitted": False}] + with messages.factory.begin() as session: + session.add_all( + [ + MessageFeedback( + app_id=newest.app_id, + conversation_id=conversation.id, + message_id=newest.id, + rating=rating, + from_source=source, + ) + for rating, source in ( + (FeedbackRating.LIKE, FeedbackFromSource.ADMIN), + (FeedbackRating.DISLIKE, FeedbackFromSource.USER), + ) + ] ) - - with ( - app.test_request_context( - "/", - query_string={"conversation_id": "11111111-1111-1111-1111-111111111111"}, - ), - patch.object( - module.MessageService, - "pagination_by_first_id", - return_value=pagination, - ), - ): - result = method(self.account, installed_app) - - assert result["limit"] == 20 - assert result["has_more"] is False - assert len(result["data"]) == 2 - - def test_get_not_chat_app(self): - api = module.MessageListApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with pytest.raises(NotChatAppError): - method(self.account, installed_app) - - def test_conversation_not_exists(self, app: Flask): - api = module.MessageListApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - app.test_request_context( - "/", - query_string={"conversation_id": "11111111-1111-1111-1111-111111111111"}, - ), - patch.object( - module.MessageService, - "pagination_by_first_id", - side_effect=ConversationNotExistsError(), - ), - ): - with pytest.raises(NotFound): - method(self.account, installed_app) - - def test_first_message_not_exists(self, app: Flask): - api = module.MessageListApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - app.test_request_context( - "/", - query_string={"conversation_id": "11111111-1111-1111-1111-111111111111"}, - ), - patch.object( - module.MessageService, - "pagination_by_first_id", - side_effect=FirstMessageNotExistsError(), - ), - ): - with pytest.raises(NotFound): - method(self.account, installed_app) - - -class TestMessageFeedbackApi(_UsesSQLiteSession): - def test_post_success(self, app: Flask): - api = module.MessageFeedbackApi() - method = unwrap(api.post) - - installed_app = make_installed_app(self.sqlite_session) - - with ( - app.test_request_context("/", json={"rating": "like"}), - patch.object( - module.MessageService, - "create_feedback", - ), - ): - result = method( - module.MessageFeedbackPayload.model_validate({"rating": "like"}), self.account, installed_app, "mid" - ) - - assert result["result"] == "success" - - def test_message_not_exists(self, app: Flask): - api = module.MessageFeedbackApi() - method = unwrap(api.post) - - installed_app = make_installed_app(self.sqlite_session) - - with ( - app.test_request_context("/", json={}), - patch.object( - module.MessageService, - "create_feedback", - side_effect=MessageNotExistsError(), - ), - ): - with pytest.raises(NotFound): - method(module.MessageFeedbackPayload.model_validate({}), self.account, installed_app, "mid") - - -class TestMessageMoreLikeThisApi(_UsesSQLiteSession): - def test_get_success(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - return_value={"ok": True}, - ), - patch.object( - module.helper, - "compact_generate_response", - return_value=("ok", 200), - ), - ): - resp = method(self.sqlite_session, self.account, installed_app, "mid") - - assert resp == ("ok", 200) - - def test_not_completion_app(self): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with pytest.raises(NotCompletionAppError): - method(self.sqlite_session, self.account, installed_app, "mid") - - def test_more_like_this_disabled(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - side_effect=module.MoreLikeThisDisabledError(), - ), - ): - with pytest.raises(AppMoreLikeThisDisabledError): - method(self.sqlite_session, self.account, installed_app, "mid") - - def test_message_not_exists_more_like_this(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - side_effect=MessageNotExistsError(), - ), - ): - with pytest.raises(NotFound): - method(self.sqlite_session, self.account, installed_app, "mid") - - def test_provider_not_init_more_like_this(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - side_effect=ProviderTokenNotInitError("test"), - ), - ): - with pytest.raises(ProviderNotInitializeError): - method(self.sqlite_session, self.account, installed_app, "mid") - - def test_quota_exceeded_more_like_this(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - side_effect=QuotaExceededError(), - ), - ): - with pytest.raises(ProviderQuotaExceededError): - method(self.sqlite_session, self.account, installed_app, "mid") - - def test_model_not_support_more_like_this(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - side_effect=ModelCurrentlyNotSupportError(), - ), - ): - with pytest.raises(ProviderModelCurrentlyNotSupportError): - method(self.sqlite_session, self.account, installed_app, "mid") - - def test_invoke_error_more_like_this(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - side_effect=InvokeError("test error"), - ), - ): - with pytest.raises(CompletionRequestError): - method(self.sqlite_session, self.account, installed_app, "mid") - - def test_unexpected_error_more_like_this(self, app: Flask): - api = module.MessageMoreLikeThisApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with ( - app.test_request_context( - "/", - query_string={"response_mode": "blocking"}, - ), - patch.object( - module.AppGenerateService, - "generate_more_like_this", - side_effect=Exception("unexpected"), - ), - ): - with pytest.raises(InternalServerError): - method(self.sqlite_session, self.account, installed_app, "mid") - - -class TestMessageSuggestedQuestionApi(_UsesSQLiteSession): - def test_get_success(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - return_value=["q1", "q2"], - ), - ): - result = method(self.account, installed_app, "mid") - - assert result["data"] == ["q1", "q2"] - - def test_not_chat_app(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="completion") - - with pytest.raises(NotChatAppError): - method(self.account, installed_app, "mid") - - def test_disabled(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=SuggestedQuestionsAfterAnswerDisabledError(), - ), - ): - with pytest.raises(AppSuggestedQuestionsAfterAnswerDisabledError): - method(self.account, installed_app, "mid") - - def test_message_not_exists_suggested_question(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=MessageNotExistsError(), - ), - ): - with pytest.raises(NotFound): - method(self.account, installed_app, "mid") - - def test_conversation_not_exists_suggested_question(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=ConversationNotExistsError(), - ), - ): - with pytest.raises(NotFound): - method(self.account, installed_app, "mid") - - def test_provider_not_init_suggested_question(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=ProviderTokenNotInitError("test"), - ), - ): - with pytest.raises(ProviderNotInitializeError): - method(self.account, installed_app, "mid") - - def test_quota_exceeded_suggested_question(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=QuotaExceededError(), - ), - ): - with pytest.raises(ProviderQuotaExceededError): - method(self.account, installed_app, "mid") - - def test_model_not_support_suggested_question(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=ModelCurrentlyNotSupportError(), - ), - ): - with pytest.raises(ProviderModelCurrentlyNotSupportError): - method(self.account, installed_app, "mid") - - def test_invoke_error_suggested_question(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=InvokeError("test error"), - ), - ): - with pytest.raises(CompletionRequestError): - method(self.account, installed_app, "mid") - - def test_unexpected_error_suggested_question(self): - api = module.MessageSuggestedQuestionApi() - method = unwrap(api.get) - - installed_app = make_installed_app(self.sqlite_session, mode="chat") - - with ( - patch.object( - module.MessageService, - "get_suggested_questions_after_answer", - side_effect=Exception("unexpected"), - ), - ): - with pytest.raises(InternalServerError): - method(self.account, installed_app, "mid") + response = messages.request("list", conversation_id=conversation.id, query="limit=2") + body = response.get_json() + assert response.status_code == 200 + assert body["limit"] == 2 + assert body["has_more"] is True + assert [item["id"] for item in body["data"]] == [middle.id, newest.id] + assert body["data"][0]["extra_contents"] == [ + { + "type": "human_input", + "workflow_run_id": "workflow-1", + "submitted": False, + "form_definition": None, + "form_submission_data": None, + } + ] + assert body["data"][1] == { + "id": newest.id, + "conversation_id": conversation.id, + "parent_message_id": None, + "inputs": {"zero": 0, "disabled": False, "null": None, "empty": []}, + "query": "Hello", + "answer": "你好", + "feedback": {"rating": "dislike"}, + "retriever_resources": [], + "created_at": int(_CREATED_AT.timestamp()), + "agent_thoughts": [], + "message_files": [], + "message_tokens": 2, + "answer_tokens": 3, + "provider_response_latency": 0.0, + "total_price": "0E-7", + "currency": "USD", + "status": "normal", + "error": None, + "extra_contents": [], + "total_tokens": 5, + "metadata": {"retriever_resources": [], "zero": 0}, + } + _assert_json_response(response, status=200, body=body) + response = messages.request("list", conversation_id=conversation.id, query=f"limit=2&first_id={middle.id}") + assert response.get_json()["has_more"] is False + assert [item["id"] for item in response.get_json()["data"]] == [oldest.id] + assert messages.extra_calls == [[middle.id, newest.id], [oldest.id]] + messages.assert_sessions_closed() + + +@pytest.mark.parametrize("empty_conversation_id", [False, True]) +def test_empty_list_keeps_defaults_and_avoids_extra_content_io( + messages: _Messages, empty_conversation_id: bool +) -> None: + conversation = _conversation(messages) + _assert_json_response( + messages.request("list", conversation_id="" if empty_conversation_id else conversation.id), + status=200, + body={"limit": 20, "has_more": False, "data": []}, + ) + assert messages.extra_calls == [] + + +@pytest.mark.parametrize("query", ["limit=0", "limit=101", "limit=invalid", "first_id=invalid"]) +def test_invalid_list_query_uses_shared_422(messages: _Messages, query: str) -> None: + _error(messages.request("list", conversation_id=str(uuid4()), query=query), status=422, code="unprocessable_entity") + assert messages.extra_calls == [] + + +@pytest.mark.parametrize("owner", ["app", "account", "source", "end_user", "deleted", "missing"]) +def test_list_requires_conversation_ownership(messages: _Messages, owner: str) -> None: + conversation = _conversation( + messages, + app_id=str(uuid4()) if owner == "app" else None, + account_id=str(uuid4()) if owner == "account" else None, + source=ConversationFromSource.API if owner == "source" else ConversationFromSource.CONSOLE, + end_user_id=str(uuid4()) if owner == "end_user" else None, + is_deleted=owner == "deleted", + ) + _error( + messages.request("list", conversation_id=str(uuid4()) if owner == "missing" else conversation.id), + status=404, + code="conversation_not_found", + ) + assert messages.extra_calls == [] + + +def test_list_rejects_cursor_from_another_conversation(messages: _Messages) -> None: + conversation = _conversation(messages) + foreign = _message(messages, _conversation(messages)) + _error( + messages.request("list", conversation_id=conversation.id, query=f"first_id={foreign.id}"), + status=404, + code="message_cursor_not_found", + ) + assert messages.extra_calls == [] + + +def test_feedback_create_update_revoke_persists_before_telemetry(messages: _Messages) -> None: + message = _message(messages, _conversation(messages)) + for rating, content in (("like", ""), ("dislike", "Changed my mind")): + _assert_json_response( + messages.request("feedback", message_id=message.id, body={"rating": rating, "content": content}), + status=200, + body={"result": "success"}, + ) + with messages.factory() as session: + rows = session.scalars(select(MessageFeedback)).all() + assert len(rows) == 1 + assert rows[0].rating.value == rating + assert rows[0].content == content + assert rows[0].from_source == FeedbackFromSource.ADMIN + assert rows[0].from_account_id == messages.harness.account.id + assert rows[0].from_end_user_id is None + _assert_json_response( + messages.request("feedback", message_id=message.id, body={"rating": None}), + status=200, + body={"result": "success"}, + ) + with messages.factory() as session: + assert session.scalars(select(MessageFeedback)).all() == [] + assert [feedback.rating for feedback in messages.feedback_events] == ["like", "dislike"] + assert all( + feedback.tenant_id == messages.harness.target_app.tenant_id + and feedback.app_id == message.app_id + and feedback.conversation_id == message.conversation_id + and feedback.message_id == message.id + and feedback.account_id == messages.harness.account.id + for feedback in messages.feedback_events + ) + + +def test_revoke_without_feedback_has_specific_400(messages: _Messages) -> None: + message = _message(messages, _conversation(messages)) + _error( + messages.request("feedback", message_id=message.id, body={"rating": None}), + status=400, + code="message_feedback_rating_required", + ) + assert messages.feedback_events == [] + + +@pytest.mark.parametrize("owner", ["app", "account", "source", "end_user", "missing"]) +def test_feedback_requires_complete_message_ownership(messages: _Messages, owner: str) -> None: + message = _message( + messages, + _conversation(messages), + app_id=str(uuid4()) if owner == "app" else None, + account_id=str(uuid4()) if owner == "account" else None, + source=ConversationFromSource.API if owner == "source" else ConversationFromSource.CONSOLE, + end_user_id=str(uuid4()) if owner == "end_user" else None, + ) + _error( + messages.request( + "feedback", message_id=str(uuid4()) if owner == "missing" else message.id, body={"rating": "like"} + ), + status=404, + code="message_not_found", + ) + assert messages.feedback_events == [] + with messages.factory() as session: + assert session.scalars(select(MessageFeedback)).all() == [] + + +@pytest.mark.parametrize("mode", list(AppMode)) +def test_feedback_accepts_every_app_mode(messages: _Messages, mode: AppMode) -> None: + _set_mode(messages, mode) + message = _message(messages, _conversation(messages)) + _assert_json_response( + messages.request("feedback", message_id=message.id, body={"rating": "like"}), + status=200, + body={"result": "success"}, + ) + + +@pytest.mark.parametrize("body", [{"rating": "invalid"}, {"content": 123}]) +def test_invalid_feedback_uses_shared_422(messages: _Messages, body: dict[str, object]) -> None: + _error(messages.request("feedback", body=body), status=422, code="unprocessable_entity") + assert messages.feedback_events == [] + + +@pytest.mark.parametrize("operation", _OPERATIONS) +@pytest.mark.parametrize("admission", ["missing", "denied"]) +def test_all_message_handlers_apply_installed_app_admission( + messages: _Messages, operation: _Operation, admission: str +) -> None: + if admission == "denied": + messages.harness.state.allowed = False + _error( + messages.request(operation, installed_app_id=str(uuid4()) if admission == "missing" else None), + status=404 if admission == "missing" else 403, + code="installed_app_not_found" if admission == "missing" else "access_denied", + ) + assert ( + messages.extra_calls == messages.feedback_events == messages.question_calls == messages.generation_calls == [] + ) + + +@pytest.mark.parametrize("operation", ["list", "more-like-this", "suggested-questions"]) +def test_mode_rejection_precedes_message_queries_and_generation(messages: _Messages, operation: _Operation) -> None: + _set_mode(messages, AppMode.CHAT if operation == "more-like-this" else AppMode.COMPLETION) + _error( + messages.request(operation), + status=400, + code="not_completion_app" if operation == "more-like-this" else "not_chat_app", + ) + assert messages.extra_calls == messages.question_calls == messages.generation_calls == [] + + +def test_more_like_this_preserves_blocking_response_and_does_not_update_usage(messages: _Messages) -> None: + _set_mode(messages, AppMode.COMPLETION) + message_id = str(uuid4()) + response = messages.request("more-like-this", message_id=message_id) + assert response.status_code == 200 + assert response.get_json() == _BLOCKING + assert dict(response.headers) == { + "Content-Type": "application/json; charset=utf-8", + "Content-Length": str(len(response.data)), + } + assert messages.generation_calls == [ + (messages.harness.target_app.id, messages.harness.account.id, message_id, False) + ] + messages.assert_unused() + + +@pytest.mark.parametrize("consume_all", [False, True]) +def test_more_like_this_stream_keeps_sse_and_closes_iterator(messages: _Messages, consume_all: bool) -> None: + _set_mode(messages, AppMode.COMPLETION) + closed: list[bool] = [] + + def chunks() -> Generator[str]: + try: + yield 'data: {"answer":"你好"}\n\n' + yield "data: [DONE]\n\n" + finally: + closed.append(True) + + messages.generation_response = chunks() + response = messages.request("more-like-this", query="response_mode=streaming") + assert response.status_code == 200 + assert dict(response.headers) == {"Content-Type": "text/event-stream; charset=utf-8"} + if consume_all: + assert response.data == 'data: {"answer":"你好"}\n\ndata: [DONE]\n\n'.encode() + else: + assert next(iter(response.response)) == 'data: {"answer":"你好"}\n\n'.encode() + assert closed == [] + response.close() + assert closed == [True] + assert messages.generation_calls[0][-1] is True + messages.assert_unused() + + +@pytest.mark.parametrize("query", ["", "response_mode=invalid"]) +def test_more_like_this_invalid_query_uses_shared_422(messages: _Messages, query: str) -> None: + _set_mode(messages, AppMode.COMPLETION) + _error(messages.request("more-like-this", query=query), status=422, code="unprocessable_entity") + assert messages.generation_calls == [] + + +@pytest.mark.parametrize("questions", [["Next question?", "还有吗？"], []]) +def test_suggested_questions_preserves_results_and_empty_fallback(messages: _Messages, questions: list[str]) -> None: + messages.questions = questions + message_id = str(uuid4()) + _assert_json_response( + messages.request("suggested-questions", message_id=message_id), status=200, body={"data": questions} + ) + installed_app, account_id, called_message_id = messages.question_calls[0] + assert installed_app.id == messages.harness.installed_app.id + assert installed_app.app_id == messages.harness.target_app.id + assert installed_app.tenant_id == messages.harness.installed_app.tenant_id + assert installed_app.app_mode == "chat" + assert account_id == messages.harness.account.id + assert called_message_id == message_id + + +@pytest.mark.parametrize("operation", ["more-like-this", "suggested-questions"]) +@pytest.mark.parametrize( + ("failure", "status", "code", "description"), + [ + (ProviderTokenNotInitError("Missing credentials"), 400, "provider_not_initialize", "Missing credentials"), + (QuotaExceededError(), 400, "provider_quota_exceeded", None), + (ModelCurrentlyNotSupportError(), 400, "model_currently_not_support", None), + (InvokeError("Provider rejected input"), 400, "completion_request_error", "Provider rejected input"), + (RuntimeError("Unexpected failure"), 500, "internal_server_error", None), + ], +) +def test_generation_failures_keep_provider_errors_and_context( + messages: _Messages, operation: _Operation, failure: Exception, status: int, code: str, description: str | None +) -> None: + if operation == "more-like-this": + _set_mode(messages, AppMode.COMPLETION) + messages.external_error = failure + _error(messages.request(operation), status=status, code=code, message=description) + assert len(messages.generation_calls) + len(messages.question_calls) == 1 + messages.assert_unused() + + +@pytest.mark.parametrize( + ("operation", "failure", "status", "code"), + [ + ("more-like-this", MessageNotExistsError(), 404, "message_not_found"), + ("more-like-this", MoreLikeThisDisabledError(), 403, "app_more_like_this_disabled"), + ("more-like-this", ValueError("Bad runtime arguments"), 400, "invalid_param"), + ("suggested-questions", MessageNotExistsError(), 404, "message_not_found"), + ("suggested-questions", ConversationNotExistsError(), 404, "conversation_not_found"), + ( + "suggested-questions", + SuggestedQuestionsAfterAnswerDisabledError(), + 403, + "app_suggested_questions_after_answer_disabled", + ), + ], +) +def test_generation_resource_errors_remain_specific( + messages: _Messages, operation: _Operation, failure: Exception, status: int, code: str +) -> None: + if operation == "more-like-this": + _set_mode(messages, AppMode.COMPLETION) + messages.external_error = failure + _error(messages.request(operation), status=status, code=code) + messages.assert_unused() diff --git a/api/tests/unit_tests/controllers/console/explore/test_workflow.py b/api/tests/unit_tests/controllers/console/explore/test_workflow.py index 3c93a4d0261..b3a5e1fe5fc 100644 --- a/api/tests/unit_tests/controllers/console/explore/test_workflow.py +++ b/api/tests/unit_tests/controllers/console/explore/test_workflow.py @@ -65,6 +65,11 @@ class _Runtime: raise self.error return self.response + def generate_more_like_this( + self, *, app_id: str, account_id: str, message_id: str, streaming: bool + ) -> GenerationResponse: + pytest.fail(f"Unexpected more-like-this call: {app_id=}, {account_id=}, {message_id=}, {streaming=}") + def assert_usage_unchanged(self) -> None: with self.session_factory() as session: installation = session.get(InstalledApp, self.installed_app_id) diff --git a/api/tests/unit_tests/repositories/test_installed_app_message_repository.py b/api/tests/unit_tests/repositories/test_installed_app_message_repository.py new file mode 100644 index 00000000000..0af872c65a4 --- /dev/null +++ b/api/tests/unit_tests/repositories/test_installed_app_message_repository.py @@ -0,0 +1,385 @@ +import json +from dataclasses import replace +from datetime import datetime, timedelta +from decimal import Decimal +from typing import Literal +from uuid import uuid4 + +import pytest +from sqlalchemy import Engine, event, inspect, select, update +from sqlalchemy.orm import Session, sessionmaker + +from models.enums import ConversationFromSource, CreatorUserRole, FeedbackFromSource, FeedbackRating +from models.model import App, AppMode, Conversation, InstalledApp, Message, MessageAgentThought, MessageFeedback +from repositories.installed_app_message_repository import SQLAlchemyInstalledAppMessageRepository +from services.errors.conversation import ConversationNotExistsError +from services.errors.message import FirstMessageNotExistsError, MessageNotExistsError +from services.installed_app_access_service import InstalledAppNotFoundError, InstalledAppRef +from services.installed_app_message_service import FeedbackRatingRequiredError, MessagePage, MessageRating + +_ACCOUNT_ID = "11111111-1111-4111-8111-111111111111" +_OWNER_TENANT_ID = "22222222-2222-4222-8222-222222222222" +_TIME = datetime(2026, 9, 8, 12, 30) + + +@pytest.fixture +def installation(sqlite_session_factory: sessionmaker[Session]) -> InstalledAppRef: + with sqlite_session_factory.begin() as session: + app = App(tenant_id=_OWNER_TENANT_ID, name="Shared chat", mode=AppMode.CHAT, enable_site=True, enable_api=True) + session.add(app) + session.flush() + installation = InstalledApp( + tenant_id=str(uuid4()), app_id=app.id, app_owner_tenant_id=app.tenant_id, position=0, is_pinned=False + ) + session.add(installation) + session.flush() + return InstalledAppRef(id=installation.id, tenant_id=installation.tenant_id, app_id=app.id, app_mode="chat") + + +def _conversation(session: Session, installation: InstalledAppRef) -> Conversation: + conversation = Conversation( + app_id=installation.app_id, + mode=AppMode.CHAT, + name="Chat", + inputs={}, + from_source=ConversationFromSource.CONSOLE, + from_account_id=_ACCOUNT_ID, + from_end_user_id=None, + is_deleted=False, + ) + session.add(conversation) + session.flush() + return conversation + + +def _message(session: Session, conversation: Conversation, *, created_at: datetime = _TIME) -> Message: + message = Message( + app_id=conversation.app_id, + conversation_id=conversation.id, + inputs={"empty": "", "list": [], "zero": 0, "false": False, "none": None}, + query="Question", + message={}, + answer="Answer", + message_tokens=2, + message_unit_price=Decimal(0), + answer_tokens=3, + answer_unit_price=Decimal(0), + provider_response_latency=0.5, + total_price=Decimal("0.0000001"), + currency="USD", + from_source=ConversationFromSource.CONSOLE, + from_account_id=_ACCOUNT_ID, + from_end_user_id=None, + created_at=created_at, + ) + session.add(message) + session.flush() + return message + + +def _feedback(session: Session, message: Message, *, source: FeedbackFromSource) -> MessageFeedback: + feedback = MessageFeedback( + app_id=message.app_id, + conversation_id=message.conversation_id, + message_id=message.id, + rating=FeedbackRating.LIKE, + content="Original", + from_source=source, + from_account_id=str(uuid4()), + ) + session.add(feedback) + session.flush() + return feedback + + +def test_page_keeps_cursor_order_and_detaches_complete_message_data( + sqlite_session_factory: sessionmaker[Session], installation: InstalledAppRef +) -> None: + with sqlite_session_factory.begin() as session: + conversation = _conversation(session, installation) + old = _message(session, conversation, created_at=_TIME - timedelta(hours=1)) + middle = _message(session, conversation) + newest = _message(session, conversation, created_at=_TIME + timedelta(hours=1)) + middle.parent_message_id = old.id + middle.message_metadata = json.dumps( + {"retriever_resources": [{"position": 1, "content": "context"}], "usage": {"total_tokens": 5}} + ) + _feedback(session, middle, source=FeedbackFromSource.USER) + admin = _feedback(session, middle, source=FeedbackFromSource.ADMIN) + admin.rating = FeedbackRating.DISLIKE + later = MessageAgentThought( + message_id=middle.id, position=2, created_by_role=CreatorUserRole.ACCOUNT, created_by=_ACCOUNT_ID + ) + earlier = MessageAgentThought( + message_id=middle.id, + message_chain_id=str(uuid4()), + position=1, + thought="Thinking", + answer="Partial answer", + tool="search", + tool_labels_str='{"search":{"en_US":"Search"}}', + tool_input='{"query":"query"}', + observation="Found", + message_files='["file-id"]', + created_by_role=CreatorUserRole.ACCOUNT, + created_by=_ACCOUNT_ID, + ) + session.add_all([later, earlier]) + repository = SQLAlchemyInstalledAppMessageRepository(session_factory=sqlite_session_factory) + page = repository.get_page( + installed_app=installation, account_id=_ACCOUNT_ID, conversation_id=conversation.id, first_id=None, limit=2 + ) + final = repository.get_page( + installed_app=installation, account_id=_ACCOUNT_ID, conversation_id=conversation.id, first_id=middle.id, limit=2 + ) + + assert [item.id for item in page.data] == [middle.id, newest.id] + assert (page.limit, page.has_more) == (2, True) + assert [item.id for item in final.data] == [old.id] + assert final.has_more is False + record = page.data[0] + assert inspect(record, raiseerr=False) is None + assert (record.conversation_id, record.parent_message_id) == (conversation.id, old.id) + assert record.inputs == {"empty": "", "list": [], "zero": 0, "false": False, "none": None} + assert (record.query, record.answer, record.created_at) == ("Question", "Answer", _TIME) + assert record.feedback is not None + assert record.feedback.rating == "like" + assert inspect(record.feedback, raiseerr=False) is None + assert record.retriever_resources == [{"position": 1, "content": "context"}] + assert record.metadata == { + "retriever_resources": [{"position": 1, "content": "context"}], + "usage": {"total_tokens": 5}, + } + assert record.message_files == [] + assert record.extra_contents == [] + assert (record.message_tokens, record.answer_tokens, record.provider_response_latency) == (2, 3, 0.5) + assert (record.total_price, record.currency, record.status, record.error) == ( + Decimal("0.0000001"), + "USD", + "normal", + None, + ) + assert [thought.position for thought in record.agent_thoughts] == [1, 2] + thought = record.agent_thoughts[0] + assert inspect(thought, raiseerr=False) is None + assert (thought.id, thought.message_chain_id, thought.message_id) == ( + earlier.id, + earlier.message_chain_id, + middle.id, + ) + assert (thought.thought, thought.answer, thought.tool) == ("Thinking", "Partial answer", "search") + assert (thought.tool_labels, thought.tool_input, thought.observation) == ( + {"search": {"en_US": "Search"}}, + '{"query":"query"}', + "Found", + ) + assert thought.files == ["file-id"] + assert thought.created_at == earlier.created_at + + +def test_empty_conversation_and_strict_timestamp_ties_preserve_legacy_page_behavior( + sqlite_session_factory: sessionmaker[Session], installation: InstalledAppRef +) -> None: + repository = SQLAlchemyInstalledAppMessageRepository(session_factory=sqlite_session_factory) + assert repository.get_page( + installed_app=installation, account_id=_ACCOUNT_ID, conversation_id="", first_id=str(uuid4()), limit=20 + ) == MessagePage(limit=20, has_more=False, data=()) + with sqlite_session_factory.begin() as session: + conversation = _conversation(session, installation) + _message(session, conversation) + _message(session, conversation) + page = repository.get_page( + installed_app=installation, account_id=_ACCOUNT_ID, conversation_id=conversation.id, first_id=None, limit=1 + ) + assert page.has_more is True + assert len(page.data) == 1 + assert repository.get_page( + installed_app=installation, + account_id=_ACCOUNT_ID, + conversation_id=conversation.id, + first_id=page.data[0].id, + limit=1, + ) == MessagePage(limit=1, has_more=False, data=()) + + +@pytest.mark.parametrize("mismatch", ["app_id", "from_account_id", "from_source", "from_end_user_id", "is_deleted"]) +def test_listing_checks_each_conversation_ownership_predicate( + sqlite_session_factory: sessionmaker[Session], installation: InstalledAppRef, mismatch: str +) -> None: + with sqlite_session_factory.begin() as session: + conversation = _conversation(session, installation) + _message(session, conversation) + value: str | bool = True if mismatch == "is_deleted" else str(uuid4()) + if mismatch == "from_source": + value = ConversationFromSource.API + session.execute(update(Conversation).where(Conversation.id == conversation.id).values({mismatch: value})) + with pytest.raises(ConversationNotExistsError): + SQLAlchemyInstalledAppMessageRepository(session_factory=sqlite_session_factory).get_page( + installed_app=installation, account_id=_ACCOUNT_ID, conversation_id=conversation.id, first_id=None, limit=20 + ) + + +def test_cursor_must_belong_to_the_requested_conversation( + sqlite_session_factory: sessionmaker[Session], installation: InstalledAppRef +) -> None: + with sqlite_session_factory.begin() as session: + conversation = _conversation(session, installation) + other_message = _message(session, _conversation(session, installation)) + repository = SQLAlchemyInstalledAppMessageRepository(session_factory=sqlite_session_factory) + for cursor in (other_message.id, str(uuid4())): + with pytest.raises(FirstMessageNotExistsError): + repository.get_page( + installed_app=installation, + account_id=_ACCOUNT_ID, + conversation_id=conversation.id, + first_id=cursor, + limit=20, + ) + + +@pytest.mark.parametrize("mismatch", ["id", "tenant_id", "app_id", "deleted_app", "deleted_installation"]) +def test_read_and_feedback_revalidate_installation_after_admission( + sqlite_session_factory: sessionmaker[Session], installation: InstalledAppRef, mismatch: str +) -> None: + with sqlite_session_factory.begin() as session: + conversation = _conversation(session, installation) + message = _message(session, conversation) + if mismatch == "deleted_app": + app = session.get(App, installation.app_id) + assert app is not None + session.delete(app) + elif mismatch == "deleted_installation": + installed = session.get(InstalledApp, installation.id) + assert installed is not None + session.delete(installed) + ref = installation + if mismatch == "id": + ref = replace(ref, id=str(uuid4())) + elif mismatch == "tenant_id": + ref = replace(ref, tenant_id=str(uuid4())) + elif mismatch == "app_id": + ref = replace(ref, app_id=str(uuid4())) + repository = SQLAlchemyInstalledAppMessageRepository(session_factory=sqlite_session_factory) + with pytest.raises(InstalledAppNotFoundError): + repository.get_page( + installed_app=ref, account_id=_ACCOUNT_ID, conversation_id=conversation.id, first_id=None, limit=20 + ) + with pytest.raises(InstalledAppNotFoundError): + repository.set_feedback( + installed_app=ref, account_id=_ACCOUNT_ID, message_id=message.id, rating="like", content=None + ) + + +@pytest.mark.parametrize("mismatch", ["app_id", "from_account_id", "from_source", "from_end_user_id"]) +def test_feedback_checks_each_message_ownership_predicate( + sqlite_session_factory: sessionmaker[Session], installation: InstalledAppRef, mismatch: str +) -> None: + with sqlite_session_factory.begin() as session: + message = _message(session, _conversation(session, installation)) + value = ConversationFromSource.API if mismatch == "from_source" else str(uuid4()) + session.execute(update(Message).where(Message.id == message.id).values({mismatch: value})) + repository = SQLAlchemyInstalledAppMessageRepository(session_factory=sqlite_session_factory) + with pytest.raises(MessageNotExistsError): + repository.set_feedback( + installed_app=installation, account_id=_ACCOUNT_ID, message_id=message.id, rating="like", content=None + ) + with sqlite_session_factory() as session: + assert session.scalars(select(MessageFeedback)).all() == [] + + +def test_feedback_create_update_and_delete_preserve_admin_scope_and_owner_tenant( + sqlite_session_factory: sessionmaker[Session], installation: InstalledAppRef +) -> None: + with sqlite_session_factory.begin() as session: + message = _message(session, _conversation(session, installation)) + user_feedback = _feedback(session, message, source=FeedbackFromSource.USER) + repository = SQLAlchemyInstalledAppMessageRepository(session_factory=sqlite_session_factory) + with pytest.raises(FeedbackRatingRequiredError, match=message.id): + repository.set_feedback( + installed_app=installation, account_id=_ACCOUNT_ID, message_id=message.id, rating=None, content=None + ) + created = repository.set_feedback( + installed_app=installation, account_id=_ACCOUNT_ID, message_id=message.id, rating="like", content="" + ) + assert created is not None + assert ( + created.tenant_id, + created.app_id, + created.conversation_id, + created.message_id, + created.account_id, + created.rating, + created.content, + ) == (_OWNER_TENANT_ID, installation.app_id, message.conversation_id, message.id, _ACCOUNT_ID, "like", "") + with sqlite_session_factory.begin() as session: + feedback = message.admin_feedback_with_session(session=session) + assert feedback is not None + assert (feedback.from_source, feedback.from_account_id, feedback.from_end_user_id, feedback.content) == ( + FeedbackFromSource.ADMIN, + _ACCOUNT_ID, + None, + "", + ) + feedback.from_account_id = original_owner = str(uuid4()) + feedback_id = feedback.id + updated = repository.set_feedback( + installed_app=installation, account_id=_ACCOUNT_ID, message_id=message.id, rating="dislike", content=None + ) + assert updated is not None + assert (updated.rating, updated.content) == ("dislike", None) + with sqlite_session_factory() as session: + feedback = session.get(MessageFeedback, feedback_id) + assert feedback is not None + assert (feedback.rating, feedback.content, feedback.from_account_id) == ( + FeedbackRating.DISLIKE, + None, + original_owner, + ) + assert ( + repository.set_feedback( + installed_app=installation, account_id=_ACCOUNT_ID, message_id=message.id, rating=None, content=None + ) + is None + ) + with sqlite_session_factory() as session: + remaining = session.scalars(select(MessageFeedback)).all() + assert [feedback.id for feedback in remaining] == [user_feedback.id] + assert (remaining[0].rating, remaining[0].content) == (FeedbackRating.LIKE, "Original") + + +@pytest.mark.parametrize("operation", ["create", "update", "delete"]) +def test_failed_feedback_commit_rolls_back_and_preserves_original_error( + sqlite_session_factory: sessionmaker[Session], + sqlite_engine: Engine, + installation: InstalledAppRef, + operation: Literal["create", "update", "delete"], +) -> None: + with sqlite_session_factory.begin() as session: + message = _message(session, _conversation(session, installation)) + if operation != "create": + _feedback(session, message, source=FeedbackFromSource.ADMIN) + failing_factory = sessionmaker(bind=sqlite_engine) + failure = RuntimeError("database commit unavailable") + + @event.listens_for(failing_factory, "before_commit") + def fail_commit(session: Session) -> None: + session.flush() + raise failure + + rating: MessageRating | None = None if operation == "delete" else "dislike" + with pytest.raises(RuntimeError) as error: + SQLAlchemyInstalledAppMessageRepository(session_factory=failing_factory).set_feedback( + installed_app=installation, + account_id=_ACCOUNT_ID, + message_id=message.id, + rating=rating, + content="Discarded", + ) + assert error.value is failure + with sqlite_session_factory() as session: + feedback = message.admin_feedback_with_session(session=session) + if operation == "create": + assert feedback is None + else: + assert feedback is not None + assert (feedback.rating, feedback.content) == (FeedbackRating.LIKE, "Original") diff --git a/api/tests/unit_tests/services/test_installed_app_generation_adapters.py b/api/tests/unit_tests/services/test_installed_app_generation_adapters.py index 451c7dfe118..d1e2b3e87ca 100644 --- a/api/tests/unit_tests/services/test_installed_app_generation_adapters.py +++ b/api/tests/unit_tests/services/test_installed_app_generation_adapters.py @@ -1,6 +1,7 @@ import json from collections.abc import Callable, Generator, Mapping from dataclasses import dataclass, field +from inspect import GEN_CLOSED, getgeneratorstate from typing import cast, override from unittest.mock import MagicMock from uuid import uuid4 @@ -17,17 +18,20 @@ from core.app.features.rate_limiting.rate_limit import RateLimit, RateLimitGener from enums import DeploymentEdition from extensions.ext_database import db from libs.broadcast_channel.channel import BroadcastChannel, Subscription, SupportsPreparedSubscription, Topic -from models import Account, App, AppMode, AppModelConfig, Conversation, Workflow +from models import Account, App, AppMode, AppModelConfig, Conversation, Message, Workflow from models.enums import ConversationFromSource from models.workflow import WorkflowType from services.account_errors import AccountNotFoundError from services.app_definition_query_service import AppDefinitionUnavailableError +from services.errors.app import MoreLikeThisDisabledError from services.errors.conversation import ConversationNotExistsError +from services.errors.message import MessageNotExistsError from services.installed_app_generation_adapters import AppGenerateServiceRuntime from services.installed_app_generation_service import GenerationResponse _ARGS: dict[str, object] = {"inputs": {"count": 0}, "query": "hello", "auto_generate_name": False} _WORKFLOW_ARGS: dict[str, object] = {"inputs": {"count": 0}, "files": []} +_MESSAGE_ID = str(uuid4()) @dataclass @@ -80,19 +84,19 @@ def harness(sqlite_session_factory: sessionmaker[Session]) -> _RuntimeHarness: def _patch_generation( monkeypatch: pytest.MonkeyPatch, harness: _RuntimeHarness, - generate: Callable[[Session], GenerationResponse], + generate: Callable[[Session], GenerationResponse | Generator[Mapping[str, object] | str, None, None]], *, streaming: bool, + more_like_this: bool = False, ) -> None: - def legacy_generate( + def assert_context( *, session: Session, app_model: App, user: Account, - args: Mapping[str, object], invoke_from: InvokeFrom, streaming: bool, - ) -> GenerationResponse: + ) -> GenerationResponse | Generator[Mapping[str, object] | str, None, None]: assert len(harness.closed_sessions) == 1 read_session = harness.closed_sessions[0] assert not read_session.in_transaction() @@ -103,13 +107,62 @@ def _patch_generation( assert inspect(user).detached assert (app_model.id, user.id) == (harness.app_id, harness.account_id) assert (app_model.name, user.name) == ("Original app", "Viewer") - assert args == _ARGS assert invoke_from == InvokeFrom.EXPLORE assert streaming is expected_streaming return generate(session) + def legacy_generate( + *, + session: Session, + app_model: App, + user: Account, + args: Mapping[str, object], + invoke_from: InvokeFrom, + streaming: bool, + ) -> GenerationResponse | Generator[Mapping[str, object] | str, None, None]: + assert args == _ARGS + return assert_context( + session=session, app_model=app_model, user=user, invoke_from=invoke_from, streaming=streaming + ) + + def legacy_generate_more_like_this( + *, + session: Session, + app_model: App, + user: Account, + message_id: str, + invoke_from: InvokeFrom, + streaming: bool, + ) -> GenerationResponse | Generator[Mapping[str, object] | str, None, None]: + assert message_id == _MESSAGE_ID + return assert_context( + session=session, app_model=app_model, user=user, invoke_from=invoke_from, streaming=streaming + ) + expected_streaming = streaming - monkeypatch.setattr(generation_module.AppGenerateService, "generate", legacy_generate) + if more_like_this: + monkeypatch.setattr( + generation_module.AppGenerateService, "generate_more_like_this", legacy_generate_more_like_this + ) + else: + monkeypatch.setattr(generation_module.AppGenerateService, "generate", legacy_generate) + + +def _invoke_generation( + harness: _RuntimeHarness, + *, + streaming: bool, + more_like_this: bool, + app_id: str | None = None, + account_id: str | None = None, +) -> GenerationResponse: + app_id = app_id or harness.app_id + account_id = account_id or harness.account_id + if more_like_this: + return harness.runtime.generate_more_like_this( + app_id=app_id, account_id=account_id, message_id=_MESSAGE_ID, streaming=streaming + ) + return harness.runtime.generate(app_id=app_id, account_id=account_id, args=_ARGS, streaming=streaming) @dataclass @@ -127,10 +180,12 @@ def _rate_limited_stream(source: Generator[str, None, None], rate: _RateLimitExi return RateLimitGenerator(rate_limit=cast(RateLimit, rate), generator=source, request_id="request-1") +@pytest.mark.parametrize("more_like_this", [False, True]) def test_runtime_loads_detached_entities_then_commits_work_and_returns_the_same_mapping( harness: _RuntimeHarness, monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessionmaker[Session], + more_like_this: bool, ) -> None: response: dict[str, object] = {"answer": "", "metadata": {"tokens": 0}} @@ -138,8 +193,8 @@ def test_runtime_loads_detached_entities_then_commits_work_and_returns_the_same_ session.execute(update(App).where(App.id == harness.app_id).values(name="Generated app")) return response - _patch_generation(monkeypatch, harness, generate, streaming=False) - result = harness.runtime.generate(app_id=harness.app_id, account_id=harness.account_id, args=_ARGS, streaming=False) + _patch_generation(monkeypatch, harness, generate, streaming=False, more_like_this=more_like_this) + result = _invoke_generation(harness, streaming=False, more_like_this=more_like_this) assert result is response assert len(harness.closed_sessions) == 2 @@ -194,10 +249,55 @@ def test_runtime_preserves_stream_identity_and_original_close_lifecycle( assert events == ["start", "close"] +@pytest.mark.parametrize("finish", ["complete", "cancel", "error", "close_before_iteration"]) +def test_more_like_this_serializes_raw_events_and_closes_source( + harness: _RuntimeHarness, monkeypatch: pytest.MonkeyPatch, finish: str +) -> None: + events: list[str] = [] + failure = RuntimeError("more-like-this stream failed") + + def source() -> Generator[Mapping[str, object] | str, None, None]: + assert len(harness.closed_sessions) == 2 + events.append("start") + try: + yield {"event": "message", "answer": "Hello"} + if finish == "error": + raise failure + yield "ping" + finally: + events.append("close") + + stream = source() + _patch_generation(monkeypatch, harness, lambda _session: stream, streaming=True, more_like_this=True) + result = _invoke_generation(harness, streaming=True, more_like_this=True) + + assert not isinstance(result, Mapping) + assert events == [] + assert harness.committed_sessions == [harness.closed_sessions[1]] + if finish != "close_before_iteration": + assert next(result) == 'data: {"event":"message","answer":"Hello"}\n\n' + if finish == "complete": + assert list(result) == ["event: ping\n\n"] + elif finish == "error": + with pytest.raises(RuntimeError) as raised: + next(result) + assert raised.value is failure + else: + result.close() + + result.close() + result.close() + assert list(result) == [] + assert getgeneratorstate(stream) == GEN_CLOSED + assert events == ([] if finish == "close_before_iteration" else ["start", "close"]) + + +@pytest.mark.parametrize("more_like_this", [False, True]) def test_streaming_preparation_errors_raise_eagerly_and_rollback_runtime_writes( harness: _RuntimeHarness, monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessionmaker[Session], + more_like_this: bool, ) -> None: failure = ValueError("query must be a string") @@ -205,9 +305,9 @@ def test_streaming_preparation_errors_raise_eagerly_and_rollback_runtime_writes( session.execute(update(App).where(App.id == harness.app_id).values(name="Uncommitted app")) raise failure - _patch_generation(monkeypatch, harness, generate, streaming=True) + _patch_generation(monkeypatch, harness, generate, streaming=True, more_like_this=more_like_this) with pytest.raises(ValueError) as raised: - harness.runtime.generate(app_id=harness.app_id, account_id=harness.account_id, args=_ARGS, streaming=True) + _invoke_generation(harness, streaming=True, more_like_this=more_like_this) assert raised.value is failure assert len(harness.closed_sessions) == 2 @@ -219,19 +319,21 @@ def test_streaming_preparation_errors_raise_eagerly_and_rollback_runtime_writes( @pytest.mark.parametrize("missing", ["app", "account"]) +@pytest.mark.parametrize("more_like_this", [False, True]) def test_missing_runtime_entities_keep_precise_errors_and_do_not_enter_generation( - harness: _RuntimeHarness, monkeypatch: pytest.MonkeyPatch, missing: str + harness: _RuntimeHarness, monkeypatch: pytest.MonkeyPatch, missing: str, more_like_this: bool ) -> None: def unexpected_generation(**_kwargs: object) -> GenerationResponse: pytest.fail("Missing app or account must not reach generation") monkeypatch.setattr(generation_module.AppGenerateService, "generate", unexpected_generation) + monkeypatch.setattr(generation_module.AppGenerateService, "generate_more_like_this", unexpected_generation) app_id = str(uuid4()) if missing == "app" else harness.app_id account_id = str(uuid4()) if missing == "account" else harness.account_id error_type = AppDefinitionUnavailableError if missing == "app" else AccountNotFoundError with pytest.raises(error_type, match=app_id if missing == "app" else account_id): - harness.runtime.generate(app_id=app_id, account_id=account_id, args=_ARGS, streaming=True) + _invoke_generation(harness, app_id=app_id, account_id=account_id, streaming=True, more_like_this=more_like_this) assert len(harness.closed_sessions) == 1 assert harness.committed_sessions == [] @@ -239,12 +341,14 @@ def test_missing_runtime_entities_keep_precise_errors_and_do_not_enter_generatio @pytest.mark.parametrize("failure_phase", ["commit", "session_close"]) @pytest.mark.parametrize("stream_close_fails", [False, True]) +@pytest.mark.parametrize("more_like_this", [False, True]) def test_failure_before_stream_handoff_closes_stream_without_masking_primary_error( harness: _RuntimeHarness, monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessionmaker[Session], failure_phase: str, stream_close_fails: bool, + more_like_this: bool, ) -> None: failure = RuntimeError(f"runtime {failure_phase} failed") rate = _RateLimitExit(failure=ValueError("stream close failed") if stream_close_fails else None) @@ -266,9 +370,9 @@ def test_failure_before_stream_handoff_closes_stream_without_masking_primary_err session.info["close_failure"] = failure return stream - _patch_generation(monkeypatch, harness, generate, streaming=True) + _patch_generation(monkeypatch, harness, generate, streaming=True, more_like_this=more_like_this) with pytest.raises(RuntimeError) as raised: - harness.runtime.generate(app_id=harness.app_id, account_id=harness.account_id, args=_ARGS, streaming=True) + _invoke_generation(harness, streaming=True, more_like_this=more_like_this) assert raised.value is failure assert stream.closed is True @@ -279,6 +383,113 @@ def test_failure_before_stream_handoff_closes_stream_without_masking_primary_err assert stored.name == ("Original app" if failure_phase == "commit" else "Generated app") +def _seed_more_like_this_message(harness: _RuntimeHarness, session_factory: sessionmaker[Session]) -> str: + with session_factory.begin() as session: + conversation = Conversation( + app_id=harness.app_id, + mode=AppMode.COMPLETION, + name="Earlier completion", + inputs={}, + from_source=ConversationFromSource.CONSOLE, + from_account_id=harness.account_id, + ) + session.add(conversation) + session.flush() + message = Message( + app_id=harness.app_id, + conversation_id=conversation.id, + inputs={"count": 0}, + query="Earlier query", + message={}, + answer="Earlier answer", + message_unit_price=0, + answer_unit_price=0, + currency="USD", + from_source=ConversationFromSource.CONSOLE, + from_account_id=harness.account_id, + ) + session.add(message) + session.flush() + return message.id + + +@pytest.mark.parametrize("inaccessible", ["missing", "other_app", "other_account", "api_source", "end_user"]) +def test_more_like_this_checks_real_message_ownership_before_feature_config( + harness: _RuntimeHarness, sqlite_session_factory: sessionmaker[Session], inaccessible: str +) -> None: + message_id = _seed_more_like_this_message(harness, sqlite_session_factory) + with sqlite_session_factory.begin() as session: + message = session.get(Message, message_id) + assert message is not None + if inaccessible == "missing": + message_id = str(uuid4()) + elif inaccessible == "other_app": + message.app_id = str(uuid4()) + elif inaccessible == "other_account": + message.from_account_id = str(uuid4()) + elif inaccessible == "api_source": + message.from_source = ConversationFromSource.API + else: + message.from_end_user_id = str(uuid4()) + + with pytest.raises(MessageNotExistsError): + harness.runtime.generate_more_like_this( + app_id=harness.app_id, account_id=harness.account_id, message_id=message_id, streaming=True + ) + + assert len(harness.closed_sessions) == 2 + assert harness.committed_sessions == [] + + +@pytest.mark.parametrize("current_config", ["missing", "null", "empty", "disabled"]) +def test_more_like_this_disabled_config_fails_eagerly_before_history_lookup( + harness: _RuntimeHarness, sqlite_session_factory: sessionmaker[Session], current_config: str +) -> None: + message_id = _seed_more_like_this_message(harness, sqlite_session_factory) + if current_config != "missing": + with sqlite_session_factory.begin() as session: + app = session.get(App, harness.app_id) + assert app is not None + feature_config = {"null": None, "empty": "{}", "disabled": '{"enabled":false}'}[current_config] + config = AppModelConfig(app_id=app.id, more_like_this=feature_config) + session.add(config) + app.app_model_config_id = config.id + + with pytest.raises(MoreLikeThisDisabledError): + harness.runtime.generate_more_like_this( + app_id=harness.app_id, account_id=harness.account_id, message_id=message_id, streaming=True + ) + + assert len(harness.closed_sessions) == 2 + assert harness.committed_sessions == [] + + +@pytest.mark.parametrize("missing_config_id", [None, "00000000-0000-0000-0000-000000000001"]) +def test_more_like_this_requires_historical_config_even_when_current_config_is_enabled( + harness: _RuntimeHarness, sqlite_session_factory: sessionmaker[Session], missing_config_id: str | None +) -> None: + message_id = _seed_more_like_this_message(harness, sqlite_session_factory) + with sqlite_session_factory.begin() as session: + app = session.get(App, harness.app_id) + assert app is not None + config = AppModelConfig(app_id=app.id, more_like_this='{"enabled":true}') + session.add(config) + app.app_model_config_id = config.id + message = session.get(Message, message_id) + assert message is not None + conversation = session.get(Conversation, message.conversation_id) + assert conversation is not None + conversation.app_model_config_id = missing_config_id + + with pytest.raises(ValueError, match="Message app_model_config is None"): + harness.runtime.generate_more_like_this( + app_id=harness.app_id, account_id=harness.account_id, message_id=message_id, streaming=True + ) + + assert len(harness.closed_sessions) == 2 + assert harness.committed_sessions == [] + + @pytest.mark.parametrize("legacy_agent", [False, True]) def test_runtime_keeps_shared_completion_and_legacy_agent_dispatch( harness: _RuntimeHarness, diff --git a/api/tests/unit_tests/services/test_installed_app_message_adapters.py b/api/tests/unit_tests/services/test_installed_app_message_adapters.py new file mode 100644 index 00000000000..83dafad0ed2 --- /dev/null +++ b/api/tests/unit_tests/services/test_installed_app_message_adapters.py @@ -0,0 +1,392 @@ +import json +from collections.abc import Iterator +from dataclasses import dataclass, replace +from datetime import datetime, timedelta +from decimal import Decimal +from typing import cast, override +from unittest.mock import MagicMock +from uuid import uuid4 + +import pytest +from flask import Flask +from sqlalchemy import update +from sqlalchemy.orm import Session, sessionmaker + +import services.message_service as message_module +from core import telemetry +from core.credit_usage import CreditUsageAppType +from core.model_context import get_credit_usage_metadata, use_credit_usage_metadata +from core.model_manager import ModelInstance, ModelManager +from core.ops.ops_trace_manager import TraceQueueManager +from core.telemetry import FeedbackCreatedEvent +from core.workflow.nodes.human_input.entities import FormDefinition, UserActionConfig +from extensions.ext_database import db +from models import Account, App, AppMode, AppModelConfig, Conversation, InstalledApp, Message +from models.enums import ConversationFromSource +from models.execution_extra_content import HumanInputContent +from models.human_input import HumanInputForm +from services.account_errors import AccountNotFoundError +from services.errors.conversation import ConversationNotExistsError +from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError +from services.installed_app_access_service import InstalledAppNotFoundError, InstalledAppRef +from services.installed_app_message_adapters import InstalledAppMessageRuntime, emit_installed_app_feedback +from services.installed_app_message_service import MessageFeedbackEvent + + +@dataclass +class _Harness: + runtime: InstalledAppMessageRuntime + installation: InstalledAppRef + account_id: str + message_id: str + conversation_id: str + config_id: str + read_sessions: list[Session] + legacy_sessions: list[Session] + model_manager: MagicMock + + def suggested_questions(self) -> list[str]: + return self.runtime.get_suggested_questions( + installed_app=self.installation, account_id=self.account_id, message_id=self.message_id + ) + + +@pytest.fixture +def harness(sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch) -> Iterator[_Harness]: + with sqlite_session_factory.begin() as session: + app = App(tenant_id=str(uuid4()), name="Original", mode=AppMode.CHAT, enable_site=True, enable_api=True) + account = Account(name="Viewer", email="message-viewer@example.com") + session.add_all([app, account]) + session.flush() + installed = InstalledApp( + tenant_id=str(uuid4()), app_id=app.id, app_owner_tenant_id=app.tenant_id, position=0, is_pinned=False + ) + config = AppModelConfig( + app_id=app.id, + suggested_questions_after_answer=json.dumps( + { + "enabled": True, + "prompt": "Ask a follow-up", + "model": {"provider": "vendor", "name": "question-model"}, + } + ), + ) + session.add_all([installed, config]) + session.flush() + app.app_model_config_id = config.id + conversation = Conversation( + app_id=app.id, + app_model_config_id=config.id, + mode=AppMode.CHAT, + name="Conversation", + inputs={}, + from_source=ConversationFromSource.CONSOLE, + from_account_id=account.id, + from_end_user_id=None, + is_deleted=False, + ) + session.add(conversation) + session.flush() + message = Message( + app_id=app.id, + conversation_id=conversation.id, + inputs={}, + query="Original question", + answer="Original answer", + message={}, + message_unit_price=Decimal(0), + answer_unit_price=Decimal(0), + currency="USD", + from_source=ConversationFromSource.CONSOLE, + from_account_id=account.id, + from_end_user_id=None, + ) + session.add(message) + session.flush() + + read_sessions: list[Session] = [] + legacy_sessions: list[Session] = [] + + class TrackedSession(Session): + @override + def close(self) -> None: + super().close() + read_sessions.append(self) + + factory = sessionmaker(bind=sqlite_session_factory.kw["bind"], class_=TrackedSession, expire_on_commit=True) + runtime_app = Flask(__name__) + runtime_app.config["SQLALCHEMY_DATABASE_URI"] = str(sqlite_session_factory.kw["bind"].url) + db.init_app(runtime_app) + + @runtime_app.teardown_appcontext + def capture_legacy_session(_error: BaseException | None) -> None: + # Registered after db.init_app, so this runs before its session removal. + legacy_sessions.append(db.session()) + + manager = MagicMock(spec=ModelManager) + model = MagicMock(spec=ModelInstance) + model.get_llm_num_tokens.return_value = 4 + manager.get_default_model_instance.return_value = model + + def provider(*, tenant_id: str) -> ModelManager: + assert tenant_id == app.tenant_id + return manager + + monkeypatch.setattr(message_module.ModelManager, "for_tenant", provider) + with runtime_app.app_context(): + try: + yield _Harness( + runtime=InstalledAppMessageRuntime(session_factory=cast(sessionmaker[Session], factory)), + installation=InstalledAppRef( + id=installed.id, tenant_id=installed.tenant_id, app_id=app.id, app_mode="chat" + ), + account_id=account.id, + message_id=message.id, + conversation_id=conversation.id, + config_id=config.id, + read_sessions=read_sessions, + legacy_sessions=legacy_sessions, + model_manager=manager, + ) + finally: + db.session.remove() + db.engine.dispose() + + +@pytest.mark.parametrize("fail", [False, True]) +def test_suggestions_keep_real_history_and_release_isolated_legacy_session_on_success_or_failure( + harness: _Harness, monkeypatch: pytest.MonkeyPatch, fail: bool +) -> None: + outer = db.session() + app = outer.get(App, harness.installation.app_id) + assert app is not None + app.name = "Pending outer edit" + owner_tenant_id = app.tenant_id + failure = RuntimeError("Suggestion provider unavailable") + calls: list[str] = [] + tracing = MagicMock(spec=TraceQueueManager) + + def trace_manager(*, app_id: str) -> TraceQueueManager: + assert app_id == harness.installation.app_id + return tracing + + def generate( + *, tenant_id: str, histories: str, instruction_prompt: str | None = None, model_config: object | None = None + ) -> list[str]: + assert tenant_id == owner_tenant_id + assert histories == "Human: Original question\nAssistant: Original answer" + assert instruction_prompt == "Ask a follow-up" + assert model_config == {"provider": "vendor", "name": "question-model"} + assert len(harness.read_sessions) == 1 + assert not harness.read_sessions[0].in_transaction() + assert not harness.read_sessions[0].identity_map + assert db.session() is not outer + assert db.session().in_transaction() + assert get_credit_usage_metadata() == { + "app_type": CreditUsageAppType.CHATBOT, + "request_id": "suggestions-request", + } + calls.append(histories) + if fail: + raise failure + return ["Follow-up one?", "Follow-up two?"] + + monkeypatch.setattr(message_module.LLMGenerator, "generate_suggested_questions_after_answer", generate) + monkeypatch.setattr(message_module, "TraceQueueManager", trace_manager) + previous_metadata = get_credit_usage_metadata() + with use_credit_usage_metadata({"request_id": "suggestions-request"}): + inherited_metadata = get_credit_usage_metadata() + if fail: + with pytest.raises(RuntimeError) as error: + harness.suggested_questions() + assert error.value is failure + else: + assert harness.suggested_questions() == ["Follow-up one?", "Follow-up two?"] + assert get_credit_usage_metadata() == inherited_metadata + assert get_credit_usage_metadata() == previous_metadata + assert len(calls) == 1 + assert len(harness.legacy_sessions) == 1 + legacy = harness.legacy_sessions[0] + assert legacy is not outer + assert not legacy.in_transaction() + assert not legacy.identity_map + assert db.session() is outer + assert outer.in_transaction() + assert app in outer.dirty + assert app.name == "Pending outer edit" + assert tracing.add_trace_task.call_count == (0 if fail else 1) + + +@pytest.mark.parametrize("case", ["disabled", "missing_workflow", "history_provider_failure"]) +def test_suggestions_keep_disabled_feature_and_missing_workflow_or_model_behavior( + harness: _Harness, sqlite_session_factory: sessionmaker[Session], case: str +) -> None: + with sqlite_session_factory.begin() as session: + if case == "disabled": + config = session.get(AppModelConfig, harness.config_id) + assert config is not None + config.suggested_questions_after_answer = '{"enabled":false}' + elif case == "missing_workflow": + app = session.get(App, harness.installation.app_id) + assert app is not None + app.mode = AppMode.ADVANCED_CHAT + else: + harness.model_manager.get_default_model_instance.side_effect = RuntimeError("No history model") + if case == "disabled": + with pytest.raises(SuggestedQuestionsAfterAnswerDisabledError): + harness.suggested_questions() + else: + assert harness.suggested_questions() == [] + assert len(harness.legacy_sessions) == 1 + assert not harness.legacy_sessions[0].in_transaction() + assert not harness.legacy_sessions[0].identity_map + + +@pytest.mark.parametrize("entity", ["message", "conversation"]) +@pytest.mark.parametrize("mismatch", ["app_id", "from_account_id", "from_source", "from_end_user_id"]) +def test_suggestions_preserve_each_message_and_conversation_ownership_predicate( + harness: _Harness, sqlite_session_factory: sessionmaker[Session], entity: str, mismatch: str +) -> None: + value = ConversationFromSource.API if mismatch == "from_source" else str(uuid4()) + with sqlite_session_factory.begin() as session: + if entity == "message": + session.execute(update(Message).where(Message.id == harness.message_id).values({mismatch: value})) + else: + session.execute( + update(Conversation).where(Conversation.id == harness.conversation_id).values({mismatch: value}) + ) + expected = MessageNotExistsError if entity == "message" else ConversationNotExistsError + with pytest.raises(expected): + harness.suggested_questions() + harness.model_manager.get_default_model_instance.assert_not_called() + assert len(harness.legacy_sessions) == 1 + assert not harness.legacy_sessions[0].in_transaction() + + +def test_suggestions_reject_deleted_conversation( + harness: _Harness, sqlite_session_factory: sessionmaker[Session] +) -> None: + with sqlite_session_factory.begin() as session: + session.execute(update(Conversation).where(Conversation.id == harness.conversation_id).values(is_deleted=True)) + with pytest.raises(ConversationNotExistsError): + harness.suggested_questions() + + +@pytest.mark.parametrize("missing", ["installation", "app", "account", "viewer_tenant"]) +def test_missing_admission_entities_fail_before_legacy_runtime( + harness: _Harness, sqlite_session_factory: sessionmaker[Session], missing: str +) -> None: + with sqlite_session_factory.begin() as session: + if missing == "installation": + entity = session.get(InstalledApp, harness.installation.id) + assert entity is not None + session.delete(entity) + elif missing == "app": + app = session.get(App, harness.installation.app_id) + assert app is not None + session.delete(app) + elif missing == "account": + account = session.get(Account, harness.account_id) + assert account is not None + session.delete(account) + else: + harness.installation = replace(harness.installation, tenant_id=str(uuid4())) + expected = AccountNotFoundError if missing == "account" else InstalledAppNotFoundError + with pytest.raises(expected): + harness.suggested_questions() + assert len(harness.read_sessions) == 1 + assert harness.legacy_sessions == [] + + +def test_extra_contents_use_real_repository_preserve_message_order_and_omit_none( + harness: _Harness, sqlite_session_factory: sessionmaker[Session] +) -> None: + assert harness.runtime.get_extra_contents(message_ids=[]) == {} + assert harness.read_sessions == [] + second_message_id = str(uuid4()) + absent_message_id = str(uuid4()) + expiration = datetime(2026, 9, 10) + with sqlite_session_factory.begin() as session: + for index, (message_id, title) in enumerate([(second_message_id, "Second"), (harness.message_id, "First")]): + definition = FormDefinition( + form_content="Approve?", + inputs=[], + user_actions=[UserActionConfig(id="approve", title="Approve")], + rendered_content="Approve?", + expiration_time=expiration, + node_title=title, + display_in_ui=True, + ) + form = HumanInputForm( + tenant_id=harness.installation.tenant_id, + app_id=harness.installation.app_id, + workflow_run_id=str(uuid4()), + node_id=f"node-{index}", + form_definition=definition.model_dump_json(), + rendered_content="Approve?", + expiration_time=expiration, + ) + session.add(form) + session.flush() + content = HumanInputContent.new( + workflow_run_id=form.workflow_run_id or "", form_id=form.id, message_id=message_id + ) + content.created_at = expiration - timedelta(days=index) + session.add(content) + contents = harness.runtime.get_extra_contents( + message_ids=[harness.message_id, absent_message_id, second_message_id] + ) + assert list(contents) == [harness.message_id, absent_message_id, second_message_id] + assert contents[absent_message_id] == [] + for message_id, title in [(harness.message_id, "First"), (second_message_id, "Second")]: + assert len(contents[message_id]) == 1 + content = contents[message_id][0] + assert content["type"] == "human_input" + assert content["submitted"] is False + assert "form_submission_data" not in content + definition = content["form_definition"] + assert isinstance(definition, dict) + assert definition["node_title"] == title + assert "form_token" not in definition + assert len(harness.read_sessions) == 1 + assert not harness.read_sessions[0].in_transaction() + assert not harness.read_sessions[0].identity_map + + +@pytest.mark.parametrize("fail", [False, True]) +def test_feedback_telemetry_preserves_event_fields_and_suppresses_sink_failure( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, fail: bool +) -> None: + events: list[FeedbackCreatedEvent] = [] + + def emit(event: FeedbackCreatedEvent) -> None: + events.append(event) + if fail: + raise RuntimeError("Telemetry unavailable") + + monkeypatch.setattr(telemetry, "emit", emit) + emit_installed_app_feedback( + feedback=MessageFeedbackEvent( + tenant_id="owner-tenant", + app_id="app", + conversation_id="conversation", + message_id="message", + account_id="account", + rating="dislike", + content="", + ) + ) + assert len(events) == 1 + assert events[0].context.tenant_id == "owner-tenant" + assert events[0].payload == { + "message_id": "message", + "app_id": "app", + "conversation_id": "conversation", + "from_end_user_id": None, + "from_account_id": "account", + "rating": "dislike", + "from_source": "admin", + "content": "", + } + if fail: + assert "Failed to emit feedback telemetry for message message" in caplog.text diff --git a/api/tests/unit_tests/services/test_installed_app_message_service.py b/api/tests/unit_tests/services/test_installed_app_message_service.py new file mode 100644 index 00000000000..d02a449f414 --- /dev/null +++ b/api/tests/unit_tests/services/test_installed_app_message_service.py @@ -0,0 +1,226 @@ +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field, replace +from datetime import datetime +from decimal import Decimal + +import pytest +from pydantic import JsonValue + +from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError +from services.installed_app_access_service import InstalledAppRef +from services.installed_app_message_service import ( + InstalledAppMessageService, + MessageFeedbackEvent, + MessageNotChatAppError, + MessagePage, + MessageRating, + MessageRecord, +) + +_REF = InstalledAppRef(id="installed", app_id="app", tenant_id="viewer-workspace", app_mode="chat") + + +def _message(message_id: str) -> MessageRecord: + return MessageRecord( + id=message_id, + conversation_id="conversation", + parent_message_id=None, + inputs={"empty": "", "zero": 0, "none": None, "list": []}, + query="Question", + answer="Answer", + feedback=None, + retriever_resources=[], + created_at=datetime(2026, 9, 8), + agent_thoughts=[], + message_files=[], + message_tokens=2, + answer_tokens=3, + provider_response_latency=0, + total_price=Decimal(0), + currency="USD", + status="normal", + error=None, + metadata={"usage": {"total_tokens": 5}}, + ) + + +@dataclass +class _Store: + page: MessagePage = field( + default_factory=lambda: MessagePage(limit=2, has_more=True, data=(_message("first"), _message("second"))) + ) + events: list[str] = field(default_factory=list) + write_error: Exception | None = None + rating: MessageRating | None = None + + def get_page( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + conversation_id: str, + first_id: str | None, + limit: int, + ) -> MessagePage: + assert (installed_app, account_id, conversation_id, first_id, limit) == ( + _REF, + "account", + "conversation", + None, + 2, + ) + self.events.append("read closed") + return self.page + + def set_feedback( + self, + *, + installed_app: InstalledAppRef, + account_id: str, + message_id: str, + rating: MessageRating | None, + content: str | None, + ) -> MessageFeedbackEvent | None: + if self.write_error is not None: + raise self.write_error + self.rating = rating + self.events.append("write committed and closed") + if rating is None: + return None + return MessageFeedbackEvent( + tenant_id="owner-workspace", + app_id=installed_app.app_id, + conversation_id="conversation", + message_id=message_id, + account_id=account_id, + rating=rating, + content=content, + ) + + +@dataclass +class _Callbacks: + store: _Store + queried_ids: list[tuple[str, ...]] = field(default_factory=list) + feedback: list[MessageFeedbackEvent] = field(default_factory=list) + suggestion_error: Exception | None = None + + def get_extra_contents(self, *, message_ids: Sequence[str]) -> Mapping[str, list[dict[str, JsonValue]]]: + assert self.store.events == ["read closed"] + self.store.events.append("extra contents") + self.queried_ids.append(tuple(message_ids)) + return {"first": [{"type": "workflow", "workflow_run_id": "run"}], "not-in-page": [{"type": "ignored"}]} + + def suggested_questions(self, *, installed_app: InstalledAppRef, account_id: str, message_id: str) -> list[str]: + assert installed_app.app_id == _REF.app_id + assert (account_id, message_id) == ("account", "message") + self.store.events.append("suggested questions") + if self.suggestion_error is not None: + raise self.suggestion_error + return ["Next question?"] + + def emit_feedback(self, *, feedback: MessageFeedbackEvent) -> None: + assert self.store.events[-1] == "write committed and closed" + assert self.store.rating == feedback.rating + self.store.events.append("emit feedback") + self.feedback.append(feedback) + + def service(self) -> InstalledAppMessageService: + return InstalledAppMessageService( + messages=self.store, + get_extra_contents=self.get_extra_contents, + suggested_questions=self.suggested_questions, + emit_feedback=self.emit_feedback, + ) + + +def test_list_batches_extra_contents_after_read_and_preserves_detached_page() -> None: + state = _Callbacks(store=_Store()) + page = state.service().get_page( + installed_app=_REF, account_id="account", conversation_id="conversation", first_id=None, limit=2 + ) + + assert state.queried_ids == [("first", "second")] + assert (page.limit, page.has_more) == (2, True) + assert [item.id for item in page.data] == ["first", "second"] + assert page.data[0].extra_contents == [{"type": "workflow", "workflow_run_id": "run"}] + assert page.data[1].extra_contents == [] + assert replace(page.data[0], extra_contents=[]) == state.store.page.data[0] + assert state.store.page.data[0].extra_contents == [] + + +def test_empty_page_skips_extra_content_query() -> None: + state = _Callbacks(store=_Store(page=MessagePage(limit=2, has_more=False, data=()))) + page = state.service().get_page( + installed_app=_REF, account_id="account", conversation_id="conversation", first_id=None, limit=2 + ) + assert page.data == () + assert state.store.events == ["read closed"] + assert state.queried_ids == [] + + +@pytest.mark.parametrize("rating", ["like", "dislike", None]) +def test_feedback_emits_only_for_committed_rating_and_allows_completion_apps(rating: MessageRating | None) -> None: + state = _Callbacks(store=_Store()) + state.service().set_feedback( + installed_app=replace(_REF, app_mode="completion"), + account_id="account", + message_id="message", + rating=rating, + content="", + ) + if rating is None: + assert state.feedback == [] + assert state.store.events == ["write committed and closed"] + else: + assert state.feedback == [ + MessageFeedbackEvent( + tenant_id="owner-workspace", + app_id="app", + conversation_id="conversation", + message_id="message", + account_id="account", + rating=rating, + content="", + ) + ] + assert state.store.events == ["write committed and closed", "emit feedback"] + + +def test_feedback_write_error_propagates_without_emitting() -> None: + failure = MessageNotExistsError("Message was deleted") + state = _Callbacks(store=_Store(write_error=failure)) + with pytest.raises(MessageNotExistsError) as error: + state.service().set_feedback( + installed_app=_REF, account_id="account", message_id="message", rating="like", content=None + ) + assert error.value is failure + assert state.store.rating is None + assert state.feedback == [] + + +@pytest.mark.parametrize("mode", ["completion", "workflow"]) +def test_non_chat_admission_rejects_reads_before_database_or_generation(mode: str) -> None: + state = _Callbacks(store=_Store()) + ref = replace(_REF, app_mode=mode) + with pytest.raises(MessageNotChatAppError): + state.service().get_page( + installed_app=ref, account_id="account", conversation_id="conversation", first_id=None, limit=2 + ) + with pytest.raises(MessageNotChatAppError): + state.service().get_suggested_questions(installed_app=ref, account_id="account", message_id="message") + assert state.store.events == [] + + +@pytest.mark.parametrize("mode", ["chat", "agent-chat", "advanced-chat"]) +def test_suggested_questions_accepts_all_chat_modes_and_preserves_errors(mode: str) -> None: + state = _Callbacks(store=_Store()) + ref = replace(_REF, app_mode=mode) + assert state.service().get_suggested_questions(installed_app=ref, account_id="account", message_id="message") == [ + "Next question?" + ] + failure = SuggestedQuestionsAfterAnswerDisabledError("Suggested questions are disabled") + state.suggestion_error = failure + with pytest.raises(SuggestedQuestionsAfterAnswerDisabledError) as error: + state.service().get_suggested_questions(installed_app=ref, account_id="account", message_id="message") + assert error.value is failure