Repository Public
1import { sql, j } from "../lib/db";2import { assertCanSpend, recordCost } from "../lib/budget";3import { hashOf } from "../lib/hash";4import { log, redact } from "../lib/log";5import { ProviderError, type Usage } from "../providers/types";67/**8 * Every paid provider call goes through here.9 * - idempotent: the same (kind, model, input) is paid for once; later calls reuse the stored result10 * - resumable: a video job keeps its provider request_id, so a crashed worker polls instead of resubmitting11 * - budgeted: the gate runs before any new spend12 * - audited: prompt, settings, model, timestamps, cost and output are stored13 */14export interface JobRow {15 id: string; idem_key: string; status: "queued" | "submitted" | "done" | "failed"; request_id: string | null; attempts: number;16 output: unknown; asset_id: string | null; cost_usd: number | null; error: string | null; started_at: Date | null;17}18export interface JobSpec<T> {19 projectId: string | null;20 shotId?: string | null;21 stage: string;22 kind: "llm" | "image" | "video" | "tts" | "search";23 provider: string;24 model: string;25 settings?: Record<string, unknown>;26 prompt?: string;27 /** Everything that determines the output. Changing it (e.g. a revision counter) buys a new generation. */28 input: unknown;29 estimateUsd: number | null;30 /** Operator explicitly accepted an unknown price. */31 approvedUnknown?: boolean;32 exec: (ctx: { job: JobRow; setRequestId: (id: string) => Promise<void> }) => Promise<{ output: T; assetId?: string | null; usage: Usage }>;33}34export interface JobResult<T> { output: T; assetId: string | null; costUsd: number | null; reused: boolean; jobId: string }3536export const MAX_ATTEMPTS_PER_KEY = 3;3738/** Who claims jobs in this process ("worker:<host>" in the worker, "cli" elsewhere). Used to recover after a restart. */39let claimer = "cli";40export function setClaimer(id: string) { claimer = id; }41/**42 * Called once when a worker starts: calls that were in flight in the previous worker process on this host and43 * have no provider request id (LLM, image, speech) can never complete, so they are released for a clean retry.44 * Video jobs keep their request id and are resumed instead.45 */46export async function releaseOrphans(id: string): Promise<number> {47 const rows = await sql`UPDATE provider_jobs SET status = 'queued', started_at = NULL48 WHERE status = 'submitted' AND request_id IS NULL AND settings->>'claimedBy' = ${id} RETURNING id`;49 return rows.length;50}51const STALE_MS = 15 * 60_000;5253export const idemKey = (kind: string, model: string, input: unknown) => hashOf({ kind, model, input });5455export class JobInProgressError extends Error {56 constructor() { super("This generation is already running"); this.name = "JobInProgressError"; }57}5859export async function runProviderJob<T>(spec: JobSpec<T>): Promise<JobResult<T>> {60 const key = idemKey(spec.kind, spec.model, spec.input);61 await sql`62 INSERT INTO provider_jobs (idem_key, project_id, shot_id, stage, kind, provider, model, settings, prompt, input_hash, est_usd)63 VALUES (${key}, ${spec.projectId}, ${spec.shotId ?? null}, ${spec.stage}, ${spec.kind}, ${spec.provider}, ${spec.model}, ${j(spec.settings ?? {})}, ${spec.prompt ?? null}, ${hashOf(spec.input)}, ${spec.estimateUsd})64 ON CONFLICT (idem_key) DO NOTHING`;65 const [existing] = await sql<JobRow[]>`SELECT * FROM provider_jobs WHERE idem_key = ${key}`;6667 if (existing.status === "done") {68 log("info", "job.reused", { key: key.slice(0, 12), kind: spec.kind, stage: spec.stage });69 return { output: existing.output as T, assetId: existing.asset_id, costUsd: 0, reused: true, jobId: existing.id };70 }7172 const resuming = existing.status === "submitted" && !!existing.request_id;73 if (existing.status === "submitted" && !resuming) {74 const age = Date.now() - (existing.started_at?.getTime() ?? 0);75 if (age < STALE_MS) throw new JobInProgressError();76 }77 if (!resuming) {78 if (existing.attempts >= MAX_ATTEMPTS_PER_KEY) {79 throw new ProviderError(`This exact generation already failed ${existing.attempts} times (${existing.error ?? "unknown error"}). Change the prompt or settings before trying again.`, { retryable: false, code: "attempts_exhausted" });80 }81 await assertCanSpend({ projectId: spec.projectId, shotId: spec.shotId ?? null, estimate: spec.estimateUsd, approvedUnknown: spec.approvedUnknown });82 // Atomic claim: only one worker moves the row to "submitted".83 // A failed video job keeps its provider request id when the failure was on our side of the wire (network,84 // download): the clip exists and is paid for, so the retry resumes it instead of buying it again.85 const keep = spec.kind === "video" && !!existing.request_id && /network error|download failed|timed out/i.test(existing.error ?? "");86 const claimed = await sql<JobRow[]>`87 UPDATE provider_jobs SET status = 'submitted', started_at = now(), attempts = attempts + 1, error = NULL, request_id = ${keep ? existing.request_id : null}, est_usd = ${spec.estimateUsd},88 settings = settings || ${j({ claimedBy: claimer })}89 WHERE id = ${existing.id} AND (status IN ('queued','failed') OR (status = 'submitted' AND request_id IS NULL AND started_at < now() - interval '15 minutes'))90 RETURNING *`;91 if (!claimed.length) throw new JobInProgressError();92 existing.status = "submitted";93 existing.request_id = keep ? existing.request_id : null;94 }9596 const job = existing;97 try {98 const res = await spec.exec({99 job,100 setRequestId: async (id) => { job.request_id = id; await sql`UPDATE provider_jobs SET request_id = ${id} WHERE id = ${job.id}`; },101 });102 const cost = res.usage.costUsd;103 await sql`104 UPDATE provider_jobs SET status = 'done', finished_at = now(), output = ${j(res.output)}, asset_id = ${res.assetId ?? null},105 cost_usd = ${cost}, cost_source = ${res.usage.source}106 WHERE id = ${job.id}`;107 if (cost !== null) await recordCost({ jobId: job.id, projectId: spec.projectId, shotId: spec.shotId ?? null, kind: spec.kind, model: spec.model, usd: cost, source: res.usage.source });108 log("info", "job.done", { kind: spec.kind, stage: spec.stage, model: spec.model, usd: cost, resumed: resuming });109 return { output: res.output, assetId: res.assetId ?? null, costUsd: cost, reused: false, jobId: job.id };110 } catch (e) {111 const err = e as Error & { usage?: Usage };112 await sql`UPDATE provider_jobs SET status = 'failed', finished_at = now(), error = ${redact(err.message).slice(0, 1000)}, cost_usd = ${err.usage?.costUsd ?? null}, cost_source = ${err.usage?.source ?? null} WHERE id = ${job.id}`;113 if (err.usage?.costUsd) await recordCost({ jobId: job.id, projectId: spec.projectId, shotId: spec.shotId ?? null, kind: spec.kind, model: spec.model, usd: err.usage.costUsd, source: err.usage.source, failed: true });114 log("error", "job.failed", { kind: spec.kind, stage: spec.stage, model: spec.model, error: err.message });115 throw e;116 }117}118