john
7f8d692d16
Fix DEV logout cookie clearing, materialize selected MindSpace OA assets before agent runs, and recover zombie runs from synced workspace pages. Add client run wait timeout, harness retry limits, page-edit asset forwarding, and logout/john2 scenario tests. Co-authored-by: Cursor <cursoragent@cursor.com>
377 lines
12 KiB
JavaScript
377 lines
12 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,
|
|
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 });
|
|
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' && !codeRunsEnabled) {
|
|
response.status(403).json({ message: '代码任务灰度未开启' });
|
|
return;
|
|
}
|
|
if (
|
|
toolMode === 'code' &&
|
|
codeRunUserIds.size > 0 &&
|
|
!codeRunUserIds.has(request.currentUser.id)
|
|
) {
|
|
response.status(403).json({ message: '当前用户未开启代码任务灰度' });
|
|
return;
|
|
}
|
|
if (
|
|
toolMode === 'code' &&
|
|
codeRunTaskTypes.size > 0 &&
|
|
(!taskType || !codeRunTaskTypes.has(taskType.toLowerCase()))
|
|
) {
|
|
response.status(403).json({ message: '当前代码任务类型未开启灰度' });
|
|
return;
|
|
}
|
|
if (toolMode === 'code' && requireCodeRunValidation && !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);
|
|
}
|
|
};
|
|
}
|