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