diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9e3b1cd..00288fe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -131,5 +131,5 @@ jobs: node-version: '22' cache: npm - run: npm ci - - name: Live MySQL concurrent upload and memory admission - run: node --test --test-concurrency=1 tests/uploadAdmission.mysql.integration.test.js tests/userMemoryAdmission.mysql.integration.test.js + - name: Live MySQL persistence concurrency acceptance + run: node --test --test-concurrency=1 tests/uploadAdmission.mysql.integration.test.js tests/userMemoryAdmission.mysql.integration.test.js tests/dataSourceTestAtomic.mysql.integration.test.js diff --git a/config/size-baseline.json b/config/size-baseline.json index a3e569a..4557542 100644 --- a/config/size-baseline.json +++ b/config/size-baseline.json @@ -82,7 +82,7 @@ "src/datasourceCrypto.js": 55, "src/datasourceLifecycle.js": 15, "src/datasourcePool.js": 24, - "src/dataSourceRepository.js": 184, + "src/dataSourceRepository.js": 180, "src/derivedMetrics.js": 391, "src/dialects/common.js": 25, "src/dialects/index.js": 27, @@ -129,7 +129,7 @@ "src/localIndicatorValidation.js": 38, "src/mcp/audit.js": 103, "src/mcp/auth.js": 54, - "src/mcp/config.js": 73, + "src/mcp/config.js": 57, "src/mcp/connectorPackage.js": 115, "src/mcp/index.js": 93, "src/mcp/modeGate.js": 46, @@ -160,7 +160,7 @@ "src/platformDatabase.js": 61, "src/platformLogRepository.js": 215, "src/presentationFallbacks.js": 63, - "src/processArtifacts.js": 173, + "src/processArtifacts.js": 89, "src/promptSanitizer.js": 123, "src/queryContract.js": 358, "src/queryContractCompiler.js": 2285, @@ -203,7 +203,7 @@ "src/turnRouter.js": 143, "src/uploads.js": 505, "src/uploadAdmission.js": 94, - "src/userMemory.js": 642, + "src/userMemory.js": 637, "src/userMemoryDistiller.js": 499, "src/userMemoryMaintenance.js": 156, "src/userMemoryModel.js": 138, @@ -212,11 +212,14 @@ "src/userRepository.js": 141, "src/valueClarification.js": 186, "src/workflow.js": 230, + "src/workspace.js": 464, "src/workspaceTransforms.js": 189, - "src/workspace.js": 566, "src/workspaceRepository.js": 287, "src/xlsx.js": 369, "src/xlsxReader.js": 480, - "src/platformAsyncLogRepository.js": 168 - "src/skillAudits.js": 119 + "src/skillAudits.js": 119, + "src/platformAsyncLogRepository.js": 168, + "src/dataSourceRow.js": 25, + "src/mcp/configValues.js": 44, + "src/processArtifactDraft.js": 119 } diff --git a/src/dataSourceRepository.js b/src/dataSourceRepository.js index e2967c8..520e17c 100644 --- a/src/dataSourceRepository.js +++ b/src/dataSourceRepository.js @@ -1,3 +1,4 @@ +import { mapRow } from './dataSourceRow.js'; import { normalizeDataSource } from './dialects/index.js'; import { dataSourceTestState } from './datasourceLifecycle.js'; @@ -5,40 +6,17 @@ function nowIso() { return new Date().toISOString(); } -function parseJson(value, fallback) { - if (value === null || value === undefined || value === '') { - return fallback; - } - try { - return JSON.parse(value); - } catch { - return fallback; - } -} - function asJson(value) { return JSON.stringify(value ?? {}); } -function asBool(value) { - return value === 1 || value === true; -} - export class DataSourceRepository { constructor(db, asyncStore = db) { this.db = db; this.asyncStore = asyncStore; } - mapRow(row) { - return row ? { - ...row, - options: parseJson(row.optionsJson, {}), - lastTestOk: row.lastTestOk === null || row.lastTestOk === undefined - ? null - : asBool(row.lastTestOk), - } : null; - } + mapRow(row) { return mapRow(row); } async listDataSourcesAsync() { const rows = await this.asyncStore.query(`SELECT id, code, name, db_type AS dbType, @@ -84,17 +62,15 @@ export class DataSourceRepository { async updateDataSourceTestAsync(id, { ok, message }, source = null) { const current = source ?? await this.getDataSourceAsync(id); if (!current) return null; - const unchanged = (await this.asyncStore.query(`SELECT COUNT(*) AS count FROM data_sources - WHERE id=? AND code=? AND db_type=? AND host=? AND port=? AND database_name=? - AND username=? AND encrypted_password=? AND options_json=?`, [ - Number(id), current.code, current.dbType, current.host, Number(current.port), - current.databaseName, current.username, current.encryptedPassword ?? '', - JSON.stringify(current.options ?? {}), - ]))[0]?.count; - if (Number(unchanged) > 0) { - await this.asyncStore.execute(`UPDATE data_sources SET last_test_at=?, last_test_ok=?, - last_test_message=? WHERE id=?`, [nowIso(), ok ? 1 : 0, String(message ?? ''), Number(id)]); - } + const fields = ['code', 'db_type', 'host', 'database_name', 'username', 'encrypted_password', 'options_json']; + const match = fields.map(field => this.asyncStore.backend === 'mysql' + ? `BINARY ${field} = BINARY ?` : `${field} = ?`).join(' AND '); + await this.asyncStore.execute(`UPDATE data_sources SET last_test_at=?, last_test_ok=?, + last_test_message=? WHERE id=? AND port=? AND ${match}`, [ + nowIso(), ok ? 1 : 0, String(message ?? ''), Number(id), Number(current.port), + current.code, current.dbType, current.host, current.databaseName, current.username, + current.encryptedPassword ?? '', JSON.stringify(current.options ?? {}), + ]); return this.getDataSourceAsync(id); } diff --git a/src/dataSourceRow.js b/src/dataSourceRow.js new file mode 100644 index 0000000..b8fd8b7 --- /dev/null +++ b/src/dataSourceRow.js @@ -0,0 +1,24 @@ +function parseJson(value, fallback) { + if (value === null || value === undefined || value === '') { + return fallback; + } + try { + return JSON.parse(value); + } catch { + return fallback; + } +} + +function asBool(value) { + return value === 1 || value === true; +} + +export function mapRow(row) { + return row ? { + ...row, + options: parseJson(row.optionsJson, {}), + lastTestOk: row.lastTestOk === null || row.lastTestOk === undefined + ? null + : asBool(row.lastTestOk), + } : null; + } diff --git a/src/mcp/config.js b/src/mcp/config.js index b498546..1627cee 100644 --- a/src/mcp/config.js +++ b/src/mcp/config.js @@ -2,88 +2,55 @@ // 默认拒绝原则：配置无效时回退到只读模式（disabled 关闭 / read 最小可见）。 import { callPlatformDatabase } from '../platformDatabaseAccess.js'; -export const DEFAULT_MCP_CONFIG = { - enabled: false, - mode: 'read', - toolOverrides: {}, - tenantScope: [], - rateLimit: { rpm: 60, burst: 10 }, - audit: { enabled: true, level: 'full' }, - sessionMode: 'stateful', -}; +import { coerce } from './configValues.js'; +export { DEFAULT_MCP_CONFIG, getDefaultMcpConfig } from './configValues.js'; -const VALID_MODES = ['disabled', 'read', 'standard', 'full']; const CACHE_TTL = 5000; - -let cache = { value: null, at: 0 }; - -export function getDefaultMcpConfig() { - return JSON.parse(JSON.stringify(DEFAULT_MCP_CONFIG)); -} - -function coerce(raw) { - const cfg = getDefaultMcpConfig(); - if (!raw || typeof raw !== 'object') { - return cfg; - } - if (typeof raw.enabled === 'boolean') cfg.enabled = raw.enabled; - if (VALID_MODES.includes(raw.mode)) cfg.mode = raw.mode; - if (raw.toolOverrides && typeof raw.toolOverrides === 'object') { - cfg.toolOverrides = { ...raw.toolOverrides }; - } - if (Array.isArray(raw.tenantScope)) cfg.tenantScope = raw.tenantScope.map(String); - if (raw.rateLimit && typeof raw.rateLimit === 'object') { - const rpm = Number(raw.rateLimit.rpm); - const burst = Number(raw.rateLimit.burst); - if (Number.isFinite(rpm) && rpm > 0) cfg.rateLimit.rpm = rpm; - if (Number.isFinite(burst) && burst > 0) cfg.rateLimit.burst = burst; - } - if (raw.audit && typeof raw.audit === 'object') { - if (typeof raw.audit.enabled === 'boolean') cfg.audit.enabled = raw.audit.enabled; - if (['full', 'summary', 'off'].includes(raw.audit.level)) cfg.audit.level = raw.audit.level; - } - if (raw.sessionMode === 'stateless' || raw.sessionMode === 'stateful') { - cfg.sessionMode = raw.sessionMode; - } - return cfg; +let cache = new WeakMap(); +function entryFor(database) { + if (!cache.has(database)) cache.set(database, { value: null, at: 0 }); + return cache.get(database); } export function getMcpConfig(database) { + const entry = entryFor(database); const now = Date.now(); - if (cache.value && now - cache.at < CACHE_TTL) { - return cache.value; + if (entry.value && now - entry.at < CACHE_TTL) { + return entry.value; } const record = database.getPlatformSetting('mcp', null); const raw = record && record.value && typeof record.value === 'object' ? record.value : null; const cfg = coerce(raw); - cache = { value: cfg, at: now }; + cache.set(database, { value: cfg, at: now }); return cfg; } export function setMcpConfig(database, patch) { const next = coerce(patch && typeof patch === 'object' ? patch : {}); database.savePlatformSetting('mcp', next); - cache = { value: next, at: Date.now() }; + cache.set(database, { value: next, at: Date.now() }); return next; } export async function getMcpConfigAsync(database) { + const entry = entryFor(database); const now = Date.now(); - if (cache.value && now - cache.at < CACHE_TTL) return cache.value; + if (entry.value && now - entry.at < CACHE_TTL) return entry.value; const record = await callPlatformDatabase(database, 'getPlatformSettingAsync', 'getPlatformSetting', 'mcp', null); const cfg = coerce(record?.value); - cache = { value: cfg, at: now }; + if (cache.get(database) !== entry) return cache.get(database)?.value ?? cfg; + cache.set(database, { value: cfg, at: now }); return cfg; } export async function setMcpConfigAsync(database, patch) { const next = coerce(patch && typeof patch === 'object' ? patch : {}); await callPlatformDatabase(database, 'savePlatformSettingAsync', 'savePlatformSetting', 'mcp', next); - cache = { value: next, at: Date.now() }; + cache.set(database, { value: next, at: Date.now() }); return next; } // 仅用于测试或进程内重置，避免缓存陈旧。 export function _resetMcpConfigCache() { - cache = { value: null, at: 0 }; + cache = new WeakMap(); } diff --git a/src/mcp/configValues.js b/src/mcp/configValues.js new file mode 100644 index 0000000..c55e489 --- /dev/null +++ b/src/mcp/configValues.js @@ -0,0 +1,43 @@ +export const DEFAULT_MCP_CONFIG = { + enabled: false, + mode: 'read', + toolOverrides: {}, + tenantScope: [], + rateLimit: { rpm: 60, burst: 10 }, + audit: { enabled: true, level: 'full' }, + sessionMode: 'stateful', +}; + +const VALID_MODES = ['disabled', 'read', 'standard', 'full']; + + +export function getDefaultMcpConfig() { + return JSON.parse(JSON.stringify(DEFAULT_MCP_CONFIG)); +} + +export function coerce(raw) { + const cfg = getDefaultMcpConfig(); + if (!raw || typeof raw !== 'object') { + return cfg; + } + if (typeof raw.enabled === 'boolean') cfg.enabled = raw.enabled; + if (VALID_MODES.includes(raw.mode)) cfg.mode = raw.mode; + if (raw.toolOverrides && typeof raw.toolOverrides === 'object') { + cfg.toolOverrides = { ...raw.toolOverrides }; + } + if (Array.isArray(raw.tenantScope)) cfg.tenantScope = raw.tenantScope.map(String); + if (raw.rateLimit && typeof raw.rateLimit === 'object') { + const rpm = Number(raw.rateLimit.rpm); + const burst = Number(raw.rateLimit.burst); + if (Number.isFinite(rpm) && rpm > 0) cfg.rateLimit.rpm = rpm; + if (Number.isFinite(burst) && burst > 0) cfg.rateLimit.burst = burst; + } + if (raw.audit && typeof raw.audit === 'object') { + if (typeof raw.audit.enabled === 'boolean') cfg.audit.enabled = raw.audit.enabled; + if (['full', 'summary', 'off'].includes(raw.audit.level)) cfg.audit.level = raw.audit.level; + } + if (raw.sessionMode === 'stateless' || raw.sessionMode === 'stateful') { + cfg.sessionMode = raw.sessionMode; + } + return cfg; +} diff --git a/src/processArtifactDraft.js b/src/processArtifactDraft.js new file mode 100644 index 0000000..8293890 --- /dev/null +++ b/src/processArtifactDraft.js @@ -0,0 +1,118 @@ +// 工作区过程文件：把 agent 每个产出数据的环节落成用户可直接读懂的数据文件。 +// 只保留数据行和人话元信息，不保存环节参数、原始响应等技术载荷，也不承载业务口径。 + +const ROW_LIMIT = 200; +const MAX_STEPS_PER_RUN = 60; + +const KIND_LABELS = { + TABLE: '数据表', + SQL_RESULT: '查询结果', +}; + +function isRecord(value) { + return Boolean(value) && typeof value === 'object' && !Array.isArray(value); +} + +// 工具返回的数据形态可能是完整结果、抽样结果或嵌套在 data 下，这里统一识别。 +function extractTable(result) { + if (!isRecord(result)) { + return null; + } + const columns = Array.isArray(result.columns) ? result.columns : null; + const rows = Array.isArray(result.rows) + ? result.rows + : Array.isArray(result.data?.rows) ? result.data.rows : null; + const sampleRows = Array.isArray(result.sampleRows) ? result.sampleRows : null; + const source = rows ?? sampleRows; + if (!columns || !source || columns.length === 0 || source.length === 0) { + return null; + } + return { + columns, + rows: source.slice(0, ROW_LIMIT), + truncated: source.length > ROW_LIMIT, + sampled: !rows, + }; +} + +function collectRelatedArtifactIds(result) { + if (!isRecord(result)) { + return []; + } + const ids = [ + result.workspaceArtifactId, + result.codeArtifactId, + ...(Array.isArray(result.artifactIds) ? result.artifactIds : []), + ...(Array.isArray(result.inputArtifactIds) ? result.inputArtifactIds : []), + ...(Array.isArray(result.outputs) ? result.outputs.map((item) => item?.artifactId) : []), + ]; + return [...new Set(ids.filter(Boolean).map(String))]; +} + +// 把一次工具调用整理成过程文件草稿。 +// 只有产出可读数据表的环节才落盘；目录、契约、原始 JSON 等纯技术载荷直接跳过。 +function buildProcessArtifact({ + tool, + result = null, + error = null, + step = {}, + previousArtifactId = null, +}) { + const table = error ? null : extractTable(result); + if (!table) { + return null; + } + const sequence = Number(step.sequence) > 0 ? Number(step.sequence) : null; + const label = String(step.title ?? tool ?? '执行步骤'); + const dataKind = result?.sql ? 'SQL_RESULT' : 'TABLE'; + const relatedArtifactIds = collectRelatedArtifactIds(result); + const inputArtifactIds = [ + ...new Set([...(previousArtifactId ? [previousArtifactId] : []), ...relatedArtifactIds]), + ]; + const rowCount = Number(result?.rowCount ?? table.rows.length); + const durationMs = Number(step.durationMs); + const metadata = { + process: true, + dataKind, + dataKindLabel: KIND_LABELS[dataKind], + stepId: step.id ?? null, + sequence, + tool: tool ?? null, + toolTitle: label, + stageCode: step.stageCode ?? null, + skillCode: step.skillCode ?? null, + skillName: step.skillName ?? null, + status: error ? 'error' : 'success', + startedAt: step.startedAt ?? null, + durationMs: Number.isFinite(durationMs) ? durationMs : null, + summary: String(step.detail ?? ''), + rowCount, + columnCount: table.columns.length, + capturedRowCount: table.rows.length, + sampled: Boolean(table.sampled), + truncated: Boolean(table.truncated), + inputArtifactIds, + parentArtifactIds: inputArtifactIds, + outputArtifactIds: relatedArtifactIds, + }; + const payload = { + step: { + id: step.id ?? null, + sequence, + tool: tool ?? null, + title: label, + stageCode: step.stageCode ?? null, + status: metadata.status, + durationMs: metadata.durationMs, + }, + dataKind, + summary: metadata.summary, + truncated: { rows: Boolean(table.truncated) }, + data: { columns: table.columns, rows: table.rows }, + }; + const rowLabel = ` · ${rowCount} 行`; + const title = `${sequence ? `步骤 ${String(sequence).padStart(2, '0')} · ` : ''}${label}${rowLabel}`; + return { title, metadata, payload }; +} + +export { buildProcessArtifact, KIND_LABELS, MAX_STEPS_PER_RUN }; diff --git a/src/processArtifacts.js b/src/processArtifacts.js index f920d38..c625d66 100644 --- a/src/processArtifacts.js +++ b/src/processArtifacts.js @@ -1,119 +1,4 @@ -// 工作区过程文件：把 agent 每个产出数据的环节落成用户可直接读懂的数据文件。 -// 只保留数据行和人话元信息，不保存环节参数、原始响应等技术载荷，也不承载业务口径。 - -const ROW_LIMIT = 200; -const MAX_STEPS_PER_RUN = 60; - -const KIND_LABELS = { - TABLE: '数据表', - SQL_RESULT: '查询结果', -}; - -function isRecord(value) { - return Boolean(value) && typeof value === 'object' && !Array.isArray(value); -} - -// 工具返回的数据形态可能是完整结果、抽样结果或嵌套在 data 下，这里统一识别。 -function extractTable(result) { - if (!isRecord(result)) { - return null; - } - const columns = Array.isArray(result.columns) ? result.columns : null; - const rows = Array.isArray(result.rows) - ? result.rows - : Array.isArray(result.data?.rows) ? result.data.rows : null; - const sampleRows = Array.isArray(result.sampleRows) ? result.sampleRows : null; - const source = rows ?? sampleRows; - if (!columns || !source || columns.length === 0 || source.length === 0) { - return null; - } - return { - columns, - rows: source.slice(0, ROW_LIMIT), - truncated: source.length > ROW_LIMIT, - sampled: !rows, - }; -} - -function collectRelatedArtifactIds(result) { - if (!isRecord(result)) { - return []; - } - const ids = [ - result.workspaceArtifactId, - result.codeArtifactId, - ...(Array.isArray(result.artifactIds) ? result.artifactIds : []), - ...(Array.isArray(result.inputArtifactIds) ? result.inputArtifactIds : []), - ...(Array.isArray(result.outputs) ? result.outputs.map((item) => item?.artifactId) : []), - ]; - return [...new Set(ids.filter(Boolean).map(String))]; -} - -// 把一次工具调用整理成过程文件草稿。 -// 只有产出可读数据表的环节才落盘；目录、契约、原始 JSON 等纯技术载荷直接跳过。 -function buildProcessArtifact({ - tool, - result = null, - error = null, - step = {}, - previousArtifactId = null, -}) { - const table = error ? null : extractTable(result); - if (!table) { - return null; - } - const sequence = Number(step.sequence) > 0 ? Number(step.sequence) : null; - const label = String(step.title ?? tool ?? '执行步骤'); - const dataKind = result?.sql ? 'SQL_RESULT' : 'TABLE'; - const relatedArtifactIds = collectRelatedArtifactIds(result); - const inputArtifactIds = [ - ...new Set([...(previousArtifactId ? [previousArtifactId] : []), ...relatedArtifactIds]), - ]; - const rowCount = Number(result?.rowCount ?? table.rows.length); - const durationMs = Number(step.durationMs); - const metadata = { - process: true, - dataKind, - dataKindLabel: KIND_LABELS[dataKind], - stepId: step.id ?? null, - sequence, - tool: tool ?? null, - toolTitle: label, - stageCode: step.stageCode ?? null, - skillCode: step.skillCode ?? null, - skillName: step.skillName ?? null, - status: error ? 'error' : 'success', - startedAt: step.startedAt ?? null, - durationMs: Number.isFinite(durationMs) ? durationMs : null, - summary: String(step.detail ?? ''), - rowCount, - columnCount: table.columns.length, - capturedRowCount: table.rows.length, - sampled: Boolean(table.sampled), - truncated: Boolean(table.truncated), - inputArtifactIds, - parentArtifactIds: inputArtifactIds, - outputArtifactIds: relatedArtifactIds, - }; - const payload = { - step: { - id: step.id ?? null, - sequence, - tool: tool ?? null, - title: label, - stageCode: step.stageCode ?? null, - status: metadata.status, - durationMs: metadata.durationMs, - }, - dataKind, - summary: metadata.summary, - truncated: { rows: Boolean(table.truncated) }, - data: { columns: table.columns, rows: table.rows }, - }; - const rowLabel = ` · ${rowCount} 行`; - const title = `${sequence ? `步骤 ${String(sequence).padStart(2, '0')} · ` : ''}${label}${rowLabel}`; - return { title, metadata, payload }; -} +import { buildProcessArtifact, KIND_LABELS, MAX_STEPS_PER_RUN } from './processArtifactDraft.js'; // 记录器：agent 每完成一个数据环节就调用一次；没有可读数据的环节直接跳过， // 写入失败也只跳过，不影响主流程。 @@ -168,7 +53,7 @@ function createProcessArtifactRecorder({ function createProcessArtifactRecorderAsync(options) { const { service, workspaceId, userId, sessionId, emit = () => {}, artifactIds = null, maxSteps = MAX_STEPS_PER_RUN } = options; const state = { previousArtifactId: null, count: 0 }; - return async ({ tool, result = null, error = null, step = {} }) => { + const record = async ({ tool, result = null, error = null, step = {} }) => { if (!service || !workspaceId || state.count >= maxSteps) return null; try { const draft = buildProcessArtifact({ tool, result, error, step, previousArtifactId: state.previousArtifactId }); @@ -186,6 +71,12 @@ function createProcessArtifactRecorderAsync(options) { return null; } }; + let pending = Promise.resolve(); + return input => { + const result = pending.then(() => record(input)); + pending = result.catch(() => {}); + return result; + }; } export { diff --git a/tests/dataSourceTestAtomic.mysql.integration.test.js b/tests/dataSourceTestAtomic.mysql.integration.test.js new file mode 100644 index 0000000..eb83556 --- /dev/null +++ b/tests/dataSourceTestAtomic.mysql.integration.test.js @@ -0,0 +1,49 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { initializePlatformDatabase } from '../src/platformDatabase.js'; +import { DataSourceRepository } from '../src/dataSourceRepository.js'; + +const databaseName = process.env.YOLO_MYSQL_UPLOAD_TEST_DATABASE; +test('live MySQL rejects stale connection results including case-only credential changes', { skip: !databaseName }, async t => { + const platform = await initializePlatformDatabase({ backend: 'mysql', mysql: { + host: process.env.YOLO_MYSQL_UPLOAD_TEST_HOST ?? '127.0.0.1', + port: Number(process.env.YOLO_MYSQL_UPLOAD_TEST_PORT ?? 3306), database: databaseName, + user: process.env.YOLO_MYSQL_UPLOAD_TEST_USER ?? 'root', + password: process.env.YOLO_MYSQL_UPLOAD_TEST_PASSWORD ?? 'test', + } }); + let sourceId; + t.after(async () => { + try { if (sourceId) await platform.database.execute('DELETE FROM data_sources WHERE id=?', [sourceId]); } + finally { await platform.close(); } + }); + const database = platform.database; + const repo = database.dataSourceRepository; + const source = await repo.saveDataSourceAsync({ code: 'cas_source', name: 'CAS source', dbType: 'MYSQL', + host: 'old.example', username: 'reader', encryptedPassword: 'Cipher', options: { tag: 'Case' } }); + sourceId = source.id; + for (const [field, replacement] of [['host', 'OLD.EXAMPLE'], ['username', 'READER'], + ['encrypted_password', 'cipher'], ['options_json', '{"tag":"case"}']]) { + const original = await repo.getDataSourceAsync(source.id); + let edited = false; + const interleaved = new DataSourceRepository(null, { + backend: 'mysql', query: (...args) => database.query(...args), + async execute(sql, bindings) { + if (!edited) { + edited = true; + await database.execute(`UPDATE data_sources SET ${field}=?, last_test_at=NULL, + last_test_ok=NULL, last_test_message='' WHERE id=?`, [replacement, source.id]); + } + return database.execute(sql, bindings); + }, + }); + const result = await interleaved.updateDataSourceTestAsync(source.id, { ok: true, message: 'stale' }, original); + assert.equal(edited, true); + assert.equal(result.lastTestAt, null, field); + assert.equal(result.lastTestOk, null, field); + assert.equal(result.lastTestMessage, '', field); + const fresh = await repo.updateDataSourceTestAsync(source.id, { ok: false, message: 'fresh' }, result); + assert.equal(fresh.lastTestOk, false, field); + assert.equal(fresh.lastTestMessage, 'fresh', field); + assert.ok(fresh.lastTestAt, field); + } +}); diff --git a/tests/dataSourceTestAtomic.test.js b/tests/dataSourceTestAtomic.test.js new file mode 100644 index 0000000..8ab9e55 --- /dev/null +++ b/tests/dataSourceTestAtomic.test.js @@ -0,0 +1,38 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { DataSourceRepository } from '../src/dataSourceRepository.js'; + +for (const ok of [true, false]) { + test(`a stale ${ok ? 'successful' : 'failed'} test cannot overwrite a connection edited immediately before persistence`, async t => { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + const original = database.saveDataSource({ code: 'atomic_test', name: 'Atomic test', dbType: 'MYSQL', + host: 'old.example', username: 'reader', options: { connectTimeout: 1234 } }); + let updated = false; + const repository = new DataSourceRepository(null, { + async query(sql, bindings = []) { return database.db.prepare(sql).all(...bindings); }, + async execute(sql, bindings = []) { + if (!updated) { + updated = true; + database.saveDataSource({ ...original, host: 'new.example' }, original.id); + } + const result = database.db.prepare(sql).run(...bindings); + return { affectedRows: Number(result.changes) }; + }, + }); + const result = await repository.updateDataSourceTestAsync(original.id, { ok, message: 'old connection result' }, original); + assert.equal(updated, true); + assert.equal(result.host, 'new.example'); + assert.equal(result.lastTestAt, null); + assert.equal(result.lastTestOk, null); + assert.equal(result.lastTestMessage, ''); + const current = await repository.getDataSourceAsync(original.id); + const fresh = await repository.updateDataSourceTestAsync(original.id, { ok: true, message: 'fresh connection' }, current); + assert.equal(fresh.lastTestOk, true); + assert.equal(fresh.lastTestMessage, 'fresh connection'); + assert.match(fresh.lastTestAt, /^\d{4}-/); + database.db.prepare('DELETE FROM data_sources WHERE id = ?').run(original.id); + assert.equal(await repository.updateDataSourceTestAsync(original.id, { ok }, current), null); + }); +} diff --git a/tests/mcpConfigCache.test.js b/tests/mcpConfigCache.test.js new file mode 100644 index 0000000..79019cf --- /dev/null +++ b/tests/mcpConfigCache.test.js @@ -0,0 +1,85 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { getMcpConfig, getMcpConfigAsync, setMcpConfig, setMcpConfigAsync, + _resetMcpConfigCache } from '../src/mcp/config.js'; + +function fixture(t) { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + _resetMcpConfigCache(); + t.after(_resetMcpConfigCache); + return database; +} + +for (const asyncOnly of [false, true]) { + test(`MCP config cache isolates ${asyncOnly ? 'async-only' : 'synchronous'} databases`, async t => { + const a = fixture(t); + const b = fixture(t); + const wrap = database => asyncOnly ? { + isAsyncPlatformStore: true, + async getPlatformSettingAsync(...args) { return database.getPlatformSetting(...args); }, + async savePlatformSettingAsync(...args) { return database.savePlatformSetting(...args); }, + } : database; + const first = wrap(a), second = wrap(b); + const read = asyncOnly ? getMcpConfigAsync : getMcpConfig; + const save = asyncOnly ? setMcpConfigAsync : setMcpConfig; + await save(first, { enabled: true, mode: 'full', tenantScope: ['first'] }); + assert.equal((await read(second)).enabled, false); + await save(second, { enabled: true, mode: 'read', tenantScope: ['second'] }); + assert.deepEqual((await read(first)).tenantScope, ['first']); + assert.deepEqual((await read(second)).tenantScope, ['second']); + }); +} + +test('an older async config read cannot overwrite a completed administrator save', async t => { + const database = fixture(t); + database.savePlatformSetting('mcp', { enabled: true, mode: 'full' }); + let release, started; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { started = resolve; }); + const adapter = { + isAsyncPlatformStore: true, + async getPlatformSettingAsync(...args) { + const snapshot = database.getPlatformSetting(...args); + started(); await gate; return snapshot; + }, + async savePlatformSettingAsync(...args) { return database.savePlatformSetting(...args); }, + }; + const pending = getMcpConfigAsync(adapter); + await ready; + await setMcpConfigAsync(adapter, { enabled: false, mode: 'disabled' }); + release(); + assert.equal((await pending).enabled, false); + assert.equal((await getMcpConfigAsync(adapter)).mode, 'disabled'); + assert.equal(database.getPlatformSetting('mcp').value.enabled, false); +}); + +test('config cache expires after five seconds and reset invalidates every database', t => { + const database = fixture(t); + let now = 1000; + t.mock.method(Date, 'now', () => now); + setMcpConfig(database, { enabled: true, mode: 'full' }); + database.savePlatformSetting('mcp', { enabled: false, mode: 'disabled' }); + now = 5999; + assert.equal(getMcpConfig(database).enabled, true); + now = 6000; + assert.equal(getMcpConfig(database).enabled, false); + database.savePlatformSetting('mcp', { enabled: true, mode: 'read' }); + _resetMcpConfigCache(); + assert.equal(getMcpConfig(database).enabled, true); +}); + +test('failed config saves leave the last successful cached policy intact', async t => { + const database = fixture(t); + const failure = new Error('settings write failed'); + const adapter = { + isAsyncPlatformStore: true, + async getPlatformSettingAsync(...args) { return database.getPlatformSetting(...args); }, + async savePlatformSettingAsync() { throw failure; }, + }; + database.savePlatformSetting('mcp', { enabled: false, mode: 'disabled' }); + await getMcpConfigAsync(adapter); + await assert.rejects(setMcpConfigAsync(adapter, { enabled: true, mode: 'full' }), error => error === failure); + assert.equal((await getMcpConfigAsync(adapter)).enabled, false); +}); diff --git a/tests/processArtifacts.concurrent.test.js b/tests/processArtifacts.concurrent.test.js new file mode 100644 index 0000000..5c83e55 --- /dev/null +++ b/tests/processArtifacts.concurrent.test.js @@ -0,0 +1,76 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { WorkspaceService } from '../src/workspace.js'; +import { createProcessArtifactRecorderAsync } from '../src/processArtifacts.js'; + +function fixture(t) { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'process-concurrency-')); + const database = new PlatformDatabase(path.join(directory, 'test.db')); + t.after(async () => { + await database.closeAsync(); + fs.rmSync(directory, { recursive: true, force: true }); + }); + const service = new WorkspaceService(database); + const userId = database.getUserByUsername('admin').id; + const session = database.createChatSession({ userId, themeId: 1 }); + const workspace = service.ensureForSession({ userId, themeId: 1, sessionId: session.id }); + return { database, service, options: { workspaceId: workspace.id, userId, sessionId: session.id } }; +} + +const input = sequence => ({ tool: 'query', result: { columns: [{ name: 'sales' }], rows: [{ sales: sequence }] }, + step: { sequence, title: `Step ${sequence}` } }); + +for (const maxSteps of [2, 3]) { + test(`overlapping process writes preserve ${maxSteps === 2 ? 'the step cap' : 'ordered lineage'}`, async t => { + const { service, options } = fixture(t); + let release, entered, writes = 0; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const adapter = { async createProcessArtifactAsync(args) { + if (++writes === 1) { entered(); await gate; } + return service.createProcessArtifactAsync(args); + } }; + const artifactIds = [], events = []; + const record = createProcessArtifactRecorderAsync({ ...options, service: adapter, maxSteps, + artifactIds, emit: event => events.push(event) }); + assert.equal(await record({ tool: 'catalog', result: {} }), null); + const first = record(input(1)); + await ready; + const second = record(input(2)); + const third = record(input(3)); + release(); + const results = await Promise.all([first, second, third]); + const stored = await service.listArtifactsAsync(options); + assert.equal(stored.length, maxSteps); + const created = results.filter(Boolean); + assert.equal(created.length, maxSteps); + assert.deepEqual(artifactIds, created.map(item => item.id)); + assert.deepEqual(events.map(event => event.artifactId), artifactIds); + for (let index = 1; index < created.length; index++) { + assert.deepEqual(created[index].metadata.inputArtifactIds, [created[index - 1].id]); + } + if (maxSteps === 2) assert.equal(results[2], null); + }); +} + +test('a failed process write does not poison the queue or consume its cap', async t => { + const { service, options } = fixture(t); + let attempts = 0; + const record = createProcessArtifactRecorderAsync({ ...options, maxSteps: 1, + service: { async createProcessArtifactAsync(args) { + if (++attempts === 1) throw new Error('write rejected'); + return service.createProcessArtifactAsync(args); + } }, + }); + const [failed, saved, capped] = await Promise.all([record(input(1)), record(input(2)), record(input(3))]); + assert.equal(failed, null); + assert.ok(saved.id); + assert.equal(capped, null); + assert.equal(attempts, 2); + assert.equal((await service.listArtifactsAsync(options)).length, 1); + assert.deepEqual(saved.metadata.inputArtifactIds, []); +});