From 06cd47807cd99ae0fccda06cc796dcd24d55ec00 Mon Sep 17 00:00:00 2001 From: FFXN Date: Thu, 3 Sep 2026 21:01:21 +0800 Subject: [PATCH] feat(knowledge-fs): add content handling for initial source preview and cleanup tasks --- api/knowledge-fs-contract.lock.json | 4 +- .../knowledge_fs/initial_source_preview.py | 1 + .../initial_source_preview_job.py | 94 +++++++++++++++++++ api/services/knowledge_fs/product_dto.py | 18 ++++ ...owledge_fs_initial_source_preview_tasks.py | 15 ++- .../knowledge_fs_initial_source_tasks.py | 19 +++- ...knowledge_fs_initial_source_preview_job.py | 67 ++++++++++++- ...owledge_fs_initial_source_preview_tasks.py | 9 ++ .../api/src/source-product-handlers.ts | 1 + .../packages/api/src/source-product-routes.ts | 12 +++ ...ce-product-workflow-database-repository.ts | 3 + ...urce-product-workflow-memory-repository.ts | 7 ++ .../source-product-workflow-runtime.test.ts | 38 ++++++++ .../src/source-product-workflow-runtime.ts | 32 +++++++ .../api/src/source-product-workflow.ts | 21 ++++- .../api/console/knowledge-fs/types.gen.ts | 9 ++ .../api/console/knowledge-fs/zod.gen.ts | 26 +++-- .../new-rag/create/__tests__/page.spec.tsx | 1 + web/features/new-rag/create/source-setup.tsx | 12 ++- 19 files changed, 370 insertions(+), 19 deletions(-) diff --git a/api/knowledge-fs-contract.lock.json b/api/knowledge-fs-contract.lock.json index ccc8db37231..5c40bbbcf17 100644 --- a/api/knowledge-fs-contract.lock.json +++ b/api/knowledge-fs-contract.lock.json @@ -1,7 +1,7 @@ { "schemaVersion": 5, - "subtreeTree": "c843ce519c9cbca708acdc37bc1900a7513469a4", - "openapiSha256": "bb5c1f6529ee216059b4aa1e01d8079a0bb175bb6adba7a0821fec8b6fa707f1", + "subtreeTree": "c24990b5756696f79ab432125dce9f521d9fc873", + "openapiSha256": "5f9bee6a3593d8bd05308c7a57ab3d02ef451133cde9347a40543fae934a9a59", "capabilityV2AuthManifestSha256": "e322a2fa779d1f40b95c54c1021cffecaec77abbd7b34899573dcdf4ff353109", "capabilityV2AuthTestVectorSha256": "ae0de37b1ff05c40f905cf17a7b410d8971acacf64db07d5ee3d6fecfa559ce3", "productOperationManifestSha256": "5d1241a83bcca12ebbd848928dd3cdda0d2ecaeb5f5336e955eb24a8c8db175b", diff --git a/api/services/knowledge_fs/initial_source_preview.py b/api/services/knowledge_fs/initial_source_preview.py index 0644801bd12..71356e47d93 100644 --- a/api/services/knowledge_fs/initial_source_preview.py +++ b/api/services/knowledge_fs/initial_source_preview.py @@ -110,6 +110,7 @@ class KnowledgeFSInitialSourcePreviewService: _raise_if_canceled(is_canceled) for website_page in website_message.result.web_info_list or []: pages_by_url[website_page.source_url] = KnowledgeFSInitialSourcePreviewPageResponse( + content=website_page.content, description=website_page.description or None, source_url=website_page.source_url, title=website_page.title or None, diff --git a/api/services/knowledge_fs/initial_source_preview_job.py b/api/services/knowledge_fs/initial_source_preview_job.py index 55297d24e96..b3da1a34996 100644 --- a/api/services/knowledge_fs/initial_source_preview_job.py +++ b/api/services/knowledge_fs/initial_source_preview_job.py @@ -3,12 +3,15 @@ from __future__ import annotations import json +import logging import uuid +from hashlib import sha256 from typing import Literal from celery import current_app as celery_app from extensions.ext_redis import redis_client +from extensions.ext_storage import storage from models.account import Account from services.knowledge_fs.initial_source_preview import KnowledgeFSInitialSourcePreviewService from services.knowledge_fs.product_dto import ( @@ -19,6 +22,8 @@ from services.knowledge_fs.product_dto import ( ) _JOB_TTL_SECONDS = 60 * 60 +_CONTENT_MANIFEST_TTL_SECONDS = 2 * _JOB_TTL_SECONDS +logger = logging.getLogger(__name__) _RELEASE_ACTIVE_JOB_SCRIPT = """ local current = redis.call('GET', KEYS[1]) if current == ARGV[1] then @@ -58,6 +63,10 @@ def _active_job_key(*, tenant_id: str, account_id: str) -> str: return f"knowledge_fs:initial_source_preview:{tenant_id}:{account_id}:active" +def _content_key(*, tenant_id: str, account_id: str, job_id: str) -> str: + return f"knowledge_fs:initial_source_preview:{tenant_id}:{account_id}:{job_id}:content" + + class KnowledgeFSInitialSourcePreviewJobService: def __init__(self, session_maker) -> None: self._preview_service = KnowledgeFSInitialSourcePreviewService(session_maker) @@ -107,6 +116,91 @@ class KnowledgeFSInitialSourcePreviewJobService: raise KnowledgeFSInitialSourcePreviewJobNotFoundError(job_id) return KnowledgeFSInitialSourcePreviewJobResponse.model_validate_json(raw) + @staticmethod + def store_content( + *, tenant_id: str, account_id: str, job_id: str, result: KnowledgeFSInitialSourcePreviewResponse + ) -> None: + pages: dict[str, dict[str, object]] = {} + saved_keys: list[str] = [] + try: + for page in result.pages or []: + if page.content is None: + continue + page_digest = sha256(page.source_url.encode()).hexdigest() + object_key = f"knowledge_fs/initial_source_previews/{tenant_id}/{account_id}/{job_id}/{page_digest}.md" + storage.save(object_key, page.content.encode()) + saved_keys.append(object_key) + pages[page.source_url] = { + "description": page.description, + "objectKey": object_key, + "sourceUrl": page.source_url, + "title": page.title, + } + redis_client.setex( + _content_key(tenant_id=tenant_id, account_id=account_id, job_id=job_id), + _CONTENT_MANIFEST_TTL_SECONDS, + json.dumps(pages, ensure_ascii=False, separators=(",", ":")), + ) + except Exception: + for object_key in saved_keys: + storage.delete(object_key) + raise + + @classmethod + def selected_content( + cls, + *, + tenant_id: str, + account_id: str, + job_id: str, + source_urls: list[str], + configuration_fingerprint: str, + ) -> list[dict[str, object]]: + job = cls.get(tenant_id=tenant_id, account_id=account_id, job_id=job_id) + if ( + job.status != "completed" + or job.result is None + or job.result.configuration_fingerprint != configuration_fingerprint + ): + raise KnowledgeFSInitialSourcePreviewJobNotFoundError(job_id) + raw = redis_client.get(_content_key(tenant_id=tenant_id, account_id=account_id, job_id=job_id)) + if raw is None: + raise KnowledgeFSInitialSourcePreviewJobNotFoundError(job_id) + pages = json.loads(raw) + try: + selected = [pages[source_url] for source_url in source_urls] + return [ + { + "content": storage.load_once(page["objectKey"]).decode(), + "description": page["description"], + "sourceUrl": page["sourceUrl"], + "title": page["title"], + } + for page in selected + ] + except (KeyError, TypeError, UnicodeDecodeError) as exc: + raise KnowledgeFSInitialSourcePreviewJobNotFoundError(job_id) from exc + + @staticmethod + def cleanup_content(*, tenant_id: str, account_id: str, job_id: str) -> None: + key = _content_key(tenant_id=tenant_id, account_id=account_id, job_id=job_id) + raw = redis_client.get(key) + if raw is None: + return + pages = json.loads(raw) + cleanup_failed = False + for page in pages.values(): + try: + storage.delete(page["objectKey"]) + except Exception: + cleanup_failed = True + logger.exception( + "KnowledgeFS initial source preview content cleanup failed", + extra={"account_id": account_id, "job_id": job_id, "tenant_id": tenant_id}, + ) + if not cleanup_failed: + redis_client.delete(key) + @classmethod def cancel(cls, *, tenant_id: str, account_id: str, job_id: str) -> KnowledgeFSInitialSourcePreviewJobResponse: current = cls.get(tenant_id=tenant_id, account_id=account_id, job_id=job_id) diff --git a/api/services/knowledge_fs/product_dto.py b/api/services/knowledge_fs/product_dto.py index 0b0e1bb8834..4369c3e5051 100644 --- a/api/services/knowledge_fs/product_dto.py +++ b/api/services/knowledge_fs/product_dto.py @@ -232,6 +232,7 @@ class KnowledgeFSInitialWebsiteSourcePayload(KnowledgeFSInitialSyncPolicyPayload preview_configuration_fingerprint: str | None = Field( default=None, min_length=64, max_length=64, alias="previewConfigurationFingerprint" ) + preview_job_id: str | None = Field(default=None, min_length=1, max_length=255, alias="previewJobId") root_url: str = Field(min_length=1, max_length=4_096) crawl_options: KnowledgeFSInitialWebsiteCrawlOptionsPayload selection: list[KnowledgeFSInitialWebsiteSelectionPayload] = Field(min_length=1, max_length=200) @@ -284,6 +285,7 @@ def knowledge_fs_initial_preview_configuration_fingerprint( class KnowledgeFSInitialSourcePreviewPageResponse(ResponseModel): + content: str | None = Field(default=None, exclude=True) description: str | None = None source_url: str = Field(validation_alias=AliasChoices("source_url", "sourceUrl")) title: str | None = None @@ -2537,8 +2539,18 @@ class KnowledgeFSAsyncSourceImportPayload(RootModel[KnowledgeFSAsyncSourceImport pass +class KnowledgeFSCrawlImportPagePayload(BaseModel): + content: str = Field(max_length=10_000_000) + description: str | None = None + source_url: str = Field(min_length=1, max_length=4_096, alias="sourceUrl") + title: str | None = Field(default=None, max_length=500) + + model_config = ConfigDict(extra="forbid", validate_by_alias=True, validate_by_name=True) + + class KnowledgeFSCrawlImportPayload(BaseModel): source_urls: list[str] = Field(min_length=1, max_length=200, alias="sourceUrls") + pages: list[KnowledgeFSCrawlImportPagePayload] | None = Field(default=None, min_length=1, max_length=200) model_config = ConfigDict(extra="forbid", validate_by_alias=True, validate_by_name=True) @@ -2550,6 +2562,12 @@ class KnowledgeFSCrawlImportPayload(BaseModel): raise ValueError("source URLs must be non-empty and at most 4096 characters") return normalized + @model_validator(mode="after") + def validate_pages(self) -> KnowledgeFSCrawlImportPayload: + if self.pages is not None and [page.source_url for page in self.pages] != self.source_urls: + raise ValueError("crawl import pages must match source URLs in order") + return self + class KnowledgeFSCrawledPageResponse(ResponseModel): content: str diff --git a/api/tasks/knowledge_fs_initial_source_preview_tasks.py b/api/tasks/knowledge_fs_initial_source_preview_tasks.py index da7c80b4f90..eb102fe2748 100644 --- a/api/tasks/knowledge_fs_initial_source_preview_tasks.py +++ b/api/tasks/knowledge_fs_initial_source_preview_tasks.py @@ -72,7 +72,8 @@ def run_knowledge_fs_initial_source_preview( ), ) result.configuration_fingerprint = knowledge_fs_initial_preview_configuration_fingerprint(preview_payload) - job_service.transition_status( + job_service.store_content(tenant_id=tenant_id, account_id=account_id, job_id=job_id, result=result) + completed = job_service.transition_status( tenant_id=tenant_id, account_id=account_id, job_id=job_id, @@ -80,6 +81,11 @@ def run_knowledge_fs_initial_source_preview( allowed_from=("running",), result=result, ) + if completed: + cleanup_knowledge_fs_initial_source_preview.apply_async( + kwargs={"account_id": account_id, "job_id": job_id, "tenant_id": tenant_id}, + countdown=60 * 60, + ) except KnowledgeFSInitialSourcePreviewCanceledError: logger.info( "KnowledgeFS initial source preview canceled", @@ -118,4 +124,9 @@ def _preview_was_canceled( return True -__all__ = ["run_knowledge_fs_initial_source_preview"] +@shared_task(queue="dataset") +def cleanup_knowledge_fs_initial_source_preview(*, tenant_id: str, account_id: str, job_id: str) -> None: + KnowledgeFSInitialSourcePreviewJobService.cleanup_content(tenant_id=tenant_id, account_id=account_id, job_id=job_id) + + +__all__ = ["cleanup_knowledge_fs_initial_source_preview", "run_knowledge_fs_initial_source_preview"] diff --git a/api/tasks/knowledge_fs_initial_source_tasks.py b/api/tasks/knowledge_fs_initial_source_tasks.py index cf4da15b82b..d635d0eb393 100644 --- a/api/tasks/knowledge_fs_initial_source_tasks.py +++ b/api/tasks/knowledge_fs_initial_source_tasks.py @@ -280,16 +280,33 @@ def _start_workflow( payload: KnowledgeFSInitialSourcePayload, ): if isinstance(payload, KnowledgeFSInitialWebsiteSourcePayload): - return facade.import_selected_source_crawl( + pages = None + if payload.preview_job_id: + from services.knowledge_fs.initial_source_preview_job import KnowledgeFSInitialSourcePreviewJobService + + pages = KnowledgeFSInitialSourcePreviewJobService.selected_content( + tenant_id=tenant_id, + account_id=account_id, + job_id=payload.preview_job_id, + source_urls=[selection.source_url for selection in payload.selection], + configuration_fingerprint=knowledge_fs_initial_preview_configuration_fingerprint(payload), + ) + workflow = facade.import_selected_source_crawl( tenant_id=tenant_id, account_id=account_id, control_space_id=control_space_id, source_id=source_id, payload=KnowledgeFSCrawlImportPayload( sourceUrls=[selection.source_url for selection in payload.selection], + pages=pages, ), idempotency_key=f"{request_id}:crawl-import", ) + if payload.preview_job_id: + KnowledgeFSInitialSourcePreviewJobService.cleanup_content( + tenant_id=tenant_id, account_id=account_id, job_id=payload.preview_job_id + ) + return workflow if isinstance(payload, KnowledgeFSInitialOnlineDocumentSourcePayload): import_payload = KnowledgeFSSourceWorkflowImportPayload( KnowledgeFSOnlineDocumentWorkflowImportPayload( diff --git a/api/tests/unit_tests/services/test_knowledge_fs_initial_source_preview_job.py b/api/tests/unit_tests/services/test_knowledge_fs_initial_source_preview_job.py index c235323ab55..2276be74e18 100644 --- a/api/tests/unit_tests/services/test_knowledge_fs_initial_source_preview_job.py +++ b/api/tests/unit_tests/services/test_knowledge_fs_initial_source_preview_job.py @@ -11,7 +11,11 @@ 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 ( + KnowledgeFSInitialSourcePreviewPageResponse, + KnowledgeFSInitialSourcePreviewResponse, + KnowledgeFSInitialWebsiteSourcePreviewPayload, +) def _payload() -> KnowledgeFSInitialWebsiteSourcePreviewPayload: @@ -109,9 +113,7 @@ def test_start_releases_the_active_preview_slot_when_enqueue_fails() -> None: service.start(tenant_id="tenant-1", account=account, payload=_payload()) job_id = redis_eval.call_args.args[3] - delete.assert_called_once_with( - f"knowledge_fs:initial_source_preview:tenant-1:account-1:{job_id}" - ) + delete.assert_called_once_with(f"knowledge_fs:initial_source_preview:tenant-1:account-1:{job_id}") assert redis_eval.call_args.args[2:] == ( "knowledge_fs:initial_source_preview:tenant-1:account-1:active", job_id, @@ -201,3 +203,60 @@ def test_terminal_canceled_status_cannot_be_overwritten() -> None: assert transitioned is False assert '"status":"completed"' in eval_status.call_args.args[3] assert eval_status.call_args.args[5:] == ("running",) + + +def test_completed_preview_keeps_content_private_and_selectable() -> None: + result = KnowledgeFSInitialSourcePreviewResponse( + configurationFingerprint="a" * 64, + kind="website_crawl", + pages=[ + KnowledgeFSInitialSourcePreviewPageResponse( + content="# Preview body", + source_url="https://docs.dify.ai/page", + title="Page", + ) + ], + ) + cache: dict[str, str] = {} + objects: dict[str, bytes] = {} + + with ( + patch( + "services.knowledge_fs.initial_source_preview_job.redis_client.setex", + side_effect=lambda key, _ttl, value: cache.__setitem__(key, value), + ), + patch( + "services.knowledge_fs.initial_source_preview_job.storage.save", + side_effect=lambda key, value: objects.__setitem__(key, value), + ), + ): + KnowledgeFSInitialSourcePreviewJobService.set_status( + tenant_id="tenant-1", account_id="account-1", job_id="job-1", status="completed", result=result + ) + KnowledgeFSInitialSourcePreviewJobService.store_content( + tenant_id="tenant-1", account_id="account-1", job_id="job-1", result=result + ) + + public_value = cache["knowledge_fs:initial_source_preview:tenant-1:account-1:job-1"] + assert "Preview body" not in public_value + + with ( + patch("services.knowledge_fs.initial_source_preview_job.redis_client.get", side_effect=cache.get), + patch("services.knowledge_fs.initial_source_preview_job.storage.load_once", side_effect=objects.get), + ): + pages = KnowledgeFSInitialSourcePreviewJobService.selected_content( + tenant_id="tenant-1", + account_id="account-1", + job_id="job-1", + source_urls=["https://docs.dify.ai/page"], + configuration_fingerprint="a" * 64, + ) + + assert pages == [ + { + "content": "# Preview body", + "description": None, + "sourceUrl": "https://docs.dify.ai/page", + "title": "Page", + } + ] diff --git a/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_preview_tasks.py b/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_preview_tasks.py index 78139098c31..01981dfa917 100644 --- a/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_preview_tasks.py +++ b/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_preview_tasks.py @@ -54,6 +54,9 @@ def test_preview_task_persists_running_and_completed_states() -> None: return_value=session_context, ), patch("tasks.knowledge_fs_initial_source_preview_tasks.session_factory.get_session_maker"), + patch( + "tasks.knowledge_fs_initial_source_preview_tasks.cleanup_knowledge_fs_initial_source_preview.apply_async" + ) as schedule_cleanup, ): run_knowledge_fs_initial_source_preview.run( tenant_id="tenant-1", @@ -79,6 +82,12 @@ def test_preview_task_persists_running_and_completed_states() -> None: result=result, ), ] + job_service.store_content.assert_called_once_with( + tenant_id="tenant-1", account_id="account-1", job_id="job-1", result=result + ) + schedule_cleanup.assert_called_once_with( + kwargs={"account_id": "account-1", "job_id": "job-1", "tenant_id": "tenant-1"}, countdown=3600 + ) job_service.release_active_job.assert_called_once_with( tenant_id="tenant-1", account_id="account-1", diff --git a/knowledge-fs/packages/api/src/source-product-handlers.ts b/knowledge-fs/packages/api/src/source-product-handlers.ts index 191dbd84d05..404d182e752 100644 --- a/knowledge-fs/packages/api/src/source-product-handlers.ts +++ b/knowledge-fs/packages/api/src/source-product-handlers.ts @@ -373,6 +373,7 @@ export function registerSourceProductHandlers(input: { ...principal(context), idempotencyKey: headers["Idempotency-Key"], knowledgeSpaceId: params.id, + ...(body.pages ? { pages: body.pages } : {}), sourceId: params.sourceId, sourceUrls: body.sourceUrls, }), diff --git a/knowledge-fs/packages/api/src/source-product-routes.ts b/knowledge-fs/packages/api/src/source-product-routes.ts index fa88a803123..fedbec2fb01 100644 --- a/knowledge-fs/packages/api/src/source-product-routes.ts +++ b/knowledge-fs/packages/api/src/source-product-routes.ts @@ -319,6 +319,18 @@ export const createSourceCrawlImportWorkflowRoute = createRoute({ "application/json": { schema: z .object({ + pages: z + .array( + z.object({ + content: z.string().max(10_000_000), + description: z.string().nullable().optional(), + sourceUrl: z.string().url().max(4096), + title: z.string().nullable().optional(), + }), + ) + .min(1) + .max(200) + .optional(), sourceUrls: z.array(z.string().url().max(4096)).min(1).max(200), }) .strict(), diff --git a/knowledge-fs/packages/api/src/source-product-workflow-database-repository.ts b/knowledge-fs/packages/api/src/source-product-workflow-database-repository.ts index 1ed9a535b32..2a1ef463527 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-database-repository.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-database-repository.ts @@ -622,6 +622,9 @@ export function createDatabaseSourceProductWorkflowRepository(input: { const terminal = state === "completed" || state === "zero_results"; const next = await writeFenced(database, tx, current, { ...current, + ...(terminal && current.payload.stagedPages + ? { payload: Object.fromEntries(Object.entries(current.payload).filter(([key]) => key !== "stagedPages")) } + : {}), activeSlot: terminal ? undefined : current.activeSlot, checkpoint: state === "preview_ready" diff --git a/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.ts b/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.ts index b24080f1404..3b73de39325 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.ts @@ -365,6 +365,13 @@ export function createInMemorySourceProductWorkflowRepository(input?: { claimableAt.delete(run.id); return save({ ...run, + ...(terminal && run.payload.stagedPages + ? { + payload: Object.fromEntries( + Object.entries(run.payload).filter(([key]) => key !== "stagedPages"), + ), + } + : {}), activeSlot: terminal ? undefined : run.activeSlot, checkpoint: state === "preview_ready" 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 8a68e0332b6..1b944f43236 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 @@ -1105,6 +1105,44 @@ describe("source-product workflow provider imports", () => { }); }); + it("imports staged preview pages without calling the website provider", async () => { + const source = sourceRecord("staged-crawl-import", { type: "web" }); + const bodies = new Map(); + const crawl = vi.fn(); + const fixture = await createFixture({ + contentStore: { + deleteRun: vi.fn(async () => ({ deleted: bodies.size, hasMore: false })), + get: vi.fn(async ({ contentObjectKey }) => bodies.get(contentObjectKey) ?? null), + put: vi.fn(async ({ body, pageId }) => { + const key = `staged/${pageId}`; + bodies.set(key, body); + return key; + }), + }, + inventory: [], + run: providerRun(source.id, "crawl-import", { + selectedSourceUrls: ["https://example.test/selected"], + stagedPages: [ + { + content: "preview body", + sourceUrl: "https://example.test/selected", + title: "Selected", + }, + ], + }), + source, + websiteCrawl: { crawl }, + }); + + await expect(fixture.runtime.tick()).resolves.toMatchObject({ completed: 1, failed: 0 }); + expect(crawl).not.toHaveBeenCalled(); + await expect(fixture.getRun()).resolves.toMatchObject({ + payload: { selectedSourceUrls: ["https://example.test/selected"] }, + progressCompleted: 1, + state: "completed", + }); + }); + it("fetches every selected URL even when multiple selections resolve to the same page", async () => { const source = sourceRecord("ambiguous-crawl-import", { type: "web" }); const crawl = vi.fn(async (_input: WebsiteCrawlInput) => ({ 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 b8b269363a5..861c83f0e6d 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts @@ -780,6 +780,26 @@ async function processSelectedCrawlImport( execution: RuntimeExecution, source: Source, ): Promise { + const stagedPages = selectedStagedPages(execution.run()); + if (stagedPages.length) { + const selectedPages = await stageCrawlPages(input, execution, stagedPages, (page) => + createHash("sha256").update(page.sourceUrl, "utf8").digest("hex"), + ); + await execution.mutate((current) => + input.repository.checkpoint({ + checkpoint: "selection-frozen", + fence: fence(current), + now: iso((input.now ?? Date.now)()), + progressCompleted: 0, + progressFailed: 0, + progressSkipped: 0, + progressTotal: selectedPages.length, + state: "importing", + }), + ); + await importCrawlPages(input, execution, source, selectedPages); + return; + } if (!input.websiteCrawl) { throw runtimeError( "SOURCE_CRAWL_PROVIDER_UNAVAILABLE", @@ -3264,6 +3284,18 @@ function selectedSourceUrls(run: SourceWorkflowRun): readonly string[] { return value as string[]; } +function selectedStagedPages(run: SourceWorkflowRun): readonly CrawledPage[] { + const value = run.payload.stagedPages; + if (!Array.isArray(value)) return []; + return value.filter( + (page): page is CrawledPage => + typeof page === "object" && + page !== null && + typeof page.content === "string" && + typeof page.sourceUrl === "string", + ); +} + function requiredPayloadString(record: Record, key: string): string { const value = record[key]; if (typeof value !== "string" || !value.trim() || value.length > 8_192) { diff --git a/knowledge-fs/packages/api/src/source-product-workflow.ts b/knowledge-fs/packages/api/src/source-product-workflow.ts index 8f34558265e..5ec7e4e292d 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow.ts @@ -500,6 +500,12 @@ export interface SourceProductWorkflowService { input: SourceWorkflowPrincipal & { readonly idempotencyKey: string; readonly knowledgeSpaceId: string; + readonly pages?: readonly { + readonly content: string; + readonly description?: string | null; + readonly sourceUrl: string; + readonly title?: string | null; + }[]; readonly sourceId: string; readonly sourceUrls: readonly string[]; }, @@ -764,11 +770,24 @@ export function createSourceProductWorkflowService(input: { `Crawl import must contain 1-${maxImportItems} source URLs`, ); } + if ( + request.pages && + (request.pages.length !== sourceUrls.length || + request.pages.some((page, index) => page.sourceUrl !== sourceUrls[index])) + ) { + throw new SourceWorkflowError( + "SOURCE_IMPORT_ITEMS_INVALID", + "Crawl import pages must match source URLs in order", + ); + } return start(request, { idempotencyKey: request.idempotencyKey, knowledgeSpaceId: request.knowledgeSpaceId, kind: "crawl-import", - payload: { selectedSourceUrls: sourceUrls }, + payload: { + selectedSourceUrls: sourceUrls, + ...(request.pages ? { stagedPages: request.pages } : {}), + }, progressTotal: sourceUrls.length, requiredPermissionScope: requiredSourceScope(source), sourceId: request.sourceId, diff --git a/packages/contracts/generated/api/console/knowledge-fs/types.gen.ts b/packages/contracts/generated/api/console/knowledge-fs/types.gen.ts index 02ffc6f0923..da963337a57 100644 --- a/packages/contracts/generated/api/console/knowledge-fs/types.gen.ts +++ b/packages/contracts/generated/api/console/knowledge-fs/types.gen.ts @@ -1023,6 +1023,7 @@ export type KnowledgeFsAsyncSourceImportPayload = } & KnowledgeFsAsyncOnlineDriveImportPayload) export type KnowledgeFsCrawlImportPayload = { + pages?: Array | null sourceUrls: Array } @@ -1345,6 +1346,7 @@ export type KnowledgeFsInitialWebsiteSourcePayload = { } pluginId?: string | null previewConfigurationFingerprint?: string | null + previewJobId?: string | null provider: string providerDisplayName?: string | null root_url: string @@ -1994,6 +1996,13 @@ export type KnowledgeFsAsyncOnlineDriveImportPayload = { syncPolicy: KnowledgeFsDeferredSyncPolicyPayload } +export type KnowledgeFsCrawlImportPagePayload = { + content: string + description?: string | null + sourceUrl: string + title?: string | null +} + export type KnowledgeFsSourceFileBucketResponse = { bucket?: string | null continuation_token?: string | null diff --git a/packages/contracts/generated/api/console/knowledge-fs/zod.gen.ts b/packages/contracts/generated/api/console/knowledge-fs/zod.gen.ts index 49021c1d34d..a8bb604b00e 100644 --- a/packages/contracts/generated/api/console/knowledge-fs/zod.gen.ts +++ b/packages/contracts/generated/api/console/knowledge-fs/zod.gen.ts @@ -479,13 +479,6 @@ export const zKnowledgeFsSourceDeletePayload = z.object({ expectedRevision: z.int().gte(1), }) -/** - * KnowledgeFSCrawlImportPayload - */ -export const zKnowledgeFsCrawlImportPayload = z.object({ - sourceUrls: z.array(z.string()).min(1).max(200), -}) - /** * KnowledgeFSSourceSyncPolicyResponse */ @@ -1943,6 +1936,24 @@ export const zKnowledgeFsSourceEditSyncPolicyPayload = z.object({ mode: z.enum(['custom', 'interval', 'manual']), }) +/** + * KnowledgeFSCrawlImportPagePayload + */ +export const zKnowledgeFsCrawlImportPagePayload = z.object({ + content: z.string().max(10000000), + description: z.string().nullish(), + sourceUrl: z.string().min(1).max(4096), + title: z.string().max(500).nullish(), +}) + +/** + * KnowledgeFSCrawlImportPayload + */ +export const zKnowledgeFsCrawlImportPayload = z.object({ + pages: z.array(zKnowledgeFsCrawlImportPagePayload).min(1).max(200).nullish(), + sourceUrls: z.array(z.string()).min(1).max(200), +}) + /** * KnowledgeFSSourceImportPagePayload */ @@ -2282,6 +2293,7 @@ export const zKnowledgeFsInitialWebsiteSourcePayload = z.object({ parameters: z.record(z.string(), zJsonValue).optional(), pluginId: z.string().min(1).max(255).nullish(), previewConfigurationFingerprint: z.string().length(64).nullish(), + previewJobId: z.string().min(1).max(255).nullish(), provider: z.string().min(1).max(255), providerDisplayName: z.string().min(1).max(255).nullish(), root_url: z.string().min(1).max(4096), diff --git a/web/features/new-rag/create/__tests__/page.spec.tsx b/web/features/new-rag/create/__tests__/page.spec.tsx index 40accac0b6d..d3f5838782c 100644 --- a/web/features/new-rag/create/__tests__/page.spec.tsx +++ b/web/features/new-rag/create/__tests__/page.spec.tsx @@ -1664,6 +1664,7 @@ describe('CreateKnowledgePage', () => { url: 'https://docs.dify.ai', }, pluginId: 'langgenius/firecrawl_datasource', + previewJobId: 'website-preview-1', provider: 'firecrawl', providerDisplayName: 'Firecrawl', root_url: 'https://docs.dify.ai/', diff --git a/web/features/new-rag/create/source-setup.tsx b/web/features/new-rag/create/source-setup.tsx index 73415e20f4e..12466377ac0 100644 --- a/web/features/new-rag/create/source-setup.tsx +++ b/web/features/new-rag/create/source-setup.tsx @@ -201,6 +201,7 @@ function CreateSourceSetupSession({ const [selectedPageIds, setSelectedPageIds] = useState>(() => new Set()) const crawlAttemptRef = useRef(0) const previewJobIdRef = useRef(undefined) + const previewFingerprintRef = useRef(undefined) const providerOptions = useMemo( () => discoverSourceProviderOptions(draft.sourceType, datasourcePluginsQuery.data ?? []), [datasourcePluginsQuery.data, draft.sourceType], @@ -257,6 +258,7 @@ function CreateSourceSetupSession({ crawlAttemptRef.current += 1 const jobId = previewJobIdRef.current previewJobIdRef.current = undefined + previewFingerprintRef.current = undefined if (jobId) void consoleClient.knowledgeFs.sourceProviderPreview.jobs.byJobId .delete({ @@ -278,7 +280,6 @@ function CreateSourceSetupSession({ params: { job_id: jobId }, }) if (response.status === 'completed' && response.result) { - if (previewJobIdRef.current === jobId) previewJobIdRef.current = undefined setPreviewPages( (response.result.pages ?? []).map((page) => ({ description: page.description ?? undefined, @@ -287,6 +288,7 @@ function CreateSourceSetupSession({ title: page.title ?? page.source_url, })), ) + previewFingerprintRef.current = response.result.configuration_fingerprint ?? undefined setCrawlState('success') } else if (response.status === 'canceled') { if (previewJobIdRef.current === jobId) previewJobIdRef.current = undefined @@ -382,8 +384,8 @@ function CreateSourceSetupSession({ }) } if (crawlAttemptRef.current !== attempt) return - previewJobIdRef.current = undefined if (response.status !== 'completed' || !response.result) { + previewJobIdRef.current = undefined setCrawlState(response.status === 'canceled' ? 'stopped' : 'error') return } @@ -395,11 +397,13 @@ function CreateSourceSetupSession({ title: page.title ?? page.source_url, })), ) + previewFingerprintRef.current = response.result.configuration_fingerprint ?? undefined setCrawlState('success') } catch { if (crawlAttemptRef.current === attempt) { const jobId = previewJobIdRef.current previewJobIdRef.current = undefined + previewFingerprintRef.current = undefined if (jobId) void consoleClient.knowledgeFs.sourceProviderPreview.jobs.byJobId .delete({ @@ -447,6 +451,10 @@ function CreateSourceSetupSession({ provider: installedProviderOption.plugin.provider, providerDisplayName: installedProviderOption.label, parameters, + ...(previewJobIdRef.current ? { previewJobId: previewJobIdRef.current } : {}), + ...(previewFingerprintRef.current + ? { previewConfigurationFingerprint: previewFingerprintRef.current } + : {}), root_url: sourceUri, selection: selectedPages.map((page) => ({ source_url: page.sourceUrl,