john
239c41f935
Persist code-run gates in admin DB and expose them via /auth/status so H5 can honor runtime policy without VITE rebuilds. Co-authored-by: Cursor <cursoragent@cursor.com>
396 lines
13 KiB
JavaScript
396 lines
13 KiB
JavaScript
import { normalizeAgentRunToolMode } from './agent-run-gateway.mjs';
|
|
import { createSessionAccess } from './session-broker.mjs';
|
|
import {
|
|
extractRunFromStreamEvent,
|
|
formatRunStreamSseChunk,
|
|
isRunStreamReplayEnabled,
|
|
isTerminalRunStatus,
|
|
parseRunStreamLastEventId,
|
|
shouldEmitRunUpdateForStreamEvent,
|
|
} from './agent-run-stream.mjs';
|
|
import { wrapRunStreamPayload, writeSseErrorAndEnd } from './sse-event-taxonomy.mjs';
|
|
|
|
function envFlag(value) {
|
|
return ['1', 'true', 'yes', 'on'].includes(String(value ?? '').trim().toLowerCase());
|
|
}
|
|
|
|
function parseUserIdSet(value) {
|
|
return new Set(
|
|
String(value ?? '')
|
|
.split(',')
|
|
.map((item) => item.trim())
|
|
.filter(Boolean),
|
|
);
|
|
}
|
|
|
|
function parseTaskTypeSet(value) {
|
|
return new Set(
|
|
String(value ?? '')
|
|
.split(',')
|
|
.map((item) => item.trim().toLowerCase())
|
|
.filter(Boolean),
|
|
);
|
|
}
|
|
|
|
function hasExpectedFileValidation(userMessage) {
|
|
const metadata = userMessage?.metadata;
|
|
const runMetadata = metadata?.memindRun ?? metadata?.agentRun ?? {};
|
|
const validation = runMetadata.validation ?? metadata?.toolGatewayValidation;
|
|
if (!validation || typeof validation !== 'object' || Array.isArray(validation)) return false;
|
|
const expectedFile = validation.expectedFile ?? validation.expectedPath;
|
|
if (typeof expectedFile === 'string' && expectedFile.trim()) return true;
|
|
if (expectedFile && typeof expectedFile === 'object' && !Array.isArray(expectedFile)) {
|
|
const filePath = String(expectedFile.path ?? expectedFile.file ?? expectedFile.relativePath ?? '').trim();
|
|
if (filePath) return true;
|
|
}
|
|
if (!Array.isArray(validation.expectedFiles)) return false;
|
|
return validation.expectedFiles.some((item) => {
|
|
if (typeof item === 'string') return Boolean(item.trim());
|
|
if (!item || typeof item !== 'object' || Array.isArray(item)) return false;
|
|
return Boolean(String(item.path ?? item.file ?? item.relativePath ?? '').trim());
|
|
});
|
|
}
|
|
|
|
export function createPostAgentRunsHandler({
|
|
userAuth,
|
|
sessionAccess = null,
|
|
agentRunGateway,
|
|
mindSpaceAssetAgent = null,
|
|
codeRunPolicyService = null,
|
|
codeRunsEnabled = envFlag(process.env.MEMIND_AGENT_CODE_RUNS_ENABLED),
|
|
codeRunUserIds = parseUserIdSet(process.env.MEMIND_AGENT_CODE_RUNS_USER_IDS),
|
|
codeRunTaskTypes = parseTaskTypeSet(process.env.MEMIND_AGENT_CODE_RUN_TASK_TYPES),
|
|
requireCodeRunValidation = envFlag(process.env.MEMIND_AGENT_CODE_RUNS_REQUIRE_VALIDATION),
|
|
}) {
|
|
const sessionStore = sessionAccess ?? createSessionAccess({ userAuth, enabled: false });
|
|
|
|
async function resolveCodeRunPolicy(userId) {
|
|
if (codeRunPolicyService?.getEffectivePolicy) {
|
|
return codeRunPolicyService.getEffectivePolicy(userId);
|
|
}
|
|
return {
|
|
source: 'env',
|
|
enabled: codeRunsEnabled,
|
|
clientEnabled: codeRunsEnabled,
|
|
generalAutodetect: false,
|
|
pageDataDevAutodetect: false,
|
|
requireValidation: requireCodeRunValidation,
|
|
userAllowlist: [...codeRunUserIds],
|
|
taskTypeAllowlist: [...codeRunTaskTypes],
|
|
userAllowed:
|
|
!codeRunUserIds.size || codeRunUserIds.has(String(userId ?? '').trim()),
|
|
};
|
|
}
|
|
|
|
return async function postAgentRuns(request, response) {
|
|
try {
|
|
const sessionId = String(request.body?.session_id ?? '').trim() || null;
|
|
const requestId = String(request.body?.request_id ?? '').trim();
|
|
const userMessage = request.body?.user_message ?? null;
|
|
const rawToolMode = request.body?.tool_mode ?? request.body?.toolMode ?? 'chat';
|
|
const taskType = String(request.body?.task_type ?? request.body?.taskType ?? '').trim() || null;
|
|
const forceDeepReasoning = request.body?.force_deep_reasoning === true || request.body?.forceDeepReasoning === true;
|
|
if (!requestId) {
|
|
response.status(400).json({ message: '缺少 request_id' });
|
|
return;
|
|
}
|
|
if (!userMessage) {
|
|
response.status(400).json({ message: '缺少 user_message' });
|
|
return;
|
|
}
|
|
let toolMode = 'chat';
|
|
try {
|
|
toolMode = normalizeAgentRunToolMode(rawToolMode);
|
|
} catch (err) {
|
|
response.status(400).json({
|
|
message: err instanceof Error ? err.message : '不支持的 tool_mode',
|
|
});
|
|
return;
|
|
}
|
|
if (toolMode === 'code') {
|
|
const codeRunPolicy = await resolveCodeRunPolicy(request.currentUser.id);
|
|
if (!codeRunPolicy.enabled) {
|
|
response.status(403).json({ message: '代码任务灰度未开启' });
|
|
return;
|
|
}
|
|
if (!codeRunPolicy.userAllowed) {
|
|
response.status(403).json({ message: '当前用户未开启代码任务灰度' });
|
|
return;
|
|
}
|
|
const taskTypeAllowlist = codeRunPolicy.taskTypeAllowlist ?? [];
|
|
if (
|
|
taskTypeAllowlist.length > 0 &&
|
|
(!taskType || !taskTypeAllowlist.map((item) => String(item).toLowerCase()).includes(taskType.toLowerCase()))
|
|
) {
|
|
response.status(403).json({ message: '当前代码任务类型未开启灰度' });
|
|
return;
|
|
}
|
|
if (codeRunPolicy.requireValidation && !hasExpectedFileValidation(userMessage)) {
|
|
response.status(400).json({ message: '代码任务必须声明产物校验规则' });
|
|
return;
|
|
}
|
|
}
|
|
if (sessionId) {
|
|
const owns = await sessionStore.validateOwnership(request.currentUser.id, sessionId);
|
|
if (!owns) {
|
|
response.status(403).json({ message: '无权访问该会话' });
|
|
return;
|
|
}
|
|
}
|
|
const selectedAssetIds = [
|
|
...new Set(
|
|
(Array.isArray(request.body?.selected_asset_ids)
|
|
? request.body.selected_asset_ids
|
|
: Array.isArray(request.body?.selectedAssetIds)
|
|
? request.body.selectedAssetIds
|
|
: []
|
|
)
|
|
.map((item) => String(item ?? '').trim())
|
|
.filter(Boolean),
|
|
),
|
|
];
|
|
if (
|
|
selectedAssetIds.length > 0 &&
|
|
mindSpaceAssetAgent?.materializeAssetsForSession
|
|
) {
|
|
try {
|
|
await mindSpaceAssetAgent.materializeAssetsForSession({
|
|
sessionId,
|
|
userId: request.currentUser.id,
|
|
assetIds: selectedAssetIds,
|
|
});
|
|
} catch (err) {
|
|
response.status(400).json({
|
|
message: err instanceof Error ? err.message : '勾选资料落盘失败',
|
|
code: err?.code ?? 'asset_materialize_failed',
|
|
});
|
|
return;
|
|
}
|
|
}
|
|
const run = await agentRunGateway.createRun(request.currentUser.id, {
|
|
sessionId,
|
|
requestId,
|
|
userMessage,
|
|
toolMode,
|
|
taskType,
|
|
...(forceDeepReasoning ? { forceDeepReasoning: true } : {}),
|
|
});
|
|
response.status(202).json({ run });
|
|
} catch (err) {
|
|
if (err?.code === 'SESSION_RUN_CONFLICT') {
|
|
response.status(409).json({
|
|
message: err.message,
|
|
code: 'SESSION_RUN_CONFLICT',
|
|
});
|
|
return;
|
|
}
|
|
response.status(500).json({
|
|
message: err instanceof Error ? err.message : '创建任务失败',
|
|
});
|
|
}
|
|
};
|
|
}
|
|
|
|
export function createGetAgentRunHandler({ agentRunGateway }) {
|
|
return async function getAgentRun(request, response) {
|
|
try {
|
|
const run = await agentRunGateway.getRunForUser(
|
|
request.currentUser.id,
|
|
request.params.runId,
|
|
);
|
|
if (!run) {
|
|
response.status(404).json({ message: '任务不存在' });
|
|
return;
|
|
}
|
|
if (run.status !== 'succeeded' && run.status !== 'failed') {
|
|
agentRunGateway.dispatchRun(run.id);
|
|
}
|
|
response.json({ run });
|
|
} catch (err) {
|
|
response.status(500).json({
|
|
message: err instanceof Error ? err.message : '读取任务失败',
|
|
});
|
|
}
|
|
};
|
|
}
|
|
|
|
function createRunStreamWriter(response, { wrapPayload = true } = {}) {
|
|
let lastSentEventId = null;
|
|
let lastRunPayload = null;
|
|
|
|
const writeChunk = (eventId, eventName, data) => {
|
|
const payload = wrapPayload ? wrapRunStreamPayload(eventName, data) : data;
|
|
response.write(formatRunStreamSseChunk({ id: eventId, event: eventName, data: payload }));
|
|
if (eventId) lastSentEventId = eventId;
|
|
};
|
|
|
|
const emitRunIfChanged = (run, eventId = null) => {
|
|
const nextPayload = JSON.stringify(run);
|
|
if (nextPayload === lastRunPayload) return false;
|
|
lastRunPayload = nextPayload;
|
|
writeChunk(eventId, 'run', { run });
|
|
return true;
|
|
};
|
|
|
|
return {
|
|
getLastSentEventId: () => lastSentEventId,
|
|
setLastSentEventId: (value) => {
|
|
lastSentEventId = value;
|
|
},
|
|
emitRunIfChanged,
|
|
writeError: (message) => {
|
|
writeChunk(null, 'error', { message });
|
|
},
|
|
};
|
|
}
|
|
|
|
export function createAgentRunEventsHandler({
|
|
agentRunGateway,
|
|
pollIntervalMs = 1000,
|
|
keepaliveIntervalMs = 20000,
|
|
replayEnabled = isRunStreamReplayEnabled(),
|
|
}) {
|
|
return async function getAgentRunEvents(request, response) {
|
|
const userId = request.currentUser.id;
|
|
const runId = request.params.runId;
|
|
let closed = false;
|
|
let pollTimer = null;
|
|
let keepaliveTimer = null;
|
|
const initialLastEventId = parseRunStreamLastEventId(request.get?.('last-event-id'));
|
|
|
|
const cleanup = () => {
|
|
closed = true;
|
|
if (pollTimer) clearInterval(pollTimer);
|
|
if (keepaliveTimer) clearInterval(keepaliveTimer);
|
|
pollTimer = null;
|
|
keepaliveTimer = null;
|
|
};
|
|
|
|
const firstRun = await agentRunGateway.getRunForUser(userId, runId);
|
|
if (!firstRun) {
|
|
response.status(404).json({ message: '任务不存在' });
|
|
return;
|
|
}
|
|
|
|
response.status(200);
|
|
response.setHeader('Content-Type', 'text/event-stream');
|
|
response.setHeader('Cache-Control', 'no-cache');
|
|
response.setHeader('Connection', 'keep-alive');
|
|
|
|
request.on('close', cleanup);
|
|
keepaliveTimer = setInterval(() => {
|
|
if (!closed) response.write(': keepalive\n\n');
|
|
}, keepaliveIntervalMs);
|
|
|
|
const stream = createRunStreamWriter(response);
|
|
|
|
const publishLegacy = async (prefetchedRun = null) => {
|
|
if (closed) return;
|
|
try {
|
|
const run = prefetchedRun ?? await agentRunGateway.getRunForUser(userId, runId);
|
|
if (!run) {
|
|
stream.writeError('任务不存在');
|
|
cleanup();
|
|
response.end();
|
|
return;
|
|
}
|
|
stream.emitRunIfChanged(run);
|
|
if (!isTerminalRunStatus(run.status)) {
|
|
agentRunGateway.dispatchRun(run.id);
|
|
return;
|
|
}
|
|
cleanup();
|
|
response.end();
|
|
} catch (err) {
|
|
if (response.headersSent) {
|
|
writeSseErrorAndEnd(response, err instanceof Error ? err.message : '读取任务失败');
|
|
} else {
|
|
stream.writeError(err instanceof Error ? err.message : '读取任务失败');
|
|
response.end();
|
|
}
|
|
cleanup();
|
|
}
|
|
};
|
|
|
|
const publishReplay = async () => {
|
|
if (closed) return;
|
|
try {
|
|
const cursor = stream.getLastSentEventId() ?? initialLastEventId;
|
|
const batch = await agentRunGateway.listRunEventsForUser(userId, runId, {
|
|
afterEventId: cursor,
|
|
});
|
|
if (!batch?.run) {
|
|
stream.writeError('任务不存在');
|
|
cleanup();
|
|
response.end();
|
|
return;
|
|
}
|
|
|
|
let latestRun = batch.run;
|
|
for (const event of batch.events) {
|
|
if (closed) return;
|
|
const snapshotRun = extractRunFromStreamEvent(event);
|
|
if (snapshotRun) {
|
|
latestRun = snapshotRun;
|
|
stream.emitRunIfChanged(snapshotRun, event.id);
|
|
continue;
|
|
}
|
|
if (shouldEmitRunUpdateForStreamEvent(event.eventType)) {
|
|
latestRun = await agentRunGateway.getRunForUser(userId, runId) ?? latestRun;
|
|
stream.emitRunIfChanged(latestRun, event.id);
|
|
continue;
|
|
}
|
|
response.write(formatRunStreamSseChunk({
|
|
id: event.id,
|
|
event: 'run_event',
|
|
data: wrapRunStreamPayload('run_event', {
|
|
eventId: event.id,
|
|
eventType: event.eventType,
|
|
data: event.data,
|
|
createdAt: event.createdAt,
|
|
}),
|
|
}));
|
|
stream.setLastSentEventId(event.id);
|
|
}
|
|
|
|
if (batch.cursorMiss && !cursor) {
|
|
stream.emitRunIfChanged(batch.run);
|
|
} else if (batch.cursorMiss && cursor) {
|
|
latestRun = await agentRunGateway.getRunForUser(userId, runId) ?? latestRun;
|
|
stream.emitRunIfChanged(latestRun);
|
|
} else {
|
|
stream.emitRunIfChanged(latestRun);
|
|
}
|
|
|
|
if (!isTerminalRunStatus(latestRun.status)) {
|
|
agentRunGateway.dispatchRun(runId);
|
|
return;
|
|
}
|
|
cleanup();
|
|
response.end();
|
|
} catch (err) {
|
|
if (response.headersSent) {
|
|
writeSseErrorAndEnd(response, err instanceof Error ? err.message : '读取任务失败');
|
|
} else {
|
|
stream.writeError(err instanceof Error ? err.message : '读取任务失败');
|
|
response.end();
|
|
}
|
|
cleanup();
|
|
}
|
|
};
|
|
|
|
const publish = replayEnabled && agentRunGateway.listRunEventsForUser
|
|
? publishReplay
|
|
: publishLegacy;
|
|
|
|
pollTimer = setInterval(() => {
|
|
void publish();
|
|
}, pollIntervalMs);
|
|
if (replayEnabled && agentRunGateway.listRunEventsForUser) {
|
|
await publishReplay();
|
|
} else {
|
|
await publishLegacy(firstRun);
|
|
}
|
|
};
|
|
}
|