spb/autoscholar

1import {2  appendEvent,3  expireStaleLeases,4  getEnv,5  markOfflineWorkers,6  requeueRetrying,7  transition,8} from "@autoscholar/core";9import { type Db, researchRuns } from "@autoscholar/db";10import type { Logger } from "@autoscholar/observability";11import { publishRun } from "@autoscholar/publisher";12import { evaluateRun } from "@autoscholar/qa";13import type { FilesystemStorage } from "@autoscholar/storage";14import { and, eq, inArray, lt, sql } from "drizzle-orm";1516export interface TickReport {17  expiredLeases: string[];18  requeued: number;19  offlineWorkers: number;20  evaluated: string[];21  published: string[];22  timedOut: string[];23  errors: Array<{ runId?: string; step: string; error: string }>;24}2526/** One scheduler pass. Every step is idempotent, so overlapping or restarted schedulers are safe. */27export async function schedulerTick(db: Db, storage: FilesystemStorage, log: Logger): Promise<TickReport> {28  const env = getEnv();29  const report: TickReport = {30    expiredLeases: [],31    requeued: 0,32    offlineWorkers: 0,33    evaluated: [],34    published: [],35    timedOut: [],36    errors: [],37  };38  try {39    report.expiredLeases = await expireStaleLeases(db);40  } catch (e) {41    report.errors.push({ step: "expireStaleLeases", error: (e as Error).message });42  }43  try {44    report.requeued = await requeueRetrying(db);45  } catch (e) {46    report.errors.push({ step: "requeueRetrying", error: (e as Error).message });47  }48  try {49    report.offlineWorkers = await markOfflineWorkers(db);50  } catch (e) {51    report.errors.push({ step: "markOfflineWorkers", error: (e as Error).message });52  }53  // runtime budget enforced outside the LLM: request cancellation when wall clock exceeds the cap (+10% grace)54  try {55    const running = await db56      .select()57      .from(researchRuns)58      .where(59        and(60          inArray(researchRuns.status, ["running", "provisioning"]),61          eq(researchRuns.cancelRequested, false),62        ),63      );64    for (const r of running) {65      if (!r.startedAt) continue;66      const limitMs = r.maxRuntimeMinutes * 60_000 * 1.1;67      if (Date.now() - r.startedAt.getTime() > limitMs) {68        await db.transaction(async (tx) => {69          await tx70            .update(researchRuns)71            .set({ cancelRequested: true, updatedAt: new Date() })72            .where(eq(researchRuns.id, r.id));73          await transition(tx, r.id, "cancelling", {74            actor: "scheduler",75            reason: `runtime cap ${r.maxRuntimeMinutes} min exceeded`,76          });77        });78        report.timedOut.push(r.id);79      }80    }81  } catch (e) {82    report.errors.push({ step: "runtimeCap", error: (e as Error).message });83  }84  // Stage 9: quality decisions85  const toEvaluate = await db86    .select({ id: researchRuns.id })87    .from(researchRuns)88    .where(eq(researchRuns.status, "quality_check"));89  for (const r of toEvaluate) {90    try {91      const { nextStatus } = await evaluateRun(db, storage, r.id);92      report.evaluated.push(`${r.id}:${nextStatus}`);93    } catch (e) {94      report.errors.push({ runId: r.id, step: "evaluate", error: (e as Error).message });95      log.error({ runId: r.id, err: (e as Error).message }, "quality evaluation crashed");96      try {97        await db.transaction((tx) =>98          appendEvent(tx, r.id, {99            kind: "qa.error",100            level: "error",101            message: `quality evaluation error: ${(e as Error).message}`.slice(0, 2000),102            at: new Date().toISOString(),103          }),104        );105      } catch {106        /* ignore */107      }108    }109  }110  // Stage 10: publication for runs in `publishing` (auto-publish or admin-approved); retried every tick until done111  if (env.PUBLISHING_ENABLED === 1) {112    const toPublish = await db113      .select({ id: researchRuns.id })114      .from(researchRuns)115      .where(eq(researchRuns.status, "publishing"));116    for (const r of toPublish) {117      try {118        const res = await publishRun(db, storage, r.id, { actor: { type: "system" } });119        report.published.push(`${r.id}:${res.publicId}`);120      } catch (e) {121        report.errors.push({ runId: r.id, step: "publish", error: (e as Error).message });122        log.error({ runId: r.id, err: (e as Error).message }, "publication failed; will retry");123      }124    }125  }126  // stuck cancelling runs without lease for too long → cancelled127  try {128    const stuck = await db129      .select({ id: researchRuns.id })130      .from(researchRuns)131      .where(132        and(133          eq(researchRuns.status, "cancelling"),134          lt(researchRuns.updatedAt, new Date(Date.now() - 3 * env.LEASE_TTL_SECONDS * 1000)),135          sql`not exists (select 1 from job_leases l where l.run_id = ${researchRuns.id})`,136        ),137      );138    for (const r of stuck)139      await db.transaction((tx) =>140        transition(tx, r.id, "cancelled", {141          actor: "scheduler",142          reason: "no lease held; cancellation finalized",143        }),144      );145  } catch (e) {146    report.errors.push({ step: "finalizeCancel", error: (e as Error).message });147  }148  return report;149}150