diff --git a/api/controllers/console/knowledge_fs/resources.py b/api/controllers/console/knowledge_fs/resources.py index 2a434981b5b..ef1606587ed 100644 --- a/api/controllers/console/knowledge_fs/resources.py +++ b/api/controllers/console/knowledge_fs/resources.py @@ -84,6 +84,7 @@ from services.knowledge_fs.product_dto import ( KnowledgeFSAppBindingListResponse, KnowledgeFSAppBindingPayload, KnowledgeFSAppBindingResponse, + KnowledgeFSAsyncSourceImportPayload, KnowledgeFSBackgroundTaskListQuery, KnowledgeFSBackgroundTaskListResponse, KnowledgeFSBackgroundTaskResponse, @@ -252,6 +253,10 @@ from services.knowledge_fs.product_remote import ( ) from services.knowledge_fs.query_images import KnowledgeFSQueryImageError, validate_query_image_references from services.knowledge_fs.runtime import KnowledgeFSRuntime, get_knowledge_fs_runtime +from services.knowledge_fs.source_import_commit_service import ( + commit_source_import, + resume_committed_source_import, +) from services.knowledge_fs.space_tag_service import KnowledgeFSSpaceTagValidationError from services.knowledge_fs.staged_upload_service import ( KnowledgeFSStagedUploadConflictError, @@ -268,6 +273,7 @@ from services.knowledge_fs_capability import ( register_schema_models( console_ns, KnowledgeFSAppBindingPayload, + KnowledgeFSAsyncSourceImportPayload, KnowledgeFSBackgroundTaskListQuery, KnowledgeFSBadCaseCreatePayload, KnowledgeFSBadCaseUpdatePayload, @@ -2746,12 +2752,20 @@ class KnowledgeFSSourceWorkflowRetryApi(Resource): @_knowledge_fs_errors def post(self, control_space_id: str, run_id: str): actor_id, tenant_id = _actor() - result = _console_services().facade.retry_source_workflow( + facade = _console_services().facade + result = facade.retry_source_workflow( tenant_id=tenant_id, account_id=actor_id, control_space_id=control_space_id, run_id=run_id, ) + resume_committed_source_import( + facade=facade, + tenant_id=tenant_id, + account_id=actor_id, + control_space_id=control_space_id, + workflow=result, + ) return dump_response(KnowledgeFSSourceWorkflowResponse, result) @@ -2807,6 +2821,34 @@ class KnowledgeFSSourceWorkflowSelectionApi(Resource): return dump_response(KnowledgeFSSourceWorkflowResponse, result), HTTPStatus.ACCEPTED +@console_ns.route("/knowledge-fs/spaces//sources//async-import") +class KnowledgeFSSourceAsyncImportApi(Resource): + @console_ns.expect(console_ns.models[KnowledgeFSAsyncSourceImportPayload.__name__]) + @console_ns.doc(params=_IDEMPOTENCY_HEADER_PARAMS) + @console_ns.response( + HTTPStatus.ACCEPTED, + "KnowledgeFS Source import accepted for asynchronous reconciliation", + console_ns.models[KnowledgeFSSourceWorkflowResponse.__name__], + ) + @setup_required + @login_required + @account_initialization_required + @_knowledge_fs_errors + def post(self, control_space_id: str, source_id: str): + actor_id, tenant_id = _actor() + payload = _payload(KnowledgeFSAsyncSourceImportPayload) + result = commit_source_import( + facade=_console_services().facade, + tenant_id=tenant_id, + account_id=actor_id, + control_space_id=control_space_id, + source_id=source_id, + payload=payload.root, + idempotency_key=_idempotency_key(), + ) + return dump_response(KnowledgeFSSourceWorkflowResponse, result), HTTPStatus.ACCEPTED + + @console_ns.route("/knowledge-fs/spaces//sources//pages") class KnowledgeFSSpaceSourcePagesApi(Resource): @console_ns.doc(params=query_params_from_model(KnowledgeFSSourcePagesQuery)) diff --git a/api/services/knowledge_fs/product_dto.py b/api/services/knowledge_fs/product_dto.py index d5b8970dc27..b4cab5b8401 100644 --- a/api/services/knowledge_fs/product_dto.py +++ b/api/services/knowledge_fs/product_dto.py @@ -2,10 +2,13 @@ from __future__ import annotations +import hashlib +import json import math import re from datetime import UTC, datetime from typing import Annotated, ClassVar, Literal +from urllib.parse import urlsplit, urlunsplit from uuid import UUID from pydantic import AliasChoices, BaseModel, ConfigDict, Field, JsonValue, RootModel, field_validator, model_validator @@ -109,8 +112,33 @@ class KnowledgeFSInitialWebsiteCrawlOptionsPayload(BaseModel): model_config = ConfigDict(extra="forbid") +def normalize_knowledge_fs_source_url(source_url: str) -> str: + """Normalize URL identity without dropping query parameters that may select content.""" + + value = source_url.strip() + parsed = urlsplit(value) + if parsed.scheme.lower() not in {"http", "https"} or parsed.hostname is None: + return value + host = parsed.hostname.lower() + if ":" in host: + host = f"[{host}]" + try: + port = parsed.port + except ValueError: + return value + default_port = (parsed.scheme.lower() == "http" and port == 80) or ( + parsed.scheme.lower() == "https" and port == 443 + ) + netloc = host if port is None or default_port else f"{host}:{port}" + path = parsed.path or "/" + if path != "/": + path = path.rstrip("/") + return urlunsplit((parsed.scheme.lower(), netloc, path, parsed.query, "")) + + class KnowledgeFSInitialWebsiteSelectionPayload(BaseModel): source_url: str = Field(min_length=1, max_length=4_096) + canonical_url: str | None = Field(default=None, min_length=1, max_length=4_096) title: str | None = Field(default=None, max_length=500) model_config = ConfigDict(extra="forbid") @@ -120,6 +148,12 @@ class KnowledgeFSInitialWebsiteSelectionPayload(BaseModel): def normalize_source_url(cls, source_url: str) -> str: return source_url.strip() + @model_validator(mode="after") + def populate_canonical_url(self) -> KnowledgeFSInitialWebsiteSelectionPayload: + canonical_url = normalize_knowledge_fs_source_url(self.canonical_url or self.source_url) + object.__setattr__(self, "canonical_url", canonical_url) + return self + class KnowledgeFSOnlineDocumentWorkflowImportItemPayload(BaseModel): etag: str | None = Field(default=None, max_length=1_024) @@ -185,6 +219,9 @@ class KnowledgeFSInitialWebsiteSourcePayload(KnowledgeFSInitialSyncPolicyPayload credential_id: str | None = Field(default=None, min_length=1, max_length=255, alias="credentialId") provider_display_name: str | None = Field(default=None, min_length=1, max_length=255, alias="providerDisplayName") parameters: dict[str, JsonValue] = Field(default_factory=dict, max_length=50) + preview_configuration_fingerprint: str | None = Field( + default=None, min_length=64, max_length=64, alias="previewConfigurationFingerprint" + ) root_url: str = Field(min_length=1, max_length=4_096) crawl_options: KnowledgeFSInitialWebsiteCrawlOptionsPayload selection: list[KnowledgeFSInitialWebsiteSelectionPayload] = Field(min_length=1, max_length=200) @@ -192,9 +229,16 @@ class KnowledgeFSInitialWebsiteSourcePayload(KnowledgeFSInitialSyncPolicyPayload @model_validator(mode="after") def validate_selection(self) -> KnowledgeFSInitialWebsiteSourcePayload: - source_urls = [item.source_url for item in self.selection] + source_urls = [item.canonical_url for item in self.selection] if len(set(source_urls)) != len(source_urls): - raise ValueError("initial website selection URLs must be unique") + raise ValueError("initial website selection canonical URLs must be unique") + if self.crawl_options.limit < len(self.selection): + raise ValueError("initial website crawl limit must cover every selected URL") + if ( + self.preview_configuration_fingerprint is not None + and self.preview_configuration_fingerprint != knowledge_fs_initial_preview_configuration_fingerprint(self) + ): + raise ValueError("initial website preview configuration no longer matches") return self @@ -218,6 +262,20 @@ class KnowledgeFSInitialWebsiteSourcePreviewPayload(KnowledgeFSInitialDatasource kind: Literal["website_crawl"] +def knowledge_fs_initial_preview_configuration_fingerprint( + payload: KnowledgeFSInitialWebsiteSourcePayload | KnowledgeFSInitialWebsiteSourcePreviewPayload, +) -> str: + configuration = { + "credentialId": payload.credential_id, + "datasource": payload.datasource, + "parameters": payload.parameters, + "pluginId": payload.plugin_id, + "provider": payload.provider, + } + encoded = json.dumps(configuration, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode() + return hashlib.sha256(encoded).hexdigest() + + class KnowledgeFSInitialSourcePreviewPageResponse(ResponseModel): description: str | None = None source_url: str = Field(validation_alias=AliasChoices("source_url", "sourceUrl")) @@ -247,6 +305,10 @@ class KnowledgeFSInitialSourcePreviewFileResponse(ResponseModel): class KnowledgeFSInitialSourcePreviewResponse(ResponseModel): + configuration_fingerprint: str | None = Field( + default=None, + validation_alias=AliasChoices("configuration_fingerprint", "configurationFingerprint"), + ) documents: list[KnowledgeFSInitialSourcePreviewDocumentResponse] = Field(default_factory=list) files: list[KnowledgeFSInitialSourcePreviewFileResponse] = Field(default_factory=list) kind: Literal["online_document", "online_drive", "website_crawl"] @@ -2324,6 +2386,59 @@ class KnowledgeFSCrawlPreviewSelectionPayload(BaseModel): model_config = ConfigDict(extra="forbid", validate_by_alias=True, validate_by_name=True) +class KnowledgeFSDeferredSyncPolicyPayload(BaseModel): + custom_interval_seconds: int | None = Field(default=None, ge=3_600, le=2_592_000, alias="customIntervalSeconds") + enabled: bool + mode: Literal["provider", "manual", "interval", "custom"] + + model_config = ConfigDict(extra="forbid", validate_by_alias=True, validate_by_name=True) + + @model_validator(mode="after") + def validate_custom_interval(self) -> KnowledgeFSDeferredSyncPolicyPayload: + if self.mode == "custom" and self.custom_interval_seconds is None: + raise ValueError("customIntervalSeconds is required for a custom sync policy") + if self.mode != "custom" and self.custom_interval_seconds is not None: + raise ValueError("customIntervalSeconds is only valid for a custom sync policy") + return self + + +class KnowledgeFSAsyncCrawlPreviewImportPayload(BaseModel): + kind: Literal["crawl-preview-selection"] + page_ids: list[str] = Field(min_length=1, max_length=200, alias="pageIds") + preview_workflow_id: str = Field(min_length=1, max_length=255, alias="previewWorkflowId") + sync_policy: KnowledgeFSDeferredSyncPolicyPayload = Field(alias="syncPolicy") + + model_config = ConfigDict(extra="forbid", validate_by_alias=True, validate_by_name=True) + + +class KnowledgeFSAsyncOnlineDocumentImportPayload(BaseModel): + items: list[KnowledgeFSOnlineDocumentWorkflowImportItemPayload] = Field(min_length=1, max_length=200) + kind: Literal["online-document-import"] + sync_policy: KnowledgeFSDeferredSyncPolicyPayload = Field(alias="syncPolicy") + + model_config = ConfigDict(extra="forbid", validate_by_alias=True, validate_by_name=True) + + +class KnowledgeFSAsyncOnlineDriveImportPayload(BaseModel): + items: list[KnowledgeFSOnlineDriveWorkflowImportItemPayload] = Field(min_length=1, max_length=200) + kind: Literal["online-drive-import"] + sync_policy: KnowledgeFSDeferredSyncPolicyPayload = Field(alias="syncPolicy") + + model_config = ConfigDict(extra="forbid", validate_by_alias=True, validate_by_name=True) + + +KnowledgeFSAsyncSourceImport = Annotated[ + KnowledgeFSAsyncCrawlPreviewImportPayload + | KnowledgeFSAsyncOnlineDocumentImportPayload + | KnowledgeFSAsyncOnlineDriveImportPayload, + Field(discriminator="kind"), +] + + +class KnowledgeFSAsyncSourceImportPayload(RootModel[KnowledgeFSAsyncSourceImport]): + pass + + class KnowledgeFSCrawlImportPayload(BaseModel): source_urls: list[str] = Field(min_length=1, max_length=200, alias="sourceUrls") @@ -3649,6 +3764,11 @@ __all__ = [ "KnowledgeFSAppBindingListResponse", "KnowledgeFSAppBindingPayload", "KnowledgeFSAppBindingResponse", + "KnowledgeFSAsyncCrawlPreviewImportPayload", + "KnowledgeFSAsyncOnlineDocumentImportPayload", + "KnowledgeFSAsyncOnlineDriveImportPayload", + "KnowledgeFSAsyncSourceImport", + "KnowledgeFSAsyncSourceImportPayload", "KnowledgeFSBackgroundTaskFailureResponse", "KnowledgeFSBackgroundTaskListQuery", "KnowledgeFSBackgroundTaskListResponse", @@ -3670,6 +3790,7 @@ __all__ = [ "KnowledgeFSCredentialItemResponse", "KnowledgeFSCredentialListResponse", "KnowledgeFSCursorQuery", + "KnowledgeFSDeferredSyncPolicyPayload", "KnowledgeFSDocumentAvailabilityPayload", "KnowledgeFSDocumentBatchDownloadPayload", "KnowledgeFSDocumentChunkListQuery", diff --git a/api/services/knowledge_fs/source_import_commit_service.py b/api/services/knowledge_fs/source_import_commit_service.py new file mode 100644 index 00000000000..b32fc7f1374 --- /dev/null +++ b/api/services/knowledge_fs/source_import_commit_service.py @@ -0,0 +1,177 @@ +"""Server-owned commit boundary for durable Add source imports.""" + +from __future__ import annotations + +from services.knowledge_fs.product_dto import ( + KnowledgeFSAsyncCrawlPreviewImportPayload, + KnowledgeFSAsyncOnlineDocumentImportPayload, + KnowledgeFSAsyncOnlineDriveImportPayload, + KnowledgeFSAsyncSourceImport, + KnowledgeFSCrawlPreviewSelectionPayload, + KnowledgeFSOnlineDocumentWorkflowImportPayload, + KnowledgeFSOnlineDriveWorkflowImportPayload, + KnowledgeFSSourceUpdatePayload, + KnowledgeFSSourceWorkflowImportPayload, + KnowledgeFSSourceWorkflowResponse, +) + +_ASYNC_IMPORT_KINDS = { + "crawl-preview-selection", + "online-document-import", + "online-drive-import", +} +_PENDING_IMPORT_KEY = "pendingImport" + + +def commit_source_import( + *, + facade, + tenant_id: str, + account_id: str, + control_space_id: str, + source_id: str, + payload: KnowledgeFSAsyncSourceImport, + idempotency_key: str, +) -> KnowledgeFSSourceWorkflowResponse: + """Start an import and transfer reconciliation ownership to the backend.""" + + preview_workflow_id: str | None = None + if isinstance(payload, KnowledgeFSAsyncCrawlPreviewImportPayload): + preview_workflow_id = payload.preview_workflow_id + preview_workflow = facade.get_source_workflow( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + run_id=preview_workflow_id, + ) + if preview_workflow.source_id != source_id: + raise ValueError("KnowledgeFS crawl preview workflow does not belong to this Source") + import_workflow = facade.select_crawl_preview_pages( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + run_id=preview_workflow_id, + payload=KnowledgeFSCrawlPreviewSelectionPayload(pageIds=payload.page_ids), + idempotency_key=idempotency_key, + ) + elif isinstance(payload, KnowledgeFSAsyncOnlineDocumentImportPayload): + import_workflow = facade.import_source_workflow( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + payload=KnowledgeFSSourceWorkflowImportPayload( + KnowledgeFSOnlineDocumentWorkflowImportPayload( + kind="online-document-import", + items=payload.items, + ) + ), + idempotency_key=idempotency_key, + ) + elif isinstance(payload, KnowledgeFSAsyncOnlineDriveImportPayload): + import_workflow = facade.import_source_workflow( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + payload=KnowledgeFSSourceWorkflowImportPayload( + KnowledgeFSOnlineDriveWorkflowImportPayload( + kind="online-drive-import", + items=payload.items, + ) + ), + idempotency_key=idempotency_key, + ) + else: + raise TypeError("Unsupported KnowledgeFS async Source import") + + if import_workflow.source_id != source_id: + raise RuntimeError("KnowledgeFS import workflow returned a different Source") + source = facade.get_source( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + ) + pending_import = { + "kind": payload.kind, + **({"previewWorkflowId": preview_workflow_id} if preview_workflow_id is not None else {}), + "workflowId": import_workflow.id, + "syncPolicy": payload.sync_policy.model_dump(mode="json", by_alias=True), + } + if source.metadata.get(_PENDING_IMPORT_KEY) != pending_import or source.status != "syncing": + facade.update_source( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source.id, + payload=KnowledgeFSSourceUpdatePayload( + expectedVersion=source.version, + metadata={**source.metadata, "preview": False, _PENDING_IMPORT_KEY: pending_import}, + status="syncing", + ), + ) + + from tasks.knowledge_fs_source_import_tasks import finalize_source_import + + finalize_source_import.delay( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + workflow_id=import_workflow.id, + ) + return import_workflow + + +def resume_committed_source_import( + *, facade, tenant_id: str, account_id: str, control_space_id: str, workflow: KnowledgeFSSourceWorkflowResponse +) -> None: + """Restore server reconciliation when a failed committed import is retried.""" + + if workflow.source_id is None: + return + source = facade.get_source( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=workflow.source_id, + ) + last_import = source.metadata.get("lastImport") + if not isinstance(last_import, dict) or last_import.get("kind") not in _ASYNC_IMPORT_KINDS: + return + pending_import = { + "kind": last_import.get("kind"), + **( + {"previewWorkflowId": last_import.get("previewWorkflowId")} + if last_import.get("previewWorkflowId") is not None + else {} + ), + "workflowId": workflow.id, + "syncPolicy": last_import.get("syncPolicy"), + } + metadata = dict(source.metadata) + metadata.pop("lastImport", None) + facade.update_source( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source.id, + payload=KnowledgeFSSourceUpdatePayload( + expectedVersion=source.version, + metadata={**metadata, "preview": False, _PENDING_IMPORT_KEY: pending_import}, + status="syncing", + ), + ) + from tasks.knowledge_fs_source_import_tasks import finalize_source_import + + finalize_source_import.delay( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source.id, + workflow_id=workflow.id, + ) + + +__all__ = ["commit_source_import", "resume_committed_source_import"] diff --git a/api/tasks/knowledge_fs_initial_source_preview_tasks.py b/api/tasks/knowledge_fs_initial_source_preview_tasks.py index df4e264a12e..da7c80b4f90 100644 --- a/api/tasks/knowledge_fs_initial_source_preview_tasks.py +++ b/api/tasks/knowledge_fs_initial_source_preview_tasks.py @@ -17,7 +17,10 @@ from services.knowledge_fs.initial_source_preview_job import ( KnowledgeFSInitialSourcePreviewJobNotFoundError, KnowledgeFSInitialSourcePreviewJobService, ) -from services.knowledge_fs.product_dto import KnowledgeFSInitialWebsiteSourcePreviewPayload +from services.knowledge_fs.product_dto import ( + KnowledgeFSInitialWebsiteSourcePreviewPayload, + knowledge_fs_initial_preview_configuration_fingerprint, +) logger = logging.getLogger(__name__) @@ -56,10 +59,11 @@ def run_knowledge_fs_initial_source_preview( if account.current_tenant_id != tenant_id: raise PermissionError("Datasource preview account is not a tenant member") session.expunge(account) + preview_payload = KnowledgeFSInitialWebsiteSourcePreviewPayload.model_validate(payload) result = KnowledgeFSInitialSourcePreviewService(session_factory.get_session_maker()).preview( tenant_id=tenant_id, account=account, - payload=KnowledgeFSInitialWebsiteSourcePreviewPayload.model_validate(payload), + payload=preview_payload, is_canceled=lambda: _preview_was_canceled( job_service=job_service, tenant_id=tenant_id, @@ -67,6 +71,7 @@ def run_knowledge_fs_initial_source_preview( job_id=job_id, ), ) + result.configuration_fingerprint = knowledge_fs_initial_preview_configuration_fingerprint(preview_payload) job_service.transition_status( tenant_id=tenant_id, account_id=account_id, diff --git a/api/tasks/knowledge_fs_initial_source_tasks.py b/api/tasks/knowledge_fs_initial_source_tasks.py index 5b236110f44..805797a8bb0 100644 --- a/api/tasks/knowledge_fs_initial_source_tasks.py +++ b/api/tasks/knowledge_fs_initial_source_tasks.py @@ -31,6 +31,7 @@ from services.knowledge_fs.product_dto import ( KnowledgeFSSourceSyncPolicyPayload, KnowledgeFSSourceUpdatePayload, KnowledgeFSSourceWorkflowImportPayload, + knowledge_fs_initial_preview_configuration_fingerprint, ) from services.knowledge_fs.product_remote import KnowledgeFSProductRemoteError, KnowledgeFSProductResourceNotFoundError from services.knowledge_fs.runtime import get_knowledge_fs_runtime @@ -248,6 +249,11 @@ def _source_payload( "includeSubpages": payload.crawl_options.include_subpages, "limit": payload.crawl_options.limit, } + metadata["initialPreview"] = { + "configurationFingerprint": knowledge_fs_initial_preview_configuration_fingerprint(payload), + "requestedSourceUrls": [selection.source_url for selection in payload.selection], + "canonicalSourceUrls": [selection.canonical_url for selection in payload.selection], + } source_type: Literal["connector", "web"] = "web" uri = payload.root_url else: @@ -280,7 +286,7 @@ def _start_workflow( control_space_id=control_space_id, source_id=source_id, payload=KnowledgeFSCrawlImportPayload( - sourceUrls=[selection.source_url for selection in payload.selection], + sourceUrls=[selection.canonical_url or selection.source_url for selection in payload.selection], ), idempotency_key=f"{request_id}:crawl-import", ) @@ -439,12 +445,20 @@ def start_initial_source_import( source_id=source_id, ) error_message = workflow.failure.message if workflow.failure is not None else None - initial_import = { + initial_import: dict[str, object] = { "errorCode": workflow.last_error_code, "errorMessage": error_message, "state": workflow.state, "workflowId": workflow.id, } + if isinstance(payload, KnowledgeFSInitialWebsiteSourcePayload): + initial_import.update( + { + "configurationFingerprint": knowledge_fs_initial_preview_configuration_fingerprint(payload), + "requestedSourceUrls": [selection.source_url for selection in payload.selection], + "canonicalSourceUrls": [selection.canonical_url for selection in payload.selection], + } + ) if ( failed_source.metadata.get("preview") is not False or failed_source.metadata.get("initialImport") != initial_import diff --git a/api/tasks/knowledge_fs_source_import_tasks.py b/api/tasks/knowledge_fs_source_import_tasks.py new file mode 100644 index 00000000000..6ed277f2733 --- /dev/null +++ b/api/tasks/knowledge_fs_source_import_tasks.py @@ -0,0 +1,126 @@ +"""Durable reconciliation for Add source imports submitted from preview.""" + +from __future__ import annotations + +from celery import shared_task + +from core.db.session_factory import session_factory +from services.knowledge_fs.product_dto import ( + KnowledgeFSDeferredSyncPolicyPayload, + KnowledgeFSSourceSyncPolicyPayload, + KnowledgeFSSourceUpdatePayload, +) +from services.knowledge_fs.product_remote import KnowledgeFSProductRemoteError, KnowledgeFSProductResourceNotFoundError +from services.knowledge_fs.runtime import get_knowledge_fs_runtime + +_ACTIVE_STATES = {"queued", "running", "crawling", "importing", "syncing"} +_PENDING_IMPORT_KEY = "pendingImport" + + +class KnowledgeFSSourceImportNotReadyError(RuntimeError): + pass + + +def finalize_source_import_once( + *, tenant_id: str, account_id: str, control_space_id: str, source_id: str, workflow_id: str +) -> str: + facade = get_knowledge_fs_runtime(session_factory.get_session_maker()).facade + workflow = facade.get_source_workflow( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + run_id=workflow_id, + ) + if workflow.state in _ACTIVE_STATES: + raise KnowledgeFSSourceImportNotReadyError("KnowledgeFS Source import is still running") + + source = facade.get_source( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + ) + pending = source.metadata.get(_PENDING_IMPORT_KEY) + if not isinstance(pending, dict) or pending.get("workflowId") != workflow_id: + return workflow_id + + metadata = dict(source.metadata) + metadata.pop(_PENDING_IMPORT_KEY, None) + if workflow.state != "completed": + failure = { + "errorCode": workflow.last_error_code, + "errorMessage": workflow.failure.message if workflow.failure is not None else None, + "kind": pending.get("kind"), + "previewWorkflowId": pending.get("previewWorkflowId"), + "state": workflow.state, + "syncPolicy": pending.get("syncPolicy"), + "workflowId": workflow.id, + } + facade.update_source( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + payload=KnowledgeFSSourceUpdatePayload( + expectedVersion=source.version, + metadata={**metadata, "lastImport": failure, "preview": False}, + status="error", + ), + ) + return workflow_id + + desired_policy = KnowledgeFSDeferredSyncPolicyPayload.model_validate(pending.get("syncPolicy")) + try: + current_policy = facade.get_source_sync_policy( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + ) + expected_revision = current_policy.revision + except KnowledgeFSProductResourceNotFoundError: + expected_revision = 0 + facade.update_source_sync_policy( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + payload=KnowledgeFSSourceSyncPolicyPayload( + enabled=desired_policy.enabled, + mode=desired_policy.mode, + customIntervalSeconds=desired_policy.custom_interval_seconds, + expectedRevision=expected_revision, + expectedSourceVersion=source.version, + ), + ) + facade.update_source( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + payload=KnowledgeFSSourceUpdatePayload( + expectedVersion=source.version, + metadata={**metadata, "preview": False}, + status="active", + ), + ) + return workflow_id + + +@shared_task(bind=True, queue="knowledge_fs_lifecycle", max_retries=300, default_retry_delay=2) +def finalize_source_import( + self, *, tenant_id: str, account_id: str, control_space_id: str, source_id: str, workflow_id: str +) -> str: + try: + return finalize_source_import_once( + tenant_id=tenant_id, + account_id=account_id, + control_space_id=control_space_id, + source_id=source_id, + workflow_id=workflow_id, + ) + except (KnowledgeFSSourceImportNotReadyError, KnowledgeFSProductRemoteError) as exc: + raise self.retry(exc=exc) + + +__all__ = ["finalize_source_import", "finalize_source_import_once"] diff --git a/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py b/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py index be644d81dab..b209346caf7 100644 --- a/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py +++ b/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py @@ -94,6 +94,7 @@ def test_console_and_service_api_routes_are_registered() -> None: "/knowledge-fs/spaces//source-workflows//retry", "/knowledge-fs/spaces//source-workflows//pages", "/knowledge-fs/spaces//source-workflows//selection", + "/knowledge-fs/spaces//sources//async-import", "/knowledge-fs/spaces//source-providers", "/knowledge-fs/spaces//queries", "/knowledge-fs/spaces//research-tasks", diff --git a/api/tests/unit_tests/services/test_knowledge_fs_initial_source_reliability.py b/api/tests/unit_tests/services/test_knowledge_fs_initial_source_reliability.py new file mode 100644 index 00000000000..a4a66b465fa --- /dev/null +++ b/api/tests/unit_tests/services/test_knowledge_fs_initial_source_reliability.py @@ -0,0 +1,60 @@ +import pytest +from pydantic import ValidationError + +from services.knowledge_fs.product_dto import ( + KnowledgeFSInitialWebsiteSourcePayload, + knowledge_fs_initial_preview_configuration_fingerprint, +) + + +def _payload(*, limit: int = 2, urls: list[str] | None = None) -> dict[str, object]: + selected = urls or ["HTTPS://Docs.Dify.AI:443/a/#section", "https://docs.dify.ai/b/"] + return { + "kind": "website_crawl", + "name": "Dify docs", + "provider": "firecrawl", + "root_url": "https://docs.dify.ai", + "crawl_options": {"include_subpages": True, "limit": limit}, + "selection": [{"source_url": url} for url in selected], + } + + +def test_initial_website_selection_populates_stable_canonical_urls() -> None: + payload = KnowledgeFSInitialWebsiteSourcePayload.model_validate(_payload()) + + assert [item.canonical_url for item in payload.selection] == [ + "https://docs.dify.ai/a", + "https://docs.dify.ai/b", + ] + + +def test_initial_website_selection_rejects_canonical_duplicates() -> None: + with pytest.raises(ValidationError, match="canonical URLs must be unique"): + KnowledgeFSInitialWebsiteSourcePayload.model_validate( + _payload(urls=["https://docs.dify.ai/a/", "https://DOCS.dify.ai:443/a#section"]) + ) + + +def test_initial_website_selection_requires_crawl_limit_to_cover_selection() -> None: + with pytest.raises(ValidationError, match="crawl limit must cover every selected URL"): + KnowledgeFSInitialWebsiteSourcePayload.model_validate(_payload(limit=1)) + + +def test_initial_website_source_rejects_stale_preview_configuration() -> None: + payload = _payload() + payload["previewConfigurationFingerprint"] = "0" * 64 + + with pytest.raises(ValidationError, match="preview configuration no longer matches"): + KnowledgeFSInitialWebsiteSourcePayload.model_validate(payload) + + +def test_initial_website_source_accepts_matching_preview_configuration() -> None: + payload = KnowledgeFSInitialWebsiteSourcePayload.model_validate(_payload()) + raw = _payload() + raw["previewConfigurationFingerprint"] = knowledge_fs_initial_preview_configuration_fingerprint(payload) + + validated = KnowledgeFSInitialWebsiteSourcePayload.model_validate(raw) + + assert validated.preview_configuration_fingerprint == knowledge_fs_initial_preview_configuration_fingerprint( + validated + ) diff --git a/api/tests/unit_tests/services/test_knowledge_fs_source_import_commit_service.py b/api/tests/unit_tests/services/test_knowledge_fs_source_import_commit_service.py new file mode 100644 index 00000000000..6067ebb014e --- /dev/null +++ b/api/tests/unit_tests/services/test_knowledge_fs_source_import_commit_service.py @@ -0,0 +1,162 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from services.knowledge_fs.product_dto import KnowledgeFSAsyncSourceImportPayload +from services.knowledge_fs.source_import_commit_service import ( + commit_source_import, + resume_committed_source_import, +) + + +def test_commit_crawl_preview_selection_makes_source_visible_and_dispatches_reconciler() -> None: + facade = MagicMock() + facade.get_source_workflow.return_value = SimpleNamespace(source_id="source-1") + facade.select_crawl_preview_pages.return_value = SimpleNamespace(id="import-1", source_id="source-1") + facade.get_source.return_value = SimpleNamespace( + id="source-1", metadata={"preview": True}, status="disabled", version=3 + ) + payload = KnowledgeFSAsyncSourceImportPayload.model_validate( + { + "kind": "crawl-preview-selection", + "pageIds": ["page-1"], + "previewWorkflowId": "preview-1", + "syncPolicy": {"enabled": False, "mode": "manual"}, + } + ).root + + with patch("tasks.knowledge_fs_source_import_tasks.finalize_source_import.delay") as delay: + result = commit_source_import( + facade=facade, + tenant_id="tenant-1", + account_id="account-1", + control_space_id="control-1", + source_id="source-1", + payload=payload, + idempotency_key="request-1", + ) + + assert result.id == "import-1" + selection = facade.select_crawl_preview_pages.call_args.kwargs + assert selection["payload"].page_ids == ["page-1"] + source_update = facade.update_source.call_args.kwargs["payload"] + assert source_update.status == "syncing" + assert source_update.metadata["preview"] is False + assert source_update.metadata["pendingImport"] == { + "kind": "crawl-preview-selection", + "previewWorkflowId": "preview-1", + "workflowId": "import-1", + "syncPolicy": {"customIntervalSeconds": None, "enabled": False, "mode": "manual"}, + } + delay.assert_called_once_with( + tenant_id="tenant-1", + account_id="account-1", + control_space_id="control-1", + source_id="source-1", + workflow_id="import-1", + ) + + +def test_commit_online_document_import_uses_same_async_reconciliation() -> None: + facade = MagicMock() + facade.import_source_workflow.return_value = SimpleNamespace(id="import-1", source_id="source-1") + facade.get_source.return_value = SimpleNamespace( + id="source-1", metadata={"preview": True}, status="disabled", version=3 + ) + payload = KnowledgeFSAsyncSourceImportPayload.model_validate( + { + "kind": "online-document-import", + "items": [ + { + "pageId": "page-1", + "providerItemId": "provider-page-1", + "type": "page", + "workspaceId": "workspace-1", + } + ], + "syncPolicy": {"enabled": True, "mode": "provider"}, + } + ).root + + with patch("tasks.knowledge_fs_source_import_tasks.finalize_source_import.delay"): + commit_source_import( + facade=facade, + tenant_id="tenant-1", + account_id="account-1", + control_space_id="control-1", + source_id="source-1", + payload=payload, + idempotency_key="request-1", + ) + + workflow_payload = facade.import_source_workflow.call_args.kwargs["payload"].root + assert workflow_payload.kind == "online-document-import" + assert workflow_payload.items[0].page_id == "page-1" + pending = facade.update_source.call_args.kwargs["payload"].metadata["pendingImport"] + assert pending["kind"] == "online-document-import" + assert "previewWorkflowId" not in pending + + +def test_commit_online_drive_import_uses_same_async_reconciliation() -> None: + facade = MagicMock() + facade.import_source_workflow.return_value = SimpleNamespace(id="import-1", source_id="source-1") + facade.get_source.return_value = SimpleNamespace( + id="source-1", metadata={"preview": True}, status="disabled", version=3 + ) + payload = KnowledgeFSAsyncSourceImportPayload.model_validate( + { + "kind": "online-drive-import", + "items": [{"id": "file-1", "name": "Plan.pdf", "providerItemId": "provider-file-1"}], + "syncPolicy": {"enabled": False, "mode": "manual"}, + } + ).root + + with patch("tasks.knowledge_fs_source_import_tasks.finalize_source_import.delay"): + commit_source_import( + facade=facade, + tenant_id="tenant-1", + account_id="account-1", + control_space_id="control-1", + source_id="source-1", + payload=payload, + idempotency_key="request-1", + ) + + workflow_payload = facade.import_source_workflow.call_args.kwargs["payload"].root + assert workflow_payload.kind == "online-drive-import" + assert workflow_payload.items[0].id == "file-1" + pending = facade.update_source.call_args.kwargs["payload"].metadata["pendingImport"] + assert pending["kind"] == "online-drive-import" + + +def test_resume_committed_source_import_restores_pending_marker() -> None: + facade = MagicMock() + facade.get_source.return_value = SimpleNamespace( + id="source-1", + metadata={ + "preview": False, + "lastImport": { + "kind": "crawl-preview-selection", + "previewWorkflowId": "preview-1", + "syncPolicy": {"enabled": False, "mode": "manual"}, + "workflowId": "import-1", + }, + }, + status="error", + version=5, + ) + workflow = SimpleNamespace(id="import-1", source_id="source-1") + + with patch("tasks.knowledge_fs_source_import_tasks.finalize_source_import.delay") as delay: + resume_committed_source_import( + facade=facade, + tenant_id="tenant-1", + account_id="account-1", + control_space_id="control-1", + workflow=workflow, + ) + + update = facade.update_source.call_args.kwargs["payload"] + assert update.status == "syncing" + assert "lastImport" not in update.metadata + assert update.metadata["pendingImport"]["workflowId"] == "import-1" + delay.assert_called_once() diff --git a/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py b/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py index 9af97d2ad13..bab22e27d52 100644 --- a/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py +++ b/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py @@ -648,12 +648,14 @@ def test_initial_website_source_import_exposes_failed_source_without_activating_ source_update_payload = facade.update_source.call_args.kwargs["payload"] assert source_update_payload.status == "disabled" assert source_update_payload.metadata["preview"] is False - assert source_update_payload.metadata["initialImport"] == { - "errorCode": "SOURCE_DOCUMENT_MATERIALIZATION_FAILED", - "errorMessage": "Source document materialization failed", - "state": "failed", - "workflowId": "workflow-1", - } + initial_import = source_update_payload.metadata["initialImport"] + assert initial_import["errorCode"] == "SOURCE_DOCUMENT_MATERIALIZATION_FAILED" + assert initial_import["errorMessage"] == "Source document materialization failed" + assert initial_import["state"] == "failed" + assert initial_import["workflowId"] == "workflow-1" + assert initial_import["requestedSourceUrls"] == ["https://docs.dify.ai/a", "https://docs.dify.ai/b"] + assert initial_import["canonicalSourceUrls"] == ["https://docs.dify.ai/a", "https://docs.dify.ai/b"] + assert len(initial_import["configurationFingerprint"]) == 64 facade.update_source_sync_policy.assert_not_called() diff --git a/api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py b/api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py new file mode 100644 index 00000000000..2a91b46b4c0 --- /dev/null +++ b/api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py @@ -0,0 +1,93 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from services.knowledge_fs.product_remote import KnowledgeFSProductResourceNotFoundError +from tasks.knowledge_fs_source_import_tasks import ( + KnowledgeFSSourceImportNotReadyError, + finalize_source_import_once, +) + + +def _source(*, status: str = "syncing") -> SimpleNamespace: + return SimpleNamespace( + id="source-1", + metadata={ + "preview": False, + "pendingImport": { + "kind": "crawl-preview-selection", + "previewWorkflowId": "preview-1", + "workflowId": "import-1", + "syncPolicy": {"enabled": False, "mode": "manual"}, + }, + }, + status=status, + version=4, + ) + + +def _run(facade: MagicMock) -> str: + with patch("tasks.knowledge_fs_source_import_tasks.get_knowledge_fs_runtime") as runtime: + runtime.return_value.facade = facade + return finalize_source_import_once( + tenant_id="tenant-1", + account_id="account-1", + control_space_id="control-1", + source_id="source-1", + workflow_id="import-1", + ) + + +def test_finalize_source_import_waits_for_terminal_workflow() -> None: + facade = MagicMock() + facade.get_source_workflow.return_value = SimpleNamespace(state="importing") + + with pytest.raises(KnowledgeFSSourceImportNotReadyError): + _run(facade) + + facade.get_source.assert_not_called() + + +def test_finalize_source_import_applies_policy_then_activates_source() -> None: + facade = MagicMock() + facade.get_source_workflow.return_value = SimpleNamespace(state="completed") + facade.get_source.return_value = _source() + facade.get_source_sync_policy.side_effect = KnowledgeFSProductResourceNotFoundError("missing") + + assert _run(facade) == "import-1" + + policy = facade.update_source_sync_policy.call_args.kwargs["payload"] + assert policy.enabled is False + assert policy.mode == "manual" + assert policy.expected_revision == 0 + assert policy.expected_source_version == 4 + update = facade.update_source.call_args.kwargs["payload"] + assert update.status == "active" + assert "pendingImport" not in update.metadata + + +def test_finalize_source_import_persists_failure_on_visible_source() -> None: + facade = MagicMock() + facade.get_source_workflow.return_value = SimpleNamespace( + failure=SimpleNamespace(message="provider failed"), + id="import-1", + last_error_code="PROVIDER_FAILED", + state="failed", + ) + facade.get_source.return_value = _source() + + assert _run(facade) == "import-1" + + update = facade.update_source.call_args.kwargs["payload"] + assert update.status == "error" + assert update.metadata["lastImport"] == { + "errorCode": "PROVIDER_FAILED", + "errorMessage": "provider failed", + "kind": "crawl-preview-selection", + "previewWorkflowId": "preview-1", + "state": "failed", + "syncPolicy": {"enabled": False, "mode": "manual"}, + "workflowId": "import-1", + } + facade.update_source_sync_policy.assert_not_called() diff --git a/knowledge-fs/packages/api/src/source-product-workflow-runtime.test.ts b/knowledge-fs/packages/api/src/source-product-workflow-runtime.test.ts index bef9e967ac7..cb95363f395 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-runtime.test.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-runtime.test.ts @@ -917,6 +917,36 @@ describe("source-product workflow provider imports", () => { }); }); + it("fails a selected URL crawl import when canonical matching is ambiguous", async () => { + const source = sourceRecord("ambiguous-crawl-import", { type: "web" }); + const fixture = await createFixture({ + contentStore: { + deleteRun: vi.fn(async () => ({ deleted: 0, hasMore: false })), + get: vi.fn(async () => null), + put: vi.fn(async () => "staged/ambiguous"), + }, + inventory: [], + run: providerRun(source.id, "crawl-import", { + selectedSourceUrls: ["https://example.test/selected#preview"], + }), + source, + websiteCrawl: { + crawl: vi.fn(async () => ({ + pages: [ + { content: "one", sourceUrl: "https://example.test/selected", title: "One" }, + { content: "two", sourceUrl: "https://example.test/selected/", title: "Two" }, + ], + })), + }, + }); + + await expect(fixture.runtime.tick()).resolves.toMatchObject({ completed: 0, failed: 1 }); + await expect(fixture.getRun()).resolves.toMatchObject({ + lastErrorCode: "SOURCE_CRAWL_PAGE_AMBIGUOUS", + state: "failed", + }); + }); + it("imports only the selected URLs from the second crawl", async () => { const source = sourceRecord("selected-url-crawl-success", { metadata: { preview: true }, @@ -953,7 +983,7 @@ describe("source-product workflow provider imports", () => { materialize, }, run: providerRun(source.id, "crawl-import", { - selectedSourceUrls: ["https://example.test/selected"], + selectedSourceUrls: ["HTTPS://EXAMPLE.TEST:443/selected/#preview"], }), source, websiteCrawl: { @@ -961,7 +991,7 @@ describe("source-product workflow provider imports", () => { pages: [ { content: "selected body", - sourceUrl: "https://example.test/selected", + sourceUrl: "https://example.test/selected/", title: "Selected", }, { diff --git a/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts b/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts index 84e13294aea..5e1d5d47027 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts @@ -760,7 +760,7 @@ async function processSelectedCrawlImport( throw runtimeError("SOURCE_CRAWL_PAGE_NOT_FOUND", "Selected crawl page is unavailable"); } - const matched = new Map(); + const crawledPages: SourceCrawlPreviewPage[] = []; let cursor: string | undefined; do { const page = await input.repository.listCrawlPages({ @@ -768,22 +768,18 @@ async function processSelectedCrawlImport( limit: 200, runId: execution.run().id, }); - for (const candidate of page.items) { - if (!requestedUrls.has(candidate.sourceUrl)) continue; - if (matched.has(candidate.sourceUrl)) { - throw runtimeError( - "SOURCE_CRAWL_PAGE_AMBIGUOUS", - "Selected crawl page matched more than one result", - ); - } - matched.set(candidate.sourceUrl, candidate); - } + crawledPages.push(...page.items); cursor = page.nextCursor; } while (cursor); - if (matched.size !== requestedUrls.size) { - throw runtimeError("SOURCE_CRAWL_PAGE_NOT_FOUND", "Selected crawl page is unavailable"); - } + const selectedPages = [...requestedUrls].map((sourceUrl) => + matchSelectedCrawlPage(sourceUrl, crawledPages), + ); + if (new Set(selectedPages.map((page) => page.pageId)).size !== selectedPages.length) + throw runtimeError( + "SOURCE_CRAWL_PAGE_AMBIGUOUS", + "Selected crawl URLs resolved to the same page", + ); await execution.mutate((current) => input.repository.checkpoint({ checkpoint: "selection-frozen", @@ -796,16 +792,47 @@ async function processSelectedCrawlImport( state: "importing", }), ); - const selectedPages = [...requestedUrls].map((sourceUrl) => { - const page = matched.get(sourceUrl); - if (!page) { - throw runtimeError("SOURCE_CRAWL_PAGE_NOT_FOUND", "Selected crawl page is unavailable"); - } - return page; - }); await importCrawlPages(input, execution, source, selectedPages); } +function matchSelectedCrawlPage( + requestedUrl: string, + candidates: readonly SourceCrawlPreviewPage[], +): SourceCrawlPreviewPage { + const exact = candidates.filter((candidate) => candidate.sourceUrl === requestedUrl); + if (exact.length > 1) + throw runtimeError( + "SOURCE_CRAWL_PAGE_AMBIGUOUS", + "Selected crawl page matched more than one result", + ); + const [exactMatch] = exact; + if (exactMatch) return exactMatch; + + const canonicalRequestedUrl = canonicalCrawlUrl(requestedUrl); + const canonical = candidates.filter( + (candidate) => canonicalCrawlUrl(candidate.sourceUrl) === canonicalRequestedUrl, + ); + if (canonical.length > 1) + throw runtimeError( + "SOURCE_CRAWL_PAGE_AMBIGUOUS", + "Selected crawl page canonical URL matched more than one result", + ); + const [canonicalMatch] = canonical; + if (canonicalMatch) return canonicalMatch; + throw runtimeError("SOURCE_CRAWL_PAGE_NOT_FOUND", "Selected crawl page is unavailable"); +} + +function canonicalCrawlUrl(value: string): string { + try { + const url = new URL(value); + url.hash = ""; + if (url.pathname !== "/") url.pathname = url.pathname.replace(/\/+$/, ""); + return url.toString(); + } catch { + return value.trim(); + } +} + async function importCrawlPages( input: Parameters[0], execution: RuntimeExecution,