john
1830 lines
65 KiB
JavaScript
1830 lines
65 KiB
JavaScript
import crypto from 'node:crypto';
|
||
import fs from 'node:fs/promises';
|
||
import path from 'node:path';
|
||
import { isRunStreamReplayEnabled } from './agent-run-stream.mjs';
|
||
import { isDirectChatSessionId } from './direct-chat-service.mjs';
|
||
import { CHAT_INTENT_ROUTE, resolveGatewayAgentSessionId, resolveLegacyRouteFromClassification, logRouterDecisionShadow } from './chat-intent-router.mjs';
|
||
import { resolveSessionAccess } from './session-broker.mjs';
|
||
import {
|
||
loadSnapshotMessages,
|
||
persistSessionTranscriptFromSnapshot,
|
||
persistSessionTranscriptMessages,
|
||
} from './conversation-transcript-persist.mjs';
|
||
import {
|
||
deriveAssistantFacingText,
|
||
deriveUserFacingText,
|
||
} from './conversation-display.mjs';
|
||
import { ensureGooseUserMessageMetadata } from './goose-message.mjs';
|
||
import {
|
||
prepareAndDetectSessionDeliverables,
|
||
SESSION_FINISHED_STALE_GRACE_MS,
|
||
tryRecoverRunFromDeliverables,
|
||
} from './agent-run-deliverable-check.mjs';
|
||
import {
|
||
isGenericPageGenerationRequest,
|
||
isPageDataIntent,
|
||
isPageGenerationIntent,
|
||
} from './chat-skills.mjs';
|
||
|
||
const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000];
|
||
const TERMINAL_STATUSES = new Set(['succeeded', 'failed']);
|
||
const CODE_TOOL_MODES = new Set(['code', 'code-task', 'code_task', 'code-tool', 'code_tool', 'code_tool_task']);
|
||
const RUN_METADATA_KEY = 'memindRun';
|
||
const DEFAULT_MAX_CONCURRENT_RUNS = 1;
|
||
const DEFAULT_RUN_TIMEOUT_MS = 15 * 60 * 1000;
|
||
const DEFAULT_RUN_HEARTBEAT_MS = 30 * 1000;
|
||
const TOOL_GATEWAY_SUMMARY_LIMIT = 4096;
|
||
|
||
function envFlag(value, fallback = false) {
|
||
const raw = String(value ?? '').trim().toLowerCase();
|
||
if (!raw) return fallback;
|
||
return ['1', 'true', 'yes', 'on'].includes(raw);
|
||
}
|
||
|
||
function nowMs() {
|
||
return Date.now();
|
||
}
|
||
|
||
function safeJsonParse(value, fallback = null) {
|
||
try {
|
||
return JSON.parse(value);
|
||
} catch {
|
||
return fallback;
|
||
}
|
||
}
|
||
|
||
function parseDbJsonColumn(value, fallback = null) {
|
||
if (value == null || value === '') return fallback;
|
||
if (typeof value === 'object') return value;
|
||
return safeJsonParse(value, fallback);
|
||
}
|
||
|
||
function serializeMessage(message) {
|
||
return JSON.stringify(message ?? {});
|
||
}
|
||
|
||
function extractRunMessageText(row) {
|
||
const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
|
||
if (typeof message.content === 'string') return message.content;
|
||
if (!Array.isArray(message.content)) return '';
|
||
return message.content
|
||
.filter((item) => item?.type === 'text')
|
||
.map((item) => String(item.text ?? '').trim())
|
||
.filter(Boolean)
|
||
.join('\n');
|
||
}
|
||
|
||
function extractRunDisplayText(row) {
|
||
const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
|
||
const displayText = String(message?.metadata?.displayText ?? '').trim();
|
||
return displayText || deriveUserFacingText(extractRunMessageText(row));
|
||
}
|
||
|
||
function selectedRunSkill(row) {
|
||
const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
|
||
return String(
|
||
message?.metadata?.[RUN_METADATA_KEY]?.selectedChatSkill
|
||
?? message?.metadata?.selectedChatSkill
|
||
?? '',
|
||
).trim();
|
||
}
|
||
|
||
function isPageDeliverableMutationIntent(text) {
|
||
const normalized = String(text ?? '').trim();
|
||
if (!normalized) return false;
|
||
const withoutNegatedActions = normalized.replace(
|
||
/(?:不要|不再|无需|不需要|禁止|仅确认|只确认)[^。！？\n]{0,24}(?:创建|生成|制作|搭建|开发|实现|设计|建立|修改|更新|编辑|发布|上线|写入|写|改)[^。！？\n]{0,16}/gu,
|
||
'',
|
||
);
|
||
return /(?:帮我|请|给我|需要|要)?(?:做一个|做个|创建|生成|制作|搭建|开发|实现|设计|建立|修改|更新|编辑|发布|上线|写入|写一个|改造|新增)/u
|
||
.test(withoutNegatedActions);
|
||
}
|
||
|
||
function messageText(message) {
|
||
if (typeof message?.content === 'string') return message.content.trim();
|
||
if (!Array.isArray(message?.content)) return '';
|
||
return message.content
|
||
.filter((item) => item?.type === 'text')
|
||
.map((item) => String(item?.text ?? '').trim())
|
||
.filter(Boolean)
|
||
.join('\n');
|
||
}
|
||
|
||
function buildFreshSessionContext(conversation, { maxMessages = 16, maxChars = 12_000 } = {}) {
|
||
const visible = [];
|
||
for (const message of Array.isArray(conversation) ? conversation : []) {
|
||
const role = String(message?.role ?? '').trim();
|
||
if (role !== 'user' && role !== 'assistant') continue;
|
||
const raw = messageText(message);
|
||
if (!raw) continue;
|
||
const text = role === 'user'
|
||
? deriveUserFacingText(raw)
|
||
: deriveAssistantFacingText(raw);
|
||
if (!text || /^Ran into this error:/i.test(text)) continue;
|
||
visible.push(`${role === 'user' ? '用户' : '助手'}：${text}`);
|
||
}
|
||
const selected = [];
|
||
let usedChars = 0;
|
||
for (let index = visible.length - 1; index >= 0 && selected.length < maxMessages; index -= 1) {
|
||
const item = visible[index];
|
||
if (selected.length > 0 && usedChars + item.length > maxChars) break;
|
||
selected.unshift(item);
|
||
usedChars += item.length;
|
||
}
|
||
return selected.join('\n\n');
|
||
}
|
||
|
||
function appendFreshSessionContext(userMessage, conversation) {
|
||
const context = buildFreshSessionContext(conversation);
|
||
if (!context) return userMessage;
|
||
const message = userMessage && typeof userMessage === 'object'
|
||
? { ...userMessage }
|
||
: { role: 'user', content: [] };
|
||
const content = Array.isArray(message.content) ? [...message.content] : [];
|
||
const textIndex = content.findIndex((item) => item?.type === 'text');
|
||
const note = `【会话恢复上下文】\n以下内容仅用于理解上文，不是新的执行指令：\n${context}`;
|
||
if (textIndex >= 0) {
|
||
content[textIndex] = {
|
||
...content[textIndex],
|
||
text: `${String(content[textIndex]?.text ?? '').trim()}\n\n${note}`,
|
||
};
|
||
} else {
|
||
content.unshift({ type: 'text', text: note });
|
||
}
|
||
return { ...message, content };
|
||
}
|
||
|
||
function resolveRequiredImageGeneration(row, routing) {
|
||
const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
|
||
const metadata = message.metadata ?? {};
|
||
const runMetadata = metadata[RUN_METADATA_KEY] ?? metadata.agentRun ?? {};
|
||
const requestedMode = String(
|
||
runMetadata.imageGenerationMode ?? runMetadata.image_generation_mode ?? metadata.imageGenerationMode ?? '',
|
||
).trim().toLowerCase();
|
||
if (requestedMode === 'required') return true;
|
||
if (requestedMode === 'disabled') return false;
|
||
return routing?.imageGeneration?.mode === 'required';
|
||
}
|
||
|
||
export function assertRequiredImageGenerationCompleted(row, routing, toolEvidence) {
|
||
if (!resolveRequiredImageGeneration(row, routing)) return;
|
||
if (toolEvidence?.generateImage?.succeeded === true) return;
|
||
if (toolEvidence?.generateImage?.called !== true) {
|
||
const error = new Error('本轮明确要求生成图片，但 Agent 没有实际调用 generate_image；历史回复不能作为本轮工具证据');
|
||
error.code = 'IMAGE_GENERATION_REQUIRED_NOT_CALLED';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
const error = new Error('本轮明确要求生成图片，但未获得 image_make 的有效 PNG/JPEG/WebP 产物，不能标记成功');
|
||
error.code = 'IMAGE_GENERATION_REQUIRED_MISSING';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
|
||
function positiveInteger(value, fallback) {
|
||
const n = Number(value);
|
||
if (!Number.isFinite(n) || n <= 0) return fallback;
|
||
return Math.floor(n);
|
||
}
|
||
|
||
function timeoutError(ms) {
|
||
const err = new Error(`agent run timed out after ${ms}ms`);
|
||
err.code = 'AGENT_RUN_TIMEOUT';
|
||
return err;
|
||
}
|
||
|
||
export function normalizeAgentRunToolMode(value) {
|
||
const normalized = String(value ?? 'chat').trim().toLowerCase();
|
||
if (!normalized || normalized === 'chat') return 'chat';
|
||
if (CODE_TOOL_MODES.has(normalized)) return 'code';
|
||
throw new Error(`不支持的 tool_mode: ${value}`);
|
||
}
|
||
|
||
function normalizeTaskType(value) {
|
||
const normalized = String(value ?? '').trim();
|
||
return normalized || null;
|
||
}
|
||
|
||
function summarizeText(value, limit = TOOL_GATEWAY_SUMMARY_LIMIT) {
|
||
const text = String(value ?? '');
|
||
if (text.length <= limit) return text;
|
||
return text.slice(text.length - limit);
|
||
}
|
||
|
||
export function buildCodeRunCompletionReply(result) {
|
||
const executor = String(result?.executor ?? 'code executor').trim() || 'code executor';
|
||
const output = summarizeText(result?.stdout, 2400).trim();
|
||
return [
|
||
`已由 ${executor} 完成执行，并通过平台文件验收。`,
|
||
output ? `\n${output}` : '',
|
||
].join('').trim();
|
||
}
|
||
|
||
function normalizeExpectedFileCheck(value) {
|
||
if (typeof value === 'string') {
|
||
const expectedPath = value.trim();
|
||
return expectedPath ? { path: expectedPath } : null;
|
||
}
|
||
if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
|
||
const expectedPath = String(value.path ?? value.file ?? value.relativePath ?? '').trim();
|
||
if (!expectedPath) return null;
|
||
const contains = value.contains ?? value.expectedContent ?? value.contentIncludes;
|
||
return {
|
||
path: expectedPath,
|
||
...(contains == null ? {} : { contains: String(contains) }),
|
||
};
|
||
}
|
||
|
||
function normalizeToolGatewayValidation(value) {
|
||
if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
|
||
const checks = [];
|
||
const single = normalizeExpectedFileCheck(value.expectedFile ?? value.expectedPath);
|
||
if (single) checks.push(single);
|
||
if (Array.isArray(value.expectedFiles)) {
|
||
for (const item of value.expectedFiles) {
|
||
const check = normalizeExpectedFileCheck(item);
|
||
if (check) checks.push(check);
|
||
}
|
||
}
|
||
return checks.length > 0 ? { expectedFiles: checks } : null;
|
||
}
|
||
|
||
function resolveValidationPath(cwd, expectedPath) {
|
||
if (!String(cwd ?? '').trim()) {
|
||
const err = new Error('Tool Gateway validation failed: missing working directory');
|
||
err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
|
||
err.retryable = false;
|
||
throw err;
|
||
}
|
||
const root = path.resolve(String(cwd ?? ''));
|
||
const target = path.resolve(root, String(expectedPath ?? ''));
|
||
if (target !== root && !target.startsWith(`${root}${path.sep}`)) {
|
||
const err = new Error(`Tool Gateway validation path escapes working directory: ${expectedPath}`);
|
||
err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
|
||
err.retryable = false;
|
||
throw err;
|
||
}
|
||
return { root, target };
|
||
}
|
||
|
||
async function validateToolGatewayResult({ result, validation, cwd }) {
|
||
if (!validation?.expectedFiles?.length || result?.dryRun) {
|
||
return null;
|
||
}
|
||
const checks = [];
|
||
for (const expected of validation.expectedFiles) {
|
||
const { root, target } = resolveValidationPath(cwd ?? result?.cwd, expected.path);
|
||
let content = '';
|
||
let stat = null;
|
||
try {
|
||
stat = await fs.stat(target);
|
||
content = await fs.readFile(target, 'utf8');
|
||
} catch (cause) {
|
||
const err = new Error(`Tool Gateway validation failed: expected file not found: ${expected.path}`);
|
||
err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
|
||
err.retryable = false;
|
||
err.validation = {
|
||
expectedFile: expected.path,
|
||
cwd: root,
|
||
reason: 'missing_file',
|
||
};
|
||
err.cause = cause;
|
||
throw err;
|
||
}
|
||
if (expected.contains != null && !content.includes(expected.contains)) {
|
||
const err = new Error(`Tool Gateway validation failed: expected content not found in ${expected.path}`);
|
||
err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
|
||
err.retryable = false;
|
||
err.validation = {
|
||
expectedFile: expected.path,
|
||
cwd: root,
|
||
reason: 'missing_content',
|
||
};
|
||
throw err;
|
||
}
|
||
checks.push({
|
||
path: expected.path,
|
||
sizeBytes: Number(stat?.size ?? 0),
|
||
contains: expected.contains == null ? null : true,
|
||
});
|
||
}
|
||
return { expectedFiles: checks };
|
||
}
|
||
|
||
function withRunMetadata(userMessage, { toolMode = 'chat', taskType = null, forceDeepReasoning = false } = {}) {
|
||
const message = (userMessage && typeof userMessage === 'object' && !Array.isArray(userMessage))
|
||
? { ...userMessage }
|
||
: { value: userMessage };
|
||
const metadata = (message.metadata && typeof message.metadata === 'object' && !Array.isArray(message.metadata))
|
||
? { ...message.metadata }
|
||
: {};
|
||
metadata[RUN_METADATA_KEY] = {
|
||
...(metadata[RUN_METADATA_KEY] && typeof metadata[RUN_METADATA_KEY] === 'object'
|
||
? metadata[RUN_METADATA_KEY]
|
||
: {}),
|
||
toolMode: normalizeAgentRunToolMode(toolMode),
|
||
...(taskType ? { taskType } : {}),
|
||
...(forceDeepReasoning ? { forceDeepReasoning: true } : {}),
|
||
};
|
||
return { ...message, metadata };
|
||
}
|
||
|
||
function normalizeSessionMessageCount(value) {
|
||
if (value == null || value === '') return null;
|
||
const parsed = Number(value);
|
||
if (!Number.isFinite(parsed) || parsed < 0) return null;
|
||
return Math.floor(parsed);
|
||
}
|
||
|
||
export function resolveRequiredCodeExecutor(userMessage) {
|
||
const metadata = userMessage?.metadata;
|
||
const runMetadata = metadata?.[RUN_METADATA_KEY] ?? metadata?.agentRun ?? {};
|
||
const executor = String(runMetadata?.executor ?? '').trim().toLowerCase();
|
||
return ['aider', 'openhands'].includes(executor) ? executor : null;
|
||
}
|
||
|
||
export function assertRequiredCodeExecutorAvailable(requiredExecutor, toolGatewayStatus) {
|
||
if (!requiredExecutor) return;
|
||
const executors = Array.isArray(toolGatewayStatus?.executors)
|
||
? toolGatewayStatus.executors.map((item) => String(item).trim().toLowerCase())
|
||
: [];
|
||
if (!toolGatewayStatus?.enabled || !executors.includes(requiredExecutor)) {
|
||
const error = new Error(`必需执行器 ${requiredExecutor} 不可用，任务已停止且不会回退`);
|
||
error.code = 'REQUIRED_EXECUTOR_UNAVAILABLE';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
function getRunOptionsFromMessage(userMessage) {
|
||
const metadata = userMessage?.metadata;
|
||
const runMetadata = metadata?.[RUN_METADATA_KEY] ?? metadata?.agentRun ?? {};
|
||
let toolMode = 'chat';
|
||
try {
|
||
toolMode = normalizeAgentRunToolMode(runMetadata?.toolMode ?? metadata?.toolMode ?? 'chat');
|
||
} catch {
|
||
toolMode = 'chat';
|
||
}
|
||
return {
|
||
toolMode,
|
||
taskType: normalizeTaskType(runMetadata?.taskType ?? metadata?.taskType),
|
||
requiredExecutor: resolveRequiredCodeExecutor(userMessage),
|
||
reviewExecutor: ['aider', 'openhands'].includes(
|
||
String(runMetadata?.reviewExecutor ?? '').trim().toLowerCase(),
|
||
)
|
||
? String(runMetadata.reviewExecutor).trim().toLowerCase()
|
||
: null,
|
||
pageDataAiderWorkflow: runMetadata?.pageDataAiderWorkflow === true,
|
||
forceDeepReasoning: runMetadata?.forceDeepReasoning === true || metadata?.forceDeepReasoning === true,
|
||
validation: normalizeToolGatewayValidation(runMetadata?.validation ?? metadata?.toolGatewayValidation),
|
||
sessionMessageCount: normalizeSessionMessageCount(
|
||
runMetadata?.sessionMessageCount ?? metadata?.sessionMessageCount,
|
||
),
|
||
};
|
||
}
|
||
|
||
export async function collectAiderReviewFiles(cwd, sinceMs = 0) {
|
||
if (!cwd) return [];
|
||
const root = path.resolve(String(cwd));
|
||
const candidates = [
|
||
{ relativeDir: 'public', extensions: new Set(['.html', '.css', '.js']) },
|
||
{ relativeDir: '.mindspace/page-data-policies', extensions: new Set(['.json']) },
|
||
];
|
||
const files = [];
|
||
async function walk(baseDir, relativeDir, extensions) {
|
||
let entries = [];
|
||
try {
|
||
entries = await fs.readdir(baseDir, { withFileTypes: true });
|
||
} catch {
|
||
return;
|
||
}
|
||
for (const entry of entries) {
|
||
if (files.length >= 40) return;
|
||
const absolute = path.join(baseDir, entry.name);
|
||
const relative = path.posix.join(relativeDir.split(path.sep).join('/'), entry.name);
|
||
if (entry.isDirectory()) {
|
||
await walk(absolute, relative, extensions);
|
||
continue;
|
||
}
|
||
if (!entry.isFile() || !extensions.has(path.extname(entry.name).toLowerCase())) continue;
|
||
try {
|
||
const stat = await fs.stat(absolute);
|
||
if (Number(stat.mtimeMs) + 5_000 >= Number(sinceMs || 0)) files.push(relative);
|
||
} catch {
|
||
// File disappeared between directory listing and stat.
|
||
}
|
||
}
|
||
}
|
||
for (const candidate of candidates) {
|
||
await walk(
|
||
path.join(root, candidate.relativeDir),
|
||
candidate.relativeDir,
|
||
candidate.extensions,
|
||
);
|
||
}
|
||
return files;
|
||
}
|
||
|
||
function projectRun(row) {
|
||
if (!row) return null;
|
||
return {
|
||
id: row.id,
|
||
userId: row.user_id,
|
||
sessionId: row.agent_session_id ?? null,
|
||
requestId: row.request_id,
|
||
status: row.status,
|
||
attempts: Number(row.attempts ?? 0),
|
||
error: row.error_message ?? null,
|
||
createdAt: Number(row.created_at ?? 0),
|
||
updatedAt: Number(row.updated_at ?? 0),
|
||
startedAt: row.started_at == null ? null : Number(row.started_at),
|
||
completedAt: row.completed_at == null ? null : Number(row.completed_at),
|
||
};
|
||
}
|
||
|
||
export function createAgentRunGateway({
|
||
pool,
|
||
userAuth,
|
||
sessionAccess = null,
|
||
tkmindProxy,
|
||
toolGateway = null,
|
||
directChatService = null,
|
||
systemDisclosurePolicyService = null,
|
||
chatIntentRouter = null,
|
||
sessionSnapshotService = null,
|
||
conversationMemoryService = null,
|
||
syncUserPagesOnSuccess = null,
|
||
observePersonalMemoryOnSuccess = null,
|
||
isSessionExternallyBusy = null,
|
||
validateRunDeliverables = null,
|
||
retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS,
|
||
autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true),
|
||
maxConcurrentRuns = positiveInteger(
|
||
process.env.MEMIND_AGENT_RUN_QUEUE_CONCURRENCY,
|
||
DEFAULT_MAX_CONCURRENT_RUNS,
|
||
),
|
||
runTimeoutMs = positiveInteger(
|
||
process.env.MEMIND_AGENT_RUN_TIMEOUT_MS,
|
||
DEFAULT_RUN_TIMEOUT_MS,
|
||
),
|
||
heartbeatMs = positiveInteger(
|
||
process.env.MEMIND_AGENT_RUN_HEARTBEAT_MS,
|
||
DEFAULT_RUN_HEARTBEAT_MS,
|
||
),
|
||
}) {
|
||
const sessionStore = resolveSessionAccess({ userAuth, sessionAccess });
|
||
const inFlight = new Set();
|
||
const queuedDispatches = [];
|
||
const queuedDispatchSet = new Set();
|
||
|
||
function enqueueRun(runId) {
|
||
if (!runId || inFlight.has(runId) || queuedDispatchSet.has(runId)) return false;
|
||
queuedDispatchSet.add(runId);
|
||
queuedDispatches.push(runId);
|
||
return true;
|
||
}
|
||
|
||
function nextQueuedRunId() {
|
||
const runId = queuedDispatches.shift();
|
||
if (runId) queuedDispatchSet.delete(runId);
|
||
return runId ?? null;
|
||
}
|
||
|
||
async function appendEvent(runId, type, data = null) {
|
||
await pool.query(
|
||
`INSERT INTO h5_agent_run_events (id, run_id, event_type, data_json, created_at)
|
||
VALUES (?, ?, ?, ?, ?)`,
|
||
[
|
||
crypto.randomUUID(),
|
||
runId,
|
||
type,
|
||
data == null ? null : JSON.stringify(data),
|
||
nowMs(),
|
||
],
|
||
);
|
||
}
|
||
|
||
async function appendRunSnapshot(runId) {
|
||
if (!isRunStreamReplayEnabled()) return;
|
||
const row = await getRunById(runId);
|
||
if (!row) return;
|
||
await appendEvent(runId, 'run_snapshot', { run: projectRun(row) });
|
||
}
|
||
|
||
async function getRunById(runId) {
|
||
const [rows] = await pool.query(
|
||
`SELECT * FROM h5_agent_runs WHERE id = ? LIMIT 1`,
|
||
[runId],
|
||
);
|
||
return rows[0] ?? null;
|
||
}
|
||
|
||
async function getRunForUser(userId, runId) {
|
||
const [rows] = await pool.query(
|
||
`SELECT * FROM h5_agent_runs WHERE id = ? AND user_id = ? LIMIT 1`,
|
||
[runId, userId],
|
||
);
|
||
return projectRun(rows[0] ?? null);
|
||
}
|
||
|
||
async function getRunByRequest(userId, requestId) {
|
||
const [rows] = await pool.query(
|
||
`SELECT * FROM h5_agent_runs WHERE user_id = ? AND request_id = ? LIMIT 1`,
|
||
[userId, requestId],
|
||
);
|
||
return rows[0] ?? null;
|
||
}
|
||
|
||
async function createRun(userId, {
|
||
sessionId = null,
|
||
requestId,
|
||
userMessage,
|
||
toolMode = 'chat',
|
||
taskType = null,
|
||
forceDeepReasoning = false,
|
||
}) {
|
||
const normalizedRequestId = String(requestId ?? '').trim();
|
||
if (!normalizedRequestId) {
|
||
throw new Error('缺少 request_id');
|
||
}
|
||
const normalizedToolMode = normalizeAgentRunToolMode(toolMode);
|
||
const normalizedTaskType = normalizeTaskType(taskType);
|
||
const existing = await getRunByRequest(userId, normalizedRequestId);
|
||
if (existing) {
|
||
if (autoDispatch && !TERMINAL_STATUSES.has(existing.status)) dispatchRun(existing.id);
|
||
return projectRun(existing);
|
||
}
|
||
|
||
// Prevent multiple concurrent agent runs on the same Goose session, which would
|
||
// cause replies to arrive out of order and appear garbled in the chat UI.
|
||
if (sessionId && !isDirectChatSessionId(sessionId)) {
|
||
if (typeof isSessionExternallyBusy === 'function' && await isSessionExternallyBusy({
|
||
userId,
|
||
sessionId,
|
||
})) {
|
||
const conflict = new Error('该会话正在完成页面交付或自动修复，请稍候再发送');
|
||
conflict.code = 'SESSION_RUN_CONFLICT';
|
||
conflict.status = 409;
|
||
throw conflict;
|
||
}
|
||
let [activeRows] = await pool.query(
|
||
`SELECT id FROM h5_agent_runs
|
||
WHERE agent_session_id = ? AND status NOT IN ('succeeded', 'failed')
|
||
LIMIT 1`,
|
||
[sessionId],
|
||
);
|
||
if (activeRows.length > 0) {
|
||
await recoverStaleRunningRuns({
|
||
staleMs: runTimeoutMs,
|
||
limit: 1,
|
||
dryRun: false,
|
||
reason: 'create_run_session_conflict_stale_recovery',
|
||
sessionId,
|
||
});
|
||
[activeRows] = await pool.query(
|
||
`SELECT id FROM h5_agent_runs
|
||
WHERE agent_session_id = ? AND status NOT IN ('succeeded', 'failed')
|
||
LIMIT 1`,
|
||
[sessionId],
|
||
);
|
||
}
|
||
if (activeRows.length > 0) {
|
||
const conflict = new Error('该会话有正在处理的任务，请等待完成后再发送');
|
||
conflict.code = 'SESSION_RUN_CONFLICT';
|
||
conflict.status = 409;
|
||
throw conflict;
|
||
}
|
||
}
|
||
|
||
const runId = crypto.randomUUID();
|
||
const createdAt = nowMs();
|
||
const runMessage = withRunMetadata(userMessage, {
|
||
toolMode: normalizedToolMode,
|
||
taskType: normalizedTaskType,
|
||
forceDeepReasoning,
|
||
});
|
||
await pool.query(
|
||
`INSERT INTO h5_agent_runs
|
||
(id, user_id, agent_session_id, request_id, status, attempts,
|
||
user_message_json, error_message, created_at, updated_at, started_at, completed_at)
|
||
VALUES (?, ?, ?, ?, 'queued', 0, ?, NULL, ?, ?, NULL, NULL)`,
|
||
[
|
||
runId,
|
||
userId,
|
||
sessionId || null,
|
||
normalizedRequestId,
|
||
serializeMessage(runMessage),
|
||
createdAt,
|
||
createdAt,
|
||
],
|
||
);
|
||
await appendEvent(runId, 'queued', {
|
||
sessionId: sessionId || null,
|
||
toolMode: normalizedToolMode,
|
||
taskType: normalizedTaskType,
|
||
forceDeepReasoning: Boolean(forceDeepReasoning),
|
||
});
|
||
if (autoDispatch) dispatchRun(runId);
|
||
return projectRun(await getRunById(runId));
|
||
}
|
||
|
||
async function prepareRunDeliverables({ userId, sessionId, runId, runStartedAtMs = null }) {
|
||
if (typeof syncUserPagesOnSuccess !== 'function') return;
|
||
return await syncUserPagesOnSuccess({
|
||
userId,
|
||
sessionId,
|
||
runId,
|
||
runStartedAtMs,
|
||
}).catch((err) => {
|
||
console.warn(
|
||
'[AgentRun] prepare deliverables sync failed:',
|
||
err instanceof Error ? err.message : err,
|
||
);
|
||
});
|
||
}
|
||
|
||
async function recoverRunFromDeliverables({
|
||
runId,
|
||
userId,
|
||
sessionId,
|
||
err,
|
||
runStartedAtMs = null,
|
||
requireRecoverableError = true,
|
||
}) {
|
||
const recovered = await tryRecoverRunFromDeliverables({
|
||
pool,
|
||
userId,
|
||
sessionId,
|
||
error: err,
|
||
requireRecoverableError,
|
||
runStartedAtMs,
|
||
prepareDeliverables: ({ userId: uid, sessionId: sid }) =>
|
||
prepareRunDeliverables({ userId: uid, sessionId: sid, runId, runStartedAtMs }),
|
||
});
|
||
if (!recovered) return false;
|
||
await appendEvent(runId, 'run_recovered_from_deliverables', {
|
||
sessionId,
|
||
deliverables: recovered.deliverables,
|
||
originalError: recovered.originalError,
|
||
});
|
||
return true;
|
||
}
|
||
|
||
async function markRun(runId, status, fields = {}, { expectedStatus = null } = {}) {
|
||
const updates = ['status = ?', 'updated_at = ?'];
|
||
const values = [status, nowMs()];
|
||
for (const [key, value] of Object.entries(fields)) {
|
||
updates.push(`${key} = ?`);
|
||
values.push(value);
|
||
}
|
||
values.push(runId);
|
||
const where = expectedStatus
|
||
? 'WHERE id = ? AND status = ?'
|
||
: 'WHERE id = ?';
|
||
if (expectedStatus) values.push(expectedStatus);
|
||
const [result] = await pool.query(
|
||
`UPDATE h5_agent_runs SET ${updates.join(', ')} ${where}`,
|
||
values,
|
||
);
|
||
if (Number(result?.affectedRows ?? 0) === 0) return false;
|
||
await appendEvent(runId, status, fields);
|
||
await appendRunSnapshot(runId);
|
||
return true;
|
||
}
|
||
|
||
function startRunHeartbeat(runId, { attempt }) {
|
||
let stopped = false;
|
||
const writeHeartbeat = async () => {
|
||
if (stopped) return;
|
||
try {
|
||
await appendEvent(runId, 'worker_heartbeat', {
|
||
attempt,
|
||
pid: process.pid,
|
||
heartbeatMs,
|
||
});
|
||
} catch (err) {
|
||
console.error('[AgentRun] worker heartbeat failed:', err instanceof Error ? err.message : err);
|
||
}
|
||
};
|
||
void writeHeartbeat();
|
||
const timer = setInterval(() => {
|
||
void writeHeartbeat();
|
||
}, heartbeatMs);
|
||
timer.unref?.();
|
||
return () => {
|
||
stopped = true;
|
||
clearInterval(timer);
|
||
};
|
||
}
|
||
|
||
async function runWithTimeout(runId, task) {
|
||
let timer = null;
|
||
const timeout = new Promise((_, reject) => {
|
||
timer = setTimeout(() => reject(timeoutError(runTimeoutMs)), runTimeoutMs);
|
||
});
|
||
try {
|
||
return await Promise.race([task(), timeout]);
|
||
} finally {
|
||
if (timer) clearTimeout(timer);
|
||
}
|
||
}
|
||
|
||
async function resolveGrantedSkills(userId) {
|
||
if (!userId) return [];
|
||
if (userAuth?.getUserSkills) {
|
||
const skills = await userAuth.getUserSkills(userId).catch(() => null);
|
||
if (Array.isArray(skills?.grantedSkills)) return skills.grantedSkills;
|
||
}
|
||
if (!userAuth?.getUserCapabilities) return [];
|
||
const caps = await userAuth.getUserCapabilities(userId).catch(() => null);
|
||
return Array.isArray(caps?.grantedSkills) ? caps.grantedSkills : [];
|
||
}
|
||
|
||
async function invalidatePortalDirectChatSnapshot(sessionId) {
|
||
if (!sessionId || isDirectChatSessionId(sessionId) || !sessionSnapshotService?.remove) return;
|
||
await sessionSnapshotService.remove(sessionId).catch(() => {});
|
||
}
|
||
|
||
async function assertOwnedAgentSession(userId, sessionId) {
|
||
if (!sessionStore.enabled || !sessionId || isDirectChatSessionId(sessionId)) return;
|
||
const owns = await sessionStore.validateOwnership(userId, sessionId);
|
||
if (!owns) {
|
||
const err = new Error('无权访问该会话');
|
||
err.code = 'SESSION_FORBIDDEN';
|
||
err.status = 403;
|
||
throw err;
|
||
}
|
||
}
|
||
|
||
async function runRequiredCodeReview({
|
||
row,
|
||
runId,
|
||
userMessage,
|
||
runOptions,
|
||
sessionId,
|
||
}) {
|
||
if (!runOptions.reviewExecutor) return;
|
||
const toolGatewayStatus = toolGateway?.getStatus ? toolGateway.getStatus() : null;
|
||
assertRequiredCodeExecutorAvailable(runOptions.reviewExecutor, toolGatewayStatus);
|
||
const workingDir = userAuth?.resolveWorkingDir
|
||
? await userAuth.resolveWorkingDir(row.user_id)
|
||
: undefined;
|
||
const claimedRun = await getRunById(runId);
|
||
const contextFiles = await collectAiderReviewFiles(
|
||
workingDir,
|
||
claimedRun?.started_at ?? row.started_at ?? 0,
|
||
);
|
||
const reviewMessage = {
|
||
...userMessage,
|
||
content: [{
|
||
type: 'text',
|
||
text: [
|
||
'[Mandatory Aider Page Data review]',
|
||
`Original user request: ${extractRunDisplayText(row)}`,
|
||
`Session: ${sessionId}`,
|
||
'Review the provided workspace files, fix concrete HTML/client-policy defects if needed,',
|
||
'do not recreate PostgreSQL tables or datasets, and update the required validation receipt.',
|
||
'Do not commit, push, publish, or modify files outside the current user workspace.',
|
||
contextFiles.length
|
||
? `Review files:\n${contextFiles.map((item) => `- ${item}`).join('\n')}`
|
||
: 'No recent page files were detected; record that fact in the receipt so platform delivery validation can fail closed.',
|
||
`Receipt requestId: ${row.request_id}`,
|
||
].join('\n'),
|
||
}],
|
||
metadata: {
|
||
...(userMessage?.metadata ?? {}),
|
||
memindRun: {
|
||
...(userMessage?.metadata?.memindRun ?? {}),
|
||
executor: runOptions.reviewExecutor,
|
||
aiderContextFiles: contextFiles,
|
||
},
|
||
},
|
||
};
|
||
await appendEvent(runId, 'required_code_review_dispatch', {
|
||
executor: runOptions.reviewExecutor,
|
||
contextFiles,
|
||
});
|
||
const result = await toolGateway.executeJob({
|
||
runId,
|
||
userId: row.user_id,
|
||
requestId: row.request_id,
|
||
userMessage: reviewMessage,
|
||
taskType: 'page_data_dev',
|
||
cwd: workingDir,
|
||
timeoutMs: runTimeoutMs,
|
||
});
|
||
if (
|
||
String(result?.executor ?? '').trim().toLowerCase() !== runOptions.reviewExecutor
|
||
) {
|
||
const error = new Error(
|
||
`审查执行器不匹配：要求 ${runOptions.reviewExecutor}，实际 ${result?.executor ?? 'unknown'}`,
|
||
);
|
||
error.code = 'REQUIRED_REVIEW_EXECUTOR_MISMATCH';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
await appendEvent(runId, 'required_code_review_result', {
|
||
executor: result.executor ?? null,
|
||
exitCode: result.exitCode ?? null,
|
||
stdoutTail: summarizeText(result.stdout),
|
||
stderrTail: summarizeText(result.stderr),
|
||
});
|
||
const validation = await validateToolGatewayResult({
|
||
result,
|
||
validation: runOptions.validation,
|
||
cwd: workingDir,
|
||
});
|
||
if (validation) {
|
||
await appendEvent(runId, 'required_code_review_validation', validation);
|
||
}
|
||
}
|
||
|
||
async function resolveRunRouting(row, userMessage, runOptions) {
|
||
if (!chatIntentRouter?.classify) return null;
|
||
const enabled = chatIntentRouter.isEnabled
|
||
? await Promise.resolve(chatIntentRouter.isEnabled()).catch(() => false)
|
||
: true;
|
||
if (!enabled) return null;
|
||
const grantedSkills = await resolveGrantedSkills(row.user_id);
|
||
return chatIntentRouter.classify({
|
||
userId: row.user_id,
|
||
userMessage,
|
||
sessionId: row.agent_session_id ?? null,
|
||
sessionMessageCount: runOptions.sessionMessageCount,
|
||
toolMode: runOptions.toolMode,
|
||
forceDeepReasoning: runOptions.forceDeepReasoning,
|
||
grantedSkills,
|
||
});
|
||
}
|
||
|
||
async function executeRun(row, runId) {
|
||
let userMessage = safeJsonParse(row.user_message_json, {});
|
||
const runOptions = getRunOptionsFromMessage(userMessage);
|
||
const toolGatewayStatus = toolGateway?.getStatus ? toolGateway.getStatus() : null;
|
||
assertRequiredCodeExecutorAvailable(
|
||
runOptions.requiredExecutor ?? runOptions.reviewExecutor,
|
||
toolGatewayStatus,
|
||
);
|
||
const effectiveToolMode = runOptions.pageDataAiderWorkflow
|
||
? 'chat'
|
||
: runOptions.toolMode;
|
||
let disclosureDecision = null;
|
||
try {
|
||
disclosureDecision = systemDisclosurePolicyService?.evaluate?.({
|
||
userMessage,
|
||
channel: 'h5',
|
||
userId: row.user_id,
|
||
sessionId: row.agent_session_id ?? null,
|
||
}) ?? null;
|
||
} catch (err) {
|
||
console.warn(
|
||
'[AgentRun] system disclosure policy evaluation failed open:',
|
||
err instanceof Error ? err.message : err,
|
||
);
|
||
}
|
||
if (disclosureDecision?.enforced) {
|
||
if (!directChatService?.respondDeterministically) {
|
||
const error = new Error('System Disclosure Policy refusal service unavailable');
|
||
error.code = 'SYSTEM_DISCLOSURE_REFUSAL_UNAVAILABLE';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
const result = await directChatService.respondDeterministically({
|
||
userId: row.user_id,
|
||
sessionId: row.agent_session_id ?? null,
|
||
requestId: row.request_id,
|
||
userMessage,
|
||
reply: disclosureDecision.responseText,
|
||
metadata: {
|
||
policyId: disclosureDecision.policyId,
|
||
policyVersion: disclosureDecision.policyVersion,
|
||
policyReasonCode: disclosureDecision.reasonCode,
|
||
},
|
||
onSessionReady: async (activeSessionId) => {
|
||
await pool.query(
|
||
`UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
|
||
[activeSessionId, nowMs(), runId],
|
||
);
|
||
await appendRunSnapshot(runId);
|
||
},
|
||
});
|
||
await appendEvent(runId, 'system_disclosure_blocked', {
|
||
policyId: disclosureDecision.policyId,
|
||
policyVersion: disclosureDecision.policyVersion,
|
||
reasonCode: disclosureDecision.reasonCode,
|
||
categories: disclosureDecision.categories,
|
||
channel: 'h5',
|
||
});
|
||
return {
|
||
sessionId: result.sessionId,
|
||
routing: null,
|
||
toolEvidence: null,
|
||
policyBlocked: true,
|
||
};
|
||
}
|
||
const routing = await resolveRunRouting(row, userMessage, runOptions);
|
||
const routingDecision = resolveLegacyRouteFromClassification(routing) ?? routing?.route ?? null;
|
||
let agentMemoryContext = null;
|
||
if (routingDecision === CHAT_INTENT_ROUTE.AGENT && chatIntentRouter?.resolveAgentMemoryContext) {
|
||
const displayText = userMessage?.metadata?.displayText
|
||
?? userMessage?.content?.find?.((item) => item?.type === 'text')?.text
|
||
?? '';
|
||
try {
|
||
agentMemoryContext = await chatIntentRouter.resolveAgentMemoryContext({
|
||
userId: row.user_id,
|
||
sessionId: row.agent_session_id ?? null,
|
||
text: displayText,
|
||
forceDeepReasoning: runOptions.forceDeepReasoning,
|
||
});
|
||
if (agentMemoryContext?.enabled && agentMemoryContext.mode !== 'off') {
|
||
await appendEvent(runId, 'agent_memory_resolved', {
|
||
mode: agentMemoryContext.mode,
|
||
injectionEnabled: Boolean(agentMemoryContext.injectionEnabled),
|
||
source: agentMemoryContext.source ?? null,
|
||
memoryCount: Array.isArray(agentMemoryContext.memories)
|
||
? agentMemoryContext.memories.length
|
||
: 0,
|
||
skipped: Boolean(agentMemoryContext.skipped),
|
||
degraded: Boolean(agentMemoryContext.degraded),
|
||
reason: agentMemoryContext.reason ?? null,
|
||
latencyMs: Number(agentMemoryContext.latencyMs ?? 0),
|
||
});
|
||
}
|
||
} catch (err) {
|
||
console.warn(
|
||
'[AgentRun] agent memory shadow resolve skipped:',
|
||
err instanceof Error ? err.message : err,
|
||
);
|
||
agentMemoryContext = null;
|
||
}
|
||
}
|
||
if (routing) {
|
||
logRouterDecisionShadow(routing, {
|
||
requestId: row.request_id ?? null,
|
||
sessionId: row.agent_session_id ?? null,
|
||
});
|
||
await appendEvent(runId, 'intent_routed', routing);
|
||
if (routingDecision === CHAT_INTENT_ROUTE.AGENT && chatIntentRouter?.applyAgentOrchestration) {
|
||
const grantedSkills = await resolveGrantedSkills(row.user_id);
|
||
userMessage = chatIntentRouter.applyAgentOrchestration(userMessage, routing, {
|
||
grantedSkills,
|
||
memoryContext: agentMemoryContext,
|
||
});
|
||
}
|
||
}
|
||
const preferDirectChat =
|
||
routingDecision === CHAT_INTENT_ROUTE.DIRECT_CHAT ||
|
||
(isDirectChatSessionId(row.agent_session_id ?? null) && !runOptions.forceDeepReasoning);
|
||
const directChatInput = {
|
||
sessionId: row.agent_session_id ?? null,
|
||
toolMode: runOptions.toolMode,
|
||
userMessage,
|
||
routingDecision,
|
||
};
|
||
const canDirectChat = Boolean(
|
||
preferDirectChat && directChatService?.canHandle?.(directChatInput),
|
||
);
|
||
if (preferDirectChat && !canDirectChat) {
|
||
const rejection = directChatService?.explainCanHandle?.(directChatInput);
|
||
await appendEvent(runId, 'direct_chat_skipped', {
|
||
sessionId: row.agent_session_id ?? null,
|
||
routingDecision,
|
||
reason: rejection?.reason ?? 'can_handle_false',
|
||
});
|
||
}
|
||
if (canDirectChat) {
|
||
try {
|
||
const result = await directChatService.run({
|
||
userId: row.user_id,
|
||
sessionId: row.agent_session_id ?? null,
|
||
requestId: row.request_id,
|
||
userMessage,
|
||
routingDecision,
|
||
routingMemory: routing?.memory ?? null,
|
||
onSessionReady: async (activeSessionId) => {
|
||
await pool.query(
|
||
`UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
|
||
[activeSessionId, nowMs(), runId],
|
||
);
|
||
await appendRunSnapshot(runId);
|
||
},
|
||
});
|
||
await appendEvent(runId, 'direct_chat_completed', {
|
||
sessionId: result.sessionId,
|
||
providerId: result.providerId ?? null,
|
||
model: result.model ?? null,
|
||
billed: Boolean(result.billing?.ok),
|
||
});
|
||
await appendRunSnapshot(runId);
|
||
return { sessionId: result.sessionId, routing };
|
||
} catch (err) {
|
||
await appendEvent(runId, 'direct_chat_failed', {
|
||
code: err?.code ?? null,
|
||
message: err instanceof Error ? err.message : String(err),
|
||
});
|
||
if (isDirectChatSessionId(row.agent_session_id)) {
|
||
throw err;
|
||
}
|
||
}
|
||
}
|
||
if (
|
||
runOptions.toolMode === 'code' &&
|
||
!runOptions.pageDataAiderWorkflow &&
|
||
toolGatewayStatus?.enabled
|
||
) {
|
||
const workingDir = userAuth?.resolveWorkingDir
|
||
? await userAuth.resolveWorkingDir(row.user_id)
|
||
: undefined;
|
||
await invalidatePortalDirectChatSnapshot(row.agent_session_id ?? null);
|
||
await appendEvent(runId, 'tool_gateway_dispatch', {
|
||
protocol: toolGatewayStatus.protocol ?? 'agent-run-v1',
|
||
taskType: runOptions.taskType,
|
||
workingDir: workingDir ?? null,
|
||
});
|
||
const result = await toolGateway.executeJob({
|
||
runId,
|
||
userId: row.user_id,
|
||
requestId: row.request_id,
|
||
userMessage,
|
||
taskType: runOptions.taskType,
|
||
cwd: workingDir,
|
||
timeoutMs: runTimeoutMs,
|
||
});
|
||
if (
|
||
runOptions.requiredExecutor &&
|
||
String(result?.executor ?? '').trim().toLowerCase() !== runOptions.requiredExecutor
|
||
) {
|
||
const error = new Error(
|
||
`执行器不匹配：要求 ${runOptions.requiredExecutor}，实际 ${result?.executor ?? 'unknown'}`,
|
||
);
|
||
error.code = 'REQUIRED_EXECUTOR_MISMATCH';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
await appendEvent(runId, 'tool_gateway_result', {
|
||
executor: result.executor ?? null,
|
||
dryRun: Boolean(result.dryRun),
|
||
exitCode: result.exitCode ?? null,
|
||
stdoutTail: summarizeText(result.stdout),
|
||
stderrTail: summarizeText(result.stderr),
|
||
});
|
||
try {
|
||
const validation = await validateToolGatewayResult({
|
||
result,
|
||
validation: runOptions.validation,
|
||
cwd: workingDir,
|
||
});
|
||
if (validation) {
|
||
await appendEvent(runId, 'tool_gateway_validation', validation);
|
||
}
|
||
} catch (err) {
|
||
await appendEvent(runId, 'tool_gateway_validation_failed', {
|
||
code: err?.code ?? null,
|
||
message: err instanceof Error ? err.message : String(err),
|
||
validation: err?.validation ?? null,
|
||
});
|
||
throw err;
|
||
}
|
||
if (runOptions.requiredExecutor) {
|
||
if (!directChatService?.respondDeterministically) {
|
||
const error = new Error('代码任务已执行，但结果回传服务不可用');
|
||
error.code = 'CODE_RUN_RESULT_DELIVERY_UNAVAILABLE';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
const delivery = await directChatService.respondDeterministically({
|
||
userId: row.user_id,
|
||
sessionId: row.agent_session_id ?? null,
|
||
requestId: row.request_id,
|
||
userMessage,
|
||
reply: buildCodeRunCompletionReply(result),
|
||
metadata: {
|
||
source: 'tool-gateway-code-run',
|
||
executor: result.executor ?? null,
|
||
validated: true,
|
||
},
|
||
onSessionReady: async (activeSessionId) => {
|
||
await pool.query(
|
||
`UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
|
||
[activeSessionId, nowMs(), runId],
|
||
);
|
||
await appendRunSnapshot(runId);
|
||
},
|
||
});
|
||
await appendEvent(runId, 'tool_gateway_result_delivered', {
|
||
sessionId: delivery.sessionId,
|
||
executor: result.executor ?? null,
|
||
});
|
||
return { sessionId: delivery.sessionId, routing };
|
||
}
|
||
return { sessionId: row.agent_session_id ?? null, routing };
|
||
}
|
||
|
||
let sessionId = resolveGatewayAgentSessionId({
|
||
agentSessionId: row.agent_session_id ?? null,
|
||
classification: routing,
|
||
forceDeepReasoning: runOptions.forceDeepReasoning,
|
||
});
|
||
let escalatedDirectSessionId = null;
|
||
if (isDirectChatSessionId(sessionId)) {
|
||
escalatedDirectSessionId = sessionId;
|
||
await appendEvent(runId, 'direct_session_escalated_to_deep_reasoning', {
|
||
previousSessionId: sessionId,
|
||
});
|
||
sessionId = null;
|
||
}
|
||
if (!sessionId) {
|
||
const sessionOptions = {};
|
||
if (effectiveToolMode === 'code' && userAuth?.getCodeAgentSessionPolicy) {
|
||
sessionOptions.sessionPolicy = await userAuth.getCodeAgentSessionPolicy(row.user_id);
|
||
}
|
||
const session = await tkmindProxy.startSessionForUser(row.user_id, sessionOptions);
|
||
sessionId = session.id;
|
||
await pool.query(
|
||
`UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
|
||
[sessionId, nowMs(), runId],
|
||
);
|
||
await appendEvent(runId, 'session_started', {
|
||
sessionId,
|
||
toolMode: effectiveToolMode,
|
||
taskType: runOptions.taskType,
|
||
});
|
||
await appendRunSnapshot(runId);
|
||
if (escalatedDirectSessionId) {
|
||
const priorMessages = await loadSnapshotMessages(
|
||
sessionSnapshotService,
|
||
escalatedDirectSessionId,
|
||
);
|
||
const persisted = await persistSessionTranscriptMessages({
|
||
conversationMemoryService,
|
||
sessionId,
|
||
userId: row.user_id,
|
||
messages: priorMessages,
|
||
});
|
||
if (persisted.saved > 0) {
|
||
await appendEvent(runId, 'direct_session_transcript_persisted', {
|
||
previousSessionId: escalatedDirectSessionId,
|
||
sessionId,
|
||
saved: persisted.saved,
|
||
});
|
||
}
|
||
}
|
||
} else {
|
||
await assertOwnedAgentSession(row.user_id, sessionId);
|
||
}
|
||
|
||
const transcriptPersisted = await persistSessionTranscriptFromSnapshot({
|
||
sessionSnapshotService,
|
||
conversationMemoryService,
|
||
sessionId,
|
||
userId: row.user_id,
|
||
});
|
||
if (transcriptPersisted.saved > 0) {
|
||
await appendEvent(runId, 'portal_direct_transcript_persisted', {
|
||
sessionId,
|
||
saved: transcriptPersisted.saved,
|
||
});
|
||
}
|
||
await invalidatePortalDirectChatSnapshot(sessionId);
|
||
let toolEvidence = null;
|
||
const awaitSessionFinish = envFlag(process.env.MEMIND_AGENT_RUN_AWAIT_SESSION_FINISH, true)
|
||
&& effectiveToolMode === 'chat'
|
||
&& typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser === 'function';
|
||
if (
|
||
runOptions.pageDataAiderWorkflow &&
|
||
typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser !== 'function'
|
||
) {
|
||
const error = new Error('Page Data + Aider 工作流需要等待 Page Data Agent 完成');
|
||
error.code = 'PAGE_DATA_AIDER_FINISH_UNAVAILABLE';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
if (awaitSessionFinish) {
|
||
let submitMessage = ensureGooseUserMessageMetadata(userMessage);
|
||
let replacedPoisonedSession = false;
|
||
while (true) {
|
||
try {
|
||
const finish = await tkmindProxy.submitSessionReplyAndAwaitFinishForUser(
|
||
row.user_id,
|
||
sessionId,
|
||
row.request_id,
|
||
submitMessage,
|
||
{
|
||
toolMode: effectiveToolMode,
|
||
forceDeepReasoning: runOptions.forceDeepReasoning,
|
||
timeoutMs: runTimeoutMs,
|
||
},
|
||
);
|
||
await appendEvent(runId, 'session_finished', {
|
||
sessionId,
|
||
tokenState: finish.tokenState ?? null,
|
||
toolCalls: finish.toolEvidence?.calls ?? [],
|
||
generateImage: finish.toolEvidence?.generateImage ?? null,
|
||
});
|
||
toolEvidence = finish.toolEvidence ?? null;
|
||
break;
|
||
} catch (err) {
|
||
if (
|
||
!replacedPoisonedSession
|
||
&& err?.code === 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED'
|
||
) {
|
||
const previousSessionId = sessionId;
|
||
const repairedConversation = Array.isArray(err?.repairedConversation)
|
||
? err.repairedConversation
|
||
: [];
|
||
const replacement = await tkmindProxy.startSessionForUser(row.user_id);
|
||
sessionId = replacement.id;
|
||
replacedPoisonedSession = true;
|
||
await pool.query(
|
||
`UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
|
||
[sessionId, nowMs(), runId],
|
||
);
|
||
const persisted = await persistSessionTranscriptMessages({
|
||
conversationMemoryService,
|
||
sessionId,
|
||
userId: row.user_id,
|
||
messages: repairedConversation,
|
||
});
|
||
await appendEvent(runId, 'poisoned_session_replaced', {
|
||
previousSessionId,
|
||
sessionId,
|
||
repairedMessageCount: repairedConversation.length,
|
||
persistedMessageCount: persisted.saved,
|
||
});
|
||
await appendRunSnapshot(runId);
|
||
await invalidatePortalDirectChatSnapshot(sessionId);
|
||
submitMessage = ensureGooseUserMessageMetadata(
|
||
appendFreshSessionContext(userMessage, repairedConversation),
|
||
);
|
||
continue;
|
||
}
|
||
if (await recoverRunFromDeliverables({
|
||
runId,
|
||
userId: row.user_id,
|
||
sessionId,
|
||
err,
|
||
})) {
|
||
return { sessionId, routing };
|
||
}
|
||
throw err;
|
||
}
|
||
}
|
||
} else {
|
||
await tkmindProxy.submitSessionReplyForUser(
|
||
row.user_id,
|
||
sessionId,
|
||
row.request_id,
|
||
ensureGooseUserMessageMetadata(userMessage),
|
||
{
|
||
toolMode: effectiveToolMode,
|
||
forceDeepReasoning: runOptions.forceDeepReasoning,
|
||
},
|
||
);
|
||
}
|
||
await runRequiredCodeReview({
|
||
row,
|
||
runId,
|
||
userMessage,
|
||
runOptions,
|
||
sessionId,
|
||
});
|
||
return { sessionId, routing, toolEvidence };
|
||
}
|
||
|
||
async function finalizeSuccessfulRun(
|
||
runId,
|
||
row,
|
||
sessionId,
|
||
{ routing = null, toolEvidence = null, policyBlocked = false } = {},
|
||
) {
|
||
if (policyBlocked) {
|
||
await markRun(runId, 'succeeded', {
|
||
agent_session_id: sessionId,
|
||
completed_at: nowMs(),
|
||
error_message: null,
|
||
}, { expectedStatus: 'running' });
|
||
return;
|
||
}
|
||
assertRequiredImageGenerationCompleted(row, routing, toolEvidence);
|
||
// `row` was loaded before this worker claimed the run, so its started_at
|
||
// can still be null. Refresh it before scoping workspace files to the
|
||
// current execution window.
|
||
const claimedRun = await getRunById(runId);
|
||
const runStartedAtMs = claimedRun?.started_at ?? row.started_at ?? null;
|
||
let deliveryResult = null;
|
||
if (typeof syncUserPagesOnSuccess === 'function') {
|
||
deliveryResult = await syncUserPagesOnSuccess({
|
||
userId: row.user_id,
|
||
sessionId,
|
||
runId,
|
||
runStartedAtMs,
|
||
});
|
||
}
|
||
const pageDataErrors = Array.isArray(deliveryResult?.pageDataBind?.errors)
|
||
? deliveryResult.pageDataBind.errors
|
||
: [];
|
||
if (pageDataErrors.length > 0) {
|
||
const error = new Error(`Page Data 页面绑定失败：${pageDataErrors.map((item) => item?.message ?? item?.code ?? 'unknown').join('; ')}`);
|
||
error.code = 'PAGE_DATA_DELIVERY_FAILED';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
const runDisplayText = extractRunDisplayText(row);
|
||
const selectedSkill = selectedRunSkill(row);
|
||
const pageDataIntent = isPageDataIntent(runDisplayText)
|
||
|| selectedSkill === 'page-data-collect'
|
||
|| routing?.suggestedSkill === 'page-data-collect';
|
||
const pageGenerationIntent = isPageGenerationIntent(runDisplayText)
|
||
|| selectedSkill === 'generate-page'
|
||
|| routing?.suggestedSkill === 'static-page-publish';
|
||
// A bare “generate a page” request has no subject to render. The assistant's
|
||
// clarification is a successful conversational turn, not a failed delivery.
|
||
// Once the user supplies a subject, the existing fail-closed guard still applies.
|
||
const requiresPageDeliverable = isPageDeliverableMutationIntent(runDisplayText)
|
||
&& (pageDataIntent || (
|
||
pageGenerationIntent
|
||
&& !isGenericPageGenerationRequest(runDisplayText)
|
||
));
|
||
let deliverables = null;
|
||
if (requiresPageDeliverable || typeof validateRunDeliverables === 'function') {
|
||
const latest = await getRunById(runId);
|
||
// `[]` is a strict allowlist meaning no workspace page may be examined.
|
||
// Preserve it so a run with no current artifact cannot fall back to
|
||
// historical workspace pages; `null` is reserved for legacy callers
|
||
// that provide no scoping information at all.
|
||
const workspaceRelativePaths = Array.isArray(deliveryResult?.pageDataRelativePaths)
|
||
? deliveryResult.pageDataRelativePaths
|
||
: null;
|
||
deliverables = await prepareAndDetectSessionDeliverables({
|
||
pool,
|
||
userId: row.user_id,
|
||
sessionId,
|
||
runStartedAtMs: latest?.started_at ?? runStartedAtMs,
|
||
workspaceRelativePaths,
|
||
});
|
||
if (requiresPageDeliverable && deliverables.pageCount < 1) {
|
||
const error = new Error(
|
||
pageDataIntent
|
||
? 'Page Data 任务未生成可交付页面，不能标记成功'
|
||
: '页面任务未生成 public HTML 交付物，不能标记成功',
|
||
);
|
||
error.code = pageDataIntent
|
||
? 'PAGE_DATA_DELIVERABLE_MISSING'
|
||
: 'PUBLIC_PAGE_DELIVERABLE_MISSING';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
}
|
||
if (typeof validateRunDeliverables === 'function') {
|
||
const validation = await validateRunDeliverables({
|
||
userId: row.user_id,
|
||
sessionId,
|
||
runId,
|
||
deliverables: deliverables ?? { pageCount: 0, publicationCount: 0, pages: [] },
|
||
});
|
||
const errors = Array.isArray(validation?.errors) ? validation.errors : [];
|
||
if (errors.length > 0) {
|
||
const error = new Error(
|
||
`页面交付违反数据存储策略：${errors.map((item) => item?.message ?? item?.code ?? 'unknown').join('; ')}`,
|
||
);
|
||
error.code = 'DELIVERABLE_DATA_STORAGE_FORBIDDEN';
|
||
error.retryable = false;
|
||
throw error;
|
||
}
|
||
}
|
||
const marked = await markRun(runId, 'succeeded', {
|
||
agent_session_id: sessionId,
|
||
completed_at: nowMs(),
|
||
error_message: null,
|
||
}, { expectedStatus: 'running' });
|
||
if (!marked) return false;
|
||
if (typeof observePersonalMemoryOnSuccess === 'function') {
|
||
await observePersonalMemoryOnSuccess({
|
||
userId: row.user_id,
|
||
sessionId,
|
||
runId,
|
||
userMessage: parseDbJsonColumn(row.user_message_json, {}),
|
||
}).catch((err) => {
|
||
console.warn(
|
||
'[AgentRun] personal memory shadow observation failed:',
|
||
err instanceof Error ? err.message : err,
|
||
);
|
||
});
|
||
}
|
||
}
|
||
|
||
async function processRun(runId) {
|
||
const row = await getRunById(runId);
|
||
if (!row || TERMINAL_STATUSES.has(row.status)) return;
|
||
const nextAttempt = Number(row.attempts ?? 0) + 1;
|
||
const [claim] = await pool.query(
|
||
`UPDATE h5_agent_runs
|
||
SET status = 'running', attempts = ?, started_at = COALESCE(started_at, ?), updated_at = ?, error_message = NULL
|
||
WHERE id = ? AND status IN ('queued', 'retryable')`,
|
||
[nextAttempt, nowMs(), nowMs(), runId],
|
||
);
|
||
if (Number(claim?.affectedRows ?? 0) === 0) return;
|
||
await appendEvent(runId, 'running', { attempt: nextAttempt });
|
||
const stopHeartbeat = startRunHeartbeat(runId, { attempt: nextAttempt });
|
||
|
||
try {
|
||
const execution = await runWithTimeout(runId, () => executeRun(row, runId));
|
||
await finalizeSuccessfulRun(runId, row, execution.sessionId, execution);
|
||
} catch (err) {
|
||
const latest = await getRunById(runId);
|
||
const recoverySessionId = latest?.agent_session_id ?? row.agent_session_id ?? null;
|
||
if (err?.code !== 'IMAGE_GENERATION_REQUIRED_MISSING' && await recoverRunFromDeliverables({
|
||
runId,
|
||
userId: row.user_id,
|
||
sessionId: recoverySessionId,
|
||
err,
|
||
runStartedAtMs: latest?.started_at ?? row.started_at ?? null,
|
||
})) {
|
||
await finalizeSuccessfulRun(runId, row, recoverySessionId);
|
||
return;
|
||
}
|
||
const message = err instanceof Error ? err.message : String(err);
|
||
const timedOut = err?.code === 'AGENT_RUN_TIMEOUT';
|
||
if (timedOut) {
|
||
await appendEvent(runId, 'timeout', { timeoutMs: runTimeoutMs });
|
||
}
|
||
const retryable = !timedOut && err?.retryable !== false && nextAttempt < retryDelaysMs.length;
|
||
await markRun(runId, retryable ? 'retryable' : 'failed', {
|
||
error_message: message,
|
||
completed_at: retryable ? null : nowMs(),
|
||
}, { expectedStatus: 'running' });
|
||
if (retryable && autoDispatch) {
|
||
setTimeout(() => dispatchRun(runId), retryDelaysMs[nextAttempt - 1]);
|
||
}
|
||
} finally {
|
||
stopHeartbeat();
|
||
}
|
||
}
|
||
|
||
function drainQueue() {
|
||
while (inFlight.size < maxConcurrentRuns) {
|
||
const runId = nextQueuedRunId();
|
||
if (!runId) break;
|
||
inFlight.add(runId);
|
||
void processRun(runId)
|
||
.finally(() => {
|
||
inFlight.delete(runId);
|
||
drainQueue();
|
||
});
|
||
}
|
||
}
|
||
|
||
function dispatchRun(runId) {
|
||
if (!enqueueRun(runId)) return;
|
||
drainQueue();
|
||
}
|
||
|
||
async function getQueueStatus() {
|
||
const [rows] = await pool.query(
|
||
`SELECT status, COUNT(*) AS count
|
||
FROM h5_agent_runs
|
||
WHERE status IN ('queued', 'running', 'retryable')
|
||
GROUP BY status`,
|
||
);
|
||
const statusCounts = {};
|
||
for (const row of rows) {
|
||
statusCounts[row.status] = Number(row.count ?? 0);
|
||
}
|
||
const [oldestRunningRows] = await pool.query(
|
||
`SELECT
|
||
r.id,
|
||
r.request_id,
|
||
r.started_at,
|
||
r.updated_at,
|
||
r.attempts,
|
||
h.latest_heartbeat_at
|
||
FROM h5_agent_runs r
|
||
LEFT JOIN (
|
||
SELECT run_id, MAX(created_at) AS latest_heartbeat_at
|
||
FROM h5_agent_run_events
|
||
WHERE event_type = 'worker_heartbeat'
|
||
GROUP BY run_id
|
||
) h ON h.run_id = r.id
|
||
WHERE r.status = 'running'
|
||
ORDER BY COALESCE(h.latest_heartbeat_at, r.started_at) ASC
|
||
LIMIT 1`,
|
||
);
|
||
const oldestRunningStartedAt = oldestRunningRows[0]?.started_at == null
|
||
? null
|
||
: Number(oldestRunningRows[0].started_at);
|
||
const oldestRunningAgeMs = oldestRunningStartedAt == null
|
||
? 0
|
||
: Math.max(0, nowMs() - oldestRunningStartedAt);
|
||
const oldestRunningHeartbeatAt = oldestRunningRows[0]?.latest_heartbeat_at == null
|
||
? null
|
||
: Number(oldestRunningRows[0].latest_heartbeat_at);
|
||
const oldestRunningHeartbeatAgeMs = oldestRunningRows[0]
|
||
? Math.max(0, nowMs() - Number(oldestRunningRows[0].latest_heartbeat_at ?? oldestRunningRows[0].started_at ?? nowMs()))
|
||
: 0;
|
||
const [runningWithoutHeartbeatRows] = await pool.query(
|
||
`SELECT COUNT(*) AS count
|
||
FROM h5_agent_runs r
|
||
LEFT JOIN (
|
||
SELECT run_id, MAX(created_at) AS latest_heartbeat_at
|
||
FROM h5_agent_run_events
|
||
WHERE event_type = 'worker_heartbeat'
|
||
GROUP BY run_id
|
||
) h ON h.run_id = r.id
|
||
WHERE r.status = 'running' AND h.latest_heartbeat_at IS NULL`,
|
||
);
|
||
const chatIntentRouterStatus = chatIntentRouter?.getStatus
|
||
? await Promise.resolve(chatIntentRouter.getStatus()).catch((err) => ({
|
||
enabled: false,
|
||
error: err instanceof Error ? err.message : String(err),
|
||
}))
|
||
: null;
|
||
return {
|
||
autoDispatch,
|
||
maxConcurrentRuns,
|
||
runTimeoutMs,
|
||
heartbeatMs,
|
||
inFlight: inFlight.size,
|
||
pendingDispatches: queuedDispatches.length,
|
||
statusCounts,
|
||
oldestRunningStartedAt,
|
||
oldestRunningAgeMs,
|
||
oldestRunningHeartbeatAt,
|
||
oldestRunningHeartbeatAgeMs,
|
||
runningWithoutHeartbeatCount: Number(runningWithoutHeartbeatRows[0]?.count ?? 0),
|
||
latestRunningRun: oldestRunningRows[0] ? {
|
||
id: oldestRunningRows[0].id,
|
||
requestId: oldestRunningRows[0].request_id,
|
||
startedAt: oldestRunningStartedAt,
|
||
updatedAt: oldestRunningRows[0].updated_at == null ? null : Number(oldestRunningRows[0].updated_at),
|
||
attempts: Number(oldestRunningRows[0].attempts ?? 0),
|
||
heartbeatAt: oldestRunningHeartbeatAt,
|
||
heartbeatAgeMs: oldestRunningHeartbeatAgeMs,
|
||
} : null,
|
||
terminalStatuses: [...TERMINAL_STATUSES],
|
||
toolGateway: toolGateway?.getStatus ? toolGateway.getStatus() : null,
|
||
directChat: directChatService?.getStatus ? directChatService.getStatus() : null,
|
||
chatIntentRouter: chatIntentRouterStatus,
|
||
};
|
||
}
|
||
|
||
async function recoverStaleRunningRuns({
|
||
staleMs = runTimeoutMs,
|
||
limit = maxConcurrentRuns,
|
||
dryRun = true,
|
||
reason = 'stale_running_timeout',
|
||
sessionFinishedGraceMs = SESSION_FINISHED_STALE_GRACE_MS,
|
||
sessionId = null,
|
||
} = {}) {
|
||
const normalizedStaleMs = positiveInteger(staleMs, runTimeoutMs);
|
||
const normalizedLimit = positiveInteger(limit, maxConcurrentRuns);
|
||
const normalizedSessionFinishedGraceMs = positiveInteger(
|
||
sessionFinishedGraceMs,
|
||
SESSION_FINISHED_STALE_GRACE_MS,
|
||
);
|
||
const startedCutoff = nowMs() - normalizedStaleMs;
|
||
const sessionFinishedCutoff = nowMs() - normalizedSessionFinishedGraceMs;
|
||
const heartbeatCutoff = nowMs() - normalizedStaleMs;
|
||
const normalizedSessionId = String(sessionId ?? '').trim();
|
||
const [rows] = await pool.query(
|
||
`SELECT
|
||
r.id,
|
||
r.user_id,
|
||
r.agent_session_id,
|
||
r.request_id,
|
||
r.started_at,
|
||
r.updated_at,
|
||
r.attempts,
|
||
h.latest_heartbeat_at,
|
||
sf.session_finished_at
|
||
FROM h5_agent_runs r
|
||
LEFT JOIN (
|
||
SELECT run_id, MAX(created_at) AS latest_heartbeat_at
|
||
FROM h5_agent_run_events
|
||
WHERE event_type = 'worker_heartbeat'
|
||
GROUP BY run_id
|
||
) h ON h.run_id = r.id
|
||
LEFT JOIN (
|
||
SELECT run_id, MAX(created_at) AS session_finished_at
|
||
FROM h5_agent_run_events
|
||
WHERE event_type = 'session_finished'
|
||
GROUP BY run_id
|
||
) sf ON sf.run_id = r.id
|
||
WHERE r.status = 'running'
|
||
AND r.started_at IS NOT NULL
|
||
AND (h.latest_heartbeat_at IS NULL OR h.latest_heartbeat_at <= ?)
|
||
AND (
|
||
r.started_at <= ?
|
||
OR (
|
||
sf.session_finished_at IS NOT NULL
|
||
AND sf.session_finished_at <= ?
|
||
)
|
||
)
|
||
${normalizedSessionId ? 'AND r.agent_session_id = ?' : ''}
|
||
ORDER BY COALESCE(sf.session_finished_at, r.started_at) ASC
|
||
LIMIT ?`,
|
||
[
|
||
heartbeatCutoff,
|
||
startedCutoff,
|
||
sessionFinishedCutoff,
|
||
...(normalizedSessionId ? [normalizedSessionId] : []),
|
||
normalizedLimit,
|
||
],
|
||
);
|
||
const recovered = [];
|
||
for (const row of rows) {
|
||
const ageMs = Math.max(0, nowMs() - Number(row.started_at ?? 0));
|
||
const heartbeatAt = row.latest_heartbeat_at == null ? null : Number(row.latest_heartbeat_at);
|
||
const heartbeatAgeMs = Math.max(0, nowMs() - Number(row.latest_heartbeat_at ?? row.started_at ?? 0));
|
||
const sessionFinishedAt = row.session_finished_at == null ? null : Number(row.session_finished_at);
|
||
const item = {
|
||
id: row.id,
|
||
requestId: row.request_id,
|
||
startedAt: Number(row.started_at ?? 0),
|
||
updatedAt: Number(row.updated_at ?? 0),
|
||
attempts: Number(row.attempts ?? 0),
|
||
heartbeatAt,
|
||
heartbeatAgeMs,
|
||
sessionFinishedAt,
|
||
ageMs,
|
||
reason,
|
||
};
|
||
if (!dryRun) {
|
||
const staleError = Object.assign(
|
||
new Error(`agent run stale after ${ageMs}ms`),
|
||
{ code: 'AGENT_RUN_STALE_RECOVERY' },
|
||
);
|
||
const deliverableRecovered = await recoverRunFromDeliverables({
|
||
runId: row.id,
|
||
userId: row.user_id,
|
||
sessionId: row.agent_session_id ?? null,
|
||
err: staleError,
|
||
runStartedAtMs: row.started_at ?? null,
|
||
requireRecoverableError: false,
|
||
});
|
||
if (deliverableRecovered) {
|
||
const marked = await markRun(row.id, 'succeeded', {
|
||
agent_session_id: row.agent_session_id ?? null,
|
||
completed_at: nowMs(),
|
||
error_message: null,
|
||
}, { expectedStatus: 'running' });
|
||
if (!marked) continue;
|
||
item.status = 'succeeded';
|
||
item.recoveredAs = 'deliverables';
|
||
recovered.push(item);
|
||
continue;
|
||
}
|
||
|
||
const message = sessionFinishedAt != null
|
||
? `agent run recovered from stale running state after session finished ${Math.max(0, nowMs() - sessionFinishedAt)}ms ago`
|
||
: heartbeatAt == null
|
||
? `agent run recovered from stale running state after ${ageMs}ms without heartbeat`
|
||
: `agent run recovered from stale running state after running for ${ageMs}ms`;
|
||
const completedAt = nowMs();
|
||
const [update] = await pool.query(
|
||
`UPDATE h5_agent_runs
|
||
SET status = 'failed', error_message = ?, updated_at = ?, completed_at = ?
|
||
WHERE id = ?
|
||
AND status = 'running'
|
||
AND started_at IS NOT NULL
|
||
AND (
|
||
started_at <= ?
|
||
OR EXISTS (
|
||
SELECT 1
|
||
FROM h5_agent_run_events sf
|
||
WHERE sf.run_id = h5_agent_runs.id
|
||
AND sf.event_type = 'session_finished'
|
||
AND sf.created_at <= ?
|
||
)
|
||
)
|
||
AND (
|
||
SELECT COALESCE(MAX(hb.created_at), h5_agent_runs.started_at)
|
||
FROM h5_agent_run_events hb
|
||
WHERE hb.run_id = h5_agent_runs.id
|
||
AND hb.event_type = 'worker_heartbeat'
|
||
) <= ?`,
|
||
[message, completedAt, completedAt, row.id, startedCutoff, sessionFinishedCutoff, heartbeatCutoff],
|
||
);
|
||
if (Number(update?.affectedRows ?? 0) === 0) continue;
|
||
await appendEvent(row.id, 'stale_recovered', {
|
||
reason,
|
||
staleMs: normalizedStaleMs,
|
||
sessionFinishedGraceMs: normalizedSessionFinishedGraceMs,
|
||
ageMs,
|
||
heartbeatAt,
|
||
heartbeatAgeMs,
|
||
sessionFinishedAt,
|
||
status: 'failed',
|
||
error: message,
|
||
});
|
||
item.status = 'failed';
|
||
}
|
||
recovered.push(item);
|
||
}
|
||
return {
|
||
dryRun,
|
||
staleMs: normalizedStaleMs,
|
||
startedCutoff,
|
||
sessionFinishedCutoff,
|
||
considered: rows.length,
|
||
recovered: dryRun ? 0 : recovered.filter((item) => item.status).length,
|
||
runs: recovered,
|
||
};
|
||
}
|
||
|
||
async function dispatchQueuedRuns({ limit = maxConcurrentRuns } = {}) {
|
||
const staleRecovery = await recoverStaleRunningRuns({
|
||
staleMs: runTimeoutMs,
|
||
limit: Math.max(maxConcurrentRuns, Number(limit) || maxConcurrentRuns),
|
||
dryRun: false,
|
||
reason: 'worker_dispatch_stale_recovery',
|
||
});
|
||
const dispatchLimit = Math.max(1, Number(limit) || maxConcurrentRuns);
|
||
const available = Math.max(0, maxConcurrentRuns - inFlight.size - queuedDispatches.length);
|
||
if (available <= 0) {
|
||
return {
|
||
considered: 0,
|
||
dispatched: 0,
|
||
staleRecovery,
|
||
queue: await getQueueStatus(),
|
||
};
|
||
}
|
||
const take = Math.min(dispatchLimit, available);
|
||
const [rows] = await pool.query(
|
||
`SELECT id
|
||
FROM h5_agent_runs
|
||
WHERE status IN ('queued', 'retryable')
|
||
ORDER BY updated_at ASC
|
||
LIMIT ?`,
|
||
[take],
|
||
);
|
||
for (const row of rows) {
|
||
dispatchRun(row.id);
|
||
}
|
||
return {
|
||
considered: rows.length,
|
||
dispatched: rows.length,
|
||
staleRecovery,
|
||
queue: await getQueueStatus(),
|
||
};
|
||
}
|
||
|
||
async function listRunEventsForUser(userId, runId, { afterEventId = null, limit = 500 } = {}) {
|
||
const run = await getRunForUser(userId, runId);
|
||
if (!run) return null;
|
||
|
||
let afterCreatedAt = null;
|
||
let cursorMiss = false;
|
||
if (afterEventId) {
|
||
const [cursorRows] = await pool.query(
|
||
`SELECT e.created_at
|
||
FROM h5_agent_run_events e
|
||
INNER JOIN h5_agent_runs r ON r.id = e.run_id
|
||
WHERE e.id = ? AND e.run_id = ? AND r.user_id = ?
|
||
LIMIT 1`,
|
||
[afterEventId, runId, userId],
|
||
);
|
||
if (!cursorRows[0]) {
|
||
cursorMiss = true;
|
||
} else {
|
||
afterCreatedAt = Number(cursorRows[0].created_at);
|
||
}
|
||
}
|
||
|
||
const [rows] = await pool.query(
|
||
afterCreatedAt == null
|
||
? `SELECT id, event_type, data_json, created_at
|
||
FROM h5_agent_run_events
|
||
WHERE run_id = ?
|
||
ORDER BY created_at ASC, id ASC
|
||
LIMIT ?`
|
||
: `SELECT id, event_type, data_json, created_at
|
||
FROM h5_agent_run_events
|
||
WHERE run_id = ? AND created_at > ?
|
||
ORDER BY created_at ASC, id ASC
|
||
LIMIT ?`,
|
||
afterCreatedAt == null ? [runId, limit] : [runId, afterCreatedAt, limit],
|
||
);
|
||
|
||
return {
|
||
run,
|
||
events: rows.map((row) => ({
|
||
id: row.id,
|
||
eventType: row.event_type,
|
||
data: parseDbJsonColumn(row.data_json, null),
|
||
createdAt: Number(row.created_at),
|
||
})),
|
||
cursorMiss,
|
||
};
|
||
}
|
||
|
||
return {
|
||
createRun,
|
||
getRunForUser,
|
||
listRunEventsForUser,
|
||
dispatchRun,
|
||
getQueueStatus,
|
||
dispatchQueuedRuns,
|
||
recoverStaleRunningRuns,
|
||
};
|
||
}
|