john
2223 lines
70 KiB
JavaScript
2223 lines
70 KiB
JavaScript
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 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('generic page request succeeds when the assistant finishes a clarification turn', async () => {
|
|
const pool = createFakePool();
|
|
const gateway = createAgentRunGateway({
|
|
pool,
|
|
userAuth: {},
|
|
tkmindProxy: {
|
|
async startSessionForUser() {
|
|
return { id: 'session-public-page-clarification' };
|
|
},
|
|
async submitSessionReplyAndAwaitFinishForUser() {
|
|
return { ok: true, finishEvent: { type: 'Finish' } };
|
|
},
|
|
},
|
|
syncUserPagesOnSuccess: async () => ({}),
|
|
retryDelaysMs: [],
|
|
});
|
|
|
|
const run = await gateway.createRun('user-1', {
|
|
requestId: 'req-public-page-clarification',
|
|
userMessage: {
|
|
role: 'user',
|
|
content: [{ type: 'text', text: '【TKMind 路由提示】使用 static-page-publish\n帮我生成一个页面吧' }],
|
|
metadata: { displayText: '帮我生成一个页面吧' },
|
|
},
|
|
});
|
|
|
|
await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
|
|
assert.equal(pool.runs.get(run.id).error_message, null);
|
|
});
|
|
|
|
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');
|
|
});
|