refactor(api): isolate installed app messages (#42016)

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

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"

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/<uuid:installed_app_id>/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/<uuid:installed_app_id>/messages/<uuid:message_id>/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/<uuid:installed_app_id>/messages/<uuid:message_id>/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/<uuid:installed_app_id>/messages/<uuid:message_id>/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})

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,

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),
+        )

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:

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:

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)

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.")

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:

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)

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")

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,

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

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