import assert from 'node:assert/strict'; import crypto from 'node:crypto'; import fs from 'node:fs/promises'; import os from 'node:os'; import path from 'node:path'; import test from 'node:test'; import { assertRequiredImageGenerationCompleted, createAgentRunGateway, } from './agent-run-gateway.mjs'; test('required image generation cannot succeed without a verified raster image_make result', () => { const row = { user_message_json: JSON.stringify({ metadata: { memindRun: { imageGenerationMode: 'required' } }, }), }; assert.throws( () => assertRequiredImageGenerationCompleted(row, null, { generateImage: { called: false, succeeded: false }, }), (error) => error?.code === 'IMAGE_GENERATION_REQUIRED_NOT_CALLED', ); assert.throws( () => assertRequiredImageGenerationCompleted(row, null, { generateImage: { called: true, succeeded: false }, }), (error) => error?.code === 'IMAGE_GENERATION_REQUIRED_MISSING', ); assert.doesNotThrow(() => assertRequiredImageGenerationCompleted(row, null, { generateImage: { called: true, succeeded: true, jobId: 'job-1', mimeType: 'image/webp' }, })); }); function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} } = {}) { const runs = new Map(); const events = []; const sessionFinishedAt = (runId) => { const timestamps = events .filter((event) => event.runId === runId && event.eventType === 'session_finished') .map((event) => Number(event.createdAt ?? 0)); return timestamps.length ? Math.max(...timestamps) : null; }; const isStaleRunningRow = (row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff) => { if (row.status !== 'running' || row.started_at == null) return false; const finishedAt = sessionFinishedAt(row.id); const heartbeatAt = latestHeartbeatAt(row.id); return ( (heartbeatAt == null || Number(heartbeatAt) <= Number(heartbeatCutoff)) && ( Number(row.started_at) <= Number(startedCutoff) || (finishedAt != null && Number(finishedAt) <= Number(sessionFinishedCutoff)) ) ); }; const latestHeartbeatAt = (runId) => { const timestamps = events .filter((event) => event.runId === runId && event.eventType === 'worker_heartbeat') .map((event) => Number(event.createdAt ?? 0)); return timestamps.length ? Math.max(...timestamps) : null; }; return { runs, events, async query(sql, params = []) { if (sql.includes('SELECT * FROM h5_agent_runs WHERE id = ? AND user_id = ?')) { const [id, userId] = params; const row = runs.get(id); return [[row && row.user_id === userId ? row : undefined].filter(Boolean)]; } if (sql.includes('SELECT * FROM h5_agent_runs WHERE user_id = ? AND request_id = ?')) { const [userId, requestId] = params; return [[...runs.values()].filter((row) => row.user_id === userId && row.request_id === requestId)]; } if (sql.includes('agent_session_id = ?') && sql.includes("status NOT IN ('succeeded', 'failed')")) { const [sessionId] = params; const active = [...runs.values()].filter( (row) => row.agent_session_id === sessionId && !['succeeded', 'failed'].includes(row.status), ); return [active.slice(0, 1).map((row) => ({ id: row.id }))]; } if (sql.includes('SELECT * FROM h5_agent_runs WHERE id = ? LIMIT 1')) { return [[runs.get(params[0])].filter(Boolean)]; } if (sql.includes('SELECT status, COUNT(*) AS count')) { const counts = new Map(); for (const row of runs.values()) { if (!['queued', 'running', 'retryable'].includes(row.status)) continue; counts.set(row.status, (counts.get(row.status) ?? 0) + 1); } return [[...counts].map(([status, count]) => ({ status, count }))]; } if (sql.includes('h.latest_heartbeat_at') && sql.includes("WHERE r.status = 'running'") && sql.includes('LIMIT 1')) { return [[...runs.values()] .filter((row) => row.status === 'running') .map((row) => ({ id: row.id, request_id: row.request_id, started_at: row.started_at, updated_at: row.updated_at, attempts: row.attempts, latest_heartbeat_at: latestHeartbeatAt(row.id), })) .sort((a, b) => Number(a.latest_heartbeat_at ?? a.started_at ?? 0) - Number(b.latest_heartbeat_at ?? b.started_at ?? 0)) .slice(0, 1)]; } if (sql.includes('SELECT COUNT(*) AS count') && sql.includes('worker_heartbeat')) { return [[{ count: [...runs.values()] .filter((row) => row.status === 'running' && latestHeartbeatAt(row.id) == null) .length, }]]; } if (sql.includes('SELECT') && sql.includes('session_finished_at') && sql.includes('r.started_at <= ?')) { const [heartbeatCutoff, startedCutoff, sessionFinishedCutoff, limit = 1] = params; return [[...runs.values()] .filter((row) => isStaleRunningRow(row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff)) .sort((a, b) => { const aKey = Number(sessionFinishedAt(a.id) ?? a.started_at ?? 0); const bKey = Number(sessionFinishedAt(b.id) ?? b.started_at ?? 0); return aKey - bKey; }) .slice(0, Number(limit)) .map((row) => ({ id: row.id, user_id: row.user_id, agent_session_id: row.agent_session_id, request_id: row.request_id, started_at: row.started_at, updated_at: row.updated_at, attempts: row.attempts, latest_heartbeat_at: latestHeartbeatAt(row.id), session_finished_at: sessionFinishedAt(row.id), }))]; } if (sql.includes('SELECT') && sql.includes('latest_heartbeat_at') && sql.includes('COALESCE(h.latest_heartbeat_at, r.started_at) <= ?')) { const [cutoff, limit = 1] = params; return [[...runs.values()] .map((row) => ({ row, heartbeatAt: latestHeartbeatAt(row.id) })) .filter(({ row, heartbeatAt }) => ( row.status === 'running' && row.started_at != null && Number(heartbeatAt ?? row.started_at) <= Number(cutoff) )) .sort((a, b) => Number(a.heartbeatAt ?? a.row.started_at) - Number(b.heartbeatAt ?? b.row.started_at)) .slice(0, Number(limit)) .map(({ row, heartbeatAt }) => ({ id: row.id, request_id: row.request_id, started_at: row.started_at, updated_at: row.updated_at, attempts: row.attempts, latest_heartbeat_at: heartbeatAt, }))]; } if (sql.includes('SELECT id') && sql.includes("status IN ('queued', 'retryable')")) { const limit = Number(params[0] ?? 1); return [[...runs.values()] .filter((row) => ['queued', 'retryable'].includes(row.status)) .sort((a, b) => Number(a.updated_at) - Number(b.updated_at)) .slice(0, limit) .map((row) => ({ id: row.id }))]; } if (sql.includes('INSERT INTO h5_agent_runs')) { const [ id, userId, sessionId, requestId, userMessageJson, createdAt, updatedAt, ] = params; runs.set(id, { id, user_id: userId, agent_session_id: sessionId, request_id: requestId, status: 'queued', attempts: 0, user_message_json: userMessageJson, error_message: null, created_at: createdAt, updated_at: updatedAt, started_at: null, completed_at: null, }); return [{ affectedRows: 1 }]; } if (sql.includes('INSERT INTO h5_agent_run_events')) { const [id, runId, eventType, dataJson, createdAt] = params; events.push({ id, runId, eventType, dataJson, createdAt }); return [{ affectedRows: 1 }]; } if (sql.includes('FROM h5_agent_run_events e') && sql.includes('INNER JOIN h5_agent_runs r')) { const [eventId, runId, userId] = params; const row = runs.get(runId); const event = events.find((item) => item.id === eventId && item.runId === runId); if (!row || row.user_id !== userId || !event) { return [[]]; } return [[{ created_at: event.createdAt }]]; } if (sql.includes('FROM h5_agent_run_events') && sql.includes('WHERE run_id = ?')) { const runId = params[0]; let filtered = events .filter((event) => event.runId === runId) .map((event) => ({ id: event.id, event_type: event.eventType, data_json: event.dataJson, created_at: event.createdAt, })) .sort((a, b) => Number(a.created_at) - Number(b.created_at) || String(a.id).localeCompare(String(b.id))); if (sql.includes('created_at > ?')) { const afterCreatedAt = Number(params[1]); filtered = filtered.filter((event) => Number(event.created_at) > afterCreatedAt); const limit = Number(params[2]); return [filtered.slice(0, limit)]; } const limit = Number(params[1]); return [filtered.slice(0, limit)]; } if (sql.includes("SET status = 'running'")) { const [attempts, startedAt, updatedAt, id] = params; const row = runs.get(id); if (!row || !['queued', 'retryable'].includes(row.status)) { return [{ affectedRows: 0 }]; } Object.assign(row, { status: 'running', attempts, started_at: row.started_at ?? startedAt, updated_at: updatedAt, error_message: null, }); return [{ affectedRows: 1 }]; } if (sql.includes("WHERE id = ?") && sql.includes("status = 'running'") && sql.includes('session_finished')) { const [errorMessage, updatedAt, completedAt, id, startedCutoff, sessionFinishedCutoff, heartbeatCutoff] = params; const row = runs.get(id); if (!row || !isStaleRunningRow(row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff)) { return [{ affectedRows: 0 }]; } Object.assign(row, { status: 'failed', error_message: errorMessage, updated_at: updatedAt, completed_at: completedAt, }); return [{ affectedRows: 1 }]; } if (sql.includes("WHERE id = ?") && sql.includes("status = 'running'") && sql.includes('worker_heartbeat')) { const [errorMessage, updatedAt, completedAt, id, cutoff] = params; const row = runs.get(id); if ( !row || row.status !== 'running' || row.started_at == null || Number(latestHeartbeatAt(id) ?? row.started_at) > Number(cutoff) ) { return [{ affectedRows: 0 }]; } Object.assign(row, { status: 'failed', error_message: errorMessage, updated_at: updatedAt, completed_at: completedAt, }); return [{ affectedRows: 1 }]; } if (sql.includes('FROM h5_page_records p') && sql.includes('source_session_id')) { const [userId, sessionId] = params; return [sessionDeliverables[`${userId}:${sessionId}`] ?? []]; } if (sql.includes('FROM h5_page_records p') && sql.includes('auto_synced')) { const userId = params[0]; const sinceMs = params.length > 1 ? Number(params[1]) : null; let rows = workspaceDeliverables[userId] ?? []; if (sinceMs != null) { rows = rows.filter((row) => Number(row.updated_at ?? 0) >= sinceMs); } return [rows]; } if (sql.includes('UPDATE h5_agent_runs SET')) { const setSql = sql.split(' WHERE ')[0]; const columns = [...setSql.matchAll(/([a-z_]+) = \?/g)].map((match) => match[1]); const id = params[columns.length]; const row = runs.get(id); if (!row) return [{ affectedRows: 0 }]; if (sql.includes("status = 'running'") && sql.includes('started_at = COALESCE(started_at, ?)')) { const [attempts, startedAt, updatedAt] = params; Object.assign(row, { status: 'running', attempts, started_at: row.started_at ?? startedAt, updated_at: updatedAt, error_message: null, }); return [{ affectedRows: 1 }]; } const expectedStatus = sql.includes('AND status = ?') ? params[columns.length + 1] : null; if (expectedStatus && row.status !== expectedStatus) return [{ affectedRows: 0 }]; for (let i = 0; i < columns.length; i += 1) { row[columns[i]] = params[i]; } return [{ affectedRows: 1 }]; } throw new Error(`Unhandled SQL: ${sql}`); }, }; } async function waitFor(predicate) { for (let i = 0; i < 50; i += 1) { if (predicate()) return; await new Promise((resolve) => setTimeout(resolve, 5)); } assert.fail('condition was not met'); } test('agent run creation is idempotent by user and request id', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, }); const first = await gateway.createRun('user-1', { requestId: 'req-1', userMessage: { role: 'user', content: [] }, }); const second = await gateway.createRun('user-1', { requestId: 'req-1', userMessage: { role: 'user', content: [] }, }); assert.equal(first.id, second.id); assert.equal(pool.runs.size, 1); assert.equal(second.status, 'queued'); }); test('agent run creation stores code tool metadata in the queued message', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, }); const run = await gateway.createRun('user-1', { requestId: 'req-code', userMessage: { role: 'user', content: [{ type: 'text', text: '修改仓库代码' }], metadata: { source: 'h5' }, }, toolMode: 'code-task', taskType: 'repo_refactor', }); const stored = JSON.parse(pool.runs.get(run.id).user_message_json); assert.equal(stored.metadata.source, 'h5'); assert.deepEqual(stored.metadata.memindRun, { toolMode: 'code', taskType: 'repo_refactor', }); }); test('agent run starts a session and marks submitted reply as succeeded', async () => { const pool = createFakePool(); const submitted = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-1' }; }, async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { submitted.push({ userId, sessionId, requestId, userMessage }); }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-1', userMessage: { role: 'user', content: [{ type: 'text', text: 'hi' }] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.deepEqual(submitted.map((item) => item.sessionId), ['session-1']); assert.equal(pool.runs.get(run.id).attempts, 1); }); test('agent run policy allow path preserves existing routing and submission behavior', async () => { const pool = createFakePool(); const submitted = []; const evaluated = []; const routed = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, systemDisclosurePolicyService: { evaluate(input) { evaluated.push(input); return { action: 'allow', matched: false, enforced: false }; }, }, chatIntentRouter: { async classify(input) { routed.push(input); return { route: 'agent_orchestration', reason: 'existing route' }; }, applyAgentOrchestration(message) { return message; }, }, tkmindProxy: { async startSessionForUser() { return { id: 'session-policy-allow' }; }, async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { submitted.push({ userId, sessionId, requestId, userMessage }); }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-policy-allow', userMessage: { role: 'user', content: [{ type: 'text', text: '请帮我安排明天的计划' }], metadata: { displayText: '请帮我安排明天的计划' }, }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(evaluated.length, 1); assert.equal(routed.length, 1); assert.equal(submitted.length, 1); assert.equal(submitted[0].sessionId, 'session-policy-allow'); assert.equal(submitted[0].userMessage.metadata.displayText, '请帮我安排明天的计划'); }); test('agent run enforced disclosure decision returns deterministic refusal before routing or tools', async () => { const pool = createFakePool(); let routed = 0; let backendCalls = 0; const deterministic = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, systemDisclosurePolicyService: { evaluate() { return { policyId: 'system-disclosure', policyVersion: 7, action: 'refuse', matched: true, enforced: true, reasonCode: 'SYSTEM_TECHNICAL_DISCLOSURE', categories: ['architecture'], responseText: '不提供内部技术信息。', }; }, }, chatIntentRouter: { async classify() { routed += 1; return { route: 'agent_orchestration' }; }, }, directChatService: { async respondDeterministically(input) { deterministic.push(input); await input.onSessionReady('h5direct_policy'); return { sessionId: 'h5direct_policy' }; }, }, tkmindProxy: { async startSessionForUser() { backendCalls += 1; return { id: 'should-not-start' }; }, async submitSessionReplyForUser() { backendCalls += 1; }, }, syncUserPagesOnSuccess: async () => { assert.fail('policy refusal must not enter page delivery'); }, validateRunDeliverables: async () => { assert.fail('policy refusal must not enter deliverable validation'); }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-policy-block', userMessage: { role: 'user', content: [{ type: 'text', text: '生成一个 TKMind 底层架构页面' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(routed, 0); assert.equal(backendCalls, 0); assert.equal(deterministic.length, 1); assert.equal(deterministic[0].reply, '不提供内部技术信息。'); assert.equal(pool.runs.get(run.id).agent_session_id, 'h5direct_policy'); assert.equal( pool.events.some((event) => event.eventType === 'system_disclosure_blocked'), true, ); }); test('agent run awaits session Finish before succeeding when proxy supports it', async () => { const pool = createFakePool(); const awaited = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-finish-1' }; }, async submitSessionReplyAndAwaitFinishForUser(userId, sessionId, requestId, userMessage) { awaited.push({ userId, sessionId, requestId, userMessage }); return { ok: true, finishEvent: { type: 'Finish' }, tokenState: { totalTokens: 9 } }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-finish-1', userMessage: { role: 'user', content: [{ type: 'text', text: '世界杯现在赛况如何' }] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(awaited.length, 1); assert.equal(awaited[0].sessionId, 'session-finish-1'); const finishEvents = pool.events.filter((item) => item.eventType === 'session_finished'); assert.equal(finishEvents.length, 1); }); test('Page Data run fails closed when Finish arrives without a generated page', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-page-data-missing' }; }, async submitSessionReplyAndAwaitFinishForUser() { return { ok: true, finishEvent: { type: 'Finish' } }; }, }, syncUserPagesOnSuccess: async () => ({ pageDataBind: { errors: [] } }), retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-page-data-missing', userMessage: { role: 'user', content: [{ type: 'text', text: '帮我创建调查问卷,保存提交记录并发布页面' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.match(pool.runs.get(run.id).error_message, /未生成可交付页面/); }); test('implicit sticky-note app run fails closed when Apps returns Finish without a public page', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'agent_orchestration', confidence: 0.96, reason: '页面需要数据交互与持久化', suggestedSkill: 'page-data-collect', source: 'rule', }; }, applyAgentOrchestration(message) { return message; }, }, tkmindProxy: { async startSessionForUser() { return { id: 'session-sticky-note-missing' }; }, async submitSessionReplyAndAwaitFinishForUser() { return { ok: true, finishEvent: { type: 'Finish' } }; }, }, syncUserPagesOnSuccess: async () => ({ pageDataBind: { errors: [] } }), retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-sticky-note-missing', userMessage: { role: 'user', content: [{ type: 'text', text: '帮我设计一个便签提醒,可以写便签提交,时间轴来显示' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.match(pool.runs.get(run.id).error_message, /Page Data 任务未生成可交付页面/); assert.ok(pool.events.some((event) => event.eventType === 'intent_routed')); }); test('static page run fails closed when Finish arrives without public HTML', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-public-page-missing' }; }, async submitSessionReplyAndAwaitFinishForUser() { return { ok: true, finishEvent: { type: 'Finish' } }; }, }, syncUserPagesOnSuccess: async () => ({}), retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-public-page-missing', userMessage: { role: 'user', content: [{ type: 'text', text: '帮我做一个秋夜诗的 H5 页面' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.match(pool.runs.get(run.id).error_message, /public HTML 交付物/); }); test('Page Data run succeeds only after a generated session page is detected', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:session-page-data-ready': [{ page_id: 'page-ready', title: '问卷', publication_id: 'pub-ready', publication_status: 'online', public_url: 'http://127.0.0.1:5173/u/john/pages/page-ready', }], }, }); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-page-data-ready' }; }, async submitSessionReplyAndAwaitFinishForUser() { return { ok: true, finishEvent: { type: 'Finish' } }; }, }, syncUserPagesOnSuccess: async () => ({ pageDataBind: { errors: [] } }), retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-page-data-ready', userMessage: { role: 'user', content: [{ type: 'text', text: '创建一个可以保存提交记录的问卷' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); }); test('agent run fails closed when a generated page violates browser storage policy', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:session-browser-storage': [{ page_id: 'page-storage', title: '页面', workspace_relative_path: 'public/page.html', }], }, }); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-browser-storage' }; }, async submitSessionReplyAndAwaitFinishForUser() { return { ok: true, finishEvent: { type: 'Finish' } }; }, }, validateRunDeliverables: async ({ deliverables }) => ({ errors: deliverables.pages.some((page) => page.workspaceRelativePath === 'public/page.html') ? [{ code: 'browser_storage_forbidden', message: 'public/page.html 使用 localStorage' }] : [], }), retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-browser-storage', userMessage: { role: 'user', content: [{ type: 'text', text: '帮我做一个展示页面' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.match(pool.runs.get(run.id).error_message, /页面交付违反数据存储策略/); assert.match(pool.runs.get(run.id).error_message, /localStorage/); }); test('agent run succeeds when Finish is missing but session pages were already created', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:session-deliverable-1': [{ page_id: 'page-front', title: '供应商数据上报', publication_id: 'pub-front', publication_status: 'online', public_url: 'http://127.0.0.1:5173/u/john/pages/page-front', }], }, }); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-deliverable-1' }; }, async submitSessionReplyAndAwaitFinishForUser() { const err = new Error('session event stream ended before Finish'); err.code = 'SESSION_REPLY_INCOMPLETE'; throw err; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-deliverable-recover', userMessage: { role: 'user', content: [{ type: 'text', text: '生成填报系统' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'run_recovered_from_deliverables'), true, ); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'session_finished'), false, ); }); test('static page run succeeds when workspace fallback reports the current HTML path', async () => { let observedRunStartedAtMs = null; const pool = createFakePool({ workspaceDeliverables: { 'user-1': [{ page_id: 'page-static-workspace', title: '都市时尚', publication_id: 'pub-static-workspace', publication_status: 'online', public_url: 'http://127.0.0.1:5173/MindSpace/user-1/public/urban-fashion.html', workspace_relative_path: 'public/urban-fashion.html', updated_at: Date.now(), }], }, }); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-static-workspace' }; }, async submitSessionReplyAndAwaitFinishForUser() { return { ok: true, finishEvent: { type: 'Finish' } }; }, }, // Static-page-publish may have no conversation artifact, but the server // reports the current run's recently modified workspace HTML explicitly. syncUserPagesOnSuccess: async ({ runStartedAtMs }) => { observedRunStartedAtMs = runStartedAtMs; return { pageDataBind: { errors: [] }, pageDataRelativePaths: ['public/urban-fashion.html'], }; }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-static-workspace-page', userMessage: { role: 'user', content: [{ type: 'text', text: '生成一个都市时尚展示页面' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.ok(Number(observedRunStartedAtMs) > 0); }); test('agent run uses direct chat service for eligible chat messages', async () => { const pool = createFakePool(); const directRuns = []; const submitted = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'agent-session-should-not-run' }; }, async submitSessionReplyForUser(...args) { submitted.push(args); }, }, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'direct_chat', confidence: 0.92, reason: '普通问候', source: 'llm', }; }, }, directChatService: { canHandle({ toolMode, userMessage, routingDecision }) { return toolMode === 'chat' && routingDecision === 'direct_chat' && userMessage?.content?.[0]?.text === 'hi'; }, async run(input) { directRuns.push(input); return { sessionId: 'h5direct_session-1', providerId: 'custom_deepseek', model: 'deepseek-chat', billing: { ok: true }, }; }, getStatus() { return { enabled: true }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-direct', userMessage: { role: 'user', content: [{ type: 'text', text: 'hi' }] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(pool.runs.get(run.id).agent_session_id, 'h5direct_session-1'); assert.equal(directRuns.length, 1); assert.equal(submitted.length, 0); assert.equal(directRuns[0].requestId, 'req-direct'); assert.ok(pool.events.some((event) => event.eventType === 'direct_chat_completed')); }); test('agent run uses direct chat on regular agent sessions when llm routes direct_chat', async () => { const pool = createFakePool(); const directRuns = []; const submitted = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async submitSessionReplyForUser(...args) { submitted.push(args); }, }, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'direct_chat', confidence: 0.95, reason: '问答', source: 'llm', }; }, }, directChatService: { canHandle({ routingDecision }) { return routingDecision === 'direct_chat'; }, explainCanHandle({ routingDecision }) { return { ok: routingDecision === 'direct_chat', reason: null }; }, async run(input) { directRuns.push(input); return { sessionId: input.sessionId ?? '20260704_11', providerId: 'custom_deepseek', model: 'deepseek-chat', billing: { ok: true }, }; }, getStatus() { return { enabled: true }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { sessionId: '20260704_11', requestId: 'req-taihu', userMessage: { role: 'user', content: [{ type: 'text', text: '我想对太湖有更多的了解' }], metadata: { displayText: '我想对太湖有更多的了解' }, }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(pool.runs.get(run.id).agent_session_id, '20260704_11'); assert.equal(directRuns.length, 1); assert.equal(submitted.length, 0); assert.ok(pool.events.some((event) => event.eventType === 'direct_chat_completed')); }); test('agent run invalidates portal direct chat snapshot before submitting to goosed', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:20260704_31': [{ page_id: 'page-essay', title: '散文页面', publication_id: 'pub-essay', publication_status: 'online', public_url: 'http://127.0.0.1:5173/MindSpace/user-1/public/essay.html', }], }, }); const submitted = []; const invalidated = []; const gateway = createAgentRunGateway({ pool, userAuth: { async getUserCapabilities() { return { grantedSkills: ['static-page-publish'] }; }, }, tkmindProxy: { async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { submitted.push({ userId, sessionId, requestId, userMessage }); }, }, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'agent_orchestration', confidence: 1, reason: '用户开启深度推理', source: 'rule', }; }, applyAgentOrchestration(userMessage) { return userMessage; }, }, sessionSnapshotService: { async remove(sessionId) { invalidated.push(sessionId); }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { sessionId: '20260704_31', requestId: 'req-essay-page', forceDeepReasoning: true, userMessage: { role: 'user', content: [{ type: 'text', text: '帮我写一个 200 字散文,做成页面' }], metadata: { displayText: '帮我写一个 200 字散文,做成页面' }, }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.deepEqual(invalidated, ['20260704_31']); assert.equal(submitted.length, 1); assert.equal(submitted[0].sessionId, '20260704_31'); }); test('agent run records direct_chat_skipped when execution is unavailable', async () => { const pool = createFakePool(); const submitted = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async submitSessionReplyForUser(...args) { submitted.push(args); }, }, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'direct_chat', confidence: 0.95, reason: '问答', source: 'llm', }; }, }, directChatService: { canHandle() { return false; }, explainCanHandle() { return { ok: false, reason: 'session_not_direct_chat' }; }, getStatus() { return { enabled: true }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { sessionId: '20260704_11', requestId: 'req-skipped', userMessage: { role: 'user', content: [{ type: 'text', text: '你好' }], }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(submitted.length, 1); const skipped = pool.events.find((event) => event.eventType === 'direct_chat_skipped'); const skippedData = typeof skipped?.dataJson === 'string' ? JSON.parse(skipped.dataJson) : skipped?.dataJson; assert.equal(skippedData?.reason, 'session_not_direct_chat'); }); test('agent run falls back to backend session when router is disabled', async () => { const pool = createFakePool(); const submitted = []; const directRuns = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'agent-session-1' }; }, async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { submitted.push({ userId, sessionId, requestId, userMessage }); }, }, directChatService: { canHandle() { return true; }, async run(input) { directRuns.push(input); return { sessionId: 'h5direct_session-1' }; }, getStatus() { return { enabled: true }; }, }, chatIntentRouter: { isEnabled() { return false; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-agent-fallback', userMessage: { role: 'user', content: [{ type: 'text', text: 'hi' }] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(pool.runs.get(run.id).agent_session_id, 'agent-session-1'); assert.equal(directRuns.length, 0); assert.equal(submitted.length, 1); assert.equal(submitted[0].sessionId, 'agent-session-1'); }); test('agent run uses chat intent router to enrich agent orchestration messages', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:agent-session-1': [{ page_id: 'page-router-agent', title: '路由页面', publication_id: 'pub-router-agent', publication_status: 'online', public_url: 'http://127.0.0.1:5173/MindSpace/user-1/public/router.html', }], }, }); const submitted = []; const gateway = createAgentRunGateway({ pool, userAuth: { async getUserCapabilities() { return { grantedSkills: ['static-page-publish'] }; }, }, tkmindProxy: { async startSessionForUser(userId) { assert.equal(userId, 'user-1'); return { id: 'agent-session-1' }; }, async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { submitted.push({ userId, sessionId, requestId, userMessage }); }, }, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'agent_orchestration', confidence: 0.93, reason: '需要生成页面', suggestedSkill: 'static-page-publish', agentBrief: '生成并发布 HTML', source: 'llm', }; }, async resolveAgentMemoryContext() { return { enabled: true, mode: 'shadow', injectionEnabled: false, skipped: false, memories: [{ label: 'preference', text: '用户喜欢完整方案' }], source: 'legacy-conversation-memory', latencyMs: 1, }; }, applyAgentOrchestration(userMessage, classification, { grantedSkills = [], memoryContext = null }) { const displayText = userMessage?.metadata?.displayText ?? userMessage?.content?.[0]?.text ?? ''; assert.equal(memoryContext?.mode, 'shadow'); assert.equal(memoryContext?.injectionEnabled, false); return { ...userMessage, content: [{ type: 'text', text: `【Memind 任务编排】${classification.reason}\n用户任务:${displayText}`, }], metadata: { ...(userMessage.metadata ?? {}), displayText, }, }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-router-agent', userMessage: { role: 'user', content: [{ type: 'text', text: '帮我做一个页面' }], metadata: { displayText: '帮我做一个页面' }, }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(submitted.length, 1); assert.match(submitted[0].userMessage.content[0].text, /Memind 任务编排/); assert.match(submitted[0].userMessage.content[0].text, /帮我做一个页面/); assert.ok(pool.events.some((event) => event.eventType === 'intent_routed')); assert.ok(pool.events.some((event) => event.eventType === 'agent_memory_resolved')); }); test('agent run escalates direct sessions to a new backend session when forced', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:deep-session-1': [{ page_id: 'page-deep', title: '深度页面', publication_id: 'pub-deep', publication_status: 'online', public_url: 'http://127.0.0.1:5173/MindSpace/user-1/public/a.html', }], }, }); const submitted = []; const gateway = createAgentRunGateway({ pool, tkmindProxy: { async startSessionForUser(userId) { assert.equal(userId, 'user-1'); return { id: 'deep-session-1' }; }, async submitSessionReplyForUser(userId, sessionId, requestId, userMessage, options = {}) { submitted.push({ userId, sessionId, requestId, userMessage, options }); }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { sessionId: 'h5direct_existing', requestId: 'req-force-deep', userMessage: { role: 'user', content: [{ type: 'text', text: '帮我生成页面 public/a.html' }] }, forceDeepReasoning: true, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(pool.runs.get(run.id).agent_session_id, 'deep-session-1'); assert.equal(submitted[0].sessionId, 'deep-session-1'); assert.ok(pool.events.some((event) => event.eventType === 'direct_session_escalated_to_deep_reasoning')); }); test('agent run persists direct session transcript before escalating to goosed', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:deep-session-1': [{ page_id: 'page-deep-transcript', title: '深度页面', publication_id: 'pub-deep-transcript', publication_status: 'online', public_url: 'http://127.0.0.1:5173/MindSpace/user-1/public/a.html', }], }, }); const submitted = []; const saved = []; const removed = []; const gateway = createAgentRunGateway({ pool, tkmindProxy: { async startSessionForUser(userId) { assert.equal(userId, 'user-1'); return { id: 'deep-session-1' }; }, async submitSessionReplyForUser(userId, sessionId, requestId, userMessage, options = {}) { submitted.push({ userId, sessionId, requestId, userMessage, options }); }, }, sessionSnapshotService: { async get(sessionId) { if (sessionId !== 'h5direct_existing') return null; return { messages: [ { role: 'user', content: [{ type: 'text', text: '中考政策' }] }, { role: 'assistant', content: [{ type: 'text', text: '政策摘要' }] }, ], }; }, async remove(sessionId) { removed.push(sessionId); }, }, conversationMemoryService: { async saveConversationMessages(sessionId, userId, messages) { saved.push({ sessionId, userId, messages }); return messages; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { sessionId: 'h5direct_existing', requestId: 'req-force-deep-transcript', userMessage: { role: 'user', content: [{ type: 'text', text: '帮我生成页面 public/a.html' }] }, forceDeepReasoning: true, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(saved.length, 1); assert.equal(saved[0].sessionId, 'deep-session-1'); assert.equal(saved[0].messages.length, 2); assert.ok(pool.events.some((event) => event.eventType === 'direct_session_transcript_persisted')); assert.equal(submitted[0].sessionId, 'deep-session-1'); assert.equal(removed.length, 1); assert.equal(removed[0], 'deep-session-1'); }); test('agent run rejects reused goosed session when broker ownership check fails', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, sessionAccess: { enabled: true, async validateOwnership(userId, sessionId) { assert.equal(userId, 'user-1'); assert.equal(sessionId, '20260705_2'); return false; }, }, tkmindProxy: { async submitSessionReplyForUser() { throw new Error('should not submit'); }, }, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'agent_orchestration', confidence: 0.9, reason: 'test', source: 'rule' }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { sessionId: '20260705_2', requestId: 'req-forbidden-session', userMessage: { role: 'user', content: [{ type: 'text', text: 'hello' }] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.match(pool.runs.get(run.id).error_message ?? '', /无权访问该会话/); }); test('agent run persists portal direct snapshot before goosed submit on same session', async () => { const pool = createFakePool({ sessionDeliverables: { 'user-1:20260705_2': [{ page_id: 'page-report', title: '报告页面', publication_id: 'pub-report', publication_status: 'online', public_url: 'http://127.0.0.1:5173/MindSpace/user-1/public/report.html', }], }, }); const submitted = []; const saved = []; const gateway = createAgentRunGateway({ pool, userAuth: { async getUserCapabilities() { return { grantedSkills: ['static-page-publish'] }; }, }, tkmindProxy: { async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { submitted.push({ userId, sessionId, requestId, userMessage }); }, }, sessionSnapshotService: { async get(sessionId) { if (sessionId !== '20260705_2') return null; return { messages: [ { role: 'user', content: [{ type: 'text', text: '深度搜索' }] }, { role: 'assistant', content: [{ type: 'text', text: '搜索结果' }], metadata: { source: 'portal-direct-chat' } }, ], }; }, async remove(sessionId) { assert.equal(sessionId, '20260705_2'); }, }, conversationMemoryService: { async saveConversationMessages(sessionId, userId, messages) { saved.push({ sessionId, userId, messages }); return messages; }, }, chatIntentRouter: { isEnabled() { return true; }, async classify() { return { route: 'agent_orchestration', confidence: 0.93, reason: '需要生成页面', suggestedSkill: 'static-page-publish', source: 'llm', }; }, applyAgentOrchestration(userMessage, classification) { const displayText = userMessage?.content?.[0]?.text ?? ''; return { ...userMessage, content: [{ type: 'text', text: `【Memind 任务编排】${classification.reason}\n用户任务:${displayText}`, }], }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { sessionId: '20260705_2', requestId: 'req-portal-direct-persist', userMessage: { role: 'user', content: [{ type: 'text', text: '生成报告' }] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(saved.length, 1); assert.equal(saved[0].sessionId, '20260705_2'); assert.equal(saved[0].messages.length, 2); assert.ok(pool.events.some((event) => event.eventType === 'portal_direct_transcript_persisted')); assert.equal(submitted.length, 1); assert.equal(submitted[0].sessionId, '20260705_2'); }); test('agent run with code tool mode starts and submits with code policy', async () => { const pool = createFakePool(); const submitted = []; const codePolicy = { extensionOverrides: { aider: { allowed: true } }, }; const gateway = createAgentRunGateway({ pool, userAuth: { async getCodeAgentSessionPolicy(userId) { assert.equal(userId, 'user-1'); return codePolicy; }, }, tkmindProxy: { async startSessionForUser(userId, options = {}) { assert.equal(userId, 'user-1'); assert.equal(options.sessionPolicy, codePolicy); return { id: 'session-code' }; }, async submitSessionReplyForUser(userId, sessionId, requestId, userMessage, options = {}) { submitted.push({ userId, sessionId, requestId, userMessage, options }); }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-code', userMessage: { role: 'user', content: [{ type: 'text', text: 'run code task' }] }, toolMode: 'code', }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.deepEqual(submitted.map((item) => item.options), [{ toolMode: 'code', forceDeepReasoning: false }]); assert.equal(submitted[0].userMessage.metadata.memindRun.toolMode, 'code'); }); test('agent run with enabled tool gateway dispatches code runs outside backend session', async () => { const pool = createFakePool(); const jobs = []; const gateway = createAgentRunGateway({ pool, userAuth: { async resolveWorkingDir(userId) { assert.equal(userId, 'user-1'); return '/tmp/memind-user-1'; }, }, tkmindProxy: { async startSessionForUser() { assert.fail('backend session should not start for external tool gateway run'); }, async submitSessionReplyForUser() { assert.fail('backend reply should not be submitted for external tool gateway run'); }, }, toolGateway: { getStatus() { return { enabled: true, protocol: 'agent-run-v1' }; }, async executeJob(job) { jobs.push(job); return { ok: true, dryRun: true, executor: 'aider' }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-code-tool-gateway', userMessage: { role: 'user', content: [{ type: 'text', text: 'run code task' }] }, toolMode: 'code', taskType: 'small_patch', }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.equal(jobs.length, 1); assert.equal(jobs[0].cwd, '/tmp/memind-user-1'); assert.equal(jobs[0].taskType, 'small_patch'); assert.equal(pool.runs.get(run.id).agent_session_id, null); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'tool_gateway_dispatch'), true, ); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'tool_gateway_result'), true, ); }); test('agent run validates expected tool gateway artifacts before succeeding', async () => { const pool = createFakePool(); const workdir = await fs.mkdtemp(path.join(os.tmpdir(), 'memind-tool-validation-')); const gateway = createAgentRunGateway({ pool, userAuth: { async resolveWorkingDir() { return workdir; }, }, tkmindProxy: { async startSessionForUser() { assert.fail('backend session should not start for external tool gateway run'); }, async submitSessionReplyForUser() { assert.fail('backend reply should not be submitted for external tool gateway run'); }, }, toolGateway: { getStatus() { return { enabled: true, protocol: 'agent-run-v1' }; }, async executeJob() { await fs.writeFile( path.join(workdir, 'RESULT.md'), 'validated artifact from tool gateway\n', 'utf8', ); return { ok: true, dryRun: false, executor: 'aider', exitCode: 0, cwd: workdir, stdout: 'created RESULT.md', stderr: '', }; }, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-code-tool-validation', userMessage: { role: 'user', content: [{ type: 'text', text: 'create validation artifact' }], metadata: { memindRun: { validation: { expectedFile: { path: 'RESULT.md', contains: 'validated artifact', }, }, }, }, }, toolMode: 'code', taskType: 'small_patch', }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); const validationEvent = pool.events.find( (event) => event.runId === run.id && event.eventType === 'tool_gateway_validation', ); assert.ok(validationEvent); assert.equal(JSON.parse(validationEvent.dataJson).expectedFiles[0].path, 'RESULT.md'); }); test('agent run fails non-retryably when tool gateway artifact validation fails', async () => { const pool = createFakePool(); const workdir = await fs.mkdtemp(path.join(os.tmpdir(), 'memind-tool-validation-missing-')); let attempts = 0; const gateway = createAgentRunGateway({ pool, userAuth: { async resolveWorkingDir() { return workdir; }, }, tkmindProxy: { async startSessionForUser() { assert.fail('backend session should not start for external tool gateway run'); }, async submitSessionReplyForUser() { assert.fail('backend reply should not be submitted for external tool gateway run'); }, }, toolGateway: { getStatus() { return { enabled: true, protocol: 'agent-run-v1' }; }, async executeJob() { attempts += 1; return { ok: true, dryRun: false, executor: 'aider', exitCode: 0, cwd: workdir, }; }, }, retryDelaysMs: [0, 0], }); const run = await gateway.createRun('user-1', { requestId: 'req-code-tool-validation-fail', userMessage: { role: 'user', content: [{ type: 'text', text: 'forget to create validation artifact' }], metadata: { memindRun: { validation: { expectedFile: 'MISSING.md', }, }, }, }, toolMode: 'code', taskType: 'small_patch', }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.equal(attempts, 1); assert.equal(pool.runs.get(run.id).attempts, 1); assert.match(pool.runs.get(run.id).error_message, /expected file not found/); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'tool_gateway_validation_failed'), true, ); }); test('agent run retries transient failures and then becomes terminal', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-1' }; }, async submitSessionReplyForUser() { throw new Error('upstream unavailable'); }, }, retryDelaysMs: [0, 0], }); const run = await gateway.createRun('user-1', { requestId: 'req-1', userMessage: { role: 'user', content: [] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.equal(pool.runs.get(run.id).attempts, 2); assert.match(pool.runs.get(run.id).error_message, /upstream unavailable/); }); test('agent run queue limits concurrent execution', async () => { const pool = createFakePool(); let active = 0; let maxActive = 0; const release = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser(_userId) { return { id: `session-${release.length + 1}` }; }, async submitSessionReplyForUser() { active += 1; maxActive = Math.max(maxActive, active); await new Promise((resolve) => release.push(resolve)); active -= 1; }, }, retryDelaysMs: [], maxConcurrentRuns: 1, }); const run1 = await gateway.createRun('user-1', { requestId: 'req-1', userMessage: { role: 'user', content: [] }, }); const run2 = await gateway.createRun('user-1', { requestId: 'req-2', userMessage: { role: 'user', content: [] }, }); await waitFor(() => active === 1 && pool.runs.get(run2.id)?.status === 'queued'); assert.equal(maxActive, 1); assert.equal((await gateway.getQueueStatus()).pendingDispatches, 1); release.shift()(); await waitFor(() => pool.runs.get(run1.id)?.status === 'succeeded' && active === 1); release.shift()(); await waitFor(() => pool.runs.get(run2.id)?.status === 'succeeded'); assert.equal(maxActive, 1); }); test('agent run timeout fails without retrying', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-timeout' }; }, async submitSessionReplyForUser() { await new Promise(() => {}); }, }, retryDelaysMs: [0, 0], maxConcurrentRuns: 1, runTimeoutMs: 5, }); const run = await gateway.createRun('user-1', { requestId: 'req-timeout', userMessage: { role: 'user', content: [] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.equal(pool.runs.get(run.id).attempts, 1); assert.match(pool.runs.get(run.id).error_message, /timed out/); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'timeout'), true, ); }); test('agent run queue status reports active database and local queue state', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, maxConcurrentRuns: 2, runTimeoutMs: 1234, }); await gateway.createRun('user-1', { requestId: 'req-status', userMessage: { role: 'user', content: [] }, }); const status = await gateway.getQueueStatus(); assert.equal(status.maxConcurrentRuns, 2); assert.equal(status.runTimeoutMs, 1234); assert.equal(status.inFlight, 0); assert.equal(status.pendingDispatches, 0); assert.equal(status.statusCounts.queued, 1); }); test('external worker dispatches queued runs through the same queue controls', async () => { const pool = createFakePool(); const submitted = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'session-worker' }; }, async submitSessionReplyForUser(_userId, _sessionId, requestId) { submitted.push(requestId); }, }, autoDispatch: false, retryDelaysMs: [], maxConcurrentRuns: 1, }); const run = await gateway.createRun('user-1', { requestId: 'req-worker', userMessage: { role: 'user', content: [] }, }); assert.equal(pool.runs.get(run.id).status, 'queued'); const result = await gateway.dispatchQueuedRuns({ limit: 10 }); assert.equal(result.dispatched, 1); await waitFor(() => pool.events.some((event) => event.runId === run.id && event.eventType === 'worker_heartbeat')); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); assert.deepEqual(submitted, ['req-worker']); }); test('external worker does not dispatch more runs when local queue is full', async () => { const pool = createFakePool(); const release = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: `session-${release.length + 1}` }; }, async submitSessionReplyForUser() { await new Promise((resolve) => release.push(resolve)); }, }, autoDispatch: false, retryDelaysMs: [], maxConcurrentRuns: 1, }); const run1 = await gateway.createRun('user-1', { requestId: 'req-full-1', userMessage: { role: 'user', content: [] }, }); const run2 = await gateway.createRun('user-1', { requestId: 'req-full-2', userMessage: { role: 'user', content: [] }, }); try { const first = await gateway.dispatchQueuedRuns({ limit: 10 }); assert.equal(first.dispatched, 1); await waitFor(() => pool.runs.get(run1.id)?.status === 'running'); assert.equal((await gateway.getQueueStatus()).inFlight, 1); const second = await gateway.dispatchQueuedRuns({ limit: 10 }); assert.equal(second.dispatched, 0); assert.equal(pool.runs.get(run2.id).status, 'queued'); } finally { while (release.length > 0) release.shift()(); } await waitFor(() => pool.runs.get(run1.id)?.status === 'succeeded'); assert.equal(pool.runs.get(run2.id).status, 'queued'); }); test('stale running recovery dry-run reports rows without mutating them', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, runTimeoutMs: 1000, }); const run = await gateway.createRun('user-1', { requestId: 'req-stale-dry-run', userMessage: { role: 'user', content: [] }, }); Object.assign(pool.runs.get(run.id), { status: 'running', attempts: 1, started_at: Date.now() - 5000, updated_at: Date.now() - 5000, }); const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: true }); assert.equal(result.considered, 1); assert.equal(result.recovered, 0); assert.equal(result.runs[0].id, run.id); assert.equal(pool.runs.get(run.id).status, 'running'); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'stale_recovered'), false, ); }); test('stale running recovery marks old running rows failed with an event', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, runTimeoutMs: 1000, }); const run = await gateway.createRun('user-1', { requestId: 'req-stale-apply', userMessage: { role: 'user', content: [] }, }); Object.assign(pool.runs.get(run.id), { status: 'running', attempts: 1, started_at: Date.now() - 5000, updated_at: Date.now() - 5000, }); const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: false }); assert.equal(result.considered, 1); assert.equal(result.recovered, 1); assert.equal(pool.runs.get(run.id).status, 'failed'); assert.match(pool.runs.get(run.id).error_message, /stale running state/); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'stale_recovered'), true, ); }); test('stale running recovery ignores old runs with a fresh heartbeat', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, runTimeoutMs: 1000, }); const run = await gateway.createRun('user-1', { requestId: 'req-stale-heartbeat-fresh', userMessage: { role: 'user', content: [] }, }); Object.assign(pool.runs.get(run.id), { status: 'running', attempts: 1, started_at: Date.now() - 5000, updated_at: Date.now() - 5000, }); pool.events.push({ id: 'heartbeat-1', runId: run.id, eventType: 'worker_heartbeat', dataJson: JSON.stringify({ attempt: 1 }), createdAt: Date.now(), }); const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: false }); assert.equal(result.considered, 0); assert.equal(result.recovered, 0); assert.equal(pool.runs.get(run.id).status, 'running'); }); test('stale running recovery succeeds when workspace pages exist after sync', async () => { const startedAt = Date.now() - 5000; const pool = createFakePool({ workspaceDeliverables: { 'user-1': [{ page_id: 'page-synced', title: '苏州攻略', publication_id: 'pub-synced', publication_status: 'online', public_url: 'http://127.0.0.1:5173/u/john/pages/page-synced', updated_at: startedAt + 1000, }], }, }); const syncCalls = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, runTimeoutMs: 1000, syncUserPagesOnSuccess: async ({ userId, sessionId, runId }) => { syncCalls.push({ userId, sessionId, runId }); }, }); const run = await gateway.createRun('user-1', { requestId: 'req-stale-deliverable', sessionId: 'session-stale-deliverable', userMessage: { role: 'user', content: [] }, }); Object.assign(pool.runs.get(run.id), { status: 'running', attempts: 1, agent_session_id: 'session-stale-deliverable', started_at: startedAt, updated_at: startedAt, }); const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: false }); assert.equal(result.considered, 1); assert.equal(result.recovered, 1); assert.equal(pool.runs.get(run.id).status, 'succeeded'); assert.equal(syncCalls.length, 1); assert.equal( pool.events.some((event) => event.runId === run.id && event.eventType === 'run_recovered_from_deliverables'), true, ); }); test('queue status reports running heartbeat age and missing heartbeat count', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, runTimeoutMs: 1000, heartbeatMs: 250, }); const runWithHeartbeat = await gateway.createRun('user-1', { requestId: 'req-heartbeat-status-1', userMessage: { role: 'user', content: [] }, }); Object.assign(pool.runs.get(runWithHeartbeat.id), { status: 'running', attempts: 1, started_at: Date.now() - 5000, updated_at: Date.now() - 5000, }); pool.events.push({ id: 'heartbeat-status-1', runId: runWithHeartbeat.id, eventType: 'worker_heartbeat', dataJson: JSON.stringify({ attempt: 1 }), createdAt: Date.now() - 100, }); const runWithoutHeartbeat = await gateway.createRun('user-1', { requestId: 'req-heartbeat-status-2', userMessage: { role: 'user', content: [] }, }); Object.assign(pool.runs.get(runWithoutHeartbeat.id), { status: 'running', attempts: 1, started_at: Date.now() - 2000, updated_at: Date.now() - 2000, }); const status = await gateway.getQueueStatus(); assert.equal(status.heartbeatMs, 250); assert.equal(status.runningWithoutHeartbeatCount, 1); assert.equal(status.latestRunningRun.id, runWithoutHeartbeat.id); assert.equal(status.oldestRunningHeartbeatAt, null); assert.ok(status.oldestRunningHeartbeatAgeMs >= 1900); }); test('listRunEventsForUser replays events after Last-Event-ID cursor', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, }); const run = await gateway.createRun('user-1', { requestId: 'req-replay-1', userMessage: { role: 'user', content: [] }, }); const queuedAt = Number(pool.events.find((event) => event.runId === run.id)?.createdAt ?? 1000); pool.events.push({ id: 'evt-running', runId: run.id, eventType: 'running', dataJson: null, createdAt: queuedAt + 1000, }); pool.events.push({ id: 'evt-snapshot', runId: run.id, eventType: 'run_snapshot', dataJson: JSON.stringify({ run: { ...run, status: 'running', agentSessionId: 'sess-1' } }), createdAt: queuedAt + 2000, }); const full = await gateway.listRunEventsForUser('user-1', run.id); assert.equal(full.events.length, 3); const snapshotEvent = full.events.find((event) => event.eventType === 'run_snapshot'); assert.ok(snapshotEvent); assert.equal(snapshotEvent.data.run.agentSessionId, 'sess-1'); assert.equal(full.cursorMiss, false); const replay = await gateway.listRunEventsForUser('user-1', run.id, { afterEventId: 'evt-running' }); assert.equal(replay.events.length, 1); assert.equal(replay.events[0].id, 'evt-snapshot'); assert.equal(replay.events[0].data.run.agentSessionId, 'sess-1'); }); test('listRunEventsForUser accepts mysql2-parsed JSON objects in data_json', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, }); const run = await gateway.createRun('user-1', { requestId: 'req-replay-json-object', userMessage: { role: 'user', content: [] }, }); const queuedAt = Number(pool.events.find((event) => event.runId === run.id)?.createdAt ?? 1000); pool.events.push({ id: 'evt-snapshot-object', runId: run.id, eventType: 'run_snapshot', dataJson: { run: { ...run, status: 'running', agentSessionId: 'sess-mysql2' } }, createdAt: queuedAt + 1000, }); const batch = await gateway.listRunEventsForUser('user-1', run.id); const snapshotEvent = batch.events.find((event) => event.id === 'evt-snapshot-object'); assert.ok(snapshotEvent); assert.equal(snapshotEvent.data.run.agentSessionId, 'sess-mysql2'); }); test('listRunEventsForUser reports cursorMiss for unknown Last-Event-ID', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, }); const run = await gateway.createRun('user-1', { requestId: 'req-replay-miss', userMessage: { role: 'user', content: [] }, }); const batch = await gateway.listRunEventsForUser('user-1', run.id, { afterEventId: 'missing-cursor' }); assert.equal(batch.cursorMiss, true); assert.ok(batch.events.length >= 1); }); test('markRun appends run_snapshot when MEMIND_RUN_STREAM_REPLAY=1', async () => { const previous = process.env.MEMIND_RUN_STREAM_REPLAY; process.env.MEMIND_RUN_STREAM_REPLAY = '1'; try { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: { async startSessionForUser() { return { id: 'sess-replay-1' }; }, async submitSessionReplyForUser() {}, }, retryDelaysMs: [], }); const run = await gateway.createRun('user-1', { requestId: 'req-replay-snapshot', userMessage: { role: 'user', content: [{ type: 'text', text: 'hello' }] }, }); await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); const snapshots = pool.events.filter((event) => ( event.runId === run.id && event.eventType === 'run_snapshot' )); assert.ok(snapshots.length >= 1); const latestSnapshot = JSON.parse(snapshots.at(-1).dataJson); assert.equal(latestSnapshot.run.status, 'succeeded'); } finally { if (previous === undefined) delete process.env.MEMIND_RUN_STREAM_REPLAY; else process.env.MEMIND_RUN_STREAM_REPLAY = previous; } }); test('createRun rejects with SESSION_RUN_CONFLICT when same session already has active run', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, }); const activeRunId = crypto.randomUUID(); const now = Date.now(); pool.runs.set(activeRunId, { id: activeRunId, user_id: 'user-1', agent_session_id: 'sess-conflict-1', request_id: 'req-active', status: 'running', attempts: 1, user_message_json: '{}', error_message: null, created_at: now, updated_at: now, started_at: now, completed_at: null, }); let conflictErr = null; try { await gateway.createRun('user-1', { sessionId: 'sess-conflict-1', requestId: 'req-conflict-2', userMessage: { role: 'user', content: [{ type: 'text', text: 'second' }] }, }); } catch (err) { conflictErr = err; } assert.ok(conflictErr, 'expected an error for duplicate active session run'); assert.equal(conflictErr.code, 'SESSION_RUN_CONFLICT'); assert.equal(conflictErr.status, 409); pool.runs.get(activeRunId).status = 'succeeded'; const run3 = await gateway.createRun('user-1', { sessionId: 'sess-conflict-1', requestId: 'req-conflict-3', userMessage: { role: 'user', content: [{ type: 'text', text: 'third' }] }, }); assert.equal(run3.requestId, 'req-conflict-3'); }); test('createRun rejects while the same session is finishing page delivery', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, isSessionExternallyBusy: ({ sessionId }) => sessionId === 'sess-repairing', }); await assert.rejects( gateway.createRun('user-1', { sessionId: 'sess-repairing', requestId: 'req-during-repair', userMessage: { role: 'user', content: [{ type: 'text', text: '继续' }] }, }), (err) => err?.code === 'SESSION_RUN_CONFLICT' && err?.status === 409 && /自动修复/.test(err.message), ); assert.equal(pool.runs.size, 0); }); test('createRun does not apply per-session conflict check for direct-chat sessions', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ pool, userAuth: {}, tkmindProxy: {}, autoDispatch: false, }); const activeRunId = crypto.randomUUID(); const now = Date.now(); pool.runs.set(activeRunId, { id: activeRunId, user_id: 'user-1', agent_session_id: 'h5direct_abc', request_id: 'req-active-dc', status: 'running', attempts: 1, user_message_json: '{}', error_message: null, created_at: now, updated_at: now, started_at: now, completed_at: null, }); const run2 = await gateway.createRun('user-1', { sessionId: 'h5direct_abc', requestId: 'req-dc-2', userMessage: { role: 'user', content: [{ type: 'text', text: 'hi2' }] }, }); assert.equal(run2.requestId, 'req-dc-2'); });