diff --git a/packages/core/flow-engine/src/__tests__/flowContext.test.ts b/packages/core/flow-engine/src/__tests__/flowContext.test.ts index c3f278690bf..dec42ad710c 100644 --- a/packages/core/flow-engine/src/__tests__/flowContext.test.ts +++ b/packages/core/flow-engine/src/__tests__/flowContext.test.ts @@ -14,6 +14,8 @@ import { FlowEngine } from '../flowEngine'; import { FlowModel } from '../models/flowModel'; import { RunJSContextRegistry } from '../runjs-context/registry'; import { setupRunJSContexts } from '../runjs-context/setup'; +import { createViewScopedEngine } from '../ViewScopedFlowEngine'; +import { DATA_SOURCE_DIRTY_EVENT } from '../views/viewEvents'; describe('FlowContext properties and methods', () => { it('should return static property value', () => { @@ -1429,6 +1431,92 @@ describe('FlowEngine context', () => { expect(engine.context.appName).toBe('NocoBase'); }); + it('ctx.api should return a dirty-aware wrapper for static api properties', async () => { + const engine = new FlowEngine(); + const update = vi.fn(async () => ({ data: { data: { id: 1 } } })); + const api = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ update })), + }; + engine.context.defineProperty('api', { value: api }); + + await engine.context.api.resource('posts').update({ filterByTk: 1, values: { title: 't' } }); + + expect(update).toHaveBeenCalledTimes(1); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(engine.context.api).toBe(engine.context.api); + }); + + it('ctx.api should stay dirty-aware when resolved from a scoped context delegate', async () => { + const root = new FlowEngine(); + const scoped = createViewScopedEngine(root); + const update = vi.fn(async () => ({ data: { data: { id: 1 } } })); + root.context.defineProperty('api', { + value: { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ update })), + }, + }); + + await scoped.context.api.resource('posts').update({ filterByTk: 1, values: { title: 't' } }); + + expect(update).toHaveBeenCalledTimes(1); + expect(root.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(scoped.context.engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + }); + + it('ctx.request should use the dirty-aware api wrapper', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { ok: true } })); + engine.context.defineProperty('api', { + value: { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }, + }); + + await engine.context.request({ + resource: 'posts', + action: 'update', + headers: { 'X-Data-Source': 'analytics' }, + params: { filterByTk: 1 }, + } as any); + + expect(request).toHaveBeenCalledTimes(1); + expect(engine.getDataSourceDirtyVersion('analytics', 'posts')).toBe(1); + }); + + it('ctx.request should use the caller context when resolved through a scoped delegate', async () => { + const root = new FlowEngine(); + const scoped = createViewScopedEngine(root); + const callerCtx = new FlowContext(); + const dirtyEvents: Array<{ dataSourceKey: string; resourceNames: string[] }> = []; + const request = vi.fn(async () => ({ data: { ok: true } })); + root.context.defineProperty('api', { + value: { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }, + }); + callerCtx.addDelegate(scoped.context); + scoped.context.engine.emitter.on(DATA_SOURCE_DIRTY_EVENT, (event) => dirtyEvents.push(event)); + + await callerCtx.request({ + resource: 'posts', + action: 'update', + params: { filterByTk: 1 }, + } as any); + + expect(request).toHaveBeenCalledTimes(1); + expect(root.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(scoped.context.engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(dirtyEvents).toEqual([{ dataSourceKey: 'main', resourceNames: ['posts'] }]); + }); + it('ctx.sql should resolve template variables from caller context in delegate chain', async () => { const engine = new FlowEngine(); const request = vi.fn(async () => ({ data: { data: [] } })); diff --git a/packages/core/flow-engine/src/__tests__/flowEngine.dataSourceDirty.test.ts b/packages/core/flow-engine/src/__tests__/flowEngine.dataSourceDirty.test.ts index db0dda90a1a..b574cbae441 100644 --- a/packages/core/flow-engine/src/__tests__/flowEngine.dataSourceDirty.test.ts +++ b/packages/core/flow-engine/src/__tests__/flowEngine.dataSourceDirty.test.ts @@ -11,6 +11,7 @@ import { describe, expect, it, vi } from 'vitest'; import { FlowEngine } from '../flowEngine'; import { MultiRecordResource } from '../resources/multiRecordResource'; import { SingleRecordResource } from '../resources/singleRecordResource'; +import { DATA_SOURCE_DIRTY_EVENT } from '../views/viewEvents'; describe('FlowEngine dataSource dirty registry', () => { it('tracks versions per dataSourceKey + resourceName', () => { @@ -60,4 +61,54 @@ describe('FlowEngine dataSource dirty registry', () => { // plus root collection (safety) expect(markSpy).toHaveBeenCalledWith('main', 'users'); }); + + it('marks dirty once for record write helpers when using the dirty-aware context api', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { data: { id: 1 }, meta: {} } })); + const dirtyEvents: Array<{ dataSourceKey: string; resourceNames: string[] }> = []; + engine.context.defineProperty('api', { + value: { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }, + }); + engine.emitter.on(DATA_SOURCE_DIRTY_EVENT, (event) => dirtyEvents.push(event)); + + const multi = engine.createResource(MultiRecordResource); + multi.setDataSourceKey('main').setResourceName('posts'); + await multi.create({ title: 't' } as any, { refresh: false }); + + expect(request).toHaveBeenCalledTimes(1); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(dirtyEvents).toEqual([{ dataSourceKey: 'main', resourceNames: ['posts'] }]); + + const single = engine.createResource(SingleRecordResource); + single.setDataSourceKey('main').setResourceName('posts').setFilterByTk(1); + await single.save({ title: 'u' } as any, { refresh: false }); + + expect(request).toHaveBeenCalledTimes(2); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(2); + expect(dirtyEvents).toEqual([ + { dataSourceKey: 'main', resourceNames: ['posts'] }, + { dataSourceKey: 'main', resourceNames: ['posts'] }, + ]); + }); + + it('still marks dirty for direct runAction writes', async () => { + const engine = new FlowEngine(); + engine.context.defineProperty('api', { + value: { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { data: { id: 1 }, meta: {} } })), + resource: vi.fn(), + }, + }); + + const multi = engine.createResource(MultiRecordResource); + multi.setDataSourceKey('main').setResourceName('posts'); + await multi.runAction('create', { data: { title: 't' } }); + + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + }); }); diff --git a/packages/core/flow-engine/src/flowContext.ts b/packages/core/flow-engine/src/flowContext.ts index b56639e6be6..73e1fe65a54 100644 --- a/packages/core/flow-engine/src/flowContext.ts +++ b/packages/core/flow-engine/src/flowContext.ts @@ -9,7 +9,7 @@ import { ISchema } from '@formily/json-schema'; import { observable } from '@formily/reactive'; -import { APIClient, RequestOptions } from '@nocobase/sdk'; +import type { APIClient, RequestOptions } from '@nocobase/sdk'; import type { Router } from '@remix-run/router'; import axios from 'axios'; import { MessageInstance } from 'antd/es/message/interface'; @@ -53,6 +53,7 @@ import { FlowExitAllException } from './utils/exceptions'; import { enqueueVariablesResolve, JSONValue } from './utils/params-resolvers'; import type { RecordRef } from './utils/serverContextParams'; import { buildServerContextParams as _buildServerContextParams } from './utils/serverContextParams'; +import { getDirtyAwareApiClient } from './utils/dirtyAwareApiClient'; import { inferRecordRef } from './utils/variablesParams'; import { FlowView, FlowViewer } from './views/FlowView'; import { RunJSContextRegistry, getModelClassName, type RunJSVersion } from './runjs-context/registry'; @@ -2911,19 +2912,20 @@ export class FlowContext { // 静态值 if ('value' in options) { - return options.value; + return key === 'api' ? getDirtyAwareApiClient(options.value, currentContext) : options.value; } // get 方法 if (options.get) { if (options.cache === false) { - return options.get(currentContext); + const value = options.get(currentContext); + return key === 'api' ? getDirtyAwareApiClient(value, currentContext) : value; } const cacheKey = options.observable ? '_observableCache' : '_cache'; if (key in this[cacheKey]) { - return this[cacheKey][key]; + return key === 'api' ? getDirtyAwareApiClient(this[cacheKey][key], currentContext) : this[cacheKey][key]; } if (this._pending[key]) return this._pending[key]; @@ -2941,7 +2943,7 @@ export class FlowContext { (v) => { this[cacheKey][key] = v; delete this._pending[key]; - return v; + return key === 'api' ? getDirtyAwareApiClient(v, currentContext) : v; }, (err) => { delete this._pending[key]; @@ -2953,7 +2955,7 @@ export class FlowContext { // sync 直接缓存 this[cacheKey][key] = result; - return result; + return key === 'api' ? getDirtyAwareApiClient(result, currentContext) : result; } return undefined; @@ -3076,7 +3078,7 @@ class BaseFlowEngineContext extends FlowContext { this.defineMethod('getModel', (modelName: string, searchInPreviousEngines?: boolean) => { return this.engine.getModel(modelName, searchInPreviousEngines); }); - this.defineMethod('request', (options: RequestOptions) => { + this.defineMethod('request', function (this: FlowContext, options: RequestOptions) { const app = this.app as { getApiUrl?: (pathname?: string) => string } | undefined; if (typeof options?.url === 'string' && shouldBypassApiClient(options.url, app)) { return axios.request(options); diff --git a/packages/core/flow-engine/src/resources/apiResource.ts b/packages/core/flow-engine/src/resources/apiResource.ts index ed908af53f4..5f1bdf5c061 100644 --- a/packages/core/flow-engine/src/resources/apiResource.ts +++ b/packages/core/flow-engine/src/resources/apiResource.ts @@ -9,6 +9,7 @@ import { APIClient } from '@nocobase/sdk'; import { FlowContext } from '../flowContext'; +import { getDirtyAwareApiClient } from '../utils/dirtyAwareApiClient'; import { FlowResource, ResourceError } from './flowResource'; export class APIResource extends FlowResource { @@ -33,7 +34,7 @@ export class APIResource extends FlowResource { } setAPIClient(api: APIClient) { - this.api = api; + this.api = getDirtyAwareApiClient(api, this.context) as APIClient; return this; } diff --git a/packages/core/flow-engine/src/resources/baseRecordResource.ts b/packages/core/flow-engine/src/resources/baseRecordResource.ts index c80443759af..7bc73492fcc 100644 --- a/packages/core/flow-engine/src/resources/baseRecordResource.ts +++ b/packages/core/flow-engine/src/resources/baseRecordResource.ts @@ -12,7 +12,7 @@ import _ from 'lodash'; import { APIResource } from './apiResource'; import { FilterItem } from './filterItem'; import { ResourceError } from './flowResource'; -import { DATA_SOURCE_DIRTY_EVENT } from '../views/viewEvents'; +import { markDataSourceDirty } from '../utils/dataSourceDirty'; export abstract class BaseRecordResource extends APIResource { protected resourceName: string; @@ -142,28 +142,11 @@ export abstract class BaseRecordResource extends APIResource * Used to coordinate "refresh on active" across view stacks. */ protected markDataSourceDirty(resourceName?: string) { - const engine = this.context.engine; - if (!engine) return; - - const dataSourceKey = this.getDataSourceKey() || 'main'; - const resName = resourceName || this.getResourceName(); - if (!resName) return; - - const affectedResourceNames = new Set([String(resName)]); - // Optional safety: association resources like "users.profile" may impact parent collection views. - if (typeof resName === 'string' && resName.includes('.')) { - affectedResourceNames.add(resName.split('.')[0]); - } - - for (const name of affectedResourceNames) { - engine.markDataSourceDirty(dataSourceKey, name); - } - - // Signal current view to re-evaluate dirty blocks (e.g., same-view sibling refresh). - // This is emitted on the *current* engine emitter (view-scoped) so it won't affect other views. - engine.emitter?.emit?.(DATA_SOURCE_DIRTY_EVENT, { - dataSourceKey, - resourceNames: Array.from(affectedResourceNames), + markDataSourceDirty({ + engine: this.context.engine, + dataSourceKey: this.getDataSourceKey(), + resourceName: resourceName || this.getResourceName(), + includePreviousEngines: true, }); } diff --git a/packages/core/flow-engine/src/resources/multiRecordResource.ts b/packages/core/flow-engine/src/resources/multiRecordResource.ts index e65c298d9c3..d85412859b3 100644 --- a/packages/core/flow-engine/src/resources/multiRecordResource.ts +++ b/packages/core/flow-engine/src/resources/multiRecordResource.ts @@ -10,6 +10,7 @@ import { observable } from '@formily/reactive'; import { AxiosRequestConfig } from 'axios'; import _ from 'lodash'; +import { SKIP_DATA_SOURCE_DIRTY } from '../utils/dirtyAwareApiClient'; import { BaseRecordResource } from './baseRecordResource'; export class MultiRecordResource extends BaseRecordResource { @@ -113,7 +114,10 @@ export class MultiRecordResource extends BaseRecordResource { const config = this.mergeRequestConfig({ data }, this.createActionOptions, options); - const res = await this.runAction('create', config); + const res = await this.runAction('create', { + ...config, + [SKIP_DATA_SOURCE_DIRTY]: true, + }); this.markDataSourceDirty(); this.emit('saved', data); if (options?.refresh !== false) { @@ -146,7 +150,10 @@ export class MultiRecordResource extends BaseRecordResource extends BaseRecordResource extends BaseRecordResource { @@ -43,6 +44,7 @@ export class SingleRecordResource extends BaseRecordResource const res = await this.runAction(actionName, { ...config, data: result, + [SKIP_DATA_SOURCE_DIRTY]: true, }); // Mark as dirty before emitting/refreshing so other views can refresh when activated. this.markDataSourceDirty(); @@ -62,7 +64,10 @@ export class SingleRecordResource extends BaseRecordResource }, options, ); - await this.runAction('destroy', config); + await this.runAction('destroy', { + ...config, + [SKIP_DATA_SOURCE_DIRTY]: true, + }); this.markDataSourceDirty(); this.setData(null); } diff --git a/packages/core/flow-engine/src/utils/__tests__/dirtyAwareApiClient.test.ts b/packages/core/flow-engine/src/utils/__tests__/dirtyAwareApiClient.test.ts new file mode 100644 index 00000000000..deb32e9493e --- /dev/null +++ b/packages/core/flow-engine/src/utils/__tests__/dirtyAwareApiClient.test.ts @@ -0,0 +1,392 @@ +/** + * This file is part of the NocoBase (R) project. + * Copyright (c) 2020-2024 NocoBase Co., Ltd. + * Authors: NocoBase Team. + * + * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License. + * For more information, please refer to: https://www.nocobase.com/agreement. + */ + +import { describe, expect, it, vi } from 'vitest'; +import { FlowContext } from '../../flowContext'; +import { FlowEngine } from '../../flowEngine'; +import { createViewScopedEngine } from '../../ViewScopedFlowEngine'; +import { DATA_SOURCE_DIRTY_EVENT } from '../../views/viewEvents'; +import { getDirtyAwareApiClient, SKIP_DATA_SOURCE_DIRTY } from '../dirtyAwareApiClient'; + +type TestRequestOptions = { + url?: string; + resource?: string; + action?: string; + headers?: Record; + params?: unknown; +}; + +type TestResource = Record Promise>; + +type TestApi = { + auth: { locale: string }; + request: (config: TestRequestOptions) => Promise; + resource: (name: string, of?: unknown, headers?: Record, cancel?: boolean) => TestResource; +}; + +function getWrappedApi(engine: FlowEngine, api: TestApi): TestApi { + return getDirtyAwareApiClient(api, engine.context) as TestApi; +} + +describe('dirtyAwareApiClient', () => { + it('should return non-api values as-is', () => { + const context = new FlowContext(); + const value = { request: vi.fn() }; + + expect(getDirtyAwareApiClient(value, context)).toBe(value); + }); + + it('should mark the resource dirty after mutating resource actions succeed', async () => { + const engine = new FlowEngine(); + const list = vi.fn(async () => ({ data: { data: [] } })); + const update = vi.fn(async () => ({ data: { data: { id: 1 } } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ list, update })), + }; + const wrappedApi = getWrappedApi(engine, api); + + await wrappedApi.resource('posts').list(); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(0); + + await wrappedApi.resource('posts').update({ filterByTk: 1, values: { title: 't' } }); + + expect(update).toHaveBeenCalledTimes(1); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(getDirtyAwareApiClient(api, engine.context)).toBe(wrappedApi); + }); + + it('should not double-wrap an already dirty-aware api', async () => { + const engine = new FlowEngine(); + const update = vi.fn(async () => ({ data: { data: { id: 1 } } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ update })), + }; + + const wrappedApi = getWrappedApi(engine, api); + const wrappedAgain = getDirtyAwareApiClient(wrappedApi, engine.context) as TestApi; + await wrappedAgain.resource('posts').update({ filterByTk: 1 }); + + expect(wrappedAgain).toBe(wrappedApi); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + }); + + it('should not mark dirty for read actions', async () => { + const nonMutatingActions = [ + 'get', + 'getSystemSettings', + 'list', + 'listByUser', + 'query', + 'count', + 'check', + 'preview', + 'test', + 'find', + 'exists', + 'aggregate', + 'listMine', + 'parents', + 'children', + 'search', + 'send', + 'testConnection', + 'refresh', + 'run', + 'runById', + 'unknownCustomAction', + ]; + + for (const actionName of nonMutatingActions) { + const engine = new FlowEngine(); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ + [actionName]: vi.fn(async () => ({ data: { data: [] } })), + })), + }; + + await getWrappedApi(engine, api).resource('posts')[actionName](); + + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(0); + } + }); + + it('should mark dirty for known mutating action variants', async () => { + const mutatingActions = [ + 'create', + 'execute', + 'updateOrCreate', + 'firstOrCreate', + 'setFields', + 'updateProfile', + 'saveAsTemplate', + 'remove/abc', + ]; + + for (const actionName of mutatingActions) { + const engine = new FlowEngine(); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ + [actionName]: vi.fn(async () => ({ data: { ok: true } })), + })), + }; + + await getWrappedApi(engine, api).resource('posts')[actionName](); + + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + } + }); + + it('should not mark dirty when a mutating action fails', async () => { + const engine = new FlowEngine(); + const update = vi.fn(async () => { + throw new Error('update failed'); + }); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ update })), + }; + + await expect(getWrappedApi(engine, api).resource('posts').update({ filterByTk: 1 })).rejects.toThrow( + 'update failed', + ); + + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(0); + }); + + it('should mark association resource and parent collection dirty', async () => { + const engine = new FlowEngine(); + const dirtyEvents: Array<{ dataSourceKey: string; resourceNames: string[] }> = []; + engine.emitter.on(DATA_SOURCE_DIRTY_EVENT, (event) => dirtyEvents.push(event)); + const add = vi.fn(async () => ({ data: { data: null } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ add })), + }; + + await getWrappedApi(engine, api) + .resource('users.roles', 1, { 'x-data-source': 'external' }) + .add({ values: [1, 2] }); + + expect(engine.getDataSourceDirtyVersion('external', 'users.roles')).toBe(1); + expect(engine.getDataSourceDirtyVersion('external', 'users')).toBe(1); + expect(dirtyEvents).toEqual([{ dataSourceKey: 'external', resourceNames: ['users.roles', 'users'] }]); + }); + + it('should mark resource-action request mutations dirty after success', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { ok: true } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }; + + await getWrappedApi(engine, api).request({ + resource: 'posts', + action: 'update', + headers: { 'X-Data-Source': 'analytics' }, + params: { filterByTk: 1 }, + }); + + expect(request).toHaveBeenCalledTimes(1); + expect(engine.getDataSourceDirtyVersion('analytics', 'posts')).toBe(1); + }); + + it('should mark URL-form resource mutations dirty after success', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { ok: true } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }; + const wrappedApi = getWrappedApi(engine, api); + + await wrappedApi.request({ url: 'posts:update' }); + await wrappedApi.request({ + url: '/api/posts:update?filterByTk=1', + headers: { 'x-data-source': 'external' }, + }); + await wrappedApi.request({ + url: '/api/posts/1/tags:set', + headers: { 'X-Data-Source': 'analytics' }, + }); + + expect(request).toHaveBeenCalledTimes(3); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(engine.getDataSourceDirtyVersion('external', 'posts')).toBe(1); + expect(engine.getDataSourceDirtyVersion('analytics', 'posts.tags')).toBe(1); + expect(engine.getDataSourceDirtyVersion('analytics', 'posts')).toBe(1); + }); + + it('should resolve data source resource URLs to the nested data source target', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { ok: true } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }; + + await getWrappedApi(engine, api).request({ + url: 'dataSources/external/collections:update', + }); + + expect(request).toHaveBeenCalledTimes(1); + expect(engine.getDataSourceDirtyVersion('external', 'collections')).toBe(1); + expect(engine.getDataSourceDirtyVersion('main', 'dataSources.collections')).toBe(0); + expect(engine.getDataSourceDirtyVersion('main', 'dataSources')).toBe(0); + }); + + it('should resolve dataSources resourceOf requests to the nested data source target', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { ok: true } })); + const update = vi.fn(async () => ({ data: { ok: true } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(() => ({ update })), + }; + const wrappedApi = getWrappedApi(engine, api); + + await wrappedApi.request({ + resource: 'dataSources.collections', + resourceOf: 'external', + action: 'update', + } as TestRequestOptions & { resourceOf: string }); + await wrappedApi.resource('dataSources.roles', 'external').update({ values: { allow: true } }); + await wrappedApi.resource('dataSources/external/roles').update({ values: { allow: false } }); + + expect(engine.getDataSourceDirtyVersion('external', 'collections')).toBe(1); + expect(engine.getDataSourceDirtyVersion('external', 'roles')).toBe(2); + expect(engine.getDataSourceDirtyVersion('main', 'dataSources.collections')).toBe(0); + expect(engine.getDataSourceDirtyVersion('main', 'dataSources.roles')).toBe(0); + }); + + it('should skip dirty marking and strip the internal skip flag from raw requests', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { ok: true } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }; + + await getWrappedApi(engine, api).request({ + url: 'posts:update', + [SKIP_DATA_SOURCE_DIRTY]: true, + } as TestRequestOptions & { [SKIP_DATA_SOURCE_DIRTY]: boolean }); + + expect(request).toHaveBeenCalledTimes(1); + expect(request.mock.calls[0][0]).not.toHaveProperty(SKIP_DATA_SOURCE_DIRTY); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(0); + }); + + it('should strip configured API base from URL-form mutations', async () => { + const engine = new FlowEngine(); + const request = vi.fn(async () => ({ data: { ok: true } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }; + engine.context.defineProperty('app', { + value: { + getApiUrl(pathname = '') { + return `https://app.example.com/foo/api/${pathname.replace(/^\//, '')}`; + }, + }, + }); + const wrappedApi = getWrappedApi(engine, api); + + await wrappedApi.request({ url: '/foo/api/posts:update' }); + await wrappedApi.request({ + url: 'https://app.example.com/foo/api/users/1/roles:set', + headers: { 'x-data-source': 'external' }, + }); + + expect(request).toHaveBeenCalledTimes(2); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(engine.getDataSourceDirtyVersion('main', 'foo.api.posts')).toBe(0); + expect(engine.getDataSourceDirtyVersion('external', 'users.roles')).toBe(1); + expect(engine.getDataSourceDirtyVersion('external', 'users')).toBe(1); + }); + + it('should not mark URL-form resource dirty for reads, failed mutations, or external URLs', async () => { + const engine = new FlowEngine(); + const request = vi + .fn() + .mockResolvedValueOnce({ data: { data: [] } }) + .mockResolvedValueOnce({ data: { data: [] } }) + .mockRejectedValueOnce(new Error('request failed')) + .mockResolvedValueOnce({ data: { ok: true } }) + .mockResolvedValueOnce({ data: { ok: true } }); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request, + resource: vi.fn(), + }; + const wrappedApi = getWrappedApi(engine, api); + + await wrappedApi.request({ url: 'posts:list' }); + await wrappedApi.request({ url: '/api/posts:parents' }); + await expect(wrappedApi.request({ url: '/api/posts:update' })).rejects.toThrow('request failed'); + await wrappedApi.request({ url: 'https://example.com/api/posts:update' }); + await wrappedApi.request({ url: '//example.com/api/posts:update' }); + + expect(request).toHaveBeenCalledTimes(5); + expect(engine.getDataSourceDirtyVersion('main', 'posts')).toBe(0); + }); + + it('should mark opener engine dirty when called from a scoped view context', async () => { + const root = new FlowEngine(); + const scoped = createViewScopedEngine(root); + const update = vi.fn(async () => ({ data: { data: { id: 1 } } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ update })), + }; + + await getWrappedApi(scoped, api) + .resource('posts') + .update({ filterByTk: 1, values: { title: 't' } }); + + expect(root.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(scoped.context.engine.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + }); + + it('should not double-mark when the context exposes a scoped engine proxy', async () => { + const root = new FlowEngine(); + const scoped = createViewScopedEngine(root); + const context = new FlowContext(); + const update = vi.fn(async () => ({ data: { data: { id: 1 } } })); + const api: TestApi = { + auth: { locale: 'zh-CN' }, + request: vi.fn(async () => ({ data: { ok: true } })), + resource: vi.fn(() => ({ update })), + }; + context.defineProperty('engine', { value: scoped }); + + await (getDirtyAwareApiClient(api, context) as TestApi).resource('posts').update({ filterByTk: 1 }); + + expect(root.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + expect(scoped.getDataSourceDirtyVersion('main', 'posts')).toBe(1); + }); +}); diff --git a/packages/core/flow-engine/src/utils/dataSourceDirty.ts b/packages/core/flow-engine/src/utils/dataSourceDirty.ts new file mode 100644 index 00000000000..cae3d516485 --- /dev/null +++ b/packages/core/flow-engine/src/utils/dataSourceDirty.ts @@ -0,0 +1,126 @@ +/** + * This file is part of the NocoBase (R) project. + * Copyright (c) 2020-2024 NocoBase Co., Ltd. + * Authors: NocoBase Team. + * + * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License. + * For more information, please refer to: https://www.nocobase.com/agreement. + */ + +import type { FlowEngine } from '../flowEngine'; +import { DATA_SOURCE_DIRTY_EVENT } from '../views/viewEvents'; + +type MarkDataSourceDirtyOptions = { + engine?: FlowEngine; + dataSourceKey?: unknown; + resourceName?: unknown; + includePreviousEngines?: boolean; +}; + +export function getHeaderValue(headers: unknown, name: string): unknown { + if (!headers || typeof headers !== 'object') { + return undefined; + } + + const maybeHeaders = headers as { get?: (key: string) => unknown }; + if (typeof maybeHeaders.get === 'function') { + const value = maybeHeaders.get(name); + if (value != null && value !== '') { + return value; + } + } + + const lowerName = name.toLowerCase(); + for (const [key, value] of Object.entries(headers as Record)) { + if (key.toLowerCase() === lowerName) { + return value; + } + } + + return undefined; +} + +export function getDataSourceKeyFromHeaders(headers: unknown): string { + const value = getHeaderValue(headers, 'x-data-source'); + if (Array.isArray(value)) { + return String(value[0] || 'main'); + } + return value == null || value === '' ? 'main' : String(value); +} + +export function getAffectedResourceNames(resourceName: unknown): string[] { + const name = String(resourceName || '').trim(); + if (!name) { + return []; + } + + const names = new Set([name]); + if (name.includes('.')) { + names.add(name.split('.')[0]); + } + return Array.from(names); +} + +function getDirtyTargetEngines(engine: FlowEngine, includePreviousEngines?: boolean): FlowEngine[] { + if (!includePreviousEngines) { + return [engine]; + } + + const engines: FlowEngine[] = []; + const seen = new Set(); + let current: FlowEngine | undefined = engine; + let guard = 0; + + while (current && guard++ < 50) { + if (!seen.has(current)) { + engines.push(current); + seen.add(current); + } + current = current.previousEngine; + } + + return engines; +} + +export function markDataSourceDirty(options: MarkDataSourceDirtyOptions): string[] { + const { engine, resourceName, includePreviousEngines } = options; + if (!engine?.markDataSourceDirty) { + return []; + } + + const resourceNames = getAffectedResourceNames(resourceName); + if (!resourceNames.length) { + return []; + } + + const dataSourceKey = String(options.dataSourceKey || 'main'); + const targetEngines = getDirtyTargetEngines(engine, includePreviousEngines); + const beforeVersions = new Map>(); + + for (const targetEngine of targetEngines) { + const versions = new Map(); + beforeVersions.set(targetEngine, versions); + for (const name of resourceNames) { + versions.set(name, targetEngine.getDataSourceDirtyVersion?.(dataSourceKey, name) || 0); + } + } + + for (const targetEngine of targetEngines) { + const versions = beforeVersions.get(targetEngine); + for (const name of resourceNames) { + const before = versions?.get(name) || 0; + const current = targetEngine.getDataSourceDirtyVersion?.(dataSourceKey, name) || 0; + if (current !== before) { + continue; + } + targetEngine.markDataSourceDirty(dataSourceKey, name); + } + } + + engine.emitter?.emit?.(DATA_SOURCE_DIRTY_EVENT, { + dataSourceKey, + resourceNames, + }); + + return resourceNames; +} diff --git a/packages/core/flow-engine/src/utils/dirtyAwareApiClient.ts b/packages/core/flow-engine/src/utils/dirtyAwareApiClient.ts new file mode 100644 index 00000000000..87e1c72ccd4 --- /dev/null +++ b/packages/core/flow-engine/src/utils/dirtyAwareApiClient.ts @@ -0,0 +1,430 @@ +/** + * This file is part of the NocoBase (R) project. + * Copyright (c) 2020-2024 NocoBase Co., Ltd. + * Authors: NocoBase Team. + * + * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License. + * For more information, please refer to: https://www.nocobase.com/agreement. + */ + +import type { ActionParams, APIClient, IResource, RequestOptions } from '@nocobase/sdk'; +import type { FlowContext } from '../flowContext'; +import { getDataSourceKeyFromHeaders, markDataSourceDirty } from './dataSourceDirty'; + +type ResourceActionFn = (params?: ActionParams, opts?: unknown) => Promise; + +export const SKIP_DATA_SOURCE_DIRTY = '__nocobaseSkipDataSourceDirty'; + +type ResourceRequestOptions = RequestOptions & { + resource?: unknown; + resourceOf?: unknown; + action?: unknown; + headers?: unknown; + [SKIP_DATA_SOURCE_DIRTY]?: boolean; +}; + +type DirtyResourceAction = { + dataSourceKey?: string; + resourceName: string; + actionName: string; +}; + +type ApiUrlProvider = { + getApiUrl?: (pathname?: string) => string; +}; + +type DirtyAwareAPIClient = APIClient & { + resource: APIClient['resource']; + request: APIClient['request']; +}; + +type APIClientRequestConfig = Parameters[0]; + +const dirtyAwareApiClientCache = new WeakMap>(); +const dirtyAwareApiClientProxies = new WeakSet(); + +const MUTATING_RESOURCE_ACTIONS = [ + 'add', + 'bulkdestroy', + 'bulkupdate', + 'create', + 'delete', + 'destroy', + 'execute', + 'firstorcreate', + 'import', + 'move', + 'remove', + 'save', + 'saveastemplate', + 'set', + 'setfields', + 'submit', + 'update', + 'updateorcreate', + 'upsert', +]; + +function isApiClientLike(value: unknown): value is DirtyAwareAPIClient { + if (!value || typeof value !== 'object') { + return false; + } + const candidate = value as { resource?: unknown; request?: unknown }; + return typeof candidate.resource === 'function' && typeof candidate.request === 'function'; +} + +function isMutatingResourceAction(actionName: string): boolean { + const normalized = String(actionName || '').trim(); + if (!normalized) { + return false; + } + const baseActionName = normalized.split('/')[0]; + const lowerBaseActionName = baseActionName.toLowerCase(); + return MUTATING_RESOURCE_ACTIONS.some((action) => { + if (lowerBaseActionName === action) { + return true; + } + if (!lowerBaseActionName.startsWith(action) || baseActionName.length <= action.length) { + return false; + } + + const nextChar = baseActionName[action.length]; + return nextChar === '-' || nextChar === '_' || (nextChar >= 'A' && nextChar <= 'Z'); + }); +} + +function getCurrentOrigin(): string | undefined { + return typeof window === 'undefined' ? undefined : window.location?.origin; +} + +function parseUrl(value: string, base?: string): URL | undefined { + try { + return new URL(value, base); + } catch { + return undefined; + } +} + +function stripSearchAndHash(path: string): string { + const index = path.search(/[?#]/); + return index === -1 ? path : path.slice(0, index); +} + +function stripKnownApiPrefix(path: string): string | undefined { + const cleanPath = stripSearchAndHash(path).trim(); + if (!cleanPath) { + return undefined; + } + + const normalizedPath = cleanPath.replace(/^\/+/, ''); + if (!normalizedPath || normalizedPath === 'api') { + return undefined; + } + if (normalizedPath.startsWith('api/')) { + return normalizedPath.slice('api/'.length); + } + + return normalizedPath; +} + +function normalizePathname(pathname: string) { + return pathname.endsWith('/') ? pathname : `${pathname}/`; +} + +function getAppApiUrl(app?: ApiUrlProvider): URL | undefined { + if (!app?.getApiUrl) { + return undefined; + } + + try { + return parseUrl(app.getApiUrl(), getCurrentOrigin()); + } catch { + return undefined; + } +} + +function stripConfiguredApiPrefix(path: string, apiPathname: string): string | undefined { + const cleanPath = stripSearchAndHash(path).trim(); + if (!cleanPath.startsWith('/')) { + return undefined; + } + + const apiPath = normalizePathname(apiPathname); + const requestPath = normalizePathname(cleanPath); + if (!requestPath.startsWith(apiPath)) { + return undefined; + } + + const apiPathWithoutTrailingSlash = apiPath.replace(/\/$/, ''); + return cleanPath.slice(apiPathWithoutTrailingSlash.length).replace(/^\/+/, '') || undefined; +} + +function getDirtyResourcePathFromAbsoluteUrl(url: URL, app?: ApiUrlProvider): string | undefined { + if (!['http:', 'https:'].includes(url.protocol)) { + return undefined; + } + + if (app?.getApiUrl) { + const apiUrl = getAppApiUrl(app); + if (!apiUrl || url.origin !== apiUrl.origin) { + return undefined; + } + + return stripConfiguredApiPrefix(url.pathname, apiUrl.pathname); + } + + const currentOrigin = getCurrentOrigin(); + if (!currentOrigin || url.origin !== currentOrigin) { + return undefined; + } + + return stripKnownApiPrefix(url.pathname); +} + +function getDirtyResourcePathFromUrl(url: unknown, context: FlowContext): string | undefined { + if (typeof url !== 'string') { + return undefined; + } + + const trimmedUrl = url.trim(); + if (!trimmedUrl || trimmedUrl.startsWith('//')) { + return undefined; + } + + if (/^https?:\/\//i.test(trimmedUrl)) { + const parsedUrl = parseUrl(trimmedUrl); + if (!parsedUrl) { + return undefined; + } + return getDirtyResourcePathFromAbsoluteUrl(parsedUrl, context.app as ApiUrlProvider | undefined); + } + + const appApiUrl = getAppApiUrl(context.app as ApiUrlProvider | undefined); + const configuredResourcePath = appApiUrl ? stripConfiguredApiPrefix(trimmedUrl, appApiUrl.pathname) : undefined; + if (configuredResourcePath) { + return configuredResourcePath; + } + + return stripKnownApiPrefix(trimmedUrl); +} + +function decodeResourcePathSegment(segment: string): string { + try { + return decodeURIComponent(segment); + } catch { + return segment; + } +} + +function getDataSourceKeyFromResourceOf(resourceOf: unknown): string | undefined { + const dataSourceKey = String(resourceOf ?? '').trim(); + return dataSourceKey || undefined; +} + +function parseResourceActionFromSegments(segments: string[]): DirtyResourceAction | undefined { + const resourceSegments: string[] = []; + let actionName: string | undefined; + let actionSegmentIndex = -1; + + for (let index = 0; index < segments.length; index += 2) { + const segment = segments[index]; + const actionDelimiterIndex = segment.lastIndexOf(':'); + const resourceSegment = actionDelimiterIndex === -1 ? segment : segment.slice(0, actionDelimiterIndex); + if (!resourceSegment) { + return undefined; + } + + resourceSegments.push(decodeResourcePathSegment(resourceSegment)); + if (actionDelimiterIndex !== -1) { + actionName = decodeResourcePathSegment(segment.slice(actionDelimiterIndex + 1)).trim(); + actionSegmentIndex = index; + break; + } + } + + if (!actionName || !resourceSegments.length) { + return undefined; + } + + if (segments.length > actionSegmentIndex + 2) { + return undefined; + } + + return { + resourceName: resourceSegments.join('.'), + actionName, + }; +} + +function parseDirtyResourceActionFromUrl(url: unknown, context: FlowContext): DirtyResourceAction | undefined { + const resourcePath = getDirtyResourcePathFromUrl(url, context); + if (!resourcePath) { + return undefined; + } + + const segments = stripSearchAndHash(resourcePath).split('/').filter(Boolean); + const firstSegment = decodeResourcePathSegment(segments[0] || ''); + if (firstSegment === 'dataSources' && segments.length >= 3) { + const dataSourceKey = getDataSourceKeyFromResourceOf(decodeResourcePathSegment(segments[1])); + const parsed = parseResourceActionFromSegments(segments.slice(2)); + if (dataSourceKey && parsed) { + return { + ...parsed, + dataSourceKey, + }; + } + } + + return parseResourceActionFromSegments(segments); +} + +function resolveDirtyResourceActionFromResource( + resourceName: string, + resourceOf: unknown, + actionName: string, + context: FlowContext, +): DirtyResourceAction | undefined { + const normalizedResourceName = resourceName.trim(); + const normalizedActionName = actionName.trim(); + if (!normalizedResourceName || !normalizedActionName) { + return undefined; + } + + if (normalizedResourceName.includes('/')) { + const parsed = parseDirtyResourceActionFromUrl(`${normalizedResourceName}:${normalizedActionName}`, context); + if (parsed) { + return parsed; + } + } + + const dataSourcesPrefix = 'dataSources.'; + if (normalizedResourceName.startsWith(dataSourcesPrefix)) { + const dataSourceKey = getDataSourceKeyFromResourceOf(resourceOf); + const nestedResourceName = normalizedResourceName.slice(dataSourcesPrefix.length).trim(); + if (dataSourceKey && nestedResourceName) { + return { + dataSourceKey, + resourceName: nestedResourceName, + actionName: normalizedActionName, + }; + } + } + + return { + resourceName: normalizedResourceName, + actionName: normalizedActionName, + }; +} + +function resolveDirtyResourceAction( + options: ResourceRequestOptions, + context: FlowContext, +): DirtyResourceAction | undefined { + const resourceName = typeof options?.resource === 'string' ? options.resource : undefined; + const actionName = typeof options?.action === 'string' ? options.action : undefined; + if (resourceName && actionName) { + return resolveDirtyResourceActionFromResource(resourceName, options.resourceOf, actionName, context); + } + + return parseDirtyResourceActionFromUrl(options?.url, context); +} + +function markResourceActionDataSourceDirty( + context: FlowContext, + dirtyResourceAction: DirtyResourceAction, + headers: unknown, +) { + markDataSourceDirty({ + engine: context.engine, + dataSourceKey: dirtyResourceAction.dataSourceKey || getDataSourceKeyFromHeaders(headers), + resourceName: dirtyResourceAction.resourceName, + includePreviousEngines: true, + }); +} + +function createDirtyAwareResource( + context: FlowContext, + resource: IResource, + resourceName: string, + resourceOf: unknown, + headers: unknown, +): IResource { + return new Proxy(resource, { + get(target, prop, receiver) { + const original = Reflect.get(target, prop, receiver); + if (typeof prop !== 'string' || typeof original !== 'function' || !isMutatingResourceAction(prop)) { + return original; + } + + const action = original as ResourceActionFn; + return async (...args: Parameters) => { + const result = await action(...args); + const dirtyResourceAction = resolveDirtyResourceActionFromResource(resourceName, resourceOf, prop, context); + if (dirtyResourceAction) { + markResourceActionDataSourceDirty(context, dirtyResourceAction, headers); + } + return result; + }; + }, + }); +} + +function createDirtyAwareApiClient(api: DirtyAwareAPIClient, context: FlowContext): APIClient { + const resource: APIClient['resource'] = (name, of, headers, cancel) => { + const targetResource = api.resource(name, of, headers, cancel); + return createDirtyAwareResource(context, targetResource, name, of, headers); + }; + + const request = ((config: APIClientRequestConfig): Promise => { + const options = config as ResourceRequestOptions; + const skipDataSourceDirty = options?.[SKIP_DATA_SOURCE_DIRTY]; + const dirtyResourceAction = skipDataSourceDirty ? undefined : resolveDirtyResourceAction(options, context); + const { [SKIP_DATA_SOURCE_DIRTY]: _skipDataSourceDirty, ...cleanConfig } = options; + return api.request(cleanConfig as typeof config).then((result) => { + if (dirtyResourceAction && isMutatingResourceAction(dirtyResourceAction.actionName)) { + markResourceActionDataSourceDirty(context, dirtyResourceAction, options.headers); + } + return result; + }); + }) as APIClient['request']; + + const proxy = new Proxy(api, { + get(target, prop, receiver) { + if (prop === 'resource') { + return resource; + } + if (prop === 'request') { + return request; + } + return Reflect.get(target, prop, receiver); + }, + }) as APIClient; + dirtyAwareApiClientProxies.add(proxy); + return proxy; +} + +export function getDirtyAwareApiClient(value: unknown, context: FlowContext): unknown { + if (!isApiClientLike(value)) { + return value; + } + + if (dirtyAwareApiClientProxies.has(value)) { + return value; + } + + const api = value as unknown as APIClient; + let contextCache = dirtyAwareApiClientCache.get(api); + if (!contextCache) { + contextCache = new WeakMap(); + dirtyAwareApiClientCache.set(api, contextCache); + } + + const cached = contextCache.get(context); + if (cached) { + return cached; + } + + const wrapped = createDirtyAwareApiClient(value, context); + contextCache.set(context, wrapped); + return wrapped; +} diff --git a/packages/plugins/@nocobase/plugin-ai/src/server/__tests__/ai-message-sanitizer.test.ts b/packages/plugins/@nocobase/plugin-ai/src/server/__tests__/ai-message-sanitizer.test.ts new file mode 100644 index 00000000000..2ea80da9779 --- /dev/null +++ b/packages/plugins/@nocobase/plugin-ai/src/server/__tests__/ai-message-sanitizer.test.ts @@ -0,0 +1,114 @@ +/** + * This file is part of the NocoBase (R) project. + * Copyright (c) 2020-2024 NocoBase Co., Ltd. + * Authors: NocoBase Team. + * + * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License. + * For more information, please refer to: https://www.nocobase.com/agreement. + */ + +import { AIMessage } from '@langchain/core/messages'; +import { describe, expect, it, vi } from 'vitest'; +import { sanitizeAdditionalKwargsForToolCalls, sanitizeLangChainAIMessage } from '../ai-employees/tool-call-sanitizer'; +import { convertAIMessage } from '../ai-employees/utils'; + +describe('AI message tool call sanitizer', () => { + const rawToolCall = { + id: 'call_bad', + type: 'function', + function: { + name: 'aiEmployeeWorkflowTaskOutput', + arguments: '{"result":{"reference_reply":":"bad json"}', + }, + }; + + it('should drop malformed raw tool calls when converting LangChain AI messages to stored messages', () => { + const logger = { warn: vi.fn() }; + const aiMessage = new AIMessage({ + id: 'ai_bad', + content: '', + additional_kwargs: { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + }); + + const values = convertAIMessage({ + aiEmployee: { + employee: { username: 'assistant' }, + skillSettings: { tools: [] }, + logger, + } as never, + providerName: 'deepseek', + model: 'deepseek-v4-flash', + aiMessage, + }); + + expect(values.metadata.additional_kwargs).toEqual({ + reasoning_content: 'keep this reasoning', + }); + expect(logger.warn).toHaveBeenCalledWith( + 'Discard malformed raw tool calls from AI message', + expect.objectContaining({ + phase: 'convertAIMessage', + messageId: 'ai_bad', + rawToolCallCount: 1, + parsedToolCallCount: 0, + rawToolCallIds: ['call_bad'], + rawToolCallNames: ['aiEmployeeWorkflowTaskOutput'], + }), + ); + expect(JSON.stringify(logger.warn.mock.calls[0][1])).not.toContain('reference_reply'); + }); + + it('should drop malformed raw tool calls when formatting stored assistant messages', () => { + const { additionalKwargs } = sanitizeAdditionalKwargsForToolCalls( + { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + null, + ); + + expect(additionalKwargs).toEqual({ + reasoning_content: 'keep this reasoning', + }); + }); + + it('should keep raw tool calls when parsed tool calls are present', () => { + const result = sanitizeAdditionalKwargsForToolCalls( + { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + [{ id: 'call_bad', name: 'aiEmployeeWorkflowTaskOutput', args: {}, type: 'tool_call' }], + ); + + expect(result.changed).toBe(false); + expect(result.additionalKwargs).toEqual({ + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }); + }); + + it('should sanitize LangChain AI messages in place without rebuilding the message instance', () => { + const aiMessage = new AIMessage({ + id: 'ai_bad', + content: '', + additional_kwargs: { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + }); + const messageWithRuntimeField = aiMessage as AIMessage & { runtimeField?: string }; + messageWithRuntimeField.runtimeField = 'keep runtime field'; + + const sanitizedMessage = sanitizeLangChainAIMessage(aiMessage); + + expect(sanitizedMessage).toBe(aiMessage); + expect(messageWithRuntimeField.runtimeField).toBe('keep runtime field'); + expect(aiMessage.additional_kwargs).toEqual({ + reasoning_content: 'keep this reasoning', + }); + }); +}); diff --git a/packages/plugins/@nocobase/plugin-ai/src/server/__tests__/tool-call-sanitizer-middleware.test.ts b/packages/plugins/@nocobase/plugin-ai/src/server/__tests__/tool-call-sanitizer-middleware.test.ts new file mode 100644 index 00000000000..f15da6b2151 --- /dev/null +++ b/packages/plugins/@nocobase/plugin-ai/src/server/__tests__/tool-call-sanitizer-middleware.test.ts @@ -0,0 +1,226 @@ +/** + * This file is part of the NocoBase (R) project. + * Copyright (c) 2020-2024 NocoBase Co., Ltd. + * Authors: NocoBase Team. + * + * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License. + * For more information, please refer to: https://www.nocobase.com/agreement. + */ + +import { AIMessage, BaseMessage, HumanMessage, ToolMessage } from '@langchain/core/messages'; +import { FakeListChatModel } from '@langchain/core/utils/testing'; +import { MemorySaver } from '@langchain/langgraph'; +import { convertMessagesToCompletionsMessageParams } from '@langchain/openai'; +import { createAgent, createMiddleware } from 'langchain'; +import { describe, expect, it, vi } from 'vitest'; +import { toolCallSanitizerMiddleware } from '../ai-employees/middleware'; + +describe('toolCallSanitizerMiddleware', () => { + const rawToolCall = { + id: 'call_bad', + type: 'function', + function: { + name: 'aiEmployeeWorkflowTaskOutput', + arguments: '{"result":{"reference_reply":":"bad json"}', + }, + }; + + type MessagePatchHook = (state: { + messages: BaseMessage[]; + }) => void | { messages?: BaseMessage[] } | Promise; + + const getMessagePatchHook = (hook: unknown): MessagePatchHook => { + if (typeof hook === 'function') { + return hook as MessagePatchHook; + } + if (hook && typeof hook === 'object' && 'hook' in hook) { + const nestedHook = (hook as { hook?: unknown }).hook; + if (typeof nestedHook === 'function') { + return nestedHook as MessagePatchHook; + } + } + throw new Error('Middleware hook is not callable'); + }; + + const getPatchMessages = (result: Awaited>) => { + if (!result || !('messages' in result)) { + return []; + } + return result.messages ?? []; + }; + + it('should remove malformed raw tool calls before checkpoint persistence and next model request', async () => { + const model = new FakeListChatModel({ + responses: [ + new AIMessage({ + id: 'ai_bad', + content: '', + additional_kwargs: { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + }), + ], + }); + const logger = { warn: vi.fn() }; + const agent = createAgent({ + model, + tools: [], + middleware: [toolCallSanitizerMiddleware({ logger })], + checkpointer: new MemorySaver(), + }); + const config = { configurable: { thread_id: 'malformed-tool-call' } }; + + await agent.invoke({ messages: [{ role: 'user', content: 'run' }] }, config); + + const state = await agent.getState(config); + const lastMessage = state.values.messages.at(-1) as AIMessage; + + expect(lastMessage.type).toBe('ai'); + expect(lastMessage.tool_calls).toEqual([]); + expect(lastMessage.invalid_tool_calls).toHaveLength(1); + expect(lastMessage.additional_kwargs).toEqual({ + reasoning_content: 'keep this reasoning', + }); + + const requestMessages = convertMessagesToCompletionsMessageParams({ + messages: state.values.messages, + }); + expect(requestMessages.at(-1)).toMatchObject({ + role: 'assistant', + content: '', + }); + expect(requestMessages.at(-1)).not.toHaveProperty('tool_calls'); + expect(logger.warn).toHaveBeenCalledWith( + 'Discard malformed raw tool calls from AI message', + expect.objectContaining({ + phase: 'afterModel', + messageId: 'ai_bad', + rawToolCallCount: 1, + parsedToolCallCount: 0, + rawToolCallIds: ['call_bad'], + rawToolCallNames: ['aiEmployeeWorkflowTaskOutput'], + }), + ); + expect(JSON.stringify(logger.warn.mock.calls[0][1])).not.toContain('reference_reply'); + }); + + it('should remove malformed raw tool calls from existing state before model requests', async () => { + let capturedMessages: BaseMessage[] = []; + const model = new FakeListChatModel({ + responses: ['done'], + }); + const agent = createAgent({ + model, + tools: [], + middleware: [ + toolCallSanitizerMiddleware(), + createMiddleware({ + name: 'CaptureMessagesMiddleware', + wrapModelCall: (request, handler) => { + capturedMessages = request.messages; + return handler(request); + }, + }), + ], + checkpointer: new MemorySaver(), + }); + + await agent.invoke( + { + messages: [ + new AIMessage({ + id: 'ai_bad_history', + content: '', + additional_kwargs: { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + }), + { role: 'user', content: 'continue' }, + ], + }, + { configurable: { thread_id: 'malformed-tool-call-history' } }, + ); + + const requestMessages = convertMessagesToCompletionsMessageParams({ + messages: capturedMessages, + }); + expect(requestMessages[0]).toMatchObject({ + role: 'assistant', + content: '', + }); + expect(requestMessages[0]).not.toHaveProperty('tool_calls'); + expect(requestMessages).toHaveLength(2); + expect(requestMessages[1]).toMatchObject({ + role: 'user', + content: 'continue', + }); + }); + + it('should build a patch without replacing non-AI messages', async () => { + const badAIMessage = new AIMessage({ + id: 'ai_bad_middle', + content: '', + additional_kwargs: { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + }); + const humanMessage = new HumanMessage({ id: 'human_1', content: 'continue' }); + const toolMessage = new ToolMessage({ + id: 'tool_1', + content: 'tool result', + tool_call_id: 'call_existing', + }); + const middleware = toolCallSanitizerMiddleware(); + const beforeModel = getMessagePatchHook(middleware.beforeModel); + const messages = [humanMessage, badAIMessage, toolMessage]; + + const result = await beforeModel({ + messages, + }); + const patchMessages = getPatchMessages(result); + + expect(messages).toEqual([humanMessage, badAIMessage, toolMessage]); + expect(patchMessages).toHaveLength(1); + expect(patchMessages[0]).not.toBe(badAIMessage); + expect(AIMessage.isInstance(patchMessages[0])).toBe(true); + expect(patchMessages[0].id).toBe(badAIMessage.id); + expect(badAIMessage.additional_kwargs).toEqual({ + reasoning_content: 'keep this reasoning', + }); + }); + + it('should sanitize malformed AI messages even when they are not the last message after model', async () => { + const badAIMessage = new AIMessage({ + id: 'ai_bad_not_last', + content: '', + additional_kwargs: { + tool_calls: [rawToolCall], + reasoning_content: 'keep this reasoning', + }, + }); + const middleware = toolCallSanitizerMiddleware(); + const afterModel = getMessagePatchHook(middleware.afterModel); + const messages = [ + new HumanMessage({ id: 'human_1', content: 'start' }), + badAIMessage, + new HumanMessage({ id: 'human_2', content: 'not last' }), + ]; + + const result = await afterModel({ + messages, + }); + const patchMessages = getPatchMessages(result); + + expect(messages[1]).toBe(badAIMessage); + expect(patchMessages).toHaveLength(1); + expect(patchMessages[0]).not.toBe(badAIMessage); + expect(AIMessage.isInstance(patchMessages[0])).toBe(true); + expect(patchMessages[0].id).toBe(badAIMessage.id); + expect(badAIMessage.additional_kwargs).toEqual({ + reasoning_content: 'keep this reasoning', + }); + }); +}); diff --git a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/ai-employee.ts b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/ai-employee.ts index 4639bd75703..9a5779b8a4d 100644 --- a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/ai-employee.ts +++ b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/ai-employee.ts @@ -23,6 +23,7 @@ import type { AIEmployee as AIEmployeeType } from '../../collections/ai-employee import { conversationMiddleware, skillToolBindingMiddleware, + toolCallSanitizerMiddleware, toolCallStatusMiddleware, toolInteractionMiddleware, workflowHistoryMiddleware, @@ -39,6 +40,7 @@ import { LLMResult } from '@langchain/core/outputs'; import { Context } from '@nocobase/actions'; import { listAccessibleAIEmployees, serializeEmployeeSummary } from '../../ai/tools/sub-agents/shared'; import { LLMStreamCached } from '../manager/llm-stream-manager'; +import { sanitizeAdditionalKwargsForToolCalls } from './tool-call-sanitizer'; export interface ModelRef { llmService: string; @@ -1288,7 +1290,15 @@ If information is missing, clearly state it in the summary.`; role: 'assistant', content, tool_calls: msg.toolCalls, - additional_kwargs: msg.metadata?.additional_kwargs, + additional_kwargs: sanitizeAdditionalKwargsForToolCalls(msg.metadata?.additional_kwargs, msg.toolCalls, { + onDiscard: (info) => { + this.logger.warn('Discard malformed raw tool calls from AI message', { + phase: 'formatMessages', + messageId: msg.metadata?.id, + ...info, + }); + }, + }).additionalKwargs, }); } @@ -1533,6 +1543,7 @@ If information is missing, clearly state it in the summary.`; toolCallStatusMiddleware(this), ...(inWorkflow ? [workflowHistoryMiddleware(this, this.db)] : []), conversationMiddleware(this, { providerName, llmService, model, messageId, agentThread }), + toolCallSanitizerMiddleware({ logger: this.logger }), ]; } diff --git a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/middleware/index.ts b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/middleware/index.ts index e6a468400ba..ab9c94dcd09 100644 --- a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/middleware/index.ts +++ b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/middleware/index.ts @@ -9,5 +9,6 @@ export * from './conversation'; export * from './skill-tools'; +export * from './tool-call-sanitizer'; export * from './tools'; export * from './workflow-history'; diff --git a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/middleware/tool-call-sanitizer.ts b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/middleware/tool-call-sanitizer.ts new file mode 100644 index 00000000000..f4e60f1ab0c --- /dev/null +++ b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/middleware/tool-call-sanitizer.ts @@ -0,0 +1,80 @@ +/** + * This file is part of the NocoBase (R) project. + * Copyright (c) 2020-2024 NocoBase Co., Ltd. + * Authors: NocoBase Team. + * + * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License. + * For more information, please refer to: https://www.nocobase.com/agreement. + */ + +import { AIMessage, BaseMessage } from '@langchain/core/messages'; +import { createMiddleware } from 'langchain'; +import { sanitizeLangChainAIMessage } from '../tool-call-sanitizer'; + +type ToolCallSanitizerLogger = { + warn: (message: string, meta?: Record) => void; +}; + +type ToolCallSanitizerMiddlewareOptions = { + logger?: ToolCallSanitizerLogger; +}; + +const buildMessageUpdates = ( + messages: readonly unknown[], + options: ToolCallSanitizerMiddlewareOptions, + phase: 'beforeModel' | 'afterModel', +): BaseMessage[] => { + const updates: BaseMessage[] = []; + + for (const message of messages) { + if (!AIMessage.isInstance(message) || !message.id) { + continue; + } + const sanitizedMessage = sanitizeLangChainAIMessage(message, { + onDiscard: (info) => { + options.logger?.warn('Discard malformed raw tool calls from AI message', { + phase, + messageId: message.id, + invalidToolCallCount: message.invalid_tool_calls?.length ?? 0, + ...info, + }); + }, + }); + if (!sanitizedMessage) { + continue; + } + updates.push( + new AIMessage({ + id: sanitizedMessage.id, + content: sanitizedMessage.content, + name: sanitizedMessage.name, + additional_kwargs: sanitizedMessage.additional_kwargs, + response_metadata: sanitizedMessage.response_metadata, + tool_calls: sanitizedMessage.tool_calls, + invalid_tool_calls: sanitizedMessage.invalid_tool_calls, + usage_metadata: sanitizedMessage.usage_metadata, + }), + ); + } + + return updates; +}; + +export const toolCallSanitizerMiddleware = (options: ToolCallSanitizerMiddlewareOptions = {}) => + createMiddleware({ + name: 'ToolCallSanitizerMiddleware', + beforeModel: (state) => { + const updates = buildMessageUpdates(state.messages ?? [], options, 'beforeModel'); + if (!updates.length) { + return; + } + return { messages: updates }; + }, + afterModel: (state) => { + const updates = buildMessageUpdates(state.messages ?? [], options, 'afterModel'); + if (!updates.length) { + return; + } + return { messages: updates }; + }, + }); diff --git a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/tool-call-sanitizer.ts b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/tool-call-sanitizer.ts new file mode 100644 index 00000000000..943fc59bb8c --- /dev/null +++ b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/tool-call-sanitizer.ts @@ -0,0 +1,95 @@ +/** + * This file is part of the NocoBase (R) project. + * Copyright (c) 2020-2024 NocoBase Co., Ltd. + * Authors: NocoBase Team. + * + * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License. + * For more information, please refer to: https://www.nocobase.com/agreement. + */ + +import { AIMessage } from '@langchain/core/messages'; + +type AdditionalKwargs = Record; + +type DiscardedToolCallsInfo = { + rawToolCallCount: number; + parsedToolCallCount: number; + rawToolCallIds: string[]; + rawToolCallNames: string[]; +}; + +type SanitizeAdditionalKwargsOptions = { + onDiscard?: (info: DiscardedToolCallsInfo) => void; +}; + +type SanitizeAdditionalKwargsResult = { + changed: boolean; + additionalKwargs?: AdditionalKwargs; +}; + +const hasRawToolCalls = (additionalKwargs?: AdditionalKwargs) => + Array.isArray(additionalKwargs?.tool_calls) && additionalKwargs.tool_calls.length > 0; + +const hasParsedToolCalls = (toolCalls?: unknown[] | null) => Array.isArray(toolCalls) && toolCalls.length > 0; + +const isRecord = (value: unknown): value is Record => + typeof value === 'object' && value !== null && !Array.isArray(value); + +const collectRawToolCallValues = (toolCalls: unknown[], key: 'id' | 'name') => + toolCalls + .map((toolCall) => { + if (!isRecord(toolCall)) { + return null; + } + if (key === 'name') { + const fn = toolCall.function; + if (isRecord(fn) && typeof fn.name === 'string') { + return fn.name; + } + } + const value = toolCall[key]; + return typeof value === 'string' ? value : null; + }) + .filter((value): value is string => Boolean(value)); + +export const sanitizeAdditionalKwargsForToolCalls = ( + additionalKwargs: AdditionalKwargs | undefined, + parsedToolCalls?: unknown[] | null, + options: SanitizeAdditionalKwargsOptions = {}, +): SanitizeAdditionalKwargsResult => { + if (!hasRawToolCalls(additionalKwargs) || hasParsedToolCalls(parsedToolCalls)) { + return { + changed: false, + additionalKwargs, + }; + } + + const rawToolCalls = additionalKwargs.tool_calls as unknown[]; + options.onDiscard?.({ + rawToolCallCount: rawToolCalls.length, + parsedToolCallCount: parsedToolCalls?.length ?? 0, + rawToolCallIds: collectRawToolCallValues(rawToolCalls, 'id'), + rawToolCallNames: collectRawToolCallValues(rawToolCalls, 'name'), + }); + + const sanitized = { ...additionalKwargs }; + delete sanitized.tool_calls; + + return { + changed: true, + additionalKwargs: Object.keys(sanitized).length > 0 ? sanitized : undefined, + }; +}; + +export const sanitizeLangChainAIMessage = (message: AIMessage, options: SanitizeAdditionalKwargsOptions = {}) => { + const sanitized = sanitizeAdditionalKwargsForToolCalls(message.additional_kwargs, message.tool_calls, options); + + if (!sanitized.changed) { + return null; + } + + message.additional_kwargs = sanitized.additionalKwargs ?? {}; + message.lc_kwargs.additional_kwargs = message.additional_kwargs; + + return message; +}; diff --git a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/utils.ts b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/utils.ts index 366937b6f13..c3a799617b8 100644 --- a/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/utils.ts +++ b/packages/plugins/@nocobase/plugin-ai/src/server/ai-employees/utils.ts @@ -10,6 +10,7 @@ import { AIMessage, HumanMessage, ToolMessage } from 'langchain'; import { AIMessageContent, AIMessageInput } from '../types'; import { AIEmployee } from './ai-employee'; +import { sanitizeAdditionalKwargsForToolCalls } from './tool-call-sanitizer'; export const convertAIMessage = ({ aiEmployee, @@ -84,8 +85,18 @@ export const convertAIMessage = ({ if (aiMessage.response_metadata) { values.metadata.response_metadata = aiMessage.response_metadata; } - if (aiMessage.additional_kwargs) { - values.metadata.additional_kwargs = aiMessage.additional_kwargs; + const additionalKwargs = sanitizeAdditionalKwargsForToolCalls(aiMessage.additional_kwargs, toolCalls, { + onDiscard: (info) => { + aiEmployee.logger?.warn('Discard malformed raw tool calls from AI message', { + phase: 'convertAIMessage', + messageId: aiMessage.id, + invalidToolCallCount: aiMessage.invalid_tool_calls?.length ?? 0, + ...info, + }); + }, + }).additionalKwargs; + if (additionalKwargs) { + values.metadata.additional_kwargs = additionalKwargs; } return values;