john
735 lines
26 KiB
JavaScript
735 lines
26 KiB
JavaScript
import crypto from 'node:crypto';
|
|
import path from 'node:path';
|
|
|
|
const JOB_TYPES = new Set(['generate_page', 'analyze_asset', 'summarize']);
|
|
const OUTPUT_TYPES = new Set(['page_draft', 'html_page', 'markdown']);
|
|
const ACTIVE_JOB_STATUSES = new Set(['queued', 'running']);
|
|
const RETRYABLE_ERROR_CODES = new Set([
|
|
'worker_crashed',
|
|
'worker_unavailable',
|
|
'model_rate_limited',
|
|
'network_temporary',
|
|
'storage_temporary',
|
|
'job_timed_out',
|
|
'invalid_agent_job_output',
|
|
]);
|
|
|
|
function asNumber(value) {
|
|
return Number(value ?? 0);
|
|
}
|
|
|
|
function agentJobError(message, code, details) {
|
|
return Object.assign(new Error(message), { code, details });
|
|
}
|
|
|
|
function dedupeStrings(values) {
|
|
return [...new Set((Array.isArray(values) ? values : []).map((value) => String(value).trim()).filter(Boolean))];
|
|
}
|
|
|
|
function normalizeInstruction(value) {
|
|
const instruction = String(value ?? '').normalize('NFKC').trim();
|
|
if (!instruction) throw agentJobError('任务说明不能为空', 'invalid_agent_job_input');
|
|
if (instruction.length > 4000) {
|
|
throw agentJobError('任务说明超过长度限制', 'invalid_agent_job_input');
|
|
}
|
|
return instruction;
|
|
}
|
|
|
|
function normalizeJobInput(input) {
|
|
const jobType = String(input.jobType ?? '').trim();
|
|
if (!JOB_TYPES.has(jobType)) {
|
|
throw agentJobError('不支持的任务类型', 'invalid_agent_job_input');
|
|
}
|
|
const outputType = String(input.outputType ?? '').trim();
|
|
if (!OUTPUT_TYPES.has(outputType)) {
|
|
throw agentJobError('不支持的输出类型', 'invalid_agent_job_input');
|
|
}
|
|
const allowedAssetIds = dedupeStrings(input.allowedAssetIds);
|
|
if (allowedAssetIds.length === 0) {
|
|
throw agentJobError('至少需要授权一个输入资产', 'invalid_agent_job_input');
|
|
}
|
|
return {
|
|
jobType,
|
|
instruction: normalizeInstruction(input.instruction),
|
|
outputType,
|
|
allowedAssetIds,
|
|
idempotencyKey: String(input.idempotencyKey ?? '').trim() || null,
|
|
outputCategoryId: input.outputCategoryId ? String(input.outputCategoryId) : null,
|
|
locale: String(input.locale ?? 'zh-CN').trim() || 'zh-CN',
|
|
timezone: String(input.timezone ?? 'Asia/Shanghai').trim() || 'Asia/Shanghai',
|
|
capabilities: {
|
|
network: Boolean(input.capabilities?.network),
|
|
shell: Boolean(input.capabilities?.shell),
|
|
createPage: input.capabilities?.createPage !== false,
|
|
},
|
|
};
|
|
}
|
|
|
|
function hashJobToken(token) {
|
|
return crypto.createHash('sha256').update(String(token)).digest('hex');
|
|
}
|
|
|
|
function jsonValue(value, fallback) {
|
|
if (value == null) return fallback;
|
|
if (typeof value === 'object') return value;
|
|
try {
|
|
return JSON.parse(String(value));
|
|
} catch {
|
|
return fallback;
|
|
}
|
|
}
|
|
|
|
function jobResponse(row, assets = []) {
|
|
if (!row) return null;
|
|
return {
|
|
id: row.id,
|
|
sessionId: row.session_id ?? null,
|
|
jobType: row.job_type,
|
|
instruction: row.instruction,
|
|
status: row.status,
|
|
outputType: row.output_type,
|
|
outputCategoryId: row.output_category_id,
|
|
outputCategoryCode: row.output_category_code ?? null,
|
|
progress: jsonValue(row.progress_json, { stage: row.status }),
|
|
permissionScope: jsonValue(row.permission_scope, {}),
|
|
userContext: jsonValue(row.user_context_json, {
|
|
locale: 'zh-CN',
|
|
timezone: 'Asia/Shanghai',
|
|
}),
|
|
resultPageId: row.result_page_id ?? null,
|
|
resultAssetId: row.result_asset_id ?? null,
|
|
errorCode: row.error_code ?? null,
|
|
errorMessage: row.error_message ?? null,
|
|
retryable: RETRYABLE_ERROR_CODES.has(row.error_code),
|
|
queuedAt: asNumber(row.queued_at),
|
|
startedAt: row.started_at == null ? null : asNumber(row.started_at),
|
|
completedAt: row.completed_at == null ? null : asNumber(row.completed_at),
|
|
expiresAt: row.expires_at == null ? null : asNumber(row.expires_at),
|
|
assets,
|
|
};
|
|
}
|
|
|
|
function outputContentFormat(outputType, input) {
|
|
if (input.contentFormat === 'html' || outputType === 'html_page') return 'html';
|
|
return 'markdown';
|
|
}
|
|
|
|
function progressPayload(input, fallbackStage) {
|
|
const stage = String(input?.stage ?? fallbackStage ?? '').trim() || fallbackStage || 'running';
|
|
const payload = { stage };
|
|
const message = String(input?.message ?? '').trim();
|
|
if (message) payload.message = message.slice(0, 500);
|
|
return payload;
|
|
}
|
|
|
|
function hasPathAccess(asset) {
|
|
return asset.status !== 'deleted' && asset.status !== 'quarantined' && asset.scan_status !== 'blocked';
|
|
}
|
|
|
|
function verifyTokenHash(expectedHash, token) {
|
|
if (!expectedHash || !token) return false;
|
|
const actualHash = hashJobToken(token);
|
|
const expected = Buffer.from(String(expectedHash), 'hex');
|
|
const actual = Buffer.from(actualHash, 'hex');
|
|
return expected.length === actual.length && crypto.timingSafeEqual(expected, actual);
|
|
}
|
|
|
|
export function createAgentJobService(pool, options = {}) {
|
|
const idFactory = options.idFactory ?? (() => crypto.randomUUID());
|
|
const nowFactory = options.nowFactory ?? (() => Date.now());
|
|
const tokenTtlMs = Number(options.tokenTtlMs ?? 30 * 60 * 1000);
|
|
const maxOutputBytes = Number(options.maxOutputBytes ?? 2 * 1024 * 1024);
|
|
const pageService = options.pageService;
|
|
const storageRoot = path.resolve(options.storageRoot ?? path.join(process.cwd(), 'data', 'mindspace'));
|
|
|
|
const absoluteStoragePath = (storageKey) => {
|
|
const resolved = path.resolve(storageRoot, storageKey);
|
|
if (resolved !== storageRoot && !resolved.startsWith(`${storageRoot}${path.sep}`)) {
|
|
throw new Error('存储路径越界');
|
|
}
|
|
return resolved;
|
|
};
|
|
|
|
const resolveStoragePathCandidates = (storageKey) => {
|
|
const normalized = String(storageKey ?? '').replace(/\\/g, '/');
|
|
const withoutMd = normalized.endsWith('.md') ? normalized.slice(0, -3) : normalized;
|
|
const match = withoutMd.match(/^users\/([^/]+)\/(assets|pages)\/([^/]+)\/(v\d+)$/);
|
|
if (!match) return [normalized];
|
|
const [, userId, scope, entityId, versionTag] = match;
|
|
return [
|
|
normalized,
|
|
`users/${userId}/${scope}/${entityId}/versions/${versionTag}.md`,
|
|
`users/${userId}/${scope}/${entityId}/versions/${versionTag}`,
|
|
].filter((candidate, index, list) => list.indexOf(candidate) === index);
|
|
};
|
|
|
|
const resolveReadableStoragePath = async (storageKey) => {
|
|
let lastError = null;
|
|
for (const candidate of resolveStoragePathCandidates(storageKey)) {
|
|
const absolutePath = absoluteStoragePath(candidate);
|
|
try {
|
|
await fs.stat(absolutePath);
|
|
return absolutePath;
|
|
} catch (error) {
|
|
if (error?.code === 'ENOENT') {
|
|
lastError = error;
|
|
continue;
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
throw lastError ?? Object.assign(new Error('存储文件不存在'), { code: 'storage_not_found' });
|
|
};
|
|
|
|
const fetchJobAssets = async (jobId) => {
|
|
const [rows] = await pool.query(
|
|
`SELECT ja.asset_id, ja.asset_version_id, ja.permission, a.display_name, a.mime_type,
|
|
a.status, v.scan_status
|
|
FROM h5_agent_job_assets ja
|
|
JOIN h5_assets a ON a.id = ja.asset_id
|
|
JOIN h5_asset_versions v ON v.id = ja.asset_version_id
|
|
WHERE ja.job_id = ?
|
|
ORDER BY ja.created_at ASC`,
|
|
[jobId],
|
|
);
|
|
return rows.map((row) => ({
|
|
assetId: row.asset_id,
|
|
assetVersionId: row.asset_version_id,
|
|
permission: row.permission,
|
|
displayName: row.display_name,
|
|
mimeType: row.mime_type,
|
|
status: row.status,
|
|
scanStatus: row.scan_status,
|
|
}));
|
|
};
|
|
|
|
const fetchJobRow = async (jobId, userId = null) => {
|
|
const params = [jobId];
|
|
let sql =
|
|
`SELECT j.*, c.category_code AS output_category_code
|
|
FROM h5_agent_jobs j
|
|
JOIN h5_space_categories c ON c.id = j.output_category_id
|
|
WHERE j.id = ?`;
|
|
if (userId) {
|
|
sql += ' AND j.user_id = ?';
|
|
params.push(userId);
|
|
}
|
|
sql += ' LIMIT 1';
|
|
const [rows] = await pool.query(sql, params);
|
|
return rows[0] ?? null;
|
|
};
|
|
|
|
const createJob = async (userId, input) => {
|
|
const normalized = normalizeJobInput(input);
|
|
if (!pageService) {
|
|
throw agentJobError('页面服务未初始化', 'internal_error');
|
|
}
|
|
if (normalized.idempotencyKey) {
|
|
const [existingRows] = await pool.query(
|
|
`SELECT j.*, c.category_code AS output_category_code
|
|
FROM h5_agent_jobs j
|
|
JOIN h5_space_categories c ON c.id = j.output_category_id
|
|
WHERE j.user_id = ? AND j.idempotency_key = ?
|
|
LIMIT 1`,
|
|
[userId, normalized.idempotencyKey],
|
|
);
|
|
if (existingRows[0]) {
|
|
const assets = await fetchJobAssets(existingRows[0].id);
|
|
return jobResponse(existingRows[0], assets);
|
|
}
|
|
}
|
|
|
|
const conn = await pool.getConnection();
|
|
try {
|
|
await conn.beginTransaction();
|
|
const categoryParams = normalized.outputCategoryId
|
|
? [normalized.outputCategoryId, userId]
|
|
: [userId];
|
|
const [categoryRows] = await conn.query(
|
|
normalized.outputCategoryId
|
|
? `SELECT id, category_code
|
|
FROM h5_space_categories
|
|
WHERE id = ? AND user_id = ?
|
|
LIMIT 1
|
|
FOR UPDATE`
|
|
: `SELECT id, category_code
|
|
FROM h5_space_categories
|
|
WHERE user_id = ? AND category_code = 'draft'
|
|
LIMIT 1
|
|
FOR UPDATE`,
|
|
categoryParams,
|
|
);
|
|
const category = categoryRows[0];
|
|
if (!category) throw agentJobError('输出分类不存在', 'category_not_found');
|
|
if (!['draft', 'oa'].includes(category.category_code)) {
|
|
throw agentJobError('该分类不允许写入 Agent 输出', 'invalid_agent_job_input');
|
|
}
|
|
|
|
const placeholders = normalized.allowedAssetIds.map(() => '?').join(', ');
|
|
const [assetRows] = await conn.query(
|
|
`SELECT a.id, a.current_version_id, a.display_name, a.mime_type, a.status, v.scan_status
|
|
FROM h5_assets a
|
|
JOIN h5_asset_versions v ON v.id = a.current_version_id
|
|
WHERE a.user_id = ? AND a.id IN (${placeholders}) AND a.status <> 'deleted'
|
|
FOR UPDATE`,
|
|
[userId, ...normalized.allowedAssetIds],
|
|
);
|
|
const assetMap = new Map(assetRows.map((row) => [row.id, row]));
|
|
const missingIds = normalized.allowedAssetIds.filter((assetId) => !assetMap.has(assetId));
|
|
if (missingIds.length > 0) {
|
|
throw agentJobError('存在未授权或不存在的资产', 'asset_not_found', {
|
|
assetIds: missingIds,
|
|
});
|
|
}
|
|
const blockedAssets = assetRows.filter((row) => !hasPathAccess(row));
|
|
if (blockedAssets.length > 0) {
|
|
throw agentJobError('有输入资产尚未通过安全检查', 'security_scan_required', {
|
|
assetIds: blockedAssets.map((row) => row.id),
|
|
});
|
|
}
|
|
|
|
const jobId = idFactory();
|
|
const now = nowFactory();
|
|
await conn.query(
|
|
`INSERT INTO h5_agent_jobs
|
|
(id, user_id, job_type, instruction, permission_scope, user_context_json,
|
|
output_category_id, output_type, status, idempotency_key, progress_json,
|
|
queued_at, expires_at, updated_at, max_output_bytes)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?)`,
|
|
[
|
|
jobId,
|
|
userId,
|
|
normalized.jobType,
|
|
normalized.instruction,
|
|
JSON.stringify({ capabilities: normalized.capabilities }),
|
|
JSON.stringify({ locale: normalized.locale, timezone: normalized.timezone }),
|
|
category.id,
|
|
normalized.outputType,
|
|
normalized.idempotencyKey,
|
|
JSON.stringify({ stage: 'queued' }),
|
|
now,
|
|
now + tokenTtlMs,
|
|
now,
|
|
maxOutputBytes,
|
|
],
|
|
);
|
|
for (const assetId of normalized.allowedAssetIds) {
|
|
const asset = assetMap.get(assetId);
|
|
await conn.query(
|
|
`INSERT INTO h5_agent_job_assets
|
|
(id, job_id, asset_id, asset_version_id, permission, created_at)
|
|
VALUES (?, ?, ?, ?, 'read', ?)`,
|
|
[idFactory(), jobId, asset.id, asset.current_version_id, now],
|
|
);
|
|
}
|
|
await conn.commit();
|
|
const created = await fetchJobRow(jobId, userId);
|
|
const assets = await fetchJobAssets(jobId);
|
|
return jobResponse(created, assets);
|
|
} catch (error) {
|
|
await conn.rollback();
|
|
throw error;
|
|
} finally {
|
|
conn.release();
|
|
}
|
|
};
|
|
|
|
const getJob = async (userId, jobId) => {
|
|
const row = await fetchJobRow(jobId, userId);
|
|
if (!row) throw agentJobError('任务不存在', 'agent_job_not_found');
|
|
const assets = await fetchJobAssets(jobId);
|
|
return jobResponse(row, assets);
|
|
};
|
|
|
|
const listJobs = async (userId, { limit = 20, offset = 0 } = {}) => {
|
|
const safeLimit = Math.min(Math.max(Number(limit) || 20, 1), 100);
|
|
const safeOffset = Math.max(Number(offset) || 0, 0);
|
|
const [countRows] = await pool.query(
|
|
`SELECT COUNT(*) AS total FROM h5_agent_jobs WHERE user_id = ?`,
|
|
[userId],
|
|
);
|
|
const total = Number(countRows[0]?.total ?? 0);
|
|
const [rows] = await pool.query(
|
|
`SELECT j.*, c.category_code AS output_category_code
|
|
FROM h5_agent_jobs j
|
|
JOIN h5_space_categories c ON c.id = j.output_category_id
|
|
WHERE j.user_id = ?
|
|
ORDER BY COALESCE(j.completed_at, j.started_at, j.queued_at) DESC
|
|
LIMIT ${safeLimit} OFFSET ${safeOffset}`,
|
|
[userId],
|
|
);
|
|
const jobs = await Promise.all(
|
|
rows.map(async (row) => jobResponse(row, await fetchJobAssets(row.id))),
|
|
);
|
|
return {
|
|
items: jobs,
|
|
total,
|
|
offset: safeOffset,
|
|
limit: safeLimit,
|
|
hasMore: safeOffset + jobs.length < total,
|
|
};
|
|
};
|
|
|
|
const cancelJob = async (userId, jobId) => {
|
|
const conn = await pool.getConnection();
|
|
try {
|
|
await conn.beginTransaction();
|
|
const [rows] = await conn.query(
|
|
`SELECT * FROM h5_agent_jobs
|
|
WHERE id = ? AND user_id = ?
|
|
LIMIT 1
|
|
FOR UPDATE`,
|
|
[jobId, userId],
|
|
);
|
|
const job = rows[0];
|
|
if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
|
|
if (!ACTIVE_JOB_STATUSES.has(job.status)) {
|
|
throw agentJobError('当前状态不允许取消任务', 'invalid_state_transition');
|
|
}
|
|
const now = nowFactory();
|
|
await conn.query(
|
|
`UPDATE h5_agent_jobs
|
|
SET status = 'cancelled', completed_at = ?, expires_at = ?, updated_at = ?,
|
|
job_token_hash = NULL, heartbeat_at = NULL, progress_json = ?
|
|
WHERE id = ?`,
|
|
[now, now, now, JSON.stringify({ stage: 'cancelled' }), jobId],
|
|
);
|
|
await conn.commit();
|
|
return getJob(userId, jobId);
|
|
} catch (error) {
|
|
await conn.rollback();
|
|
throw error;
|
|
} finally {
|
|
conn.release();
|
|
}
|
|
};
|
|
|
|
const retryJob = async (userId, jobId) => {
|
|
const conn = await pool.getConnection();
|
|
try {
|
|
await conn.beginTransaction();
|
|
const [rows] = await conn.query(
|
|
`SELECT * FROM h5_agent_jobs
|
|
WHERE id = ? AND user_id = ?
|
|
LIMIT 1
|
|
FOR UPDATE`,
|
|
[jobId, userId],
|
|
);
|
|
const job = rows[0];
|
|
if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
|
|
if (!['failed', 'timed_out', 'cancelled'].includes(job.status)) {
|
|
throw agentJobError('当前状态不允许重试任务', 'invalid_state_transition');
|
|
}
|
|
if (job.status === 'failed' && !RETRYABLE_ERROR_CODES.has(job.error_code)) {
|
|
throw agentJobError('该任务错误不可自动重试', 'invalid_state_transition');
|
|
}
|
|
const now = nowFactory();
|
|
await conn.query(
|
|
`UPDATE h5_agent_jobs
|
|
SET status = 'queued', progress_json = ?, started_at = NULL, completed_at = NULL,
|
|
error_code = NULL, error_message = NULL, result_page_id = NULL, result_asset_id = NULL,
|
|
job_token_hash = NULL, heartbeat_at = NULL, expires_at = ?, updated_at = ?
|
|
WHERE id = ?`,
|
|
[JSON.stringify({ stage: 'queued' }), now + tokenTtlMs, now, jobId],
|
|
);
|
|
await conn.commit();
|
|
return getJob(userId, jobId);
|
|
} catch (error) {
|
|
await conn.rollback();
|
|
throw error;
|
|
} finally {
|
|
conn.release();
|
|
}
|
|
};
|
|
|
|
// Marks a locked, queued job row as running and returns the claim payload.
|
|
// Caller must hold a row lock on `job` inside `conn`'s open transaction; this
|
|
// commits the transaction.
|
|
const finalizeClaim = async (conn, job) => {
|
|
const jobId = job.id;
|
|
const token = crypto.randomBytes(24).toString('base64url');
|
|
const now = nowFactory();
|
|
await conn.query(
|
|
`UPDATE h5_agent_jobs
|
|
SET status = 'running', started_at = COALESCE(started_at, ?), heartbeat_at = ?,
|
|
updated_at = ?, expires_at = ?, progress_json = ?, job_token_hash = ?
|
|
WHERE id = ?`,
|
|
[
|
|
now,
|
|
now,
|
|
now,
|
|
now + tokenTtlMs,
|
|
JSON.stringify({ stage: 'preparing_files' }),
|
|
hashJobToken(token),
|
|
jobId,
|
|
],
|
|
);
|
|
await conn.commit();
|
|
const assets = await fetchJobAssets(jobId);
|
|
const permissionScope = jsonValue(job.permission_scope, {});
|
|
const userContext = jsonValue(job.user_context_json, {
|
|
locale: 'zh-CN',
|
|
timezone: 'Asia/Shanghai',
|
|
});
|
|
return {
|
|
jobId,
|
|
userId: job.user_id,
|
|
jobToken: token,
|
|
instruction: job.instruction,
|
|
userContext,
|
|
allowedAssets: assets.map((asset) => ({
|
|
assetId: asset.assetId,
|
|
versionId: asset.assetVersionId,
|
|
permission: asset.permission,
|
|
displayName: asset.displayName,
|
|
mimeType: asset.mimeType,
|
|
downloadEndpoint: `/api/internal/agent/jobs/${jobId}/assets/${asset.assetId}`,
|
|
})),
|
|
output: {
|
|
categoryId: job.output_category_id,
|
|
categoryCode: job.output_category_code,
|
|
allowedTypes: [job.output_type],
|
|
maxBytes: asNumber(job.max_output_bytes),
|
|
},
|
|
capabilities: permissionScope.capabilities ?? {
|
|
network: false,
|
|
shell: false,
|
|
createPage: true,
|
|
},
|
|
expiresAt: now + tokenTtlMs,
|
|
};
|
|
};
|
|
|
|
// Atomically claim the oldest queued, unexpired job. Uses SKIP LOCKED so that
|
|
// multiple worker loops (across portal instances) pull distinct jobs without
|
|
// blocking each other. Returns null when the queue is empty.
|
|
const claimNextJob = async () => {
|
|
const conn = await pool.getConnection();
|
|
try {
|
|
await conn.beginTransaction();
|
|
const [rows] = await conn.query(
|
|
`SELECT j.*, c.category_code AS output_category_code
|
|
FROM h5_agent_jobs j
|
|
JOIN h5_space_categories c ON c.id = j.output_category_id
|
|
WHERE j.status = 'queued'
|
|
AND (j.expires_at IS NULL OR j.expires_at >= ?)
|
|
ORDER BY j.queued_at ASC
|
|
LIMIT 1
|
|
FOR UPDATE SKIP LOCKED`,
|
|
[nowFactory()],
|
|
);
|
|
const job = rows[0];
|
|
if (!job) {
|
|
await conn.commit();
|
|
return null;
|
|
}
|
|
return await finalizeClaim(conn, job);
|
|
} catch (error) {
|
|
await conn.rollback();
|
|
throw error;
|
|
} finally {
|
|
conn.release();
|
|
}
|
|
};
|
|
|
|
const claimJob = async (jobId) => {
|
|
const conn = await pool.getConnection();
|
|
try {
|
|
await conn.beginTransaction();
|
|
const [rows] = await conn.query(
|
|
`SELECT j.*, c.category_code AS output_category_code
|
|
FROM h5_agent_jobs j
|
|
JOIN h5_space_categories c ON c.id = j.output_category_id
|
|
WHERE j.id = ?
|
|
LIMIT 1
|
|
FOR UPDATE`,
|
|
[jobId],
|
|
);
|
|
const job = rows[0];
|
|
if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
|
|
if (job.status !== 'queued') {
|
|
throw agentJobError('任务当前不可领取', 'invalid_state_transition');
|
|
}
|
|
if (job.expires_at != null && asNumber(job.expires_at) < nowFactory()) {
|
|
throw agentJobError('任务已过期', 'agent_job_expired');
|
|
}
|
|
return await finalizeClaim(conn, job);
|
|
} catch (error) {
|
|
await conn.rollback();
|
|
throw error;
|
|
} finally {
|
|
conn.release();
|
|
}
|
|
};
|
|
|
|
|
|
const requireJobToken = async (jobId, token) => {
|
|
const [rows] = await pool.query(
|
|
`SELECT j.*, c.category_code AS output_category_code
|
|
FROM h5_agent_jobs j
|
|
JOIN h5_space_categories c ON c.id = j.output_category_id
|
|
WHERE j.id = ?
|
|
LIMIT 1`,
|
|
[jobId],
|
|
);
|
|
const job = rows[0];
|
|
if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
|
|
if (job.status !== 'running') {
|
|
throw agentJobError('任务未处于执行中状态', 'invalid_state_transition');
|
|
}
|
|
if (job.expires_at != null && asNumber(job.expires_at) < nowFactory()) {
|
|
throw agentJobError('任务 token 已过期', 'agent_job_expired');
|
|
}
|
|
if (!verifyTokenHash(job.job_token_hash, token)) {
|
|
throw agentJobError('任务 token 无效', 'agent_job_token_invalid');
|
|
}
|
|
return job;
|
|
};
|
|
|
|
const getAssetForJob = async (jobId, token, assetId) => {
|
|
await requireJobToken(jobId, token);
|
|
const [rows] = await pool.query(
|
|
`SELECT a.id, a.display_name, a.mime_type, a.status, v.id AS asset_version_id,
|
|
v.storage_key, v.scan_status
|
|
FROM h5_agent_job_assets ja
|
|
JOIN h5_assets a ON a.id = ja.asset_id
|
|
JOIN h5_asset_versions v ON v.id = ja.asset_version_id
|
|
WHERE ja.job_id = ? AND ja.asset_id = ?
|
|
LIMIT 1`,
|
|
[jobId, assetId],
|
|
);
|
|
const asset = rows[0];
|
|
if (!asset) throw agentJobError('任务资产不存在', 'asset_not_found');
|
|
if (!hasPathAccess(asset)) {
|
|
throw agentJobError('任务资产尚未通过安全检查', 'security_scan_required');
|
|
}
|
|
return {
|
|
assetId: asset.id,
|
|
assetVersionId: asset.asset_version_id,
|
|
displayName: asset.display_name,
|
|
mimeType: asset.mime_type,
|
|
path: await resolveReadableStoragePath(asset.storage_key),
|
|
};
|
|
};
|
|
|
|
const heartbeat = async (jobId, token, input) => {
|
|
const job = await requireJobToken(jobId, token);
|
|
const now = nowFactory();
|
|
const payload = progressPayload(input, 'running');
|
|
await pool.query(
|
|
`UPDATE h5_agent_jobs
|
|
SET heartbeat_at = ?, updated_at = ?, expires_at = ?, progress_json = ?
|
|
WHERE id = ?`,
|
|
[now, now, now + tokenTtlMs, JSON.stringify(payload), jobId],
|
|
);
|
|
return jobResponse({ ...job, progress_json: JSON.stringify(payload), heartbeat_at: now });
|
|
};
|
|
|
|
const completeJob = async (jobId, token, input) => {
|
|
const job = await requireJobToken(jobId, token);
|
|
const now = nowFactory();
|
|
if (input?.status === 'failed') {
|
|
const errorCode = String(input.errorCode ?? 'worker_crashed').trim() || 'worker_crashed';
|
|
const errorMessage = String(input.errorMessage ?? '任务执行失败').trim() || '任务执行失败';
|
|
await pool.query(
|
|
`UPDATE h5_agent_jobs
|
|
SET status = 'failed', error_code = ?, error_message = ?, completed_at = ?,
|
|
updated_at = ?, heartbeat_at = NULL, job_token_hash = NULL, progress_json = ?
|
|
WHERE id = ?`,
|
|
[errorCode, errorMessage.slice(0, 1000), now, now, JSON.stringify({ stage: 'failed' }), jobId],
|
|
);
|
|
return getJob(job.user_id, jobId);
|
|
}
|
|
|
|
const content = String(input?.content ?? '');
|
|
const outputBytes = Buffer.byteLength(content, 'utf8');
|
|
if (!content.trim()) {
|
|
throw agentJobError('任务输出内容不能为空', 'invalid_agent_job_output');
|
|
}
|
|
if (outputBytes > asNumber(job.max_output_bytes)) {
|
|
throw agentJobError('任务输出超过大小限制', 'page_content_too_large', {
|
|
maxBytes: asNumber(job.max_output_bytes),
|
|
});
|
|
}
|
|
const sourceAssetIds = dedupeStrings(input?.sourceAssetIds);
|
|
const boundAssets = await fetchJobAssets(jobId);
|
|
const boundAssetIdSet = new Set(boundAssets.map((asset) => asset.assetId));
|
|
const invalidSourceIds = sourceAssetIds.filter((assetId) => !boundAssetIdSet.has(assetId));
|
|
if (invalidSourceIds.length > 0) {
|
|
throw agentJobError('输出引用了未授权的输入资产', 'invalid_agent_job_output', {
|
|
assetIds: invalidSourceIds,
|
|
});
|
|
}
|
|
|
|
const page = await pageService.createFromAgent(
|
|
job.user_id,
|
|
{
|
|
title: input?.title,
|
|
summary: input?.summary,
|
|
content,
|
|
contentFormat: outputContentFormat(job.output_type, input),
|
|
pageType: input?.pageType,
|
|
templateId: input?.templateId,
|
|
categoryCode: job.output_category_code,
|
|
changeNote: `Agent job ${jobId}`,
|
|
},
|
|
{
|
|
jobId,
|
|
assetId: sourceAssetIds[0] ?? null,
|
|
assetIds: sourceAssetIds,
|
|
},
|
|
);
|
|
await pool.query(
|
|
`UPDATE h5_agent_jobs
|
|
SET status = 'completed', result_page_id = ?, completed_at = ?, updated_at = ?,
|
|
heartbeat_at = NULL, job_token_hash = NULL, progress_json = ?
|
|
WHERE id = ?`,
|
|
[page.id, now, now, JSON.stringify({ stage: 'completed' }), jobId],
|
|
);
|
|
return getJob(job.user_id, jobId);
|
|
};
|
|
|
|
// Restart resilience: a job left in 'running' by a crashed/restarted worker
|
|
// stops heartbeating. Mark such jobs failed with the retryable worker_crashed
|
|
// code so they surface to the user (and can be retried) instead of being stuck
|
|
// forever. We fail rather than auto-requeue to avoid poison-job loops, since
|
|
// there is no per-job attempt counter.
|
|
const reapStaleJobs = async (maxStaleMs = 5 * 60 * 1000) => {
|
|
const now = nowFactory();
|
|
const cutoff = now - Math.max(1, maxStaleMs);
|
|
const [result] = await pool.query(
|
|
`UPDATE h5_agent_jobs
|
|
SET status = 'failed', error_code = 'worker_crashed',
|
|
error_message = 'worker heartbeat timed out',
|
|
completed_at = ?, updated_at = ?, heartbeat_at = NULL, job_token_hash = NULL,
|
|
progress_json = ?
|
|
WHERE status = 'running'
|
|
AND heartbeat_at IS NOT NULL
|
|
AND heartbeat_at < ?`,
|
|
[now, now, JSON.stringify({ stage: 'failed' }), cutoff],
|
|
);
|
|
return result?.affectedRows ?? 0;
|
|
};
|
|
|
|
return {
|
|
createJob,
|
|
getJob,
|
|
listJobs,
|
|
cancelJob,
|
|
retryJob,
|
|
claimJob,
|
|
claimNextJob,
|
|
reapStaleJobs,
|
|
getAssetForJob,
|
|
heartbeat,
|
|
completeJob,
|
|
};
|
|
}
|
|
|
|
export const agentJobInternals = {
|
|
normalizeJobInput,
|
|
hashJobToken,
|
|
verifyTokenHash,
|
|
progressPayload,
|
|
};
|