spb/autoscholar
Public
1import { AppError, appendEvent, audit, conflict, notFound, transition } from "@autoscholar/core";2import {3 artifacts,4 corrections,5 type Db,6 paperSequences,7 papers,8 paperTopics,9 paperVersions,10 publicationEvents,11 researchRuns,12 retractions,13 runAttempts,14 topics,15} from "@autoscholar/db";16import { redact } from "@autoscholar/observability";17import { isPromotable } from "@autoscholar/research-contract";18import { ManuscriptMetadata, RunSummary } from "@autoscholar/schemas";19import { type FilesystemStorage, publicArtifactKey } from "@autoscholar/storage";20import { and, desc, eq, sql } from "drizzle-orm";2122export function slugify(title: string, publicId: string): string {23 const base = title24 .toLowerCase()25 .normalize("NFKD")26 .replace(/[̀-ͯ]/g, "")27 .replace(/[^a-z0-9]+/g, "-")28 .replace(/^-+|-+$/g, "")29 .slice(0, 70)30 .replace(/-+$/, "");31 return `${publicId.toLowerCase()}-${base || "paper"}`;32}3334/** Allocates AS-WP-<year>-NNNN inside the caller's transaction (row-locked per year). */35async function allocatePublicId(36 tx: Parameters<Parameters<Db["transaction"]>[0]>[0],37 year: number,38): Promise<string> {39 await tx.insert(paperSequences).values({ year, last: 0 }).onConflictDoNothing();40 const [row] = await tx41 .update(paperSequences)42 .set({ last: sql`${paperSequences.last} + 1` })43 .where(eq(paperSequences.year, year))44 .returning({ last: paperSequences.last });45 return `AS-WP-${year}-${String(row!.last).padStart(4, "0")}`;46}4748export interface PublishOptions {49 actor: { type: "system" | "admin"; id?: string; ip?: string | null };50 /** publish as a new version (correction) of an existing paper */51 correctsPaperId?: string;52 changelog?: string;53 override?: { rationale: string };54}5556/**57 * Stage 10 — transactional, idempotent publication. Promotes public artifacts (copy + re-hash into the public store),58 * then commits paper + version + events in one transaction. Re-running after a partial failure is safe:59 * object copies are write-once and the paper_versions.run_id unique index prevents duplicates.60 */61export async function publishRun(62 db: Db,63 storage: FilesystemStorage,64 runId: string,65 opts: PublishOptions,66): Promise<{67 paperId: string;68 publicId: string;69 slug: string;70 versionNo: number;71 alreadyPublished: boolean;72}> {73 const [run] = await db.select().from(researchRuns).where(eq(researchRuns.id, runId)).limit(1);74 if (!run) throw notFound("run");75 const [existing] = await db76 .select({ v: paperVersions, p: papers })77 .from(paperVersions)78 .innerJoin(papers, eq(paperVersions.paperId, papers.id))79 .where(eq(paperVersions.runId, runId))80 .limit(1);81 if (existing) {82 if (run.status !== "published")83 await db.transaction((tx) =>84 transition(tx, runId, "published", { actor: "publisher", reason: "already published (idempotent)" }),85 );86 return {87 paperId: existing.p.id,88 publicId: existing.p.publicId,89 slug: existing.p.slug,90 versionNo: existing.v.versionNo,91 alreadyPublished: true,92 };93 }94 if (!["publishing", "ready_to_publish", "awaiting_approval"].includes(run.status))95 throw conflict(`run status ${run.status} is not publishable`);96 if (run.status !== "publishing")97 await db.transaction((tx) =>98 transition(tx, runId, "publishing", {99 actor: opts.actor.type === "admin" ? `admin:${opts.actor.id}` : "publisher",100 reason: opts.override ? `admin override: ${opts.override.rationale}` : "admin approval",101 }),102 );103104 // latest version of each artifact of the current attempt chain105 const rows = await db106 .select()107 .from(artifacts)108 .where(eq(artifacts.runId, runId))109 .orderBy(desc(artifacts.version));110 const latest = new Map<string, (typeof rows)[number]>();111 for (const r of rows) if (!latest.has(r.path)) latest.set(r.path, r);112 const metaRow = latest.get("manuscript/metadata.json");113 const pdfRow = latest.get("manuscript/main.pdf");114 if (!metaRow || !pdfRow) throw new AppError("incomplete", "manuscript metadata or PDF missing", 409);115 if (pdfRow.synthetic || metaRow.synthetic)116 throw new AppError("synthetic", "synthetic artifacts can never be published", 409);117 const meta = ManuscriptMetadata.parse(JSON.parse((await storage.get(metaRow.storageKey)).toString("utf8")));118 const summaryRow = latest.get("provenance/run-summary.json");119 const summary = summaryRow120 ? RunSummary.parse(JSON.parse((await storage.get(summaryRow.storageKey)).toString("utf8")))121 : null;122 const qaRow = latest.get("review/qa-report.json");123 const qa = qaRow124 ? (JSON.parse((await storage.get(qaRow.storageKey)).toString("utf8")) as Record<string, unknown>)125 : {};126 const [attempt] = await db127 .select()128 .from(runAttempts)129 .where(and(eq(runAttempts.runId, runId), eq(runAttempts.attemptNo, run.currentAttemptNo)))130 .limit(1);131132 const year = new Date().getUTCFullYear();133 const result = await db.transaction(async (tx) => {134 // re-check idempotency under lock135 const [again] = await tx136 .select({ id: paperVersions.id })137 .from(paperVersions)138 .where(eq(paperVersions.runId, runId))139 .limit(1);140 if (again) throw new AppError("race", "published concurrently", 409);141 let paperId: string;142 let publicId: string;143 let slug: string;144 let versionNo: number;145 if (opts.correctsPaperId) {146 const [p] = await tx.select().from(papers).where(eq(papers.id, opts.correctsPaperId)).for("update");147 if (!p) throw notFound("paper to correct");148 if (p.status === "retracted")149 throw conflict("retracted papers cannot be corrected; publish a new paper");150 const [last] = await tx151 .select({ n: sql<number>`coalesce(max(${paperVersions.versionNo}),0)::int` })152 .from(paperVersions)153 .where(eq(paperVersions.paperId, p.id));154 paperId = p.id;155 publicId = p.publicId;156 slug = p.slug;157 versionNo = (last?.n ?? 0) + 1;158 } else {159 publicId = await allocatePublicId(tx, year);160 slug = slugify(meta.title, publicId);161 versionNo = 1;162 const [p] = await tx163 .insert(papers)164 .values({165 publicId,166 slug,167 title: meta.title,168 status: "published",169 discipline: run.discipline,170 researchMode: run.researchMode,171 synthetic: false,172 })173 .returning({ id: papers.id });174 paperId = p!.id;175 }176 // promote artifacts (write-once copies; idempotent)177 const artifactSet: Array<{178 path: string;179 kind: string;180 sha256: string;181 bytes: number;182 contentType: string;183 publicPath: string;184 }> = [];185 for (const a of latest.values()) {186 const rule = isPromotable(a.path);187 if (!rule.promote || a.synthetic) continue;188 const pubKey = publicArtifactKey(publicId, versionNo, a.path);189 const copied = await storage.copy(a.storageKey, pubKey);190 if (copied.sha256 !== a.sha256)191 throw new AppError("integrity", `promotion digest mismatch for ${a.path}`, 500);192 await tx.update(artifacts).set({ visibility: "public" }).where(eq(artifacts.id, a.id));193 artifactSet.push({194 path: a.path,195 kind: rule.kind ?? a.kind,196 sha256: a.sha256,197 bytes: a.bytes,198 contentType: rule.contentType ?? a.contentType,199 publicPath: `/papers/${publicId}/v${versionNo}/${a.path}`,200 });201 }202 const pdfPublicPath = `/papers/${publicId}/v${versionNo}/manuscript/main.pdf`;203 const provenance = redact({204 runId,205 attemptNo: run.currentAttemptNo,206 instructionSha256: run.manifestSha256207 ? (run.manifest as { instructionSha256?: string }).instructionSha256208 : undefined,209 instructionVersion: (run.manifest as { instructionVersion?: number }).instructionVersion,210 manifestSha256: run.manifestSha256,211 model: run.model,212 reviewerModel: run.reviewerModel,213 provider: run.provider,214 qualityPolicyVersion: run.qualityPolicyVersion,215 computeProfile: run.computeProfile,216 costUsd: Number(run.totalCostUsd),217 workerName: summary?.workerName ?? null,218 sdkVersion: summary?.sdkVersion ?? null,219 toolchain: summary?.toolchain ?? {},220 stagesCompleted: summary?.stagesCompleted ?? [],221 startedAt: run.startedAt?.toISOString() ?? null,222 completedAt: attempt?.endedAt?.toISOString() ?? null,223 infrastructure: "MacLustr — Apple Silicon research compute infrastructure",224 });225 const [v] = await tx226 .insert(paperVersions)227 .values({228 paperId,229 versionNo,230 runId,231 title: meta.title,232 abstract: meta.abstract,233 keywords: meta.keywords,234 metadata: { ...meta },235 artifactSet,236 pdfPublicPath,237 qaSummary: {238 publishable: qa.publishable,239 warnings: qa.warnings ?? [],240 policyVersion: qa.policyVersion,241 evaluatedAt: qa.evaluatedAt,242 gates: ((qa.gates as Array<Record<string, unknown>>) ?? []).map((g) => ({243 gate: g.gate,244 status: g.status,245 severity: g.severity,246 message: g.message,247 })),248 },249 provenance,250 changelog: opts.changelog ?? (versionNo === 1 ? "Initial release." : "Corrected version."),251 })252 .returning({ id: paperVersions.id });253 await tx254 .update(papers)255 .set({256 currentVersionId: v!.id,257 title: meta.title,258 status: versionNo > 1 ? "corrected" : "published",259 updatedAt: new Date(),260 })261 .where(eq(papers.id, paperId));262 if (versionNo > 1) {263 await tx.insert(corrections).values({264 paperId,265 fromVersionNo: versionNo - 1,266 toVersionNo: versionNo,267 notice: opts.changelog ?? "Corrected version.",268 });269 await tx.insert(publicationEvents).values({270 paperId,271 versionId: v!.id,272 kind: "corrected",273 actor: opts.actor.type === "admin" ? `admin:${opts.actor.id}` : "system",274 detail: { runId, versionNo },275 });276 } else {277 await tx.insert(publicationEvents).values({278 paperId,279 versionId: v!.id,280 kind: "published",281 actor: opts.actor.type === "admin" ? `admin:${opts.actor.id}` : "system",282 detail: { runId, publicId },283 });284 }285 // topics from discipline (+ keyword matches on existing topics)286 const topicRows = await tx.select().from(topics);287 const chosen = new Set<string>();288 for (const t of topicRows) {289 if (t.slug === run.discipline) chosen.add(t.id);290 if (meta.keywords.some((k) => k.toLowerCase() === t.name.toLowerCase() || k.toLowerCase() === t.slug))291 chosen.add(t.id);292 }293 for (const id of chosen)294 await tx.insert(paperTopics).values({ paperId, topicId: id }).onConflictDoNothing();295 await tx.update(researchRuns).set({ paperId }).where(eq(researchRuns.id, runId));296 await transition(tx, runId, "published", {297 actor: "publisher",298 reason: `${publicId} v${versionNo}`,299 data: { paperId, publicId, slug, versionNo, artifacts: artifactSet.length },300 });301 await audit(302 tx,303 { type: opts.actor.type, id: opts.actor.id ?? null, ip: opts.actor.ip ?? null },304 "paper.publish",305 { type: "paper", id: paperId },306 { publicId, versionNo, runId, override: !!opts.override },307 );308 return { paperId, publicId, slug, versionNo, alreadyPublished: false };309 });310 return result;311}312313export async function retractPaper(314 db: Db,315 paperId: string,316 reason: string,317 notice: string,318 actor: { id: string; ip?: string | null },319) {320 return db.transaction(async (tx) => {321 const [p] = await tx.select().from(papers).where(eq(papers.id, paperId)).for("update");322 if (!p) throw notFound("paper");323 if (p.status === "retracted") throw conflict("already retracted");324 await tx.insert(retractions).values({ paperId, reason, notice, retractedBy: `admin:${actor.id}` });325 await tx326 .update(papers)327 .set({ status: "retracted", retractedAt: new Date(), updatedAt: new Date() })328 .where(eq(papers.id, paperId));329 await tx.insert(publicationEvents).values({330 paperId,331 versionId: p.currentVersionId,332 kind: "retracted",333 actor: `admin:${actor.id}`,334 detail: { reason },335 });336 await audit(337 tx,338 { type: "admin", id: actor.id, ip: actor.ip },339 "paper.retract",340 { type: "paper", id: paperId },341 { reason },342 );343 return { ok: true, publicId: p.publicId };344 });345}346347/** Withdraws artifact delivery (files) while keeping the public tombstone and internal audit. */348export async function withdrawArtifacts(349 db: Db,350 paperId: string,351 actor: { id: string; ip?: string | null },352 reason: string,353) {354 return db.transaction(async (tx) => {355 const [p] = await tx.select().from(papers).where(eq(papers.id, paperId)).for("update");356 if (!p) throw notFound("paper");357 const versions = await tx.select().from(paperVersions).where(eq(paperVersions.paperId, paperId));358 for (const v of versions)359 await tx360 .update(paperVersions)361 .set({ metadata: { ...(v.metadata as object), artifactsWithdrawn: true, withdrawnReason: reason } })362 .where(eq(paperVersions.id, v.id));363 await tx.insert(publicationEvents).values({364 paperId,365 versionId: p.currentVersionId,366 kind: "unpublished",367 actor: `admin:${actor.id}`,368 detail: { reason },369 });370 await audit(371 tx,372 { type: "admin", id: actor.id, ip: actor.ip },373 "paper.withdraw_artifacts",374 { type: "paper", id: paperId },375 { reason },376 );377 return { ok: true };378 });379}380381export async function updatePaperMetadata(382 db: Db,383 paperId: string,384 patch: { title?: string; keywords?: string[]; abstract?: string },385 actor: { id: string; ip?: string | null },386 note: string,387) {388 return db.transaction(async (tx) => {389 const [p] = await tx.select().from(papers).where(eq(papers.id, paperId)).for("update");390 if (!p?.currentVersionId) throw notFound("paper");391 const [v] = await tx392 .select()393 .from(paperVersions)394 .where(eq(paperVersions.id, p.currentVersionId))395 .limit(1);396 if (!v) throw notFound("version");397 const history = ((v.metadata as { metadataHistory?: unknown[] }).metadataHistory ?? []) as unknown[];398 history.push({399 at: new Date().toISOString(),400 by: `admin:${actor.id}`,401 note,402 previous: { title: v.title, keywords: v.keywords, abstract: v.abstract },403 });404 await tx405 .update(paperVersions)406 .set({407 title: patch.title ?? v.title,408 keywords: patch.keywords ?? v.keywords,409 abstract: patch.abstract ?? v.abstract,410 metadata: { ...(v.metadata as object), metadataHistory: history },411 })412 .where(eq(paperVersions.id, v.id));413 if (patch.title)414 await tx415 .update(papers)416 .set({ title: patch.title, updatedAt: new Date() })417 .where(eq(papers.id, paperId));418 await tx.insert(publicationEvents).values({419 paperId,420 versionId: v.id,421 kind: "metadata_updated",422 actor: `admin:${actor.id}`,423 detail: { note, fields: Object.keys(patch) },424 });425 await audit(426 tx,427 { type: "admin", id: actor.id, ip: actor.ip },428 "paper.metadata_update",429 { type: "paper", id: paperId },430 { note, fields: Object.keys(patch) },431 );432 return { ok: true };433 });434}435436export async function listPapersAdmin(db: Db) {437 return db438 .select({ p: papers, v: paperVersions, runStatus: researchRuns.status })439 .from(papers)440 .leftJoin(paperVersions, eq(papers.currentVersionId, paperVersions.id))441 .leftJoin(researchRuns, eq(paperVersions.runId, researchRuns.id))442 .orderBy(desc(papers.updatedAt));443}444445export async function appendRunEvent(446 db: Db,447 runId: string,448 kind: string,449 message: string,450 data?: Record<string, unknown>,451) {452 await db.transaction((tx) => appendEvent(tx, runId, { kind, message, data, at: new Date().toISOString() }));453}454