From c4aaf57c5413bddeb18ff324d0342c111d54f1e3 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 20 Aug 2026 22:00:43 -0700 Subject: [PATCH 1/7] Fix knowledge connector sync follow-up --- apps/docs/content/docs/en/cli/knowledge.mdx | 98 ++++++------- apps/docs/content/docs/en/cli/reference.mdx | 106 +++++++------- apps/docs/openapi-v2-knowledge.json | 4 +- .../[id]/connectors/[connectorId]/route.ts | 5 +- .../knowledge/[id]/connectors/route.test.ts | 25 ++++ apps/sim/connectors/linear/linear.test.ts | 37 +++++ apps/sim/connectors/linear/linear.ts | 3 +- apps/sim/lib/api/contracts/v2/knowledge.ts | 4 +- .../lib/api/contracts/v2/openapi/knowledge.ts | 2 +- .../server/knowledge/knowledge-base.test.ts | 35 +++++ .../tools/server/knowledge/knowledge-base.ts | 25 +++- .../knowledge/application/connectors.test.ts | 10 ++ .../lib/knowledge/application/connectors.ts | 8 ++ .../orchestration/connectors.test.ts | 136 ++++++++++++++++++ .../lib/knowledge/orchestration/connectors.ts | 74 ++++++++-- .../sim-cli/src/contract/commands.test.ts | 21 +++ packages/sim-cli/src/contract/commands.ts | 20 +++ packages/sim-cli/src/generated/v2-api.ts | 3 +- packages/sim-cli/src/runtime/build.ts | 7 +- 19 files changed, 495 insertions(+), 128 deletions(-) create mode 100644 apps/sim/connectors/linear/linear.test.ts diff --git a/apps/docs/content/docs/en/cli/knowledge.mdx b/apps/docs/content/docs/en/cli/knowledge.mdx index a2fb532b6d8..d09f27e1dad 100644 --- a/apps/docs/content/docs/en/cli/knowledge.mdx +++ b/apps/docs/content/docs/en/cli/knowledge.mdx @@ -216,7 +216,7 @@ sim knowledge create [options] ## Create knowledge connector ```bash -sim knowledge connectors create [options] +sim knowledge connectors create [options] ``` **Arguments** @@ -225,7 +225,7 @@ sim knowledge connectors create [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -246,7 +246,7 @@ sim knowledge connectors create [options] ## Delete knowledge connector ```bash -sim knowledge connectors delete [options] +sim knowledge connectors delete [options] ``` **Arguments** @@ -255,7 +255,7 @@ sim knowledge connectors delete [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -275,7 +275,7 @@ sim knowledge connectors delete [options] ## Get knowledge connector ```bash -sim knowledge connectors get +sim knowledge connectors get ``` **Arguments** @@ -284,7 +284,7 @@ sim knowledge connectors get | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -292,7 +292,7 @@ sim knowledge connectors get ## List knowledge connector documents ```bash -sim knowledge connectors documents list [options] +sim knowledge connectors documents list [options] ``` **Arguments** @@ -301,7 +301,7 @@ sim knowledge connectors documents list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -318,10 +318,38 @@ sim knowledge connectors documents list [options] +## Update knowledge connector documents + +```bash +sim knowledge connectors documents update [options] +``` + +**Arguments** + + + +| Argument | Required | Description | +| --- | --- | --- | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | +| `connectorId` | Yes | Connector selected for the operation. | + + + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | +| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | + + + ## List knowledge connectors ```bash -sim knowledge connectors list [options] +sim knowledge connectors list [options] ``` **Arguments** @@ -330,7 +358,7 @@ sim knowledge connectors list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -346,10 +374,10 @@ sim knowledge connectors list [options] -## Update knowledge connector +## Queue a knowledge connector synchronization ```bash -sim knowledge connectors update [options] +sim knowledge connectors sync [options] ``` **Arguments** @@ -358,7 +386,7 @@ sim knowledge connectors update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -369,16 +397,15 @@ sim knowledge connectors update [options] | Option | Required | Description | | --- | --- | --- | -| `--source-config ` | No | Replacement source selection and filtering configuration. (JSON, or @path / @- to read a file or stdin). | -| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | -| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | +| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | +| `--no-rehydrate` | No | Send --rehydrate as false. | -## Update knowledge connector documents +## Update knowledge connector ```bash -sim knowledge connectors documents update [options] +sim knowledge connectors update [options] ``` **Arguments** @@ -387,7 +414,7 @@ sim knowledge connectors documents update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -398,8 +425,9 @@ sim knowledge connectors documents update [options] | Option | Required | Description | | --- | --- | --- | -| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | -| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | +| `--source-config ` | No | Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused. (JSON, or @path / @- to read a file or stdin). | +| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | +| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | @@ -588,34 +616,6 @@ sim knowledge search [options] -## Sync knowledge connector - -```bash -sim knowledge sync create [options] -``` - -**Arguments** - - - -| Argument | Required | Description | -| --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | -| `connectorId` | Yes | Connector selected for the operation. | - - - -**Options** - - - -| Option | Required | Description | -| --- | --- | --- | -| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | -| `--no-rehydrate` | No | Send --rehydrate as false. | - - - ## Update knowledge base ```bash diff --git a/apps/docs/content/docs/en/cli/reference.mdx b/apps/docs/content/docs/en/cli/reference.mdx index 4346ef28715..abd3276b06b 100644 --- a/apps/docs/content/docs/en/cli/reference.mdx +++ b/apps/docs/content/docs/en/cli/reference.mdx @@ -1208,7 +1208,7 @@ sim knowledge create [options] Create Knowledge Connector ```bash -sim knowledge connectors create [options] +sim knowledge connectors create [options] ``` **Arguments** @@ -1217,7 +1217,7 @@ sim knowledge connectors create [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -1240,7 +1240,7 @@ sim knowledge connectors create [options] Delete Knowledge Connector ```bash -sim knowledge connectors delete [options] +sim knowledge connectors delete [options] ``` **Arguments** @@ -1249,7 +1249,7 @@ sim knowledge connectors delete [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1271,7 +1271,7 @@ sim knowledge connectors delete [options] Get Knowledge Connector ```bash -sim knowledge connectors get +sim knowledge connectors get ``` **Arguments** @@ -1280,7 +1280,7 @@ sim knowledge connectors get | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1290,7 +1290,7 @@ sim knowledge connectors get List Knowledge Connector Documents ```bash -sim knowledge connectors documents list [options] +sim knowledge connectors documents list [options] ``` **Arguments** @@ -1299,7 +1299,7 @@ sim knowledge connectors documents list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1316,12 +1316,42 @@ sim knowledge connectors documents list [options] +### sim knowledge connectors documents update + +Update Knowledge Connector Documents + +```bash +sim knowledge connectors documents update [options] +``` + +**Arguments** + + + +| Argument | Required | Description | +| --- | --- | --- | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | +| `connectorId` | Yes | Connector selected for the operation. | + + + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | +| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | + + + ### sim knowledge connectors list List Knowledge Connectors ```bash -sim knowledge connectors list [options] +sim knowledge connectors list [options] ``` **Arguments** @@ -1330,7 +1360,7 @@ sim knowledge connectors list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -1346,12 +1376,12 @@ sim knowledge connectors list [options] -### sim knowledge connectors update +### sim knowledge connectors sync -Update Knowledge Connector +Queue a knowledge connector synchronization ```bash -sim knowledge connectors update [options] +sim knowledge connectors sync [options] ``` **Arguments** @@ -1360,7 +1390,7 @@ sim knowledge connectors update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1371,18 +1401,17 @@ sim knowledge connectors update [options] | Option | Required | Description | | --- | --- | --- | -| `--source-config ` | No | Replacement source selection and filtering configuration. (JSON, or @path / @- to read a file or stdin). | -| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | -| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | +| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | +| `--no-rehydrate` | No | Send --rehydrate as false. | -### sim knowledge connectors documents update +### sim knowledge connectors update -Update Knowledge Connector Documents +Update Knowledge Connector ```bash -sim knowledge connectors documents update [options] +sim knowledge connectors update [options] ``` **Arguments** @@ -1391,7 +1420,7 @@ sim knowledge connectors documents update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1402,8 +1431,9 @@ sim knowledge connectors documents update [options] | Option | Required | Description | | --- | --- | --- | -| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | -| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | +| `--source-config ` | No | Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused. (JSON, or @path / @- to read a file or stdin). | +| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | +| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | @@ -1610,36 +1640,6 @@ sim knowledge search [options] -### sim knowledge sync create - -Sync Knowledge Connector - -```bash -sim knowledge sync create [options] -``` - -**Arguments** - - - -| Argument | Required | Description | -| --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | -| `connectorId` | Yes | Connector selected for the operation. | - - - -**Options** - - - -| Option | Required | Description | -| --- | --- | --- | -| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | -| `--no-rehydrate` | No | Send --rehydrate as false. | - - - ### sim knowledge update Update Knowledge Base diff --git a/apps/docs/openapi-v2-knowledge.json b/apps/docs/openapi-v2-knowledge.json index 79634b38de6..496500dfba3 100644 --- a/apps/docs/openapi-v2-knowledge.json +++ b/apps/docs/openapi-v2-knowledge.json @@ -766,7 +766,7 @@ "patch": { "operationId": "updateKnowledgeConnector", "summary": "Update Knowledge Connector", - "description": "Update connector source configuration, schedule, or active state. Authentication material cannot be changed through this operation. A workspace API key is rejected with `403`; use a personal API key.", + "description": "Update connector source configuration, schedule, or active state. Replacing source configuration on a runnable connector queues an immediate synchronization; paused connectors retain the change without synchronizing until resumed. Source configuration cannot be replaced while synchronization is already in progress. Authentication material cannot be changed through this operation. A workspace API key is rejected with `403`; use a personal API key.", "tags": ["Knowledge Bases"], "parameters": [ { @@ -3988,7 +3988,7 @@ "description": "Workspace that owns the knowledge base." }, "sourceConfig": { - "description": "Replacement source selection and filtering configuration.", + "description": "Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused.", "type": "object", "propertyNames": { "type": "string" diff --git a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts index bfa2f27c880..969c620e32a 100644 --- a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts +++ b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts @@ -6,6 +6,7 @@ import { import { defineInternalJsonRoute, internalRateLimits } from '@/lib/api/server/routes' import { internalKnowledgeAnalytics, + resolveInternalKnowledgeBillingAttribution, toInternalKnowledgeConnector, toInternalKnowledgeConnectorDetail, } from '@/lib/knowledge/api/internal-route' @@ -47,10 +48,12 @@ export const PATCH = defineInternalJsonRoute({ reason: 'Preserve existing internal connector-update behavior', }), errorPolicy: internalKnowledgeErrorPolicies.connectors, - mapInput: ({ params, body }) => ({ + mapInput: ({ params, body }, { principal, request }) => ({ connectorId: params.connectorId, knowledgeBaseId: params.id, updates: body, + resolveBillingAttribution: (workspaceId: string) => + resolveInternalKnowledgeBillingAttribution(request, principal, workspaceId), source: 'ui' as const, }), useCase: updateKnowledgeConnector, diff --git a/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts b/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts index 685a477429a..d2e58cc9b38 100644 --- a/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts +++ b/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts @@ -264,6 +264,31 @@ describe('v2 knowledge connector routes', () => { expect(mocks.connectorRemoved).toHaveBeenCalledOnce() }) + it('passes source changes to application billing without an adapter resolver', async () => { + const response = await updateConnector( + request(`/api/v2/knowledge/${KNOWLEDGE_BASE_ID}/connectors/${CONNECTOR_ID}`, 'PATCH', { + workspaceId: WORKSPACE_ID, + sourceConfig: { pageIds: ['page-2'] }, + }), + connectorContext + ) + + expect(response.status).toBe(200) + expect(mocks.update).toHaveBeenCalledWith( + expect.objectContaining({ + input: expect.objectContaining({ + assertedWorkspaceId: WORKSPACE_ID, + updates: expect.objectContaining({ sourceConfig: { pageIds: ['page-2'] } }), + }), + }) + ) + expect(mocks.update).toHaveBeenCalledWith( + expect.objectContaining({ + input: expect.not.objectContaining({ resolveBillingAttribution: expect.anything() }), + }) + ) + }) + it('queues connector synchronization without an adapter billing resolver', async () => { const response = await syncConnector( request(`/api/v2/knowledge/${KNOWLEDGE_BASE_ID}/connectors/${CONNECTOR_ID}/sync`, 'POST', { diff --git a/apps/sim/connectors/linear/linear.test.ts b/apps/sim/connectors/linear/linear.test.ts new file mode 100644 index 00000000000..873612db586 --- /dev/null +++ b/apps/sim/connectors/linear/linear.test.ts @@ -0,0 +1,37 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockFetchWithRetry } = vi.hoisted(() => ({ + mockFetchWithRetry: vi.fn(), +})) + +vi.mock('@/lib/knowledge/documents/utils', () => ({ + fetchWithRetry: mockFetchWithRetry, + VALIDATE_RETRY_OPTIONS: { maxRetries: 0 }, +})) + +import { linearConnector } from '@/connectors/linear/linear' + +describe('linearConnector authentication', () => { + beforeEach(() => { + vi.clearAllMocks() + mockFetchWithRetry.mockResolvedValue( + new Response(JSON.stringify({ data: { teams: { nodes: [{ id: 'team-1' }] } } }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) + ) + }) + + it.each([ + ['personal API key', 'lin_api_example', 'lin_api_example'], + ['OAuth access token', 'opaque-oauth-token', 'Bearer opaque-oauth-token'], + ])('sends a %s with the documented authorization shape', async (_kind, token, expected) => { + await expect(linearConnector.validateConfig(token, {})).resolves.toEqual({ valid: true }) + + const request = mockFetchWithRetry.mock.calls[0]?.[1] as RequestInit | undefined + expect(new Headers(request?.headers).get('Authorization')).toBe(expected) + }) +}) diff --git a/apps/sim/connectors/linear/linear.ts b/apps/sim/connectors/linear/linear.ts index 3a1639b311f..93aab253e64 100644 --- a/apps/sim/connectors/linear/linear.ts +++ b/apps/sim/connectors/linear/linear.ts @@ -7,6 +7,7 @@ import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/document import { linearConnectorMeta } from '@/connectors/linear/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { joinTagArray, parseMultiValue, parseTagDate } from '@/connectors/utils' +import { linearAuthorizationHeader } from '@/tools/linear/utils' const logger = createLogger('LinearConnector') @@ -88,7 +89,7 @@ async function linearGraphQL( method: 'POST', headers: { 'Content-Type': 'application/json', - Authorization: `Bearer ${accessToken}`, + Authorization: linearAuthorizationHeader(accessToken), }, body: JSON.stringify({ query, variables }), }, diff --git a/apps/sim/lib/api/contracts/v2/knowledge.ts b/apps/sim/lib/api/contracts/v2/knowledge.ts index a45f021cb03..41a16829de2 100644 --- a/apps/sim/lib/api/contracts/v2/knowledge.ts +++ b/apps/sim/lib/api/contracts/v2/knowledge.ts @@ -1652,7 +1652,9 @@ export const v2UpdateKnowledgeConnectorBodySchema = z sourceConfig: z .record(z.string(), z.unknown().describe('Connector-specific source configuration value.')) .optional() - .describe('Replacement source selection and filtering configuration.'), + .describe( + 'Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused.' + ), syncIntervalMinutes: z .number() .int() diff --git a/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts b/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts index 3ac09dd84ef..978199a5eeb 100644 --- a/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts +++ b/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts @@ -354,7 +354,7 @@ const declaredRoutes = [ knowledgeOperation({ operationId: 'updateKnowledgeConnector', summary: 'Update Knowledge Connector', - description: `Update connector source configuration, schedule, or active state. Authentication material cannot be changed through this operation. ${WORKSPACE_API_KEY_DENIED}`, + description: `Update connector source configuration, schedule, or active state. Replacing source configuration on a runnable connector queues an immediate synchronization; paused connectors retain the change without synchronizing until resumed. Source configuration cannot be replaced while synchronization is already in progress. Authentication material cannot be changed through this operation. ${WORKSPACE_API_KEY_DENIED}`, errors: RESOURCE_CONFLICT_ERRORS, success: { description: 'The updated connector.' }, }), diff --git a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts index 95afa3a9066..8c3757a4cb2 100644 --- a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts +++ b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts @@ -631,6 +631,41 @@ describe('manage_knowledge_base trusted application delegation', () => { expect(result.message).not.toContain('private-db') }) + it('passes trusted billing attribution for a source-change sync', async () => { + mockUpdateKnowledgeConnector.mockResolvedValueOnce({ + connector: { + id: 'connector-1', + knowledgeBaseId: KNOWLEDGE_BASE.id, + connectorType: 'notion', + sourceConfig: { pageIds: ['page-2'] }, + status: 'active', + }, + }) + + const result = await knowledgeBaseServerTool.execute( + { + operation: 'update_connector', + args: { connectorId: 'connector-1', sourceConfig: { pageIds: ['page-2'] } }, + }, + BILLED_CONTEXT + ) + + expect(result).toMatchObject({ + success: true, + message: 'Connector updated successfully. Synchronization was queued for the source change.', + }) + const call = mockUpdateKnowledgeConnector.mock.calls[0]?.[0] as { + input?: { resolveBillingAttribution?: (workspaceId: string) => Promise } + } + expectDelegatedPrincipal(call) + if (!call.input?.resolveBillingAttribution) { + throw new Error('Copilot connector update did not provide billing attribution') + } + await expect(call.input.resolveBillingAttribution('workspace-paid')).resolves.toEqual( + BILLED_CONTEXT.billingAttribution + ) + }) + it.each([ [ 'update_document', diff --git a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts index 20d88e2840f..c65ad755b66 100644 --- a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts +++ b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts @@ -1013,16 +1013,27 @@ export const knowledgeBaseServerTool: BaseServerTool + requireKnowledgeBillingAttribution(context, canonicalWorkspaceId), + source: 'agent', + } + ) return { success: true, - message: 'Connector updated successfully', + message: + updates.sourceConfig === undefined + ? 'Connector updated successfully' + : connector.status === 'paused' || connector.status === 'disabled' + ? 'Connector updated successfully. The source change will synchronize when the connector is resumed.' + : 'Connector updated successfully. Synchronization was queued for the source change.', data: { id: args.connectorId, ...(updates.sourceConfig !== undefined && { sourceConfig: updates.sourceConfig }), diff --git a/apps/sim/lib/knowledge/application/connectors.test.ts b/apps/sim/lib/knowledge/application/connectors.test.ts index 8947d91829e..cc17ed8ba0f 100644 --- a/apps/sim/lib/knowledge/application/connectors.test.ts +++ b/apps/sim/lib/knowledge/application/connectors.test.ts @@ -117,6 +117,8 @@ const delegatedPrincipal = { resourceScope: {}, } +const BILLING = { actorUserId: 'shared-user', workspaceId: 'workspace-a' } as never + describe('knowledge connector application use cases', () => { beforeEach(() => { vi.clearAllMocks() @@ -138,6 +140,7 @@ describe('knowledge connector application use cases', () => { mocks.resolveTokenIdentity.mockResolvedValue({ kind: 'oauth', userId: 'credential-owner' }) mocks.refreshToken.mockResolvedValue('access-token') mocks.validateConnectorConfig.mockResolvedValue({ valid: true }) + mocks.resolveBilling.mockResolvedValue(BILLING) }) afterAll(resetDbChainMock) @@ -281,11 +284,13 @@ describe('knowledge connector application use cases', () => { connectorId: 'connector-b', assertedWorkspaceId: 'workspace-a', updates: { sourceConfig: { space: 'ENG' } }, + resolveBillingAttribution: mocks.resolveBilling, source: 'agent', }, }) const orchestrationInput = mocks.updateConnector.mock.calls[0]?.[0] as { + resolveBillingAttribution?: () => Promise validateSourceConfig?: ( connector: { connectorType: string @@ -298,6 +303,10 @@ describe('knowledge connector application use cases', () => { if (!orchestrationInput.validateSourceConfig) { throw new Error('Application command did not provide source-config validation') } + if (!orchestrationInput.resolveBillingAttribution) { + throw new Error('Application command did not provide sync billing attribution') + } + await expect(orchestrationInput.resolveBillingAttribution()).resolves.toBe(BILLING) await expect( orchestrationInput.validateSourceConfig( { @@ -319,6 +328,7 @@ describe('knowledge connector application use cases', () => { expect.any(String) ) expect(mocks.validateConnectorConfig).toHaveBeenCalledWith('access-token', { space: 'ENG' }) + expect(mocks.resolveBilling).toHaveBeenCalledWith('workspace-a') }) it('rejects connector creation when the writer cannot use the workspace credential', async () => { diff --git a/apps/sim/lib/knowledge/application/connectors.ts b/apps/sim/lib/knowledge/application/connectors.ts index 9a591df3f5b..1510526af86 100644 --- a/apps/sim/lib/knowledge/application/connectors.ts +++ b/apps/sim/lib/knowledge/application/connectors.ts @@ -77,6 +77,7 @@ export interface UpdateKnowledgeConnectorInput extends KnowledgeConnectorApplica syncIntervalMinutes?: number status?: 'active' | 'paused' } + resolveBillingAttribution?(workspaceId: string): Promise } export interface DeleteKnowledgeConnectorInput extends KnowledgeConnectorApplicationInput { @@ -347,6 +348,13 @@ export const updateKnowledgeConnector = defineAuthorizedKnowledgeUseCase({ knowledgeBase: connectorTarget(context), connectorId: context.connectorId, updates: input.updates, + resolveBillingAttribution: () => { + const workspaceId = requireConnectorWorkspaceId(context) + return ( + input.resolveBillingAttribution?.(workspaceId) ?? + resolveKnowledgeBillingAttribution(principal, context) + ) + }, validateSourceConfig: (connector, sourceConfig) => { const workspaceId = requireConnectorWorkspaceId(context) return validateConnectorSourceConfig({ diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index 6131c462c93..ae702b7d043 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -136,6 +136,8 @@ describe('performUpdateKnowledgeConnector', () => { beforeEach(() => { vi.clearAllMocks() resetDbChainMock() + mockDispatchSync.mockResolvedValue(undefined) + resolveBillingAttribution.mockResolvedValue(BILLING) }) afterAll(resetDbChainMock) @@ -146,6 +148,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: {}, + resolveBillingAttribution, }) expect(outcome).toMatchObject({ success: false, errorCode: 'validation' }) @@ -161,6 +164,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { syncIntervalMinutes: 5 }, + resolveBillingAttribution, }) expect(outcome).toMatchObject({ success: false, errorCode: 'forbidden' }) @@ -175,6 +179,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { sourceConfig: { database: 'gone' } }, + resolveBillingAttribution, validateSourceConfig: async () => ({ message: 'Database not found', errorCode: 'validation' as const, @@ -198,6 +203,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { sourceConfig: { database: 'x' } }, + resolveBillingAttribution, validateSourceConfig: async () => ({ message: 'Failed to refresh access token. Please reconnect your account.', errorCode: 'unauthorized' as const, @@ -218,6 +224,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { status: 'active' }, + resolveBillingAttribution, }) expect(outcome).toMatchObject({ success: true }) @@ -237,12 +244,141 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { status: 'paused' }, + resolveBillingAttribution, recordSemanticAudit: false, }) expect(outcome).toMatchObject({ success: true }) expect(mockRecordAudit).not.toHaveBeenCalled() }) + + it('queues synchronization after replacing an active connector source', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(resolveBillingAttribution).toHaveBeenCalledOnce() + expect(mockDispatchSync).toHaveBeenCalledWith('conn-1', { + billingAttribution: BILLING, + requestId: 'req-1', + }) + }) + + it('saves a paused connector source without synchronizing it', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'paused' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'paused' }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(resolveBillingAttribution).not.toHaveBeenCalled() + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('does not synchronize a schedule-only update', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 60 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(resolveBillingAttribution).not.toHaveBeenCalled() + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('rejects a source replacement while synchronization is in progress', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'syncing' }, + ]) + const validateSourceConfig = vi.fn() + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(validateSourceConfig).not.toHaveBeenCalled() + expect(resolveBillingAttribution).not.toHaveBeenCalled() + expect(dbChainMockFns.update).not.toHaveBeenCalled() + }) + + it('fails before persisting when sync billing attribution cannot be resolved', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + const rejectsBilling = vi.fn().mockRejectedValue(new Error('billing unavailable')) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution: rejectsBilling, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'internal' }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('rejects a source replacement that races with synchronization', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'active' }]) + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'syncing' }]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) }) describe('performSyncKnowledgeConnector', () => { diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index 3071f5287c7..f745308185a 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -9,7 +9,7 @@ import { } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { generateId } from '@sim/utils/id' -import { and, eq, inArray, isNull, sql } from 'drizzle-orm' +import { and, eq, inArray, isNull, ne, sql } from 'drizzle-orm' import { encryptApiKey } from '@/lib/api-key/crypto' import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution' import { hasWorkspaceLiveSyncAccess } from '@/lib/billing/core/subscription' @@ -367,6 +367,8 @@ export interface PerformUpdateKnowledgeConnectorParams extends KnowledgeOperatio syncIntervalMinutes?: number status?: 'active' | 'paused' } + /** Resolves the payer only when a source change will queue synchronization. */ + resolveBillingAttribution: () => Promise /** * Validates a replacement `sourceConfig` against the live source. Supplied by * the caller because resolving the connector's token needs the requesting @@ -409,7 +411,15 @@ export async function getKnowledgeConnector( export async function performUpdateKnowledgeConnector( params: PerformUpdateKnowledgeConnectorParams ): Promise { - const { knowledgeBase: kb, connectorId, updates, validateSourceConfig, request, source } = params + const { + knowledgeBase: kb, + connectorId, + updates, + resolveBillingAttribution, + validateSourceConfig, + request, + source, + } = params const requestId = params.requestId ?? generateRequestId() const updatedFields = Object.keys(updates).filter( @@ -427,6 +437,13 @@ export async function performUpdateKnowledgeConnector( return fail('Connector not found', 'not_found') } + if (updates.sourceConfig !== undefined && existing.status === 'syncing') { + return fail( + 'Cannot update source configuration while connector synchronization is in progress', + 'conflict' + ) + } + if (updates.syncIntervalMinutes !== undefined) { if (!kb.workspaceId && updates.syncIntervalMinutes > 0 && updates.syncIntervalMinutes < 60) { return fail('Knowledge base is missing workspace billing context', 'conflict') @@ -447,6 +464,22 @@ export async function performUpdateKnowledgeConnector( } } + const resultingStatus = updates.status ?? existing.status + const shouldDispatchSourceSync = + updates.sourceConfig !== undefined && + resultingStatus !== 'paused' && + resultingStatus !== 'disabled' + let billingAttribution: BillingAttributionSnapshot | undefined + let dispatchSourceSync: Awaited> | undefined + if (shouldDispatchSourceSync) { + try { + billingAttribution = await resolveBillingAttribution() + dispatchSourceSync = await loadDispatchSync() + } catch (error) { + return classifyKnowledgeFailure(error, requestId, `Update connector ${connectorId}`) + } + } + const values: Partial = { updatedAt: new Date() } if (updates.sourceConfig !== undefined) { values.sourceConfig = updates.sourceConfig @@ -473,20 +506,32 @@ export async function performUpdateKnowledgeConnector( let updated: ConnectorRow try { + const updateConditions = [ + eq(knowledgeConnector.id, connectorId), + eq(knowledgeConnector.knowledgeBaseId, kb.id), + isNull(knowledgeConnector.archivedAt), + isNull(knowledgeConnector.deletedAt), + ] + if (updates.sourceConfig !== undefined) { + updateConditions.push(ne(knowledgeConnector.status, 'syncing')) + } + const [row] = await db .update(knowledgeConnector) .set(values) - .where( - and( - eq(knowledgeConnector.id, connectorId), - eq(knowledgeConnector.knowledgeBaseId, kb.id), - isNull(knowledgeConnector.archivedAt), - isNull(knowledgeConnector.deletedAt) - ) - ) + .where(and(...updateConditions)) .returning() if (!row) { + if (updates.sourceConfig !== undefined) { + const current = await getKnowledgeConnector(kb.id, connectorId) + if (current?.status === 'syncing') { + return fail( + 'Cannot update source configuration while connector synchronization is in progress', + 'conflict' + ) + } + } return fail('Connector not found', 'not_found') } updated = row @@ -518,6 +563,15 @@ export async function performUpdateKnowledgeConnector( }) } + if (dispatchSourceSync && billingAttribution) { + dispatchSourceSync(connectorId, { billingAttribution, requestId }).catch((error) => { + logger.error( + `[${requestId}] Failed to dispatch source-change sync for connector ${connectorId}`, + error + ) + }) + } + return { success: true, connector: withoutSecret(updated) } } diff --git a/packages/sim-cli/src/contract/commands.test.ts b/packages/sim-cli/src/contract/commands.test.ts index da0ac01386b..42e11f84a19 100644 --- a/packages/sim-cli/src/contract/commands.test.ts +++ b/packages/sim-cli/src/contract/commands.test.ts @@ -70,6 +70,27 @@ describe('the command tree', () => { } }) + it('places connector synchronization with the other connector commands', () => { + const all = leafPaths({ includeHidden: true }) + + expect(all).toContain('knowledge connectors sync') + expect(all).not.toContain('knowledge sync create') + for (const path of [ + ['create'], + ['delete'], + ['get'], + ['list'], + ['sync'], + ['update'], + ['documents', 'list'], + ['documents', 'update'], + ]) { + expect(commandAt('knowledge', 'connectors', ...path).helpInformation()).toContain( + '' + ) + } + }) + it('spells one concept with one flag name across the contract', () => { // `predicate` was `--filter` on two row commands and `--predicate` on the // third, and the same idea was `--q` here and `--query` on knowledge search. diff --git a/packages/sim-cli/src/contract/commands.ts b/packages/sim-cli/src/contract/commands.ts index 7c9f5d97b6a..1b5886419f5 100644 --- a/packages/sim-cli/src/contract/commands.ts +++ b/packages/sim-cli/src/contract/commands.ts @@ -141,15 +141,34 @@ export const CLI_CONTRACT: CliContract = { selectAll: { boolean: true, describe: 'Apply to every document in the knowledge base' }, }, }, + createKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, + listKnowledgeConnectors: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, + getKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, listKnowledgeConnectorDocuments: { command: 'knowledge connectors documents list', + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, + updateKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, }, updateKnowledgeConnectorDocuments: { command: 'knowledge connectors documents update', + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, flags: { documentIds: { name: 'document', list: true }, }, }, + syncKnowledgeConnector: { + command: 'knowledge connectors sync', + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + describe: 'Queue a knowledge connector synchronization', + }, // `DELETE /workflows/[id]/deploy` is an undeploy, not a delete. undeployWorkflow: { command: 'workflows undeploy', @@ -179,6 +198,7 @@ export const CLI_CONTRACT: CliContract = { confirm: 'This deletes the document and its embeddings.', }, deleteKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, confirm: 'This deletes the connector; --delete-documents also deletes its synchronized documents.', }, diff --git a/packages/sim-cli/src/generated/v2-api.ts b/packages/sim-cli/src/generated/v2-api.ts index 6271ef1ed30..72a3575472f 100644 --- a/packages/sim-cli/src/generated/v2-api.ts +++ b/packages/sim-cli/src/generated/v2-api.ts @@ -9325,7 +9325,8 @@ export const V2_OPERATIONS = { }, sourceConfig: { kind: 'object', - describe: 'Replacement source selection and filtering configuration.', + describe: + 'Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused.', }, syncIntervalMinutes: { kind: 'integer', diff --git a/packages/sim-cli/src/runtime/build.ts b/packages/sim-cli/src/runtime/build.ts index 958ccce154e..df6cfcd98d8 100644 --- a/packages/sim-cli/src/runtime/build.ts +++ b/packages/sim-cli/src/runtime/build.ts @@ -235,8 +235,11 @@ function addLeafCommand( const group = groupFor(groups, groupName) if (rest.length > 1) { - const [subName, ...tail] = rest - nestedGroup(group, subName).addCommand(buildLeaf(operation, spec, tail.join(' '))) + let parent = group + for (const segment of rest.slice(0, -1)) { + parent = nestedGroup(parent, segment) + } + parent.addCommand(buildLeaf(operation, spec, rest[rest.length - 1])) return } From d57e676e00301b409cdc910436cc7bbea566ee43 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 20 Aug 2026 22:13:26 -0700 Subject: [PATCH 2/7] Fix connector sync pause race --- .../knowledge-connector-sync.test.ts | 3 ++ .../background/knowledge-connector-sync.ts | 9 ++++- .../lib/knowledge/connectors/queue.test.ts | 40 +++++++++++++++++++ apps/sim/lib/knowledge/connectors/queue.ts | 20 +++++++++- .../knowledge/connectors/sync-engine.test.ts | 11 +++++ .../lib/knowledge/connectors/sync-engine.ts | 19 ++++++++- .../orchestration/connectors.test.ts | 23 ++++++++++- .../lib/knowledge/orchestration/connectors.ts | 24 ++++++++--- 8 files changed, 137 insertions(+), 12 deletions(-) diff --git a/apps/sim/background/knowledge-connector-sync.test.ts b/apps/sim/background/knowledge-connector-sync.test.ts index 39a2d8f4ca6..0730dbc44d3 100644 --- a/apps/sim/background/knowledge-connector-sync.test.ts +++ b/apps/sim/background/knowledge-connector-sync.test.ts @@ -60,6 +60,7 @@ describe('knowledge connector sync worker', () => { connectorId: 'connector-1', requestId: 'request-1', fullSync: true, + requireRunnable: true, billingAttribution: BILLING_ATTRIBUTION, }) @@ -72,6 +73,7 @@ describe('knowledge connector sync worker', () => { expect(mockExecuteSync).toHaveBeenCalledWith('connector-1', { billingAttribution: BILLING_ATTRIBUTION, fullSync: true, + requireRunnable: true, rehydrate: undefined, }) }) @@ -94,6 +96,7 @@ describe('knowledge connector sync worker', () => { expect(mockExecuteSync).toHaveBeenCalledWith('connector-1', { billingAttribution: BILLING_ATTRIBUTION, fullSync: undefined, + requireRunnable: undefined, rehydrate: true, }) }) diff --git a/apps/sim/background/knowledge-connector-sync.ts b/apps/sim/background/knowledge-connector-sync.ts index 6efe6dbd26e..a9ee4a660e5 100644 --- a/apps/sim/background/knowledge-connector-sync.ts +++ b/apps/sim/background/knowledge-connector-sync.ts @@ -10,13 +10,18 @@ import { CONNECTOR_SYNC_MAX_DURATION_SECONDS } from '@/lib/knowledge/connectors/ const logger = createLogger('TriggerKnowledgeConnectorSync') export async function executeConnectorSyncJob(payload: unknown) { - const { connectorId, fullSync, rehydrate, requestId, billingAttribution } = + const { connectorId, fullSync, requireRunnable, rehydrate, requestId, billingAttribution } = assertConnectorSyncPayload(payload) logger.info(`[${requestId}] Starting connector sync: ${connectorId}`) try { - const result = await executeSync(connectorId, { billingAttribution, fullSync, rehydrate }) + const result = await executeSync(connectorId, { + billingAttribution, + fullSync, + requireRunnable, + rehydrate, + }) logger.info(`[${requestId}] Connector sync completed`, { connectorId, diff --git a/apps/sim/lib/knowledge/connectors/queue.test.ts b/apps/sim/lib/knowledge/connectors/queue.test.ts index 9ad039247b1..f3d818e34be 100644 --- a/apps/sim/lib/knowledge/connectors/queue.test.ts +++ b/apps/sim/lib/knowledge/connectors/queue.test.ts @@ -21,6 +21,7 @@ vi.mock('@/lib/knowledge/documents/service', () => ({ })) vi.mock('@/lib/knowledge/connectors/sync-engine', () => ({ executeSync: mockExecuteSync, + isConnectorRunnableStatus: (status: string) => status === 'active' || status === 'error', })) import { assertConnectorSyncPayload, dispatchSync } from '@/lib/knowledge/connectors/queue' @@ -53,6 +54,7 @@ describe('connector sync queue', () => { queueTableRows(schemaMock.knowledgeConnector, [ { knowledgeBaseId: 'knowledge-base-1', + connectorStatus: 'active', connectorArchivedAt: null, connectorDeletedAt: null, workspaceId: 'workspace-paid', @@ -80,6 +82,7 @@ describe('connector sync queue', () => { { connectorId: 'connector-1', fullSync: true, + requireRunnable: undefined, rehydrate: undefined, requestId: 'request-1', billingAttribution: BILLING_ATTRIBUTION, @@ -110,6 +113,43 @@ describe('connector sync queue', () => { ) }) + it('carries the runnable requirement into the queued payload', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requireRunnable: true, + requestId: 'request-1', + }) + + expect(mockTrigger).toHaveBeenCalledWith( + 'knowledge-connector-sync', + expect.objectContaining({ connectorId: 'connector-1', requireRunnable: true }), + expect.anything() + ) + }) + + it('skips automatic dispatch when the connector was paused concurrently', async () => { + resetDbChainMock() + queueTableRows(schemaMock.knowledgeConnector, [ + { + knowledgeBaseId: 'knowledge-base-1', + connectorStatus: 'paused', + connectorArchivedAt: null, + connectorDeletedAt: null, + workspaceId: 'workspace-paid', + kbDeletedAt: null, + }, + ]) + + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requireRunnable: true, + requestId: 'request-1', + }) + + expect(mockTrigger).not.toHaveBeenCalled() + expect(mockExecuteSync).not.toHaveBeenCalled() + }) + it('rejects legacy payloads without billing attribution', () => { expect(() => assertConnectorSyncPayload({ diff --git a/apps/sim/lib/knowledge/connectors/queue.ts b/apps/sim/lib/knowledge/connectors/queue.ts index 3bd73456ad5..56599de58bc 100644 --- a/apps/sim/lib/knowledge/connectors/queue.ts +++ b/apps/sim/lib/knowledge/connectors/queue.ts @@ -11,7 +11,7 @@ import { type BillingAttributionSnapshot, } from '@/lib/billing/core/billing-attribution' import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' -import { executeSync } from '@/lib/knowledge/connectors/sync-engine' +import { executeSync, isConnectorRunnableStatus } from '@/lib/knowledge/connectors/sync-engine' import { isTriggerAvailable } from '@/lib/knowledge/documents/service' const logger = createLogger('ConnectorSyncQueue') @@ -19,6 +19,8 @@ const logger = createLogger('ConnectorSyncQueue') export interface ConnectorSyncPayload { connectorId: string fullSync?: boolean + /** Skip automatic work if the connector is paused or disabled before execution starts. */ + requireRunnable?: boolean /** * Force re-hydration + re-indexing of already-synced documents for connectors * whose rendered content can drift without a hash change (see @@ -34,6 +36,7 @@ export interface ConnectorSyncPayload { export interface DispatchSyncOptions { billingAttribution: BillingAttributionSnapshot fullSync?: boolean + requireRunnable?: boolean rehydrate?: boolean requestId?: string } @@ -55,6 +58,9 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload if (value.fullSync !== undefined && typeof value.fullSync !== 'boolean') { throw new Error('Connector sync payload fullSync must be a boolean when provided') } + if (value.requireRunnable !== undefined && typeof value.requireRunnable !== 'boolean') { + throw new Error('Connector sync payload requireRunnable must be a boolean when provided') + } if (value.rehydrate !== undefined && typeof value.rehydrate !== 'boolean') { throw new Error('Connector sync payload rehydrate must be a boolean when provided') } @@ -65,6 +71,7 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload return { connectorId: value.connectorId, fullSync: value.fullSync as boolean | undefined, + requireRunnable: value.requireRunnable as boolean | undefined, rehydrate: value.rehydrate as boolean | undefined, requestId: value.requestId, billingAttribution: assertBillingAttributionSnapshot(value.billingAttribution), @@ -87,6 +94,7 @@ export async function dispatchSync( const payload = assertConnectorSyncPayload({ connectorId, fullSync: options?.fullSync, + requireRunnable: options?.requireRunnable, rehydrate: options?.rehydrate, requestId, billingAttribution: options?.billingAttribution, @@ -95,6 +103,7 @@ export async function dispatchSync( const connectorRows = await db .select({ knowledgeBaseId: knowledgeConnector.knowledgeBaseId, + connectorStatus: knowledgeConnector.status, connectorArchivedAt: knowledgeConnector.archivedAt, connectorDeletedAt: knowledgeConnector.deletedAt, workspaceId: knowledgeBase.workspaceId, @@ -134,6 +143,14 @@ export async function dispatchSync( }) return } + if (payload.requireRunnable && !isConnectorRunnableStatus(row.connectorStatus)) { + logger.info('Skipping automatic sync dispatch: connector is not runnable', { + connectorId, + status: row.connectorStatus, + requestId, + }) + return + } if (!row.workspaceId) { throw new Error(`Connector ${connectorId} is missing workspace billing context`) } @@ -161,6 +178,7 @@ export async function dispatchSync( executeSync(connectorId, { fullSync: payload.fullSync, + requireRunnable: payload.requireRunnable, rehydrate: payload.rehydrate, billingAttribution: payload.billingAttribution, }).catch((error) => { diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index 6e88d9d3814..90d656ceccb 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -7,6 +7,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { classifySuspectListing, evaluateListingSafety, + isConnectorRunnableStatus, isStuckDocumentSweepEligible, mergeHydratedDocument, type PreviousListingObservation, @@ -44,6 +45,16 @@ vi.mock('@/connectors/registry.server', () => ({ }, })) +describe('isConnectorRunnableStatus', () => { + it.each(['active', 'error'])('allows automatic sync from %s', (status) => { + expect(isConnectorRunnableStatus(status)).toBe(true) + }) + + it.each(['paused', 'disabled', 'syncing'])('blocks automatic sync from %s', (status) => { + expect(isConnectorRunnableStatus(status)).toBe(false) + }) +}) + describe('shouldReconcileDeletions', () => { it('runs on a clean full listing', async () => { const { shouldReconcileDeletions } = await import('@/lib/knowledge/connectors/sync-engine') diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index 87b89dc100a..2ffa68cd98e 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -91,6 +91,12 @@ const QUEUED_DISPATCH_GRACE_MINUTES = Math.ceil( ) const RETRY_WINDOW_DAYS = 7 const MAX_CONSECUTIVE_FAILURES = 10 +const RUNNABLE_CONNECTOR_STATUSES = ['active', 'error'] as const + +/** Whether an automatic connector sync may begin from this persisted state. */ +export function isConnectorRunnableStatus(status: string): boolean { + return RUNNABLE_CONNECTOR_STATUSES.some((runnableStatus) => runnableStatus === status) +} /** The processing state the stuck-document sweep decides on, one row at a time. */ export interface StuckDocumentSweepCandidate { @@ -777,6 +783,7 @@ export async function executeSync( options: { billingAttribution: BillingAttributionSnapshot fullSync?: boolean + requireRunnable?: boolean rehydrate?: boolean } ): Promise { @@ -808,6 +815,14 @@ export async function executeSync( const connector = connectorRows[0] + if (options.requireRunnable && !isConnectorRunnableStatus(connector.status)) { + logger.info('Skipping automatic sync: connector is not runnable', { + connectorId, + status: connector.status, + }) + return result + } + const connectorConfig = CONNECTOR_REGISTRY[connector.connectorType] if (!connectorConfig) { throw new Error(`Unknown connector type: ${connector.connectorType}`) @@ -857,7 +872,9 @@ export async function executeSync( .where( and( eq(knowledgeConnector.id, connectorId), - ne(knowledgeConnector.status, 'syncing'), + options.requireRunnable + ? inArray(knowledgeConnector.status, RUNNABLE_CONNECTOR_STATUSES) + : ne(knowledgeConnector.status, 'syncing'), isNull(knowledgeConnector.archivedAt), isNull(knowledgeConnector.deletedAt) ) diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index ae702b7d043..d0b2dd31689 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -274,6 +274,7 @@ describe('performUpdateKnowledgeConnector', () => { expect(mockDispatchSync).toHaveBeenCalledWith('conn-1', { billingAttribution: BILLING, requestId: 'req-1', + requireRunnable: true, }) }) @@ -361,10 +362,10 @@ describe('performUpdateKnowledgeConnector', () => { expect(mockDispatchSync).not.toHaveBeenCalled() }) - it('rejects a source replacement that races with synchronization', async () => { + it('rejects a source replacement that races with a pause', async () => { dbChainMockFns.limit .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'active' }]) - .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'syncing' }]) + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'paused' }]) dbChainMockFns.returning.mockResolvedValueOnce([]) const outcome = await performUpdateKnowledgeConnector({ @@ -379,6 +380,24 @@ describe('performUpdateKnowledgeConnector', () => { expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) expect(mockDispatchSync).not.toHaveBeenCalled() }) + + it('rejects a pause that races with synchronization startup', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'active' }]) + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'syncing' }]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { status: 'paused' }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) }) describe('performSyncKnowledgeConnector', () => { diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index f745308185a..1712f179e5d 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -9,7 +9,7 @@ import { } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { generateId } from '@sim/utils/id' -import { and, eq, inArray, isNull, ne, sql } from 'drizzle-orm' +import { and, eq, inArray, isNull, sql } from 'drizzle-orm' import { encryptApiKey } from '@/lib/api-key/crypto' import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution' import { hasWorkspaceLiveSyncAccess } from '@/lib/billing/core/subscription' @@ -443,6 +443,9 @@ export async function performUpdateKnowledgeConnector( 'conflict' ) } + if (updates.status !== undefined && existing.status === 'syncing') { + return fail('Cannot change connector status while synchronization is in progress', 'conflict') + } if (updates.syncIntervalMinutes !== undefined) { if (!kb.workspaceId && updates.syncIntervalMinutes > 0 && updates.syncIntervalMinutes < 60) { @@ -512,8 +515,8 @@ export async function performUpdateKnowledgeConnector( isNull(knowledgeConnector.archivedAt), isNull(knowledgeConnector.deletedAt), ] - if (updates.sourceConfig !== undefined) { - updateConditions.push(ne(knowledgeConnector.status, 'syncing')) + if (updates.sourceConfig !== undefined || updates.status !== undefined) { + updateConditions.push(eq(knowledgeConnector.status, existing.status)) } const [row] = await db @@ -523,14 +526,19 @@ export async function performUpdateKnowledgeConnector( .returning() if (!row) { - if (updates.sourceConfig !== undefined) { + if (updates.sourceConfig !== undefined || updates.status !== undefined) { const current = await getKnowledgeConnector(kb.id, connectorId) if (current?.status === 'syncing') { return fail( - 'Cannot update source configuration while connector synchronization is in progress', + updates.sourceConfig !== undefined + ? 'Cannot update source configuration while connector synchronization is in progress' + : 'Cannot change connector status while synchronization is in progress', 'conflict' ) } + if (current) { + return fail('Connector status changed during the update; retry the request', 'conflict') + } } return fail('Connector not found', 'not_found') } @@ -564,7 +572,11 @@ export async function performUpdateKnowledgeConnector( } if (dispatchSourceSync && billingAttribution) { - dispatchSourceSync(connectorId, { billingAttribution, requestId }).catch((error) => { + dispatchSourceSync(connectorId, { + billingAttribution, + requestId, + requireRunnable: true, + }).catch((error) => { logger.error( `[${requestId}] Failed to dispatch source-change sync for connector ${connectorId}`, error From 335c01546278122c8ee015e2817a22c32b93ff05 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 20 Aug 2026 22:21:56 -0700 Subject: [PATCH 3/7] fix(knowledge): surface connector sync dispatch failures --- .../orchestration/connectors.test.ts | 27 +++++++++++++++++++ .../lib/knowledge/orchestration/connectors.ts | 21 ++++++++------- 2 files changed, 39 insertions(+), 9 deletions(-) diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index d0b2dd31689..3f74e19b846 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -278,6 +278,33 @@ describe('performUpdateKnowledgeConnector', () => { }) }) + it('reports a queue failure after replacing an active connector source', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + mockDispatchSync.mockRejectedValueOnce(new Error('queue unavailable')) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ + success: false, + errorCode: 'internal', + error: 'queue unavailable', + }) + expect(dbChainMockFns.update).toHaveBeenCalledOnce() + expect(mockDispatchSync).toHaveBeenCalledOnce() + }) + it('saves a paused connector source without synchronizing it', async () => { dbChainMockFns.limit.mockResolvedValueOnce([ { id: 'conn-1', connectorType: 'notion', status: 'paused' }, diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index 1712f179e5d..97560ccf638 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -572,16 +572,19 @@ export async function performUpdateKnowledgeConnector( } if (dispatchSourceSync && billingAttribution) { - dispatchSourceSync(connectorId, { - billingAttribution, - requestId, - requireRunnable: true, - }).catch((error) => { - logger.error( - `[${requestId}] Failed to dispatch source-change sync for connector ${connectorId}`, - error + try { + await dispatchSourceSync(connectorId, { + billingAttribution, + requestId, + requireRunnable: true, + }) + } catch (error) { + return classifyKnowledgeFailure( + error, + requestId, + `Dispatch source-change sync for connector ${connectorId}` ) - }) + } } return { success: true, connector: withoutSecret(updated) } From e632415ab00d49497e3ad1a2225aff6efa90459c Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 20 Aug 2026 22:31:17 -0700 Subject: [PATCH 4/7] fix(knowledge): make connector sync recovery durable --- .../api/knowledge/connectors/sync/route.ts | 6 +++- .../lib/knowledge/connectors/sync-engine.ts | 7 +++- .../orchestration/connectors.test.ts | 36 ++++++++++++++++--- .../lib/knowledge/orchestration/connectors.ts | 8 ++++- 4 files changed, 49 insertions(+), 8 deletions(-) diff --git a/apps/sim/app/api/knowledge/connectors/sync/route.ts b/apps/sim/app/api/knowledge/connectors/sync/route.ts index e6dd8177b1f..04e4f050aec 100644 --- a/apps/sim/app/api/knowledge/connectors/sync/route.ts +++ b/apps/sim/app/api/knowledge/connectors/sync/route.ts @@ -102,7 +102,11 @@ export const GET = withRouteHandler(async (request: NextRequest) => { throw new Error(`Connector ${connector.id} is missing workspace billing context`) } const billingAttribution = await resolveSystemBillingAttribution(connector.workspaceId) - await dispatchSync(connector.id, { billingAttribution, requestId }) + await dispatchSync(connector.id, { + billingAttribution, + requestId, + requireRunnable: true, + }) } catch (error) { logger.error(`[${requestId}] Failed to dispatch sync for connector ${connector.id}`, error) } diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index 2ffa68cd98e..bb4fd2642d0 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -882,7 +882,12 @@ export async function executeSync( .returning({ id: knowledgeConnector.id }) if (lockResult.length === 0) { - logger.info('Sync already in progress, skipping', { connectorId }) + logger.info( + options.requireRunnable + ? 'Connector is not runnable or sync is already in progress, skipping' + : 'Sync already in progress, skipping', + { connectorId } + ) return result } diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index 3f74e19b846..e9846199ccd 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -254,10 +254,20 @@ describe('performUpdateKnowledgeConnector', () => { it('queues synchronization after replacing an active connector source', async () => { dbChainMockFns.limit.mockResolvedValueOnce([ - { id: 'conn-1', connectorType: 'notion', status: 'active' }, + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, ]) dbChainMockFns.returning.mockResolvedValueOnce([ - { id: 'conn-1', connectorType: 'notion', status: 'active' }, + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, ]) const outcome = await performUpdateKnowledgeConnector({ @@ -278,12 +288,22 @@ describe('performUpdateKnowledgeConnector', () => { }) }) - it('reports a queue failure after replacing an active connector source', async () => { + it('reports a queue failure and leaves the source sync due for retry', async () => { dbChainMockFns.limit.mockResolvedValueOnce([ - { id: 'conn-1', connectorType: 'notion', status: 'active' }, + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, ]) dbChainMockFns.returning.mockResolvedValueOnce([ - { id: 'conn-1', connectorType: 'notion', status: 'active' }, + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, ]) mockDispatchSync.mockRejectedValueOnce(new Error('queue unavailable')) @@ -302,6 +322,12 @@ describe('performUpdateKnowledgeConnector', () => { error: 'queue unavailable', }) expect(dbChainMockFns.update).toHaveBeenCalledOnce() + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ + sourceConfig: { database: 'next' }, + nextSyncAt: expect.any(Date), + }) + ) expect(mockDispatchSync).toHaveBeenCalledOnce() }) diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index 97560ccf638..da9b59451cf 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -483,7 +483,10 @@ export async function performUpdateKnowledgeConnector( } } - const values: Partial = { updatedAt: new Date() } + const updateTimestamp = new Date() + const values: Partial = { + updatedAt: updateTimestamp, + } if (updates.sourceConfig !== undefined) { values.sourceConfig = updates.sourceConfig } @@ -506,6 +509,9 @@ export async function performUpdateKnowledgeConnector( } } } + if (shouldDispatchSourceSync) { + values.nextSyncAt = updateTimestamp + } let updated: ConnectorRow try { From 531cde57fe50870420ec1b5dd96bbe76ce924396 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 20 Aug 2026 22:48:49 -0700 Subject: [PATCH 5/7] fix(knowledge): deduplicate connector sync dispatches --- .../api/knowledge/connectors/sync/route.ts | 2 + .../lib/knowledge/connectors/queue.test.ts | 64 ++++++++++++++++--- apps/sim/lib/knowledge/connectors/queue.ts | 28 +++++++- .../orchestration/connectors.test.ts | 1 + .../lib/knowledge/orchestration/connectors.ts | 1 + 5 files changed, 85 insertions(+), 11 deletions(-) diff --git a/apps/sim/app/api/knowledge/connectors/sync/route.ts b/apps/sim/app/api/knowledge/connectors/sync/route.ts index 04e4f050aec..233e5d5e89a 100644 --- a/apps/sim/app/api/knowledge/connectors/sync/route.ts +++ b/apps/sim/app/api/knowledge/connectors/sync/route.ts @@ -70,6 +70,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { const dueConnectors = await db .select({ id: knowledgeConnector.id, + nextSyncAt: knowledgeConnector.nextSyncAt, workspaceId: knowledgeBase.workspaceId, }) .from(knowledgeConnector) @@ -104,6 +105,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { const billingAttribution = await resolveSystemBillingAttribution(connector.workspaceId) await dispatchSync(connector.id, { billingAttribution, + expectedNextSyncAt: connector.nextSyncAt ?? undefined, requestId, requireRunnable: true, }) diff --git a/apps/sim/lib/knowledge/connectors/queue.test.ts b/apps/sim/lib/knowledge/connectors/queue.test.ts index f3d818e34be..b6b53e9ec73 100644 --- a/apps/sim/lib/knowledge/connectors/queue.test.ts +++ b/apps/sim/lib/knowledge/connectors/queue.test.ts @@ -4,15 +4,24 @@ import { queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockExecuteSync, mockIsTriggerAvailable, mockResolveTriggerRegion, mockTrigger } = - vi.hoisted(() => ({ - mockExecuteSync: vi.fn(), - mockIsTriggerAvailable: vi.fn(), - mockResolveTriggerRegion: vi.fn(), - mockTrigger: vi.fn(), - })) - -vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mockTrigger } })) +const { + mockCreateIdempotencyKey, + mockExecuteSync, + mockIsTriggerAvailable, + mockResolveTriggerRegion, + mockTrigger, +} = vi.hoisted(() => ({ + mockCreateIdempotencyKey: vi.fn(), + mockExecuteSync: vi.fn(), + mockIsTriggerAvailable: vi.fn(), + mockResolveTriggerRegion: vi.fn(), + mockTrigger: vi.fn(), +})) + +vi.mock('@trigger.dev/sdk', () => ({ + idempotencyKeys: { create: mockCreateIdempotencyKey }, + tasks: { trigger: mockTrigger }, +})) vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: mockResolveTriggerRegion, })) @@ -46,6 +55,7 @@ const BILLING_ATTRIBUTION = { periodEnd: '2026-08-01T00:00:00.000Z', }, } +const NEXT_SYNC_AT = new Date('2026-07-15T12:00:00.000Z') describe('connector sync queue', () => { beforeEach(() => { @@ -57,11 +67,13 @@ describe('connector sync queue', () => { connectorStatus: 'active', connectorArchivedAt: null, connectorDeletedAt: null, + connectorNextSyncAt: NEXT_SYNC_AT, workspaceId: 'workspace-paid', kbDeletedAt: null, }, ]) mockIsTriggerAvailable.mockReturnValue(true) + mockCreateIdempotencyKey.mockResolvedValue('idempotency-key') mockResolveTriggerRegion.mockResolvedValue('us-east-1') mockTrigger.mockResolvedValue({ id: 'run-1' }) }) @@ -116,6 +128,7 @@ describe('connector sync queue', () => { it('carries the runnable requirement into the queued payload', async () => { await dispatchSync('connector-1', { billingAttribution: BILLING_ATTRIBUTION, + expectedNextSyncAt: NEXT_SYNC_AT, requireRunnable: true, requestId: 'request-1', }) @@ -123,8 +136,38 @@ describe('connector sync queue', () => { expect(mockTrigger).toHaveBeenCalledWith( 'knowledge-connector-sync', expect.objectContaining({ connectorId: 'connector-1', requireRunnable: true }), - expect.anything() + expect.objectContaining({ idempotencyKey: 'idempotency-key' }) ) + expect(mockCreateIdempotencyKey).toHaveBeenCalledWith( + `knowledge-connector-sync:connector-1:${NEXT_SYNC_AT.toISOString()}`, + { scope: 'global' } + ) + }) + + it('skips an automatic dispatch after the due time changes', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + expectedNextSyncAt: new Date('2026-07-15T11:00:00.000Z'), + requireRunnable: true, + requestId: 'request-1', + }) + + expect(mockCreateIdempotencyKey).not.toHaveBeenCalled() + expect(mockTrigger).not.toHaveBeenCalled() + expect(mockExecuteSync).not.toHaveBeenCalled() + }) + + it('rejects automatic dispatch without an expected due time', async () => { + await expect( + dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requireRunnable: true, + requestId: 'request-1', + }) + ).rejects.toThrow('Automatic connector sync dispatch requires the expected next sync time') + + expect(mockTrigger).not.toHaveBeenCalled() + expect(mockExecuteSync).not.toHaveBeenCalled() }) it('skips automatic dispatch when the connector was paused concurrently', async () => { @@ -142,6 +185,7 @@ describe('connector sync queue', () => { await dispatchSync('connector-1', { billingAttribution: BILLING_ATTRIBUTION, + expectedNextSyncAt: NEXT_SYNC_AT, requireRunnable: true, requestId: 'request-1', }) diff --git a/apps/sim/lib/knowledge/connectors/queue.ts b/apps/sim/lib/knowledge/connectors/queue.ts index 56599de58bc..467b8adbe94 100644 --- a/apps/sim/lib/knowledge/connectors/queue.ts +++ b/apps/sim/lib/knowledge/connectors/queue.ts @@ -4,7 +4,7 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' -import { tasks } from '@trigger.dev/sdk' +import { idempotencyKeys, tasks } from '@trigger.dev/sdk' import { eq } from 'drizzle-orm' import { assertBillingAttributionSnapshot, @@ -35,6 +35,7 @@ export interface ConnectorSyncPayload { export interface DispatchSyncOptions { billingAttribution: BillingAttributionSnapshot + expectedNextSyncAt?: Date fullSync?: boolean requireRunnable?: boolean rehydrate?: boolean @@ -89,6 +90,13 @@ export async function dispatchSync( if (!isNonEmptyString(connectorId)) { throw new Error('Connector sync dispatch requires a connector ID') } + if ( + options.requireRunnable && + (!(options.expectedNextSyncAt instanceof Date) || + Number.isNaN(options.expectedNextSyncAt.getTime())) + ) { + throw new Error('Automatic connector sync dispatch requires the expected next sync time') + } const requestId = options?.requestId ?? generateId() const payload = assertConnectorSyncPayload({ @@ -106,6 +114,7 @@ export async function dispatchSync( connectorStatus: knowledgeConnector.status, connectorArchivedAt: knowledgeConnector.archivedAt, connectorDeletedAt: knowledgeConnector.deletedAt, + connectorNextSyncAt: knowledgeConnector.nextSyncAt, workspaceId: knowledgeBase.workspaceId, kbDeletedAt: knowledgeBase.deletedAt, }) @@ -151,6 +160,16 @@ export async function dispatchSync( }) return } + if ( + options.expectedNextSyncAt && + row.connectorNextSyncAt?.getTime() !== options.expectedNextSyncAt.getTime() + ) { + logger.info('Skipping stale automatic sync dispatch: next sync time changed', { + connectorId, + requestId, + }) + return + } if (!row.workspaceId) { throw new Error(`Connector ${connectorId} is missing workspace billing context`) } @@ -168,7 +187,14 @@ export async function dispatchSync( ] if (isTriggerAvailable()) { + const idempotencyKey = options.expectedNextSyncAt + ? await idempotencyKeys.create( + `knowledge-connector-sync:${connectorId}:${options.expectedNextSyncAt.toISOString()}`, + { scope: 'global' } + ) + : undefined await tasks.trigger('knowledge-connector-sync', payload, { + ...(idempotencyKey ? { idempotencyKey } : {}), tags, region: await resolveTriggerRegion(), }) diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index e9846199ccd..a6d3099e9ed 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -283,6 +283,7 @@ describe('performUpdateKnowledgeConnector', () => { expect(resolveBillingAttribution).toHaveBeenCalledOnce() expect(mockDispatchSync).toHaveBeenCalledWith('conn-1', { billingAttribution: BILLING, + expectedNextSyncAt: expect.any(Date), requestId: 'req-1', requireRunnable: true, }) diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index da9b59451cf..762cd89de2e 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -581,6 +581,7 @@ export async function performUpdateKnowledgeConnector( try { await dispatchSourceSync(connectorId, { billingAttribution, + expectedNextSyncAt: updateTimestamp, requestId, requireRunnable: true, }) From ea8b87750199324b86f6d8d44b09e07a56dfc7ac Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 21 Aug 2026 03:19:16 -0700 Subject: [PATCH 6/7] fix(knowledge): preserve pending connector syncs --- .../orchestration/connectors.test.ts | 61 +++++++++++++++++++ .../lib/knowledge/orchestration/connectors.ts | 43 ++++++++----- 2 files changed, 88 insertions(+), 16 deletions(-) diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index a6d3099e9ed..1b33f3458d0 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -375,6 +375,67 @@ describe('performUpdateKnowledgeConnector', () => { expect(mockDispatchSync).not.toHaveBeenCalled() }) + it('preserves an already-due source sync when scheduled sync is disabled', async () => { + const pendingSourceSyncAt = new Date(0) + dbChainMockFns.limit.mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + nextSyncAt: pendingSourceSyncAt, + status: 'active', + }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + nextSyncAt: pendingSourceSyncAt, + status: 'active', + }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ nextSyncAt: pendingSourceSyncAt, syncIntervalMinutes: 0 }) + ) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('rejects an interval update that races with a source-sync due marker', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', nextSyncAt: null, status: 'active' }, + ]) + .mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + nextSyncAt: new Date(), + status: 'active', + }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + it('rejects a source replacement while synchronization is in progress', async () => { dbChainMockFns.limit.mockResolvedValueOnce([ { id: 'conn-1', connectorType: 'notion', status: 'syncing' }, diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index 762cd89de2e..a995997fd6a 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -493,9 +493,11 @@ export async function performUpdateKnowledgeConnector( if (updates.syncIntervalMinutes !== undefined) { values.syncIntervalMinutes = updates.syncIntervalMinutes values.nextSyncAt = - updates.syncIntervalMinutes > 0 - ? new Date(Date.now() + updates.syncIntervalMinutes * 60 * 1000) - : null + existing.nextSyncAt && existing.nextSyncAt <= updateTimestamp + ? existing.nextSyncAt + : updates.syncIntervalMinutes > 0 + ? new Date(updateTimestamp.getTime() + updates.syncIntervalMinutes * 60 * 1000) + : null } if (updates.status !== undefined) { values.status = updates.status @@ -524,6 +526,13 @@ export async function performUpdateKnowledgeConnector( if (updates.sourceConfig !== undefined || updates.status !== undefined) { updateConditions.push(eq(knowledgeConnector.status, existing.status)) } + if (values.nextSyncAt !== undefined) { + updateConditions.push( + existing.nextSyncAt + ? eq(knowledgeConnector.nextSyncAt, existing.nextSyncAt) + : isNull(knowledgeConnector.nextSyncAt) + ) + } const [row] = await db .update(knowledgeConnector) @@ -532,19 +541,21 @@ export async function performUpdateKnowledgeConnector( .returning() if (!row) { - if (updates.sourceConfig !== undefined || updates.status !== undefined) { - const current = await getKnowledgeConnector(kb.id, connectorId) - if (current?.status === 'syncing') { - return fail( - updates.sourceConfig !== undefined - ? 'Cannot update source configuration while connector synchronization is in progress' - : 'Cannot change connector status while synchronization is in progress', - 'conflict' - ) - } - if (current) { - return fail('Connector status changed during the update; retry the request', 'conflict') - } + const current = await getKnowledgeConnector(kb.id, connectorId) + if (current?.status === 'syncing' && updates.sourceConfig !== undefined) { + return fail( + 'Cannot update source configuration while connector synchronization is in progress', + 'conflict' + ) + } + if (current?.status === 'syncing' && updates.status !== undefined) { + return fail( + 'Cannot change connector status while synchronization is in progress', + 'conflict' + ) + } + if (current) { + return fail('Connector changed during the update; retry the request', 'conflict') } return fail('Connector not found', 'not_found') } From f66b0b7afc4a412c2b1281169c6ee3b9be0a4539 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 21 Aug 2026 10:15:49 -0700 Subject: [PATCH 7/7] fix(knowledge): lock connector sync snapshot --- .../lib/knowledge/connectors/sync-engine.ts | 33 ++++++++----- .../orchestration/connectors.test.ts | 47 +++++++++++++++++++ .../lib/knowledge/orchestration/connectors.ts | 33 +++++-------- 3 files changed, 80 insertions(+), 33 deletions(-) diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index bb4fd2642d0..8e22166b6d9 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -813,30 +813,35 @@ export async function executeSync( return { ...result, error: 'connector_unavailable' } } - const connector = connectorRows[0] + const connectorBeforeLock = connectorRows[0] - if (options.requireRunnable && !isConnectorRunnableStatus(connector.status)) { + if (options.requireRunnable && !isConnectorRunnableStatus(connectorBeforeLock.status)) { logger.info('Skipping automatic sync: connector is not runnable', { connectorId, - status: connector.status, + status: connectorBeforeLock.status, }) return result } - const connectorConfig = CONNECTOR_REGISTRY[connector.connectorType] + const connectorConfig = CONNECTOR_REGISTRY[connectorBeforeLock.connectorType] if (!connectorConfig) { - throw new Error(`Unknown connector type: ${connector.connectorType}`) + throw new Error(`Unknown connector type: ${connectorBeforeLock.connectorType}`) } const kbRows = await db .select({ userId: knowledgeBase.userId, workspaceId: knowledgeBase.workspaceId }) .from(knowledgeBase) - .where(and(eq(knowledgeBase.id, connector.knowledgeBaseId), isNull(knowledgeBase.deletedAt))) + .where( + and( + eq(knowledgeBase.id, connectorBeforeLock.knowledgeBaseId), + isNull(knowledgeBase.deletedAt) + ) + ) .limit(1) if (kbRows.length === 0) { logger.warn( - `Skipping sync: knowledge base ${connector.knowledgeBaseId} is deleted (connector ${connectorId})` + `Skipping sync: knowledge base ${connectorBeforeLock.knowledgeBaseId} is deleted (connector ${connectorId})` ) await db .update(knowledgeConnector) @@ -856,7 +861,7 @@ export async function executeSync( const kbOwner: KnowledgeBaseOwner = { workspaceId: kbRows[0].workspaceId, userId } if (!kbOwner.workspaceId) { throw new Error( - `Knowledge base ${connector.knowledgeBaseId} is missing workspace billing context` + `Knowledge base ${connectorBeforeLock.knowledgeBaseId} is missing workspace billing context` ) } if (billingAttribution.workspaceId !== kbOwner.workspaceId) { @@ -864,8 +869,6 @@ export async function executeSync( `Connector sync billing attribution does not match knowledge base workspace ${kbOwner.workspaceId}` ) } - const sourceConfig = connector.sourceConfig as Record - const lockResult = await db .update(knowledgeConnector) .set({ status: 'syncing', updatedAt: new Date() }) @@ -879,7 +882,7 @@ export async function executeSync( isNull(knowledgeConnector.deletedAt) ) ) - .returning({ id: knowledgeConnector.id }) + .returning() if (lockResult.length === 0) { logger.info( @@ -891,6 +894,14 @@ export async function executeSync( return result } + /** + * The row returned by the lock is the authoritative sync snapshot. A source update + * committed before the lock is included here; one attempted after it sees `syncing` + * and conflicts instead of letting this worker process stale configuration. + */ + const connector = lockResult[0] + const sourceConfig = connector.sourceConfig as Record + const syncLogId = generateId() const syncStartedAt = new Date() await db.insert(knowledgeConnectorSyncLog).values({ diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index 1b33f3458d0..eec18765749 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -457,6 +457,53 @@ describe('performUpdateKnowledgeConnector', () => { expect(dbChainMockFns.update).not.toHaveBeenCalled() }) + it('rejects a schedule change while synchronization is in progress', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'syncing' }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ + success: false, + error: 'Cannot change connector sync interval while synchronization is in progress', + errorCode: 'conflict', + }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + }) + + it('rejects a schedule change that races with synchronization startup', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', nextSyncAt: null, status: 'active' }, + ]) + .mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', nextSyncAt: null, status: 'syncing' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ + success: false, + error: 'Cannot change connector sync interval while synchronization is in progress', + errorCode: 'conflict', + }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + it('fails before persisting when sync billing attribution cannot be resolved', async () => { dbChainMockFns.limit.mockResolvedValueOnce([ { id: 'conn-1', connectorType: 'notion', status: 'active' }, diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index a995997fd6a..d541df8f45b 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -437,14 +437,14 @@ export async function performUpdateKnowledgeConnector( return fail('Connector not found', 'not_found') } - if (updates.sourceConfig !== undefined && existing.status === 'syncing') { - return fail( - 'Cannot update source configuration while connector synchronization is in progress', - 'conflict' - ) - } - if (updates.status !== undefined && existing.status === 'syncing') { - return fail('Cannot change connector status while synchronization is in progress', 'conflict') + const syncingUpdateError = + updates.sourceConfig !== undefined + ? 'Cannot update source configuration while connector synchronization is in progress' + : updates.status !== undefined + ? 'Cannot change connector status while synchronization is in progress' + : 'Cannot change connector sync interval while synchronization is in progress' + if (existing.status === 'syncing') { + return fail(syncingUpdateError, 'conflict') } if (updates.syncIntervalMinutes !== undefined) { @@ -523,9 +523,7 @@ export async function performUpdateKnowledgeConnector( isNull(knowledgeConnector.archivedAt), isNull(knowledgeConnector.deletedAt), ] - if (updates.sourceConfig !== undefined || updates.status !== undefined) { - updateConditions.push(eq(knowledgeConnector.status, existing.status)) - } + updateConditions.push(eq(knowledgeConnector.status, existing.status)) if (values.nextSyncAt !== undefined) { updateConditions.push( existing.nextSyncAt @@ -542,17 +540,8 @@ export async function performUpdateKnowledgeConnector( if (!row) { const current = await getKnowledgeConnector(kb.id, connectorId) - if (current?.status === 'syncing' && updates.sourceConfig !== undefined) { - return fail( - 'Cannot update source configuration while connector synchronization is in progress', - 'conflict' - ) - } - if (current?.status === 'syncing' && updates.status !== undefined) { - return fail( - 'Cannot change connector status while synchronization is in progress', - 'conflict' - ) + if (current?.status === 'syncing') { + return fail(syncingUpdateError, 'conflict') } if (current) { return fail('Connector changed during the update; retry the request', 'conflict')