// apps/api/src/routes/analysis.ts // Seven REST endpoints under /api/analysis. All responses use the standard // envelope { code, msg, data? } per the system contract. import type { FastifyInstance } from "fastify"; import { z } from "zod"; import { basename, join } from "node:path"; import { mkdirSync, writeFileSync, existsSync, createReadStream } from "node:fs"; import { readdir, readFile } from "node:fs/promises"; import { randomUUID } from "node:crypto"; import JSZip from "jszip"; import type { Config } from "../config.js"; import type { Store, ItemRow } from "../queue/store.js"; import type { QueueRunner, OptionsSnapshot } from "../queue/runner.js"; import { detectModeFromZip } from "../analyzers/detect.js"; interface RouteDeps { store: Store; config: Config; runner: QueueRunner; } const ZIP_FILENAME_RE = /\.zip$/i; function boundedNumber(min: number, max: number) { return z.preprocess((value) => { if (value === undefined) return value; const parsed = Number(value); if (!Number.isFinite(parsed)) return value; return Math.min(max, Math.max(min, parsed)); }, z.number().min(min).max(max)); } function boundedInteger(min: number, max: number) { return z.preprocess((value) => { if (value === undefined) return value; const parsed = Number(value); if (!Number.isFinite(parsed)) return value; return Math.min(max, Math.max(min, Math.round(parsed))); }, z.number().int().min(min).max(max)); } // Whitelist of fields accepted for optionsSnapshot. The frontend is permitted // to override the main analyzer model's API key (used for recommended_groups // + recommended_rubrics generation). Judge keys remain backend-only via .env. const OptionsSnapshotSchema = z .object({ requestedMode: z.enum(["api", "mcp"]).optional(), model: z.string().optional(), baseURL: z.string().optional(), apiKey: z.string().optional(), apiMode: z.enum(["chat", "responses"]).optional(), concurrency: boundedInteger(1, 8).optional(), temperature: boundedNumber(0, 2).optional(), topP: boundedNumber(0, 1).optional(), maxTokens: boundedInteger(1, 131072).optional(), streaming: z.boolean().optional(), enableThinking: z.boolean().optional(), clearThinking: z.boolean().optional(), ruleProfile: z.string().optional(), optimizeTask: z.boolean().optional(), feedback: z.string().optional(), strictFullMarks: z.boolean().optional(), ignoreTaskComplexityForFullMarks: z.boolean().optional(), judge: z .object({ model: z.string().optional(), baseURL: z.string().optional(), apiMode: z.enum(["chat", "responses"]).optional(), promptTemplate: z.string().optional(), }) .strict() .optional(), rubricGeneration: z .object({ model: z.string().optional(), baseURL: z.string().optional(), apiMode: z.enum(["chat", "responses"]).optional(), }) .strict() .optional(), }) .strict(); function ok(data: unknown) { return { code: 0, msg: "ok", data }; } function errEnv(code: number, msg: string) { return { code, msg }; } function publicItemView(row: ItemRow) { const progressByStatus: Record = { queued: 0, running: 0.35, judging: 0.75, completed: 1, failed: 1, stopped: 1, }; return { id: row.id, jobId: row.job_id, ord: row.ord, filename: row.filename, detectedMode: row.detected_mode, status: row.status, progress: progressByStatus[row.status] ?? 0, errorPreview: row.error_preview, attemptCount: row.attempt_count, lastStage: row.last_stage, hasExport: Boolean(row.export_zip_path), createdAt: row.created_at, updatedAt: row.updated_at, }; } function attachmentHeader(filename: string): string { let fallback = basename(filename) .replace(/[^\w.-]+/g, "_") .replace(/^_+|_+$/g, ""); if (!fallback || fallback.startsWith(".")) fallback = "download.zip"; return `attachment; filename="${fallback}"; filename*=UTF-8''${encodeURIComponent(filename)}`; } async function readJsonArtifact(path: string | null): Promise { if (!path || !existsSync(path)) return null; return JSON.parse(await readFile(path, "utf8")); } function finiteNumber(value: unknown): number | null { const parsed = Number(value); return Number.isFinite(parsed) ? parsed : null; } async function readJudgeHistory(config: Config, item: ItemRow) { const itemDir = join(config.exportDir, item.job_id, item.id); let entries: string[]; try { entries = await readdir(itemDir); } catch { return []; } const files = entries .map((name) => { const match = /^judge-result-iter-(\d+)\.json$/.exec(name); return match ? { name, iteration: Number(match[1]) } : null; }) .filter((entry): entry is { name: string; iteration: number } => Boolean(entry)) .sort((a, b) => a.iteration - b.iteration); const history = []; for (const file of files) { const judgeResult = await readJsonArtifact(join(itemDir, file.name)); if (!judgeResult || typeof judgeResult !== "object") continue; const result = judgeResult as Record; history.push({ iteration: file.iteration, score: finiteNumber(result.total_score), maxScore: finiteNumber(result.max_score) ?? 12, verdict: typeof result.verdict === "string" ? result.verdict : "", hasZeros: Boolean(result.has_zeros), judgeResult: result, }); } return history; } interface PendingUpload { filename: string; buffer: Buffer; autoDetectedMode: "api" | "mcp"; } function normalizeRequestedMode(value: unknown): "api" | "mcp" | null { return value === "api" || value === "mcp" ? value : null; } export async function registerAnalysisRoutes( app: FastifyInstance, deps: RouteDeps ): Promise { const { store, config, runner } = deps; // ── POST /api/analysis/jobs ──────────────────────────────────────────── // Multipart upload: 1..N .zip files. The selected page can force api/mcp // mode via requestedMode; otherwise each zip is auto-detected. app.post("/api/analysis/jobs", async (req, reply) => { if (!req.isMultipart()) { reply.code(400).send(errEnv(400, "Expected multipart/form-data")); return; } // Step 1: read + validate all parts in memory (bounded by maxUploadMB + // maxBatchFiles via fastify-multipart limits). const pending: PendingUpload[] = []; let requestedMode: "api" | "mcp" | null = null; try { const parts = req.parts(); for await (const part of parts) { if (part.type === "field") { if (part.fieldname === "requestedMode") { const normalized = normalizeRequestedMode(part.value); if (!normalized) { reply.code(400).send(errEnv(400, "requestedMode 只能是 api 或 mcp")); return; } requestedMode = normalized; } continue; } if (!part.filename) continue; if (!ZIP_FILENAME_RE.test(part.filename)) { reply.code(400).send(errEnv(400, `仅支持 .zip 文件:${part.filename}`)); return; } const buf = await part.toBuffer(); let detected: { mode: "api" | "mcp" }; try { detected = await detectModeFromZip(buf); } catch (e) { const msg = e instanceof Error ? e.message : "ZIP 校验失败"; reply.code(400).send(errEnv(400, `${part.filename}: ${msg}`)); return; } pending.push({ filename: part.filename, buffer: buf, autoDetectedMode: detected.mode, }); } } catch (e) { app.log.error({ err: e }, "Failed to consume multipart upload"); reply.code(500).send(errEnv(500, "上传读取失败")); return; } if (pending.length === 0) { reply.code(400).send(errEnv(400, "未收到任何 .zip 文件")); return; } if (pending.length > config.maxBatchFiles) { reply .code(400) .send( errEnv( 400, `单次最多上传 ${config.maxBatchFiles} 个 zip,本次:${pending.length}` ) ); return; } // Step 2: create job + persist files under canonical job dir. const job = await store.createJob({}); const jobUploadDir = join(config.uploadDir, job.id); try { mkdirSync(jobUploadDir, { recursive: true }); } catch (e) { app.log.error({ err: e, jobUploadDir }, "Failed to create upload dir"); reply.code(500).send(errEnv(500, "上传目录创建失败")); return; } const itemRows: ItemRow[] = []; for (const [idx, p] of pending.entries()) { const uploadPath = join(jobUploadDir, `${randomUUID()}.zip`); writeFileSync(uploadPath, p.buffer); const row = await store.createItem({ jobId: job.id, ord: idx + 1, filename: p.filename, uploadPath, detectedMode: requestedMode ?? p.autoDetectedMode, }); itemRows.push(row); } reply.send( ok({ jobId: job.id, items: itemRows.map(publicItemView), }) ); }); // ── GET /api/analysis/jobs/:jobId ────────────────────────────────────── app.get<{ Params: { jobId: string } }>( "/api/analysis/jobs/:jobId", async (req, reply) => { const job = await store.getJob(req.params.jobId); if (!job) { reply.code(404).send(errEnv(404, "Job not found")); return; } const items = (await store.listItemsByJob(job.id)).map(publicItemView); reply.send( ok({ id: job.id, status: job.status, createdAt: job.created_at, updatedAt: job.updated_at, items, }) ); } ); // ── POST /api/analysis/jobs/:jobId/start ─────────────────────────────── app.post<{ Params: { jobId: string }; Body: { optionsSnapshot?: unknown }; }>("/api/analysis/jobs/:jobId/start", async (req, reply) => { const job = await store.getJob(req.params.jobId); if (!job) { reply.code(404).send(errEnv(404, "Job not found")); return; } if (job.status === "running") { reply.code(409).send(errEnv(409, "Job 已经在运行中")); return; } const rawSnapshot = (req.body as { optionsSnapshot?: unknown } | undefined)?.optionsSnapshot ?? {}; const parsed = OptionsSnapshotSchema.safeParse(rawSnapshot); if (!parsed.success) { reply .code(400) .send(errEnv(400, `optionsSnapshot 校验失败:${parsed.error.message}`)); return; } const snapshot: OptionsSnapshot = parsed.data; await store.updateJobOptions(job.id, snapshot); runner.scheduleJob(job.id); reply.send(ok({ jobId: job.id, status: "running" })); }); // ── POST /api/analysis/jobs/:jobId/cancel ────────────────────────────── app.post<{ Params: { jobId: string } }>( "/api/analysis/jobs/:jobId/cancel", async (req, reply) => { const job = await store.getJob(req.params.jobId); if (!job) { reply.code(404).send(errEnv(404, "Job not found")); return; } runner.cancelJob(job.id); reply.send(ok({ jobId: job.id, status: "cancel-requested" })); } ); // ── POST /api/analysis/jobs/:jobId/items/:itemId/retry ───────────────── app.post<{ Params: { jobId: string; itemId: string } }>( "/api/analysis/jobs/:jobId/items/:itemId/retry", async (req, reply) => { const item = await store.getItem(req.params.itemId); if (!item || item.job_id !== req.params.jobId) { reply.code(404).send(errEnv(404, "Item not found")); return; } await runner.retryItem(item.id); reply.send(ok({ itemId: item.id, status: "queued" })); } ); // ── GET /api/analysis/jobs/:jobId/items/:itemId/result ──────────────── // Returns the generated JSON artifacts so the frontend can render concrete // analysis details even though the heavy work runs in the backend. app.get<{ Params: { jobId: string; itemId: string } }>( "/api/analysis/jobs/:jobId/items/:itemId/result", async (req, reply) => { const item = await store.getItem(req.params.itemId); if (!item || item.job_id !== req.params.jobId) { reply.code(404).send(errEnv(404, "Item not found")); return; } try { const evidenceBundle = (await readJsonArtifact(item.evidence_path)) as | { flags?: unknown; evidence?: unknown; analyzerResult?: unknown; iterationMemory?: unknown; } | null; const taskPackage = await readJsonArtifact(item.task_package_path); const judgeHistory = await readJudgeHistory(config, item); const latestJudgeResult = judgeHistory.at(-1)?.judgeResult ?? null; const judgeResult = (await readJsonArtifact(item.judge_result_path)) ?? latestJudgeResult; reply.send( ok({ item: publicItemView(item), flags: evidenceBundle?.flags ?? [], evidence: evidenceBundle?.evidence ?? null, analyzerResult: evidenceBundle?.analyzerResult ?? null, iterationMemory: evidenceBundle?.iterationMemory ?? null, taskPackage, judgeResult, judgeHistory, }) ); } catch (e) { app.log.error({ err: e, itemId: item.id }, "Failed to read item artifacts"); reply.code(500).send(errEnv(500, "分析结果读取失败")); } } ); // ── GET /api/analysis/jobs/:jobId/items/:itemId/download ─────────────── app.get<{ Params: { jobId: string; itemId: string } }>( "/api/analysis/jobs/:jobId/items/:itemId/download", async (req, reply) => { const item = await store.getItem(req.params.itemId); if (!item || item.job_id !== req.params.jobId) { reply.code(404).send(errEnv(404, "Item not found")); return; } if (!item.export_zip_path || !existsSync(item.export_zip_path)) { reply.code(404).send(errEnv(404, "导出文件尚未生成")); return; } const downloadName = item.filename.replace(ZIP_FILENAME_RE, "") + ".task-package.zip"; reply .header("Content-Type", "application/zip") .header("Content-Disposition", attachmentHeader(downloadName)); return reply.send(createReadStream(item.export_zip_path)); } ); // ── GET /api/analysis/jobs/:jobId/download-all ───────────────────────── // Bundles all completed items into a single zip-of-zips on the fly. app.get<{ Params: { jobId: string } }>( "/api/analysis/jobs/:jobId/download-all", async (req, reply) => { const job = await store.getJob(req.params.jobId); if (!job) { reply.code(404).send(errEnv(404, "Job not found")); return; } const items = await store.listItemsByJob(job.id); const exportable = items.filter( (it) => it.export_zip_path && existsSync(it.export_zip_path) ); if (exportable.length === 0) { reply.code(404).send(errEnv(404, "暂无可下载的导出项")); return; } const zip = new JSZip(); const used = new Set(); for (const it of exportable) { const baseName = it.filename.replace(ZIP_FILENAME_RE, "") + ".task-package.zip"; let entryName = baseName; let i = 1; while (used.has(entryName)) { entryName = `${baseName.replace(/\.task-package\.zip$/i, "")}-${i}.task-package.zip`; i += 1; } used.add(entryName); const content = await readFile(it.export_zip_path!); zip.file(entryName, content); } const buf = await zip.generateAsync({ type: "nodebuffer", compression: "DEFLATE", compressionOptions: { level: 6 }, }); reply .header("Content-Type", "application/zip") .header( "Content-Disposition", attachmentHeader(`all-results-${job.id.slice(0, 8)}.zip`) ); reply.send(buf); } ); }