from http import HTTPStatus from typing import Any, Literal from uuid import UUID from flask import request from flask_restx import Resource from pydantic import BaseModel, Field, field_validator from controllers.common.errors import ( FileTooLargeError, InternalServerError, NoFileUploadedError, NotFoundError, TooManyFilesError, ) from controllers.common.rbac import PlainApp, RBACCheck 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 AnnotationIdsRequiredError, AnnotationImportRateLimitError, AppNotFoundError from controllers.console.flask_admission import console_account_admission from controllers.console.wraps import ( RBACPermission, model_validate, ) from core.helper.csv_sanitizer import CSVSanitizer from enums.account import TenantAccountRole from extensions.ext_application_services import application_services from fields.annotation_fields import ( Annotation, AnnotationExportList, AnnotationHitHistory, AnnotationHitHistoryList, AnnotationJobStatusDetailResponse, AnnotationJobStatusResponse, AnnotationList, ) from fields.base import ResponseModel from libs.external_api import error_responses from libs.helper import dump_response, uuid_value from machinery.context import RequestContext from services.annotation_command_service import AnnotationSettingNotFoundError from services.annotation_import_service import ( AnnotationImportJobNotFoundError, AnnotationImportLimitError, AnnotationImportValidationError, ) from services.annotation_query import AnnotationAppNotFoundError, AnnotationNotFoundError from services.annotation_reply_service import AnnotationReplyJobNotFoundError from services.errors.message import MessageNotExistsError class AnnotationReplyPayload(BaseModel): score_threshold: float = Field(..., description="Score threshold for annotation matching") embedding_provider_name: str = Field(..., description="Embedding provider name") embedding_model_name: str = Field(..., description="Embedding model name") class AnnotationSettingUpdatePayload(BaseModel): score_threshold: float = Field(..., description="Score threshold") class AnnotationListQuery(BaseModel): page: int = Field(default=1, ge=1, description="Page number") limit: int = Field(default=20, ge=1, description="Page size") keyword: str = Field(default="", description="Search keyword") class AnnotationBatchDeleteQuery(BaseModel): annotation_id: list[str] = Field( default_factory=list, description="Annotation IDs to delete; omitting it clears all annotations of the app", ) class CreateAnnotationPayload(BaseModel): message_id: str | None = Field(default=None, description="Message ID") question: str | None = Field(default=None, description="Question text") answer: str | None = Field(default=None, description="Answer text") content: str | None = Field(default=None, description="Content text") annotation_reply: dict[str, Any] | None = Field( default=None, description="Annotation reply data", ) @field_validator("message_id") @classmethod def validate_message_id(cls, value: str | None) -> str | None: if value is None: return value return uuid_value(value) class UpdateAnnotationPayload(BaseModel): question: str | None = None answer: str | None = None content: str | None = None annotation_reply: dict[str, Any] | None = Field(default=None) class AnnotationReplyStatusQuery(BaseModel): action: Literal["enable", "disable"] class AnnotationHitHistoryListQuery(BaseModel): page: int = Field(default=1, ge=1, description="Page number") limit: int = Field(default=20, ge=1, description="Page size") class AnnotationFilePayload(BaseModel): message_id: str = Field(..., description="Message ID") @field_validator("message_id") @classmethod def validate_message_id(cls, value: str) -> str: return uuid_value(value) class AnnotationSettingEmbeddingModelResponse(ResponseModel): embedding_provider_name: str | None = None embedding_model_name: str | None = None class AnnotationSettingResponse(ResponseModel): enabled: bool id: str | None = None score_threshold: float | None = None embedding_model: AnnotationSettingEmbeddingModelResponse | None = None class AnnotationCountResponse(ResponseModel): count: int = Field(description="Number of annotations") class AnnotationBatchImportResponse(ResponseModel): job_id: str | None = None job_status: str | None = None record_count: int | None = None error_msg: str | None = None register_schema_models( console_ns, AnnotationReplyPayload, AnnotationSettingUpdatePayload, AnnotationListQuery, CreateAnnotationPayload, UpdateAnnotationPayload, AnnotationReplyStatusQuery, AnnotationHitHistoryListQuery, AnnotationFilePayload, ) register_response_schema_models( console_ns, Annotation, AnnotationList, AnnotationExportList, AnnotationHitHistory, AnnotationHitHistoryList, AnnotationJobStatusResponse, AnnotationJobStatusDetailResponse, AnnotationSettingEmbeddingModelResponse, AnnotationSettingResponse, AnnotationBatchImportResponse, AnnotationCountResponse, ) @console_ns.route("/apps//annotations/count") class MessageAnnotationCountApi(Resource): @error_responses(AppNotFoundError) @console_ns.doc("get_annotation_count") @console_ns.doc(description="Get count of message annotations for the app") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.response( HTTPStatus.OK, "Annotation count retrieved successfully", console_ns.models[AnnotationCountResponse.__name__], ) @console_account_admission(rbac_checks=(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()),)) def get(self, context: RequestContext, app_id: UUID) -> dict[str, object]: try: count = application_services().annotation_queries.count( tenant_id=context.active_workspace_id, app_id=str(app_id) ) except AnnotationAppNotFoundError as exc: raise AppNotFoundError() from exc return dump_response(AnnotationCountResponse, {"count": count}) @console_ns.route("/apps//annotation-reply/") class AnnotationReplyActionApi(Resource): @console_ns.doc("annotation_reply_action") @console_ns.doc(description="Enable or disable annotation reply for an app") @console_ns.doc(params={"app_id": "Application ID", "action": "Action to perform (enable/disable)"}) @console_ns.expect(console_ns.models[AnnotationReplyPayload.__name__]) @console_ns.response( HTTPStatus.OK, "Action completed successfully", console_ns.models[AnnotationJobStatusResponse.__name__] ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_ns.response(HTTPStatus.NOT_FOUND, "App not found") @console_ns.response(HTTPStatus.BAD_REQUEST, "Invalid action") @console_ns.response(HTTPStatus.UNPROCESSABLE_ENTITY, "Invalid payload") @console_account_admission( billing_resource="annotation", allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_EDIT, PlainApp()),), ) @model_validate(AnnotationReplyPayload) def post( self, req_data: AnnotationReplyPayload, context: RequestContext, app_id: UUID, action: str ) -> tuple[dict[str, object], HTTPStatus]: validated_action = AnnotationReplyStatusQuery.model_validate({"action": action}).action service = application_services().annotation_reply try: if validated_action == "enable": result = service.enable( tenant_id=context.active_workspace_id, app_id=str(app_id), account_id=context.account_id, score_threshold=req_data.score_threshold, embedding_provider_name=req_data.embedding_provider_name, embedding_model_name=req_data.embedding_model_name, ) else: result = service.disable(tenant_id=context.active_workspace_id, app_id=str(app_id)) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc return dump_response(AnnotationJobStatusResponse, result), HTTPStatus.OK @console_ns.route("/apps//annotation-setting") class AppAnnotationSettingDetailApi(Resource): @console_ns.doc("get_annotation_setting") @console_ns.doc(description="Get annotation settings for an app") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.response( HTTPStatus.OK, "Annotation settings retrieved successfully", console_ns.models[AnnotationSettingResponse.__name__], ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_account_admission( allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()),), ) def get(self, context: RequestContext, app_id: UUID) -> tuple[dict[str, object], HTTPStatus]: try: result = application_services().annotation_queries.get_setting( tenant_id=context.active_workspace_id, app_id=str(app_id) ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc return dump_response(AnnotationSettingResponse, result), HTTPStatus.OK @console_ns.route("/apps//annotation-settings/") class AppAnnotationSettingUpdateApi(Resource): @console_ns.doc("update_annotation_setting") @console_ns.doc(description="Update annotation settings for an app") @console_ns.doc(params={"app_id": "Application ID", "annotation_setting_id": "Annotation setting ID"}) @console_ns.expect(console_ns.models[AnnotationSettingUpdatePayload.__name__]) @console_ns.response( HTTPStatus.OK, "Settings updated successfully", console_ns.models[AnnotationSettingResponse.__name__] ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_account_admission( allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_EDIT, PlainApp()),), ) @model_validate(AnnotationSettingUpdatePayload) def post( self, req_data: AnnotationSettingUpdatePayload, context: RequestContext, app_id: UUID, annotation_setting_id: UUID, ) -> tuple[dict[str, object], HTTPStatus]: try: result = application_services().annotation_commands.update_setting( tenant_id=context.active_workspace_id, app_id=str(app_id), setting_id=str(annotation_setting_id), account_id=context.account_id, score_threshold=req_data.score_threshold, ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except AnnotationSettingNotFoundError as exc: raise NotFoundError("App annotation not found") from exc return dump_response(AnnotationSettingResponse, result), HTTPStatus.OK @console_ns.route("/apps//annotation-reply//status/") class AnnotationReplyActionStatusApi(Resource): @console_ns.doc("get_annotation_reply_action_status") @console_ns.doc(description="Get status of annotation reply action job") @console_ns.doc(params={"app_id": "Application ID", "job_id": "Job ID", "action": "Action type"}) @console_ns.response( HTTPStatus.OK, "Job status retrieved successfully", console_ns.models[AnnotationJobStatusDetailResponse.__name__], ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_ns.response(HTTPStatus.NOT_FOUND, "App or owned job not found") @console_ns.response(HTTPStatus.BAD_REQUEST, "Invalid action") @console_account_admission( billing_resource="annotation", allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()),), ) def get( self, context: RequestContext, app_id: UUID, job_id: UUID, action: str ) -> tuple[dict[str, object], HTTPStatus]: validated_action = AnnotationReplyStatusQuery.model_validate({"action": action}).action try: result = application_services().annotation_reply.get_status( tenant_id=context.active_workspace_id, app_id=str(app_id), action=validated_action, job_id=str(job_id), ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except AnnotationReplyJobNotFoundError as exc: raise NotFoundError("The job does not exist.") from exc return dump_response(AnnotationJobStatusDetailResponse, result), HTTPStatus.OK @console_ns.route("/apps//annotations") class AnnotationApi(Resource): @console_ns.doc("list_annotations") @console_ns.doc(description="Get annotations for an app with pagination") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.doc(params=query_params_from_model(AnnotationListQuery)) @console_ns.response( HTTPStatus.OK, "Annotations retrieved successfully", console_ns.models[AnnotationList.__name__] ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_account_admission( allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()),), ) @model_validate(AnnotationListQuery) def get( self, req_data: AnnotationListQuery, context: RequestContext, app_id: UUID ) -> tuple[dict[str, object], HTTPStatus]: try: result = application_services().annotation_queries.get_page( tenant_id=context.active_workspace_id, app_id=str(app_id), page=req_data.page, limit=req_data.limit, keyword=req_data.keyword, ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc return dump_response(AnnotationList, result), HTTPStatus.OK @console_ns.doc("create_annotation") @console_ns.doc(description="Create a new annotation for an app") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.expect(console_ns.models[CreateAnnotationPayload.__name__]) @console_ns.response(HTTPStatus.CREATED, "Annotation created successfully", console_ns.models[Annotation.__name__]) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_account_admission( billing_resource="annotation", allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_EDIT, PlainApp()),), ) @model_validate(CreateAnnotationPayload) def post( self, req_data: CreateAnnotationPayload, context: RequestContext, app_id: UUID ) -> tuple[dict[str, object], HTTPStatus]: try: annotation = application_services().annotation_commands.upsert( tenant_id=context.active_workspace_id, app_id=str(app_id), account_id=context.account_id, message_id=req_data.message_id, question=req_data.question, answer=req_data.answer, content=req_data.content, ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except MessageNotExistsError as exc: raise NotFoundError("Message Not Exists.") from exc return dump_response(Annotation, annotation), HTTPStatus.CREATED @error_responses(AnnotationIdsRequiredError) @console_account_admission( allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_CREATE_AND_MANAGEMENT, PlainApp()),), ) @console_ns.doc(params=query_params_from_model(AnnotationBatchDeleteQuery)) @console_ns.response(HTTPStatus.NO_CONTENT, "Annotations deleted successfully") def delete(self, context: RequestContext, app_id: UUID) -> tuple[str, HTTPStatus]: annotation_ids = request.args.getlist("annotation_id") if not all(annotation_id.strip() for annotation_id in annotation_ids if annotation_id): raise AnnotationIdsRequiredError() try: commands = application_services().annotation_commands if annotation_ids: commands.delete_many( tenant_id=context.active_workspace_id, app_id=str(app_id), annotation_ids=annotation_ids ) else: commands.clear(tenant_id=context.active_workspace_id, app_id=str(app_id)) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc return "", HTTPStatus.NO_CONTENT @console_ns.route("/apps//annotations/export") class AnnotationExportApi(Resource): @console_ns.doc("export_annotations") @console_ns.doc(description="Export all annotations for an app with CSV injection protection") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.response( HTTPStatus.OK, "Annotations exported successfully", console_ns.models[AnnotationExportList.__name__], ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_account_admission( allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()),), ) def get(self, context: RequestContext, app_id: UUID) -> tuple[dict[str, object], HTTPStatus, dict[str, str]]: try: annotations = application_services().annotation_queries.get_all( tenant_id=context.active_workspace_id, app_id=str(app_id) ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc result = AnnotationExportList.model_validate({"data": annotations}) # The client turns this JSON into CSV; escape only the detached response. for annotation in result.data: if annotation.question: annotation.question = CSVSanitizer.sanitize_value(annotation.question) if annotation.answer: annotation.answer = CSVSanitizer.sanitize_value(annotation.answer) return ( dump_response(AnnotationExportList, result), HTTPStatus.OK, { "Content-Type": "application/json; charset=utf-8", "X-Content-Type-Options": "nosniff", }, ) @console_ns.route("/apps//annotations/") class AnnotationUpdateDeleteApi(Resource): @console_ns.doc("update_delete_annotation") @console_ns.doc(description="Update or delete an annotation") @console_ns.doc(params={"app_id": "Application ID", "annotation_id": "Annotation ID"}) @console_ns.response(HTTPStatus.OK, "Annotation updated successfully", console_ns.models[Annotation.__name__]) @console_ns.response(HTTPStatus.NO_CONTENT, "Annotation deleted successfully") @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_ns.expect(console_ns.models[UpdateAnnotationPayload.__name__]) @console_account_admission( billing_resource="annotation", allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_EDIT, PlainApp()),), ) @model_validate(UpdateAnnotationPayload) def post( self, req_data: UpdateAnnotationPayload, context: RequestContext, app_id: UUID, annotation_id: UUID ) -> dict[str, object]: try: annotation = application_services().annotation_commands.update( tenant_id=context.active_workspace_id, app_id=str(app_id), annotation_id=str(annotation_id), question=req_data.question, answer=req_data.answer, ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except AnnotationNotFoundError as exc: raise NotFoundError("Annotation not found") from exc return dump_response(Annotation, annotation) @console_account_admission( allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_EDIT, PlainApp()),), ) @console_ns.response(HTTPStatus.NO_CONTENT, "Annotation deleted successfully") def delete(self, context: RequestContext, app_id: UUID, annotation_id: UUID) -> tuple[str, HTTPStatus]: try: application_services().annotation_commands.delete( tenant_id=context.active_workspace_id, app_id=str(app_id), annotation_id=str(annotation_id) ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except AnnotationNotFoundError as exc: raise NotFoundError("Annotation not found") from exc return "", HTTPStatus.NO_CONTENT @console_ns.route("/apps//annotations/batch-import") class AnnotationBatchImportApi(Resource): @error_responses(NoFileUploadedError, TooManyFilesError) @console_ns.doc("batch_import_annotations") @console_ns.doc(description="Batch import annotations from CSV file with rate limiting and security checks") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.response( HTTPStatus.OK, "Import started, or CSV/subscription validation failed (error_msg)", console_ns.models[AnnotationBatchImportResponse.__name__], ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_ns.response(HTTPStatus.NOT_FOUND, "App not found") @console_ns.response(HTTPStatus.BAD_REQUEST, "Missing, multiple, empty or invalid file") @console_ns.response(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "File too large") @console_ns.response(HTTPStatus.TOO_MANY_REQUESTS, "Too many requests or concurrent imports") @console_ns.response(HTTPStatus.INTERNAL_SERVER_ERROR, "Import infrastructure failed") @console_account_admission( billing_resource="annotation", allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_EDIT, PlainApp()),), ) def post(self, context: RequestContext, app_id: UUID) -> tuple[dict[str, object], HTTPStatus]: from configs import dify_config if "file" not in request.files: raise NoFileUploadedError() if len(list(request.files.items(multi=True))) > 1: raise TooManyFilesError() file = request.files["file"] if not file.filename or not file.filename.lower().endswith(".csv"): raise ValueError("Invalid file type. Only CSV files are allowed") file.stream.seek(0, 2) file_size = file.stream.tell() file.stream.seek(0) if file_size > dify_config.ANNOTATION_IMPORT_FILE_SIZE_LIMIT * 1024 * 1024: raise FileTooLargeError( f"File size exceeds maximum limit of {dify_config.ANNOTATION_IMPORT_FILE_SIZE_LIMIT}MB. " "Please reduce the file size and try again." ) if file_size == 0: raise ValueError("The uploaded file is empty") try: result = application_services().annotation_imports.import_csv( tenant_id=context.active_workspace_id, app_id=str(app_id), account_id=context.account_id, stream=file.stream, ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except AnnotationImportLimitError as exc: raise AnnotationImportRateLimitError(str(exc)) from exc except AnnotationImportValidationError as exc: # Retain the existing import response for CSV and record-quota validation failures. return dump_response(AnnotationBatchImportResponse, {"error_msg": str(exc)}), HTTPStatus.OK except Exception as exc: # Infrastructure ValueErrors must not become client-side invalid_param responses. raise InternalServerError() from exc return dump_response(AnnotationBatchImportResponse, result), HTTPStatus.OK @console_ns.route("/apps//annotations/batch-import-status/") class AnnotationBatchImportStatusApi(Resource): @console_ns.doc("get_batch_import_status") @console_ns.doc(description="Get status of batch import job") @console_ns.doc(params={"app_id": "Application ID", "job_id": "Job ID"}) @console_ns.response( HTTPStatus.OK, "Job status retrieved successfully", console_ns.models[AnnotationJobStatusDetailResponse.__name__], ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_ns.response(HTTPStatus.NOT_FOUND, "App or owned job not found") @console_ns.response(HTTPStatus.INTERNAL_SERVER_ERROR, "Import status could not be read") @console_account_admission( billing_resource="annotation", allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()),), ) def get(self, context: RequestContext, app_id: UUID, job_id: UUID) -> tuple[dict[str, object], HTTPStatus]: try: result = application_services().annotation_imports.get_status( tenant_id=context.active_workspace_id, app_id=str(app_id), job_id=str(job_id) ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except AnnotationImportJobNotFoundError as exc: raise NotFoundError("The job does not exist.") from exc except Exception as exc: raise InternalServerError() from exc return dump_response(AnnotationJobStatusDetailResponse, result), HTTPStatus.OK @console_ns.route("/apps//annotations//hit-histories") class AnnotationHitHistoryListApi(Resource): @console_ns.doc("list_annotation_hit_histories") @console_ns.doc(description="Get hit histories for an annotation") @console_ns.doc(params={"app_id": "Application ID", "annotation_id": "Annotation ID"}) @console_ns.doc(params=query_params_from_model(AnnotationHitHistoryListQuery)) @console_ns.response( HTTPStatus.OK, "Hit histories retrieved successfully", console_ns.models[AnnotationHitHistoryList.__name__], ) @console_ns.response(HTTPStatus.FORBIDDEN, "Insufficient permissions") @console_account_admission( allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), rbac_checks=(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()),), ) def get(self, context: RequestContext, app_id: UUID, annotation_id: UUID) -> dict[str, object]: # Preserve this endpoint's forgiving parsing; the repository clamps bounds. page = request.args.get("page", default=1, type=int) limit = request.args.get("limit", default=20, type=int) try: result = application_services().annotation_queries.get_hit_history_page( tenant_id=context.active_workspace_id, app_id=str(app_id), annotation_id=str(annotation_id), page=page, limit=limit, ) except AnnotationAppNotFoundError as exc: raise NotFoundError("App not found") from exc except AnnotationNotFoundError as exc: raise NotFoundError("Annotation not found") from exc return dump_response(AnnotationHitHistoryList, result)