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 const reserved = (run.manifest as { paperNumber?: string }).paperNumber;160 if (reserved) {161 const [taken] = await tx162 .select({ id: papers.id })163 .from(papers)164 .where(eq(papers.publicId, reserved))165 .limit(1);166 publicId = taken ? await allocatePublicId(tx, year) : reserved;167 } else publicId = await allocatePublicId(tx, year);168 slug = slugify(meta.title, publicId);169 versionNo = 1;170 const [p] = await tx171 .insert(papers)172 .values({173 publicId,174 slug,175 title: meta.title,176 status: "published",177 discipline: run.discipline,178 researchMode: run.researchMode,179 synthetic: false,180 })181 .returning({ id: papers.id });182 paperId = p!.id;183 }184 // promote artifacts (write-once copies; idempotent)185 const artifactSet: Array<{186 path: string;187 kind: string;188 sha256: string;189 bytes: number;190 contentType: string;191 publicPath: string;192 }> = [];193 for (const a of latest.values()) {194 const rule = isPromotable(a.path);195 if (!rule.promote || a.synthetic) continue;196 const pubKey = publicArtifactKey(publicId, versionNo, a.path);197 const copied = await storage.copy(a.storageKey, pubKey);198 if (copied.sha256 !== a.sha256)199 throw new AppError("integrity", `promotion digest mismatch for ${a.path}`, 500);200 await tx.update(artifacts).set({ visibility: "public" }).where(eq(artifacts.id, a.id));201 artifactSet.push({202 path: a.path,203 kind: rule.kind ?? a.kind,204 sha256: a.sha256,205 bytes: a.bytes,206 contentType: rule.contentType ?? a.contentType,207 publicPath: `/papers/${publicId}/v${versionNo}/${a.path}`,208 });209 }210 const pdfPublicPath = `/papers/${publicId}/v${versionNo}/manuscript/main.pdf`;211 const provenance = redact({212 runId,213 attemptNo: run.currentAttemptNo,214 instructionSha256: run.manifestSha256215 ? (run.manifest as { instructionSha256?: string }).instructionSha256216 : undefined,217 instructionVersion: (run.manifest as { instructionVersion?: number }).instructionVersion,218 manifestSha256: run.manifestSha256,219 model: run.model,220 reviewerModel: run.reviewerModel,221 provider: run.provider,222 qualityPolicyVersion: run.qualityPolicyVersion,223 computeProfile: run.computeProfile,224 costUsd: Number(run.totalCostUsd),225 workerName: summary?.workerName ?? null,226 sdkVersion: summary?.sdkVersion ?? null,227 toolchain: summary?.toolchain ?? {},228 stagesCompleted: summary?.stagesCompleted ?? [],229 startedAt: run.startedAt?.toISOString() ?? null,230 completedAt: attempt?.endedAt?.toISOString() ?? null,231 infrastructure: "MacLustr — Apple Silicon research compute infrastructure",232 });233 const [v] = await tx234 .insert(paperVersions)235 .values({236 paperId,237 versionNo,238 runId,239 title: meta.title,240 abstract: meta.abstract,241 keywords: meta.keywords,242 metadata: { ...meta },243 artifactSet,244 pdfPublicPath,245 qaSummary: {246 publishable: qa.publishable,247 warnings: qa.warnings ?? [],248 policyVersion: qa.policyVersion,249 evaluatedAt: qa.evaluatedAt,250 gates: ((qa.gates as Array<Record<string, unknown>>) ?? []).map((g) => ({251 gate: g.gate,252 status: g.status,253 severity: g.severity,254 message: g.message,255 })),256 },257 provenance,258 changelog: opts.changelog ?? (versionNo === 1 ? "Initial release." : "Corrected version."),259 })260 .returning({ id: paperVersions.id });261 await tx262 .update(papers)263 .set({264 currentVersionId: v!.id,265 title: meta.title,266 status: versionNo > 1 ? "corrected" : "published",267 updatedAt: new Date(),268 })269 .where(eq(papers.id, paperId));270 if (versionNo > 1) {271 await tx.insert(corrections).values({272 paperId,273 fromVersionNo: versionNo - 1,274 toVersionNo: versionNo,275 notice: opts.changelog ?? "Corrected version.",276 });277 await tx.insert(publicationEvents).values({278 paperId,279 versionId: v!.id,280 kind: "corrected",281 actor: opts.actor.type === "admin" ? `admin:${opts.actor.id}` : "system",282 detail: { runId, versionNo },283 });284 } else {285 await tx.insert(publicationEvents).values({286 paperId,287 versionId: v!.id,288 kind: "published",289 actor: opts.actor.type === "admin" ? `admin:${opts.actor.id}` : "system",290 detail: { runId, publicId },291 });292 }293 // topics from discipline (+ keyword matches on existing topics)294 const topicRows = await tx.select().from(topics);295 const chosen = new Set<string>();296 for (const t of topicRows) {297 if (t.slug === run.discipline) chosen.add(t.id);298 if (meta.keywords.some((k) => k.toLowerCase() === t.name.toLowerCase() || k.toLowerCase() === t.slug))299 chosen.add(t.id);300 }301 for (const id of chosen)302 await tx.insert(paperTopics).values({ paperId, topicId: id }).onConflictDoNothing();303 await tx.update(researchRuns).set({ paperId }).where(eq(researchRuns.id, runId));304 await transition(tx, runId, "published", {305 actor: "publisher",306 reason: `${publicId} v${versionNo}`,307 data: { paperId, publicId, slug, versionNo, artifacts: artifactSet.length },308 });309 await audit(310 tx,311 { type: opts.actor.type, id: opts.actor.id ?? null, ip: opts.actor.ip ?? null },312 "paper.publish",313 { type: "paper", id: paperId },314 { publicId, versionNo, runId, override: !!opts.override },315 );316 return { paperId, publicId, slug, versionNo, alreadyPublished: false };317 });318 return result;319}320321export async function retractPaper(322 db: Db,323 paperId: string,324 reason: string,325 notice: string,326 actor: { id: string; ip?: string | null },327) {328 return db.transaction(async (tx) => {329 const [p] = await tx.select().from(papers).where(eq(papers.id, paperId)).for("update");330 if (!p) throw notFound("paper");331 if (p.status === "retracted") throw conflict("already retracted");332 await tx.insert(retractions).values({ paperId, reason, notice, retractedBy: `admin:${actor.id}` });333 await tx334 .update(papers)335 .set({ status: "retracted", retractedAt: new Date(), updatedAt: new Date() })336 .where(eq(papers.id, paperId));337 await tx.insert(publicationEvents).values({338 paperId,339 versionId: p.currentVersionId,340 kind: "retracted",341 actor: `admin:${actor.id}`,342 detail: { reason },343 });344 await audit(345 tx,346 { type: "admin", id: actor.id, ip: actor.ip },347 "paper.retract",348 { type: "paper", id: paperId },349 { reason },350 );351 return { ok: true, publicId: p.publicId };352 });353}354355/** Withdraws artifact delivery (files) while keeping the public tombstone and internal audit. */356export async function withdrawArtifacts(357 db: Db,358 paperId: string,359 actor: { id: string; ip?: string | null },360 reason: string,361) {362 return db.transaction(async (tx) => {363 const [p] = await tx.select().from(papers).where(eq(papers.id, paperId)).for("update");364 if (!p) throw notFound("paper");365 const versions = await tx.select().from(paperVersions).where(eq(paperVersions.paperId, paperId));366 for (const v of versions)367 await tx368 .update(paperVersions)369 .set({ metadata: { ...(v.metadata as object), artifactsWithdrawn: true, withdrawnReason: reason } })370 .where(eq(paperVersions.id, v.id));371 await tx.insert(publicationEvents).values({372 paperId,373 versionId: p.currentVersionId,374 kind: "unpublished",375 actor: `admin:${actor.id}`,376 detail: { reason },377 });378 await audit(379 tx,380 { type: "admin", id: actor.id, ip: actor.ip },381 "paper.withdraw_artifacts",382 { type: "paper", id: paperId },383 { reason },384 );385 return { ok: true };386 });387}388389export async function updatePaperMetadata(390 db: Db,391 paperId: string,392 patch: { title?: string; keywords?: string[]; abstract?: string },393 actor: { id: string; ip?: string | null },394 note: string,395) {396 return db.transaction(async (tx) => {397 const [p] = await tx.select().from(papers).where(eq(papers.id, paperId)).for("update");398 if (!p?.currentVersionId) throw notFound("paper");399 const [v] = await tx400 .select()401 .from(paperVersions)402 .where(eq(paperVersions.id, p.currentVersionId))403 .limit(1);404 if (!v) throw notFound("version");405 const history = ((v.metadata as { metadataHistory?: unknown[] }).metadataHistory ?? []) as unknown[];406 history.push({407 at: new Date().toISOString(),408 by: `admin:${actor.id}`,409 note,410 previous: { title: v.title, keywords: v.keywords, abstract: v.abstract },411 });412 await tx413 .update(paperVersions)414 .set({415 title: patch.title ?? v.title,416 keywords: patch.keywords ?? v.keywords,417 abstract: patch.abstract ?? v.abstract,418 metadata: { ...(v.metadata as object), metadataHistory: history },419 })420 .where(eq(paperVersions.id, v.id));421 if (patch.title)422 await tx423 .update(papers)424 .set({ title: patch.title, updatedAt: new Date() })425 .where(eq(papers.id, paperId));426 await tx.insert(publicationEvents).values({427 paperId,428 versionId: v.id,429 kind: "metadata_updated",430 actor: `admin:${actor.id}`,431 detail: { note, fields: Object.keys(patch) },432 });433 await audit(434 tx,435 { type: "admin", id: actor.id, ip: actor.ip },436 "paper.metadata_update",437 { type: "paper", id: paperId },438 { note, fields: Object.keys(patch) },439 );440 return { ok: true };441 });442}443444export async function listPapersAdmin(db: Db) {445 return db446 .select({ p: papers, v: paperVersions, runStatus: researchRuns.status })447 .from(papers)448 .leftJoin(paperVersions, eq(papers.currentVersionId, paperVersions.id))449 .leftJoin(researchRuns, eq(paperVersions.runId, researchRuns.id))450 .orderBy(desc(papers.updatedAt));451}452453export async function appendRunEvent(454 db: Db,455 runId: string,456 kind: string,457 message: string,458 data?: Record<string, unknown>,459) {460 await db.transaction((tx) => appendEvent(tx, runId, { kind, message, data, at: new Date().toISOString() }));461}462