diff --git a/api/controllers/console/knowledge_fs/resources.py b/api/controllers/console/knowledge_fs/resources.py index f1066d2a0f8..f1bdb0e8db4 100644 --- a/api/controllers/console/knowledge_fs/resources.py +++ b/api/controllers/console/knowledge_fs/resources.py @@ -2430,12 +2430,14 @@ class KnowledgeFSSpaceSourceApi(Resource): @_knowledge_fs_errors def patch(self, control_space_id: str, source_id: str): actor_id, tenant_id = _actor() + payload = _payload(KnowledgeFSSourceUpdatePayload) + payload.sync_after_update = True result = _console_services().facade.update_source( tenant_id=tenant_id, account_id=actor_id, control_space_id=control_space_id, source_id=source_id, - payload=_payload(KnowledgeFSSourceUpdatePayload), + payload=payload, ) return dump_response(KnowledgeFSSourceResponse, result) diff --git a/api/knowledge-fs-contract.lock.json b/api/knowledge-fs-contract.lock.json index 1f4f0ec4a9b..ab6590c830a 100644 --- a/api/knowledge-fs-contract.lock.json +++ b/api/knowledge-fs-contract.lock.json @@ -1,7 +1,7 @@ { "schemaVersion": 5, - "subtreeTree": "602f9fdb2f415093507850074b71e22571b1a4cd", - "openapiSha256": "b199b523299eff504a174bf92f8d41ab47622bef1ff30b285d49f10c0119058a", + "subtreeTree": "151d5e5d9883f06e49be6f562b32b192b2619edc", + "openapiSha256": "94b9676d6a2b27b79f7b2c42ca91902d3777a99b1e7efc5747f4a0fce6df8f91", "capabilityV2AuthManifestSha256": "e322a2fa779d1f40b95c54c1021cffecaec77abbd7b34899573dcdf4ff353109", "capabilityV2AuthTestVectorSha256": "ae0de37b1ff05c40f905cf17a7b410d8971acacf64db07d5ee3d6fecfa559ce3", "productOperationManifestSha256": "32f305d86525f31eb07a6aef1f72d286d53991f1f8ed1f9d6e52d37cf4561116", diff --git a/api/services/knowledge_fs/product_dto.py b/api/services/knowledge_fs/product_dto.py index 738b0187757..a573c286b01 100644 --- a/api/services/knowledge_fs/product_dto.py +++ b/api/services/knowledge_fs/product_dto.py @@ -2159,6 +2159,7 @@ class KnowledgeFSSourceUpdatePayload(BaseModel): provider_parameters: dict[str, bool | FiniteFloat | str] | None = Field( default=None, max_length=50, alias="providerParameters" ) + sync_after_update: bool | None = Field(default=None, alias="syncAfterUpdate") status: Literal["active", "disabled", "error", "syncing"] | None = None uri: str | None = Field(default=None, min_length=1, max_length=4_096) diff --git a/api/tests/unit_tests/services/test_knowledge_fs_product_dto.py b/api/tests/unit_tests/services/test_knowledge_fs_product_dto.py index 65477ebccf0..f6ce0281514 100644 --- a/api/tests/unit_tests/services/test_knowledge_fs_product_dto.py +++ b/api/tests/unit_tests/services/test_knowledge_fs_product_dto.py @@ -920,6 +920,17 @@ def test_source_update_payload_accepts_mutable_source_configuration() -> None: } +def test_source_update_payload_serializes_backend_sync_admission_flag() -> None: + payload = KnowledgeFSSourceUpdatePayload.model_validate( + {"name": "Renamed", "syncAfterUpdate": True} + ) + + assert payload.model_dump(mode="json", by_alias=True, exclude_none=True) == { + "name": "Renamed", + "syncAfterUpdate": True, + } + + def test_source_update_payload_requires_provider_parameters_for_parameter_updates() -> None: with pytest.raises(ValidationError): KnowledgeFSSourceUpdatePayload.model_validate({"metadata": {"parameters": {"limit": 50}}}) diff --git a/knowledge-fs/docs/api-reference.md b/knowledge-fs/docs/api-reference.md index a3f4842c156..c9e2b9ea0df 100644 --- a/knowledge-fs/docs/api-reference.md +++ b/knowledge-fs/docs/api-reference.md @@ -743,7 +743,7 @@ currently visible. > **Source object & conventions** > - `Source` = `{ id, knowledgeSpaceId, name, type: enum(upload|object-storage|connector|web), uri, status: enum(active|syncing|error|disabled), permissionScope: string[], metadata, version: int≥1, createdAt, updatedAt }`. > - **Optimistic locking**: `version` is bumped on every write. Pass it back as `expectedVersion` on PATCH to fail with `409` instead of overwriting a concurrent modification. -> - **Scheduled sync**: set `metadata.syncPolicy` to `{"everyHours": N}` (1–720) or `{"dailyAt": ["HH:MM", …], "utcOffset": "±HH:MM"?}`. Invalid policies are rejected with `400` at create/update. A background scheduler (multi-replica safe; `KNOWLEDGE_SOURCE_SYNC*` env) then re-syncs due sources: web → re-crawl with content-hash dedup, connector pages → re-import previously imported pages whose `lastEditedTime` changed, drive → re-download previously imported files. The scheduler records progress under `metadata.syncState` `{ lastSyncAt, lastSyncStatus: ok|error, lastSyncError?, nextSyncAt, syncStartedAt? }`. +> - **Scheduled sync**: set `metadata.syncPolicy` to `{"everyHours": N}` (1–720) or `{"dailyAt": ["HH:MM", …], "utcOffset": "±HH:MM"?}`. Invalid policies are rejected with `400` at create/update. A background scheduler (multi-replica safe; `KNOWLEDGE_SOURCE_SYNC*` env) then re-syncs due sources: web → re-crawl with content-hash dedup, connector pages → re-import previously imported pages whose `lastEditedTime` changed, drive → re-download previously imported files. A disabled Source pauses due scheduling without disabling or advancing its policy; after re-enabling, an overdue policy is eligible on the next scheduler pass. Source updates rebind the unchanged policy to the new Source version. The scheduler records progress under `metadata.syncState` `{ lastSyncAt, lastSyncStatus: ok|error, lastSyncError?, nextSyncAt, syncStartedAt? }`. > - **Reserved metadata keys** (managed by the platform; user PATCHes should carry them through): `tenantId` (auto-stamped on create/update), `sync` (last sync summary), `crawled`, `imported`, `importedFiles` (per-source sync state), `syncState`. ### `POST /knowledge-spaces/{id}/sources` @@ -769,9 +769,9 @@ currently visible. ### `PATCH /knowledge-spaces/{id}/sources/{sourceId}` **Description**: Update a source's mutable fields, optionally guarded by optimistic locking. **Auth**: Bearer; scope `knowledge-spaces:write`. -**Path params**: `id` (uuid); `sourceId` (uuid). **Body** (`application/json`, strict; at least one mutable field is required): `name` (1–200); `status` (enum); `metadata` (object, recursively patches ordinary metadata and cannot contain `parameters`; `tenantId` is re-stamped automatically); `providerParameters` (object containing at most 50 boolean/finite-number/string values, fully replaces `metadata.parameters`, so omitted keys are removed); `uri` (1–4096); `expectedVersion` (int ≥1) — the `version` from your last read; when provided, a concurrent modification makes the update fail with `409` instead of overwriting it. For web sources, changing `uri` also changes the provider parameter `url`, while a replacement `providerParameters.url` changes `uri`; if both are supplied, `uri` is authoritative. +**Path params**: `id` (uuid); `sourceId` (uuid). **Body** (`application/json`, strict; at least one mutable field is required): `name` (1–200); `status` (enum); `metadata` (object, recursively patches ordinary metadata and cannot contain `parameters`; `tenantId` is re-stamped automatically); `providerParameters` (object containing at most 50 boolean/finite-number/string values, fully replaces `metadata.parameters`, so omitted keys are removed); `uri` (1–4096); `expectedVersion` (int ≥1) — the `version` from your last read; when provided, a concurrent modification makes the update fail with `409` instead of overwriting it; `syncAfterUpdate` (boolean, optional) asks the product gateway to enqueue a durable sync only when URI, provider parameters, or other business metadata actually changed. Name-only and `metadata.syncPolicy`-only updates do not enqueue a sync. For web sources, changing `uri` also changes the provider parameter `url`, while a replacement `providerParameters.url` changes `uri`; if both are supplied, `uri` is authoritative. Source identity metadata (`datasource`, `pluginId`, `provider`, `providerId`, and `providerKind`) is immutable. **Responses**: -- `200`: updated `Source` (`version` bumped). +- `200`: updated `Source` (`version` bumped), optionally enriched with `syncWorkflow` when a configuration sync was accepted. - `400` invalid `metadata.syncPolicy`, provider parameters under `metadata`, credential-shaped provider parameters, or an invalid web Source URL; `404`; `409` concurrent modification (`expectedVersion` mismatch) or protected legacy credential-bearing parameters; `401`/`403`. ### `DELETE /knowledge-spaces/{id}/sources/{sourceId}` diff --git a/knowledge-fs/packages/api/src/gateway-source.test.ts b/knowledge-fs/packages/api/src/gateway-source.test.ts index e06c4521955..f4ff08e4e40 100644 --- a/knowledge-fs/packages/api/src/gateway-source.test.ts +++ b/knowledge-fs/packages/api/src/gateway-source.test.ts @@ -393,6 +393,29 @@ describe("knowledge space source CRUD", () => { } }); + it("rejects source identity changes through metadata patches", async () => { + const app = createApp(); + const spaceId = await createSpace(app); + const sourceId = await createWebSource(app, spaceId); + + for (const [key, value] of [ + ["datasource", "other-datasource"], + ["pluginId", "other/plugin"], + ["provider", "other-provider"], + ["providerId", "other-provider-id"], + ["providerKind", "online-drive"], + ] as const) { + const response = await app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, { + body: JSON.stringify({ metadata: { [key]: value } }), + headers: json(writeToken), + method: "PATCH", + }); + + expect(response.status).toBe(400); + await expect(response.json()).resolves.toEqual({ error: `Source ${key} cannot be changed` }); + } + }); + it("rejects credentials disguised as provider parameters", async () => { const app = createApp(); const spaceId = await createSpace(app); diff --git a/knowledge-fs/packages/api/src/index.ts b/knowledge-fs/packages/api/src/index.ts index e282216395b..534e1eef5d6 100644 --- a/knowledge-fs/packages/api/src/index.ts +++ b/knowledge-fs/packages/api/src/index.ts @@ -2174,6 +2174,8 @@ export function createKnowledgeGateway({ ...(sourceCredentials ? { sourceCredentials } : {}), ...(sourceProduct ? { sourceConnections: sourceProduct.connections } : {}), ...(sourceProduct ? { sourceProductWorkflows: sourceProduct.repository } : {}), + ...(sourceProductWorkflows ? { sourceProductWorkflowService: sourceProductWorkflows } : {}), + ...(sourceProduct ? { sourceSyncPolicyVersionBinder: sourceProduct.repository } : {}), legacyMutationEndpointsEnabled: sourceProduct === undefined, sourceDocumentMaterializer, sources: sourceRepository, diff --git a/knowledge-fs/packages/api/src/source-handlers-coverage.test.ts b/knowledge-fs/packages/api/src/source-handlers-coverage.test.ts index ca0f5d8edc0..9a7a8e99570 100644 --- a/knowledge-fs/packages/api/src/source-handlers-coverage.test.ts +++ b/knowledge-fs/packages/api/src/source-handlers-coverage.test.ts @@ -16,6 +16,7 @@ import { type SourceCredentialTester, type SourceDocumentMaterializer, type SourceProductWorkflowRepository, + type SourceProductWorkflowService, type SourceRepository, type SourceSecretStore, type WebsiteCrawlConnector, @@ -1420,6 +1421,11 @@ describe("source handlers without optional collaborators", () => { SourceProductWorkflowRepository, "listLatestSyncCompletions" | "listLatestSyncRuns" | "listSyncPolicies" >; + sourceProductWorkflowService?: Pick; + sourceSyncPolicyVersionBinder?: Pick< + SourceProductWorkflowRepository, + "rebindSyncPolicySourceVersion" + >; sources?: SourceRepository; websiteCrawlConnector?: WebsiteCrawlConnector; } @@ -1472,6 +1478,12 @@ describe("source handlers without optional collaborators", () => { ...(options.sourceProductWorkflows ? { sourceProductWorkflows: options.sourceProductWorkflows } : {}), + ...(options.sourceProductWorkflowService + ? { sourceProductWorkflowService: options.sourceProductWorkflowService } + : {}), + ...(options.sourceSyncPolicyVersionBinder + ? { sourceSyncPolicyVersionBinder: options.sourceSyncPolicyVersionBinder } + : {}), sources, spaces, ...(options.websiteCrawlConnector @@ -1501,6 +1513,139 @@ describe("source handlers without optional collaborators", () => { return { sourceId: source.id, spaceId: space.id }; } + it.each([ + ["website", "web", { uri: "https://docs.example.com" }], + ["online document", "connector", { providerParameters: { database: "docs" } }], + ["online drive", "connector", { providerParameters: { folder: "reports" } }], + ] as const)( + "enqueues a durable sync after a %s configuration update", + async (_kind, sourceType, update) => { + const createSync = vi.fn(async ({ knowledgeSpaceId, sourceId, idempotencyKey }) => ({ + checkpoint: "queued" as const, + createdAt: "2026-08-30T00:00:00.000Z", + executionAttempts: 0, + id: "00000000-0000-4000-8000-000000000501", + idempotencyKey, + knowledgeSpaceId, + kind: "sync" as const, + maxExecutionAttempts: 5, + payload: {}, + progressCompleted: 0, + progressFailed: 0, + progressSkipped: 0, + rowVersion: 1, + sourceId, + state: "queued" as const, + tenantId: "tenant-1", + updatedAt: "2026-08-30T00:00:00.000Z", + })); + const bare = createBareApp({ sourceProductWorkflowService: { createSync } }); + const { sourceId, spaceId } = await seedSource(bare, sourceType); + + const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, { + body: JSON.stringify({ ...update, syncAfterUpdate: true }), + headers: { "content-type": "application/json" }, + method: "PATCH", + }); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toMatchObject({ + id: sourceId, + syncWorkflow: { kind: "sync", sourceId, state: "queued" }, + }); + expect(createSync).toHaveBeenCalledOnce(); + expect(createSync).toHaveBeenCalledWith( + expect.objectContaining({ + idempotencyKey: `source-config-update:${sourceId}:2`, + knowledgeSpaceId: spaceId, + sourceId, + }), + ); + }, + ); + + it.each([ + ["website", "web", { uri: "https://docs.example.com" }], + ["online document", "connector", { providerParameters: { database: "docs" } }], + ["online drive", "connector", { providerParameters: { folder: "reports" } }], + ] as const)( + "rebinds an existing sync policy after a disabled %s configuration update without syncing", + async (_kind, sourceType, update) => { + const createSync = vi.fn(); + const rebindSyncPolicySourceVersion = vi.fn(async () => null); + const bare = createBareApp({ + sourceProductWorkflowService: { createSync }, + sourceSyncPolicyVersionBinder: { rebindSyncPolicySourceVersion }, + }); + const { sourceId, spaceId } = await seedSource(bare, sourceType); + + const disable = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, { + body: JSON.stringify({ status: "disabled", syncAfterUpdate: true }), + headers: { "content-type": "application/json" }, + method: "PATCH", + }); + expect(disable.status).toBe(200); + + const updateWhileDisabled = await bare.app.request( + `/knowledge-spaces/${spaceId}/sources/${sourceId}`, + { + body: JSON.stringify({ ...update, syncAfterUpdate: true }), + headers: { "content-type": "application/json" }, + method: "PATCH", + }, + ); + expect(updateWhileDisabled.status).toBe(200); + expect(createSync).not.toHaveBeenCalled(); + expect(rebindSyncPolicySourceVersion).toHaveBeenNthCalledWith(1, { + expectedSourceVersion: 1, + knowledgeSpaceId: spaceId, + sourceId, + sourceVersion: 2, + tenantId: "tenant-1", + }); + expect(rebindSyncPolicySourceVersion).toHaveBeenNthCalledWith(2, { + expectedSourceVersion: 2, + knowledgeSpaceId: spaceId, + sourceId, + sourceVersion: 3, + tenantId: "tenant-1", + }); + }, + ); + + it.each([ + ["name", { name: "Renamed" }], + ["sync policy", { metadata: { syncPolicy: { everyHours: 24 } } }], + ] as const)("does not enqueue a sync for a %s-only update", async (_kind, update) => { + const createSync = vi.fn(); + const bare = createBareApp({ sourceProductWorkflowService: { createSync } }); + const { sourceId, spaceId } = await seedSource(bare, "web"); + + const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, { + body: JSON.stringify({ ...update, syncAfterUpdate: true }), + headers: { "content-type": "application/json" }, + method: "PATCH", + }); + + expect(response.status).toBe(200); + expect(createSync).not.toHaveBeenCalled(); + }); + + it("does not enqueue a sync when submitted configuration is unchanged", async () => { + const createSync = vi.fn(); + const bare = createBareApp({ sourceProductWorkflowService: { createSync } }); + const { sourceId, spaceId } = await seedSource(bare, "web"); + + const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, { + body: JSON.stringify({ syncAfterUpdate: true, uri: "https://example.com" }), + headers: { "content-type": "application/json" }, + method: "PATCH", + }); + + expect(response.status).toBe(200); + expect(createSync).not.toHaveBeenCalled(); + }); + it("enriches a source list with product sync details in bulk lookups", async () => { const listSyncPolicies = vi.fn(); const listLatestSyncCompletions = vi.fn(); diff --git a/knowledge-fs/packages/api/src/source-handlers.ts b/knowledge-fs/packages/api/src/source-handlers.ts index bc0d6a26974..f95f17d19b7 100644 --- a/knowledge-fs/packages/api/src/source-handlers.ts +++ b/knowledge-fs/packages/api/src/source-handlers.ts @@ -1,3 +1,4 @@ +import { isDeepStrictEqual } from "node:util"; import type { OpenAPIHono } from "@hono/zod-openapi"; import { @@ -20,6 +21,7 @@ import { type KnowledgeFsPublicFailure, knowledgeFsFailureForCode } from "./know import type { KnowledgeSpaceRepository } from "./knowledge-space-repository"; import type { OnlineDocumentConnector } from "./online-document-connector"; import type { OnlineDriveConnector } from "./online-drive-connector"; +import type { LooseOpenApiContext } from "./openapi-handler-utils"; import type { SourceConnectionService } from "./source-connection"; import { SOURCE_DOCUMENT_REPLACEMENT_SAGA_REQUIRED, @@ -41,6 +43,7 @@ import type { import { safeSourceOperationError, sourceOperationFailureMetadata } from "./source-operation-error"; import { type SourceProductWorkflowRepository, + type SourceProductWorkflowService, type SourceWorkflowRun, toPublicSourceWorkflowRun, } from "./source-product-workflow"; @@ -88,6 +91,12 @@ export interface RegisterSourceHandlersOptions { "listLatestSyncCompletions" | "listLatestSyncRuns" | "listSyncPolicies" > | undefined; + readonly sourceSyncPolicyVersionBinder?: + | Pick + | undefined; + readonly sourceProductWorkflowService?: + | Pick + | undefined; readonly sources: SourceRepository; readonly spaces: KnowledgeSpaceRepository; readonly websiteCrawlConnector?: WebsiteCrawlConnector | undefined; @@ -105,6 +114,8 @@ export function registerSourceHandlers({ sourceCredentials, sourceDocumentMaterializer, sourceProductWorkflows, + sourceProductWorkflowService, + sourceSyncPolicyVersionBinder, sources, spaces, websiteCrawlConnector, @@ -340,6 +351,23 @@ export function registerSourceHandlers({ return context.json({ error: "Provider parameters cannot contain credentials" }, 400); } + const immutableMetadataKeys = [ + "datasource", + "pluginId", + "provider", + "providerId", + "providerKind", + ] as const; + const requestsConfigurationSync = + body.syncAfterUpdate === true && + (body.providerParameters !== undefined || + body.uri !== undefined || + (body.metadata !== undefined && + Object.keys(body.metadata).some((key) => key !== "syncPolicy"))); + if (requestsConfigurationSync && !sourceProductWorkflowService) { + return context.json({ error: "Source workflow service is unavailable" }, 503); + } + if (body.metadata?.syncPolicy !== undefined) { try { parseSourceSyncPolicy(body.metadata.syncPolicy); @@ -354,6 +382,8 @@ export function registerSourceHandlers({ try { let source = null; + let configurationChanged = false; + let previousSourceVersion: number | undefined; const maxAttempts = body.expectedVersion === undefined ? 3 : 1; for (let attempt = 0; attempt < maxAttempts; attempt += 1) { @@ -366,6 +396,16 @@ export function registerSourceHandlers({ break; } + for (const key of immutableMetadataKeys) { + if ( + body.metadata && + Object.prototype.hasOwnProperty.call(body.metadata, key) && + body.metadata[key] !== fresh.metadata[key] + ) { + return context.json({ error: `Source ${key} cannot be changed` }, 400); + } + } + if (body.expectedVersion !== undefined && body.expectedVersion !== fresh.version) { throw new SourceVersionConflictError(params.sourceId, body.expectedVersion); } @@ -416,6 +456,15 @@ export function registerSourceHandlers({ } } } + configurationChanged = + (uri !== undefined && uri !== fresh.uri) || + (body.providerParameters !== undefined && + !isDeepStrictEqual(metadata?.parameters, fresh.metadata.parameters)) || + (body.metadata !== undefined && + Object.entries(body.metadata).some( + ([key]) => + key !== "syncPolicy" && !isDeepStrictEqual(metadata?.[key], fresh.metadata[key]), + )); source = await sources.update({ expectedVersion: fresh.version, id: params.sourceId, @@ -433,6 +482,7 @@ export function registerSourceHandlers({ ...(body.status === undefined ? {} : { status: body.status }), ...(uri === undefined ? {} : { uri }), }); + if (source) previousSourceVersion = fresh.version; break; } catch (error) { if ( @@ -450,6 +500,29 @@ export function registerSourceHandlers({ if (!source) { return context.json({ error: "Source not found" }, 404); } + if (previousSourceVersion !== undefined && sourceSyncPolicyVersionBinder) { + await sourceSyncPolicyVersionBinder.rebindSyncPolicySourceVersion({ + expectedSourceVersion: previousSourceVersion, + knowledgeSpaceId: params.id, + sourceId: source.id, + sourceVersion: source.version, + tenantId: subject.tenantId, + }); + } + + if (body.syncAfterUpdate && configurationChanged && source.status !== "disabled") { + const workflow = await sourceProductWorkflowService?.createSync({ + ...sourceWorkflowPrincipal(context), + idempotencyKey: `source-config-update:${source.id}:${source.version}`, + knowledgeSpaceId: params.id, + sourceId: source.id, + }); + if (!workflow) throw new Error("Source workflow service became unavailable"); + return context.json( + { ...toSourceResponse(source), syncWorkflow: toPublicSourceWorkflowRun(workflow) }, + 200, + ); + } return context.json(toSourceResponse(source), 200); } catch (error) { @@ -1236,6 +1309,24 @@ export function registerSourceHandlers({ }); } +function sourceWorkflowPrincipal(context: Pick) { + const apiKey = context.get("authenticatedApiKey"); + const capabilityGrant = context.get("capabilityV2Grant"); + return { + ...(apiKey ? { apiKey } : {}), + ...(capabilityGrant + ? { + capability: { + contentScopeIds: capabilityGrant.contentScopeIds, + grantId: capabilityGrant.grantId, + }, + } + : {}), + callerKind: context.get("callerKind") ?? "interactive", + subject: context.get("subject"), + } as const; +} + function sourceStatusWithSyncWorkflow(source: Source, run?: SourceWorkflowRun): Source["status"] { if (source.status === "disabled" || !run) return source.status; if (run.state === "failed" || run.state === "canceled") return "error"; diff --git a/knowledge-fs/packages/api/src/source-product-workflow-database-repository.test.ts b/knowledge-fs/packages/api/src/source-product-workflow-database-repository.test.ts index c3f9b5e85a4..d6888a22357 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-database-repository.test.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-database-repository.test.ts @@ -2184,6 +2184,36 @@ describe("database source-product workflow repository edge coverage", () => { expect(insert?.params[13]).toBeNull(); }); + it("rebinds a sync policy source version without changing its schedule or revision", async () => { + const calls: DatabaseExecuteInput[] = []; + const repository = createDatabaseSourceProductWorkflowRepository({ + database: testDatabase("postgres", async (input) => { + calls.push(input); + if (input.tableName === "source_sync_policies" && input.operation === "select") { + return { rows: [syncPolicyRow()], rowsAffected: 1 }; + } + return { rows: [], rowsAffected: input.operation === "select" ? 0 : 1 }; + }), + }); + + await expect( + repository.rebindSyncPolicySourceVersion({ + expectedSourceVersion: 1, + knowledgeSpaceId, + sourceId, + sourceVersion: 4, + tenantId, + }), + ).resolves.toEqual({ ...syncPolicy(), expectedSourceVersion: 4 }); + + const update = calls.find( + (call) => call.tableName === "source_sync_policies" && call.operation === "update", + ); + expect(update?.params).toEqual([4, "sync-policy-a", 1]); + expect(update?.sql).not.toContain('"next_run_at" ='); + expect(update?.sql).not.toContain('"revision" ='); + }); + it("disables invalid due policies and handles scheduler races", async () => { await expect( createDatabaseSourceProductWorkflowRepository({ @@ -2191,11 +2221,17 @@ describe("database source-product workflow repository edge coverage", () => { }).enqueueDueSyncRuns({ limit: 1, maxExecutionAttempts: 2, now }), ).resolves.toEqual([]); + const disabledSourceCalls: DatabaseExecuteInput[] = []; await expect( createDatabaseSourceProductWorkflowRepository({ - database: duePolicyDatabase({ sourceStatus: "disabled" }), + database: duePolicyDatabase({ calls: disabledSourceCalls, sourceStatus: "disabled" }), }).enqueueDueSyncRuns({ limit: 1, maxExecutionAttempts: 2, now }), ).resolves.toEqual([]); + expect( + disabledSourceCalls.some( + (call) => call.tableName === "source_sync_policies" && call.operation === "update", + ), + ).toBe(false); await expect( createDatabaseSourceProductWorkflowRepository({ @@ -2421,6 +2457,7 @@ function bulkMutationDatabase( function duePolicyDatabase( options: { + readonly calls?: DatabaseExecuteInput[] | undefined; readonly currentPolicy?: DatabaseRow | undefined; readonly missingSource?: boolean | undefined; readonly missingSpace?: boolean | undefined; @@ -2431,6 +2468,7 @@ function duePolicyDatabase( ): DatabaseAdapter { const candidate = syncPolicyRow(); return testDatabase(dialect, async (input) => { + options.calls?.push(input); if (input.tableName === "source_sync_policies") { if (input.operation === "update") { return { rows: [], rowsAffected: options.policyUpdateRowsAffected ?? 1 }; 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 c8d708a43d2..4babb8e251a 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 @@ -1118,6 +1118,26 @@ export function createDatabaseSourceProductWorkflowRepository(input: { if (result.rowsAffected !== 1) policyConflict(); return policy; }), + rebindSyncPolicySourceVersion: (input) => + database.transaction(async (tx) => { + const policy = await getPolicy(database, tx, input, true); + if (!policy) return null; + if (policy.expectedSourceVersion !== input.expectedSourceVersion) { + throw new SourceWorkflowError( + "SOURCE_SYNC_POLICY_SOURCE_CONFLICT", + "Source sync policy changed concurrently", + ); + } + const updated = await tx.execute({ + maxRows: 0, + operation: "update", + params: [input.sourceVersion, policy.id, input.expectedSourceVersion], + sql: `UPDATE ${q(database, policyTable)} SET ${q(database, "expected_source_version")} = ${p(database, 1)} WHERE ${q(database, "id")} = ${p(database, 2)} AND ${q(database, "expected_source_version")} = ${p(database, 3)};`, + tableName: policyTable, + }); + if (updated.rowsAffected !== 1) policyConflict(); + return { ...policy, expectedSourceVersion: input.sourceVersion }; + }), listDueSyncPolicies: async ({ cursor, limit, now }) => { listLimit(limit); const readLimit = limit + 1; @@ -1273,14 +1293,14 @@ export function createDatabaseSourceProductWorkflowRepository(input: { policy.nextRunAt > now ) continue; - if ( - !sourceRow || - numberColumn(sourceRow, "version") !== policy.expectedSourceVersion || - stringColumn(sourceRow, "status") === "disabled" - ) { + if (!sourceRow || numberColumn(sourceRow, "version") !== policy.expectedSourceVersion) { await disableSyncPolicy(database, tx, policy, now); continue; } + // Disabling a Source pauses scheduling; it does not destroy the user's durable policy. + // Keep nextRunAt unchanged so an overdue policy is admitted on the first scheduler pass + // after the Source is enabled again. + if (stringColumn(sourceRow, "status") === "disabled") continue; const scheduledFor = policy.nextRunAt; const run: SourceWorkflowRun = { ...(isCapabilitySyncPolicy(policy) diff --git a/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.test.ts b/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.test.ts index 42290351848..78c5300e95e 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.test.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-memory-repository.test.ts @@ -1039,16 +1039,33 @@ describe("in-memory source product workflow repository", () => { updatedAt: "2026-03-01T00:01:00.000Z", }); expect(revised.revision).toBe(2); + const rebound = await repository.rebindSyncPolicySourceVersion({ + expectedSourceVersion: 1, + knowledgeSpaceId, + sourceId: provider.sourceId, + sourceVersion: 4, + tenantId, + }); + expect(rebound).toEqual({ ...revised, expectedSourceVersion: 4 }); + await expect( + repository.rebindSyncPolicySourceVersion({ + expectedSourceVersion: 1, + knowledgeSpaceId, + sourceId: provider.sourceId, + sourceVersion: 5, + tenantId, + }), + ).rejects.toMatchObject({ code: "SOURCE_SYNC_POLICY_SOURCE_CONFLICT" }); await expect( repository.getSyncPolicy({ knowledgeSpaceId, sourceId: provider.sourceId, tenantId }), - ).resolves.toEqual(revised); + ).resolves.toEqual(rebound); await expect( repository.listSyncPolicies({ knowledgeSpaceId, sourceIds: ["missing", provider.sourceId, provider.sourceId], tenantId, }), - ).resolves.toEqual([revised]); + ).resolves.toEqual([rebound]); await expect( repository.listSyncPolicies({ knowledgeSpaceId, 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 fa983c685cf..caa1103d062 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 @@ -646,6 +646,26 @@ export function createInMemorySourceProductWorkflowRepository(input?: { policies.set(key, clonePolicy(policy)); return clonePolicy(policy); }, + rebindSyncPolicySourceVersion: async ({ + expectedSourceVersion, + knowledgeSpaceId, + sourceId, + sourceVersion, + tenantId, + }) => { + const key = `${tenantId}\0${knowledgeSpaceId}\0${sourceId}`; + const policy = policies.get(key); + if (!policy) return null; + if (policy.expectedSourceVersion !== expectedSourceVersion) { + throw new SourceWorkflowError( + "SOURCE_SYNC_POLICY_SOURCE_CONFLICT", + "Source sync policy changed concurrently", + ); + } + const rebound = { ...policy, expectedSourceVersion: sourceVersion }; + policies.set(key, rebound); + return clonePolicy(rebound); + }, getSyncPolicy: async ({ knowledgeSpaceId, sourceId, tenantId }) => { const policy = policies.get(`${tenantId}\0${knowledgeSpaceId}\0${sourceId}`); return policy ? clonePolicy(policy) : null; 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 5e1d5d47027..f9e0349fc5d 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow-runtime.ts @@ -3089,6 +3089,13 @@ async function executeBulkAction( }), ); if (!updated) throw runtimeError("SOURCE_NOT_FOUND", "Source not found"); + await input.repository.rebindSyncPolicySourceVersion({ + expectedSourceVersion: source.version, + knowledgeSpaceId: updated.knowledgeSpaceId, + sourceId: updated.id, + sourceVersion: updated.version, + tenantId: run.tenantId, + }); return; } throw runtimeError( diff --git a/knowledge-fs/packages/api/src/source-product-workflow.ts b/knowledge-fs/packages/api/src/source-product-workflow.ts index b0fe8d1741e..3ccae748c2b 100644 --- a/knowledge-fs/packages/api/src/source-product-workflow.ts +++ b/knowledge-fs/packages/api/src/source-product-workflow.ts @@ -391,6 +391,14 @@ export interface SourceProductWorkflowRepository { readonly run: NewSourceWorkflowRun; }): Promise; upsertSyncPolicy(input: SourceSyncPolicyRecord): Promise; + /** Rebinds an unchanged policy to a successful Source CAS without changing its schedule. */ + rebindSyncPolicySourceVersion(input: { + readonly expectedSourceVersion: number; + readonly knowledgeSpaceId: string; + readonly sourceId: string; + readonly sourceVersion: number; + readonly tenantId: string; + }): Promise; listDueSyncPolicies(input: { readonly cursor?: string | undefined; readonly limit: number; diff --git a/knowledge-fs/packages/api/src/source-request-schemas.ts b/knowledge-fs/packages/api/src/source-request-schemas.ts index 6f28d59786a..87211a16c81 100644 --- a/knowledge-fs/packages/api/src/source-request-schemas.ts +++ b/knowledge-fs/packages/api/src/source-request-schemas.ts @@ -69,6 +69,12 @@ export const UpdateSourceSchema = z return keys.length <= 50 && keys.every((key) => key.length >= 1 && key.length <= 255); }, "Provider parameters must contain at most 50 bounded keys") .optional(), + /** + * Product gateways set this when a successful configuration update must enqueue a durable + * Source sync. Internal reconciliation writes leave it unset so metadata-only lifecycle + * updates do not recursively create sync work. + */ + syncAfterUpdate: z.boolean().optional(), status: z.enum(["active", "syncing", "error", "disabled"]).optional(), uri: z.string().min(1).max(SOURCE_URI_MAX_LENGTH).optional(), }) diff --git a/knowledge-fs/packages/api/src/source-routes.ts b/knowledge-fs/packages/api/src/source-routes.ts index 08d40533fdf..aae01d91a15 100644 --- a/knowledge-fs/packages/api/src/source-routes.ts +++ b/knowledge-fs/packages/api/src/source-routes.ts @@ -159,7 +159,7 @@ export const updateSourceRoute = createRoute({ }, responses: { 200: { - content: { "application/json": { schema: SourceResponseSchema } }, + content: { "application/json": { schema: SourceListItemResponseSchema } }, description: "Updated source", }, 400: InvalidRequestResponse, @@ -169,6 +169,10 @@ export const updateSourceRoute = createRoute({ description: "Source was modified concurrently, or protected legacy credential-bearing parameters cannot be replaced", }, + 503: { + content: { "application/json": { schema: ErrorResponseSchema } }, + description: "Source workflow service unavailable", + }, 401: UnauthorizedResponse, 403: ForbiddenResponse, },