spb/autoscholar
Public
1import { createHash } from "node:crypto";2import {3 artifacts,4 checkpoints,5 type Db,6 jobLeases,7 paperSequences,8 projectInstructionVersions,9 qualityGateResults,10 researchProjects,11 researchRuns,12 runAttempts,13 stageEvents,14 stageExecutions,15 type Tx,16} from "@autoscholar/db";17import { effectiveAutoPublish, QUALITY_POLICY_VERSION } from "@autoscholar/research-contract";18import {19 ACTIVE_RUN_STATUSES,20 type LaunchRunRequest,21 RUN_TRANSITIONS,22 type RunEventInput,23 RunManifest,24 type RunStatus,25} from "@autoscholar/schemas";26import { and, asc, desc, eq, gt, inArray, sql } from "drizzle-orm";27import { audit } from "./audit.js";28import { getEnv } from "./env.js";29import { AppError, conflict, invalid, notFound } from "./errors.js";3031export type RunRow = typeof researchRuns.$inferSelect;3233export function canTransition(from: RunStatus, to: RunStatus): boolean {34 return RUN_TRANSITIONS[from]?.includes(to) ?? false;35}3637/** Appends an event with a per-run monotonically increasing seq. Must be called inside a transaction holding the run row. */38export async function appendEvent(39 tx: Tx,40 runId: string,41 ev: RunEventInput & { attemptId?: string | null; workerId?: string | null },42): Promise<number> {43 const [r] = await tx44 .update(researchRuns)45 .set({ lastEventSeq: sql`${researchRuns.lastEventSeq} + 1`, updatedAt: new Date() })46 .where(eq(researchRuns.id, runId))47 .returning({ seq: researchRuns.lastEventSeq });48 if (!r) throw notFound("run");49 await tx.insert(stageEvents).values({50 runId,51 attemptId: ev.attemptId ?? null,52 workerId: ev.workerId ?? null,53 seq: r.seq,54 stage: ev.stage ?? null,55 level: ev.level ?? "info",56 kind: ev.kind,57 message: ev.message.slice(0, 4000),58 data: ev.data ?? null,59 at: new Date(ev.at),60 });61 return r.seq;62}6364/** Atomic, validated state transition with ledger entry. Locks the run row. */65export async function transition(66 tx: Tx,67 runId: string,68 to: RunStatus,69 opts: {70 reason?: string;71 actor?: string;72 data?: Record<string, unknown>;73 attemptId?: string | null;74 workerId?: string | null;75 expectFrom?: RunStatus[];76 } = {},77): Promise<RunRow> {78 const [run] = await tx.select().from(researchRuns).where(eq(researchRuns.id, runId)).for("update");79 if (!run) throw notFound("run");80 const from = run.status as RunStatus;81 if (opts.expectFrom && !opts.expectFrom.includes(from))82 throw conflict(`run is ${from}, expected ${opts.expectFrom.join("/")}`);83 if (from === to) return run;84 if (!canTransition(from, to)) throw conflict(`invalid transition ${from} -> ${to}`);85 const patch: Partial<typeof researchRuns.$inferInsert> = { status: to, updatedAt: new Date() };86 if (to === "queued") patch.queuedAt = new Date();87 if (to === "running" && !run.startedAt) patch.startedAt = new Date();88 if (["published", "cancelled", "failed"].includes(to)) patch.finishedAt = new Date();89 if (to === "failed" && opts.reason) patch.failureReason = opts.reason;90 const [updated] = await tx.update(researchRuns).set(patch).where(eq(researchRuns.id, runId)).returning();91 await appendEvent(tx, runId, {92 kind: "status.transition",93 level: to === "failed" ? "error" : "info",94 message: `${from} → ${to}${opts.reason ? `: ${opts.reason}` : ""}`,95 data: { from, to, actor: opts.actor ?? "system", ...(opts.data ?? {}) },96 at: new Date().toISOString(),97 attemptId: opts.attemptId ?? null,98 workerId: opts.workerId ?? null,99 });100 return updated!;101}102103/**104 * Idempotent launch: same idempotencyKey → same run. Builds and freezes the immutable manifest server-side.105 * draft → validated → queued in one transaction.106 */107export async function launchRun(108 db: Db,109 req: LaunchRunRequest,110 actor: { id: string; ip?: string | null },111): Promise<{ run: RunRow; created: boolean }> {112 const env = getEnv();113 const existing = await db114 .select()115 .from(researchRuns)116 .where(eq(researchRuns.idempotencyKey, req.idempotencyKey))117 .limit(1);118 if (existing[0]) return { run: existing[0], created: false };119 const [iv] = await db120 .select({ iv: projectInstructionVersions, p: researchProjects })121 .from(projectInstructionVersions)122 .innerJoin(researchProjects, eq(projectInstructionVersions.projectId, researchProjects.id))123 .where(124 and(125 eq(projectInstructionVersions.id, req.instructionVersionId),126 eq(projectInstructionVersions.projectId, req.projectId),127 ),128 )129 .limit(1);130 if (!iv) throw notFound("instruction version for this project");131 const autoPub = effectiveAutoPublish(req.researchMode, req.discipline, req.autoPublish);132 const maxBudgetUsd = req.maxBudgetUsd ?? env.DEFAULT_MAX_BUDGET_USD;133 const maxRuntimeMinutes = req.maxRuntimeMinutes ?? env.DEFAULT_MAX_RUNTIME_MINUTES;134 if (maxBudgetUsd > 2000) throw invalid("budget cap exceeds platform maximum");135 const runId = crypto.randomUUID();136 const manifest = RunManifest.parse({137 schemaVersion: 1,138 runId,139 projectId: req.projectId,140 instructionVersionId: iv.iv.id,141 instructionVersion: iv.iv.version,142 instructionSha256: iv.iv.sha256,143 researchMode: req.researchMode,144 secondaryModes: req.secondaryModes,145 discipline: req.discipline,146 computeProfile: req.computeProfile,147 provider: "anthropic",148 model: req.model ?? env.ANTHROPIC_MODEL,149 reviewerModel: env.ANTHROPIC_REVIEWER_MODEL,150 maxBudgetUsd,151 maxRuntimeMinutes,152 maxAttempts: req.maxAttempts,153 autoPublish: autoPub.allowed,154 qualityPolicyVersion: QUALITY_POLICY_VERSION,155 createdAt: new Date().toISOString(),156 createdBy: actor.id,157 title: req.title,158 });159 try {160 const run = await db.transaction(async (tx) => {161 const year = new Date().getUTCFullYear();162 await tx.insert(paperSequences).values({ year, last: 0 }).onConflictDoNothing();163 const [seq] = await tx164 .update(paperSequences)165 .set({ last: sql`${paperSequences.last} + 1` })166 .where(eq(paperSequences.year, year))167 .returning({ last: paperSequences.last });168 manifest.paperNumber = `AS-WP-${year}-${String(seq!.last).padStart(4, "0")}`;169 const manifestSha256 = createHash("sha256").update(JSON.stringify(manifest)).digest("hex");170 const [row] = await tx171 .insert(researchRuns)172 .values({173 id: runId,174 projectId: req.projectId,175 instructionVersionId: iv.iv.id,176 idempotencyKey: req.idempotencyKey,177 title: req.title,178 status: "draft",179 researchMode: req.researchMode,180 secondaryModes: req.secondaryModes,181 discipline: req.discipline,182 computeProfile: req.computeProfile,183 model: manifest.model,184 reviewerModel: manifest.reviewerModel,185 maxBudgetUsd: String(maxBudgetUsd),186 maxRuntimeMinutes,187 maxAttempts: req.maxAttempts,188 autoPublish: autoPub.allowed,189 qualityPolicyVersion: QUALITY_POLICY_VERSION,190 manifest,191 manifestSha256,192 createdBy: actor.id,193 })194 .returning();195 await appendEvent(tx, runId, {196 kind: "run.created",197 message: `Run created from instruction v${iv.iv.version} (${iv.iv.sha256.slice(0, 12)})`,198 data: { manifestSha256, autoPublishReason: autoPub.reason ?? null },199 at: new Date().toISOString(),200 });201 await transition(tx, runId, "validated", { actor: `admin:${actor.id}` });202 await transition(tx, runId, "queued", { actor: `admin:${actor.id}` });203 await audit(204 tx,205 { type: "admin", id: actor.id, ip: actor.ip },206 "run.launch",207 { type: "run", id: runId },208 { idempotencyKey: req.idempotencyKey, manifestSha256 },209 );210 return row!;211 });212 return { run: { ...run, status: "queued" }, created: true };213 } catch (e) {214 const again = await db215 .select()216 .from(researchRuns)217 .where(eq(researchRuns.idempotencyKey, req.idempotencyKey))218 .limit(1);219 if (again[0]) return { run: again[0], created: false };220 throw e;221 }222}223224export async function getRun(db: Db, runId: string) {225 const [run] = await db.select().from(researchRuns).where(eq(researchRuns.id, runId)).limit(1);226 if (!run) throw notFound("run");227 const attempts = await db228 .select()229 .from(runAttempts)230 .where(eq(runAttempts.runId, runId))231 .orderBy(asc(runAttempts.attemptNo));232 const stages = await db233 .select()234 .from(stageExecutions)235 .where(eq(stageExecutions.runId, runId))236 .orderBy(asc(stageExecutions.startedAt));237 const arts = await db238 .select({239 id: artifacts.id,240 path: artifacts.path,241 kind: artifacts.kind,242 sha256: artifacts.sha256,243 bytes: artifacts.bytes,244 contentType: artifacts.contentType,245 stage: artifacts.stage,246 version: artifacts.version,247 synthetic: artifacts.synthetic,248 createdAt: artifacts.createdAt,249 attemptId: artifacts.attemptId,250 })251 .from(artifacts)252 .where(eq(artifacts.runId, runId))253 .orderBy(asc(artifacts.path), desc(artifacts.version));254 const [lease] = await db.select().from(jobLeases).where(eq(jobLeases.runId, runId)).limit(1);255 const gates = await db256 .select()257 .from(qualityGateResults)258 .where(eq(qualityGateResults.runId, runId))259 .orderBy(desc(qualityGateResults.createdAt));260 const latestEval = gates[0]?.evaluationId;261 return {262 run,263 attempts,264 stages,265 artifacts: arts,266 lease: lease ?? null,267 gates: gates.filter((g) => g.evaluationId === latestEval),268 };269}270271export async function listRuns(272 db: Db,273 opts: { status?: RunStatus[]; limit?: number; projectId?: string } = {},274) {275 const conds = [];276 if (opts.status?.length) conds.push(inArray(researchRuns.status, opts.status));277 if (opts.projectId) conds.push(eq(researchRuns.projectId, opts.projectId));278 return db279 .select()280 .from(researchRuns)281 .where(conds.length ? and(...conds) : undefined)282 .orderBy(desc(researchRuns.createdAt))283 .limit(opts.limit ?? 100);284}285286export async function listEvents(db: Db, runId: string, afterSeq = 0, limit = 500) {287 return db288 .select()289 .from(stageEvents)290 .where(and(eq(stageEvents.runId, runId), gt(stageEvents.seq, afterSeq)))291 .orderBy(asc(stageEvents.seq))292 .limit(limit);293}294295export async function requestCancel(db: Db, runId: string, actor: { id: string; ip?: string | null }) {296 return db.transaction(async (tx) => {297 const [run] = await tx.select().from(researchRuns).where(eq(researchRuns.id, runId)).for("update");298 if (!run) throw notFound("run");299 const st = run.status as RunStatus;300 if (["published", "cancelled", "failed"].includes(st)) throw conflict(`run already ${st}`);301 if (302 [303 "queued",304 "validated",305 "draft",306 "quality_failed",307 "ready_to_publish",308 "awaiting_approval",309 "retrying",310 "revising",311 ].includes(st)312 ) {313 const updated = await transition(tx, runId, "cancelled", {314 actor: `admin:${actor.id}`,315 reason: "cancelled by admin",316 });317 await tx.delete(jobLeases).where(eq(jobLeases.runId, runId));318 await audit(tx, { type: "admin", id: actor.id, ip: actor.ip }, "run.cancel", {319 type: "run",320 id: runId,321 });322 return updated;323 }324 // running on a worker: cooperative cancel, worker acknowledges via complete(cancelled); lease expiry is the fallback325 await tx326 .update(researchRuns)327 .set({ cancelRequested: true, updatedAt: new Date() })328 .where(eq(researchRuns.id, runId));329 const updated = await transition(tx, runId, "cancelling", {330 actor: `admin:${actor.id}`,331 reason: "cancellation requested",332 });333 await audit(tx, { type: "admin", id: actor.id, ip: actor.ip }, "run.cancel", { type: "run", id: runId });334 return updated;335 });336}337338/** Retry from the last checkpoint with a new attempt; preserves frozen input and previous artifacts. */339export async function retryRun(db: Db, runId: string, actor: { id: string; ip?: string | null }) {340 return db.transaction(async (tx) => {341 const [run] = await tx.select().from(researchRuns).where(eq(researchRuns.id, runId)).for("update");342 if (!run) throw notFound("run");343 const st = run.status as RunStatus;344 if (!["failed", "quality_failed", "cancelled"].includes(st))345 throw conflict(`cannot retry a run in status ${st}`);346 if (st === "cancelled")347 throw conflict("cancelled runs cannot be retried; duplicate the project run instead");348 if (run.currentAttemptNo >= run.maxAttempts) {349 await tx350 .update(researchRuns)351 .set({ maxAttempts: run.currentAttemptNo + 1 })352 .where(eq(researchRuns.id, runId));353 }354 await tx.delete(jobLeases).where(eq(jobLeases.runId, runId));355 await tx356 .update(researchRuns)357 .set({ cancelRequested: false, failureReason: null, finishedAt: null })358 .where(eq(researchRuns.id, runId));359 if (st === "quality_failed") await transition(tx, runId, "revising", { actor: `admin:${actor.id}` });360 const updated = await transition(tx, runId, "queued", {361 actor: `admin:${actor.id}`,362 reason: "retry requested",363 });364 await audit(365 tx,366 { type: "admin", id: actor.id, ip: actor.ip },367 "run.retry",368 { type: "run", id: runId },369 { fromStatus: st },370 );371 return updated;372 });373}374375export async function latestCheckpoint(db: Db | Tx, runId: string) {376 const [c] = await db377 .select()378 .from(checkpoints)379 .where(eq(checkpoints.runId, runId))380 .orderBy(desc(checkpoints.createdAt))381 .limit(1);382 return c ?? null;383}384385export async function overviewCounts(db: Db) {386 const rows = await db387 .select({ status: researchRuns.status, n: sql<number>`count(*)::int` })388 .from(researchRuns)389 .groupBy(researchRuns.status);390 const byStatus: Record<string, number> = {};391 for (const r of rows) byStatus[r.status] = r.n;392 const [cost] = await db393 .select({ total: sql<string>`coalesce(sum(${researchRuns.totalCostUsd}), 0)::text` })394 .from(researchRuns);395 return {396 byStatus,397 active: ACTIVE_RUN_STATUSES.reduce((n, s) => n + (byStatus[s] ?? 0), 0),398 totalCostUsd: Number(cost?.total ?? 0),399 };400}401402export { AppError };403