diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 948d735..25e761b 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -25,6 +25,8 @@ import { createCloudAgentConnection, createConnectorStack, createJobReconciler, + createMediaOperations, + createMediaRuntime, createMessagingContextLoader, createMessagingTeamChatSender, createProfileConsolidationScheduler, @@ -47,6 +49,7 @@ import { isPipedreamEnabled, LocalAgentHomeStore, LocalArtifactStore, + loadMediaSettings, McpConnector, McpOAuthBroker, messagingPlatformsFromEnv, @@ -375,6 +378,15 @@ export async function createApp( CLOUD_AGENT_SPACE_ID: env.cloudAgentSpaceId, }); const shutdown = new AbortController(); + const media = createMediaRuntime({ + prisma, + events, + jobs, + artifacts, + registrations: createMediaOperations({ + loadSettings: () => loadMediaSettings({ prisma, secrets }), + }), + }); const executor = createRunExecutor({ prisma, runtime, @@ -411,6 +423,7 @@ export async function createApp( messaging: messaging ? createMessagingContextLoader(prisma) : undefined, web: createWebProvider(), cloudAgent, + media, shutdownSignal: shutdown.signal, }); @@ -428,6 +441,7 @@ export async function createApp( deploymentModelKey: env.deploymentModelKey, messaging, cloudAgent, + media, }); if (inMemoryJobs) { await inMemoryJobs.start(jobHandlers); @@ -446,6 +460,7 @@ export async function createApp( reconcileComputerUpdates: () => reconcileComputerUpdates({ prisma, jobs }), reconcileProfileConsolidation: createProfileConsolidationScheduler({ prisma, jobs }), reconcileDelegatedTasks: () => reconcileDelegatedTasks({ prisma, jobs, events }), + reconcileMediaJobs: () => media.reconcile(), }) : undefined; reconciler?.start(); diff --git a/apps/web/src/lib/thread-events.ts b/apps/web/src/lib/thread-events.ts index 51272f6..bf1fb53 100644 --- a/apps/web/src/lib/thread-events.ts +++ b/apps/web/src/lib/thread-events.ts @@ -33,6 +33,7 @@ const runTriggers = new Set([ "messaging", "cloud_agent", "delegation", + "media", ]); function runFromStartedEvent(event: ProductEvent, previous: Run | undefined): Run { diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 804281c..643cb70 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -10,6 +10,8 @@ import { createCloudAgentConnection, createConnectorStack, createJobReconciler, + createMediaOperations, + createMediaRuntime, createMessagingContextLoader, createPostgresReconciliationLeadership, createProfileConsolidationScheduler, @@ -29,6 +31,7 @@ import { isPipedreamEnabled, LocalAgentHomeStore, LocalArtifactStore, + loadMediaSettings, McpConnector, McpOAuthBroker, messagingEnvFromProcess, @@ -164,6 +167,15 @@ async function main() { }); // One provider instance so emulator launches and polls share the same Map. const cloudAgent = createCloudAgentConnection(); + const media = createMediaRuntime({ + prisma, + events, + jobs, + artifacts, + registrations: createMediaOperations({ + loadSettings: () => loadMediaSettings({ prisma, secrets }), + }), + }); const executor = createRunExecutor({ prisma, runtime, @@ -199,6 +211,7 @@ async function main() { messaging: messaging ? createMessagingContextLoader(prisma) : undefined, web: createWebProvider(), cloudAgent, + media, }); const jobHandlers = createBackgroundJobHandlers({ @@ -215,6 +228,7 @@ async function main() { deploymentModelKey, messaging, cloudAgent, + media, }); // graphile-worker run() connects through the shared pool. createPool already // retries connect() on 53300 a finite number of times. Keep retrying start @@ -244,6 +258,7 @@ async function main() { reconcileComputerUpdates: () => reconcileComputerUpdates({ prisma, jobs }), reconcileProfileConsolidation: createProfileConsolidationScheduler({ prisma, jobs }), reconcileDelegatedTasks: () => reconcileDelegatedTasks({ prisma, jobs, events }), + reconcileMediaJobs: () => media.reconcile(), }); reconciler.start(); diff --git a/packages/adapter-kit/src/background-jobs.test.ts b/packages/adapter-kit/src/background-jobs.test.ts index 13880d7..1846253 100644 --- a/packages/adapter-kit/src/background-jobs.test.ts +++ b/packages/adapter-kit/src/background-jobs.test.ts @@ -3,6 +3,9 @@ import { dispatchBackgroundJob, historyCompactJob, historyCompactJobKey, + MEDIA_JOB_QUEUE, + mediaRunJob, + mediaRunJobKey, messagingDeliverJob, parseBackgroundJob, profileConsolidateJob, @@ -22,6 +25,7 @@ function handlers(): BackgroundJobHandlers { "messaging.deliver": vi.fn(async () => undefined), "cloud_agent.poll": vi.fn(async () => undefined), "profile.consolidate": vi.fn(async () => undefined), + "media.run": vi.fn(async () => undefined), }; } @@ -112,3 +116,21 @@ describe("profileConsolidateJob", () => { expect(profileConsolidateJobKey("user-1")).not.toBe(profileConsolidateJobKey("user-2")); }); }); + +describe("mediaRunJob", () => { + it("validates, dispatches, and runs every media job on the shared media queue", async () => { + const target = handlers(); + await dispatchBackgroundJob(target, "media.run", { jobId: "job-1" }); + expect(target["media.run"]).toHaveBeenCalledWith({ jobId: "job-1" }); + expect(() => parseBackgroundJob("media.run", { jobId: "" })).toThrow(); + expect(mediaRunJob("job-1")).toEqual({ + name: "media.run", + payload: { jobId: "job-1" }, + replaceKey: mediaRunJobKey("job-1"), + queue: MEDIA_JOB_QUEUE, + }); + const at = new Date("2026-10-03T12:00:00.000Z"); + expect(mediaRunJob("job-1", at).availableAt).toBe(at); + expect(mediaRunJobKey("job-1")).not.toBe(mediaRunJobKey("job-2")); + }); +}); diff --git a/packages/adapter-kit/src/background-jobs.ts b/packages/adapter-kit/src/background-jobs.ts index 45cad56..6705ad5 100644 --- a/packages/adapter-kit/src/background-jobs.ts +++ b/packages/adapter-kit/src/background-jobs.ts @@ -23,6 +23,7 @@ const payloadSchemas = { "messaging.deliver": z.object({ runId: z.string().min(1).optional() }), "cloud_agent.poll": z.object({ agentId: z.string().min(1) }), "profile.consolidate": z.object({ userId: z.string().min(1) }), + "media.run": z.object({ jobId: z.string().min(1) }), } satisfies { [Name in BackgroundJobName]: z.ZodType }; export function parseBackgroundJob(name: string, payload: unknown): BackgroundJob { @@ -160,3 +161,20 @@ export function profileConsolidateJob(userId: string): BackgroundJob { replaceKey: profileConsolidateJobKey(userId), }; } + +/** Media jobs share one queue so a single GPU never runs two generations at once. */ +export const MEDIA_JOB_QUEUE = "media"; + +export function mediaRunJobKey(jobId: string): string { + return `media.run:${jobId}`; +} + +export function mediaRunJob(jobId: string, availableAt?: Date): BackgroundJob { + return { + name: "media.run", + payload: { jobId }, + replaceKey: mediaRunJobKey(jobId), + queue: MEDIA_JOB_QUEUE, + ...(availableAt ? { availableAt } : {}), + }; +} diff --git a/packages/adapter-kit/src/media.ts b/packages/adapter-kit/src/media.ts index 4f57c4c..c9610ea 100644 --- a/packages/adapter-kit/src/media.ts +++ b/packages/adapter-kit/src/media.ts @@ -195,13 +195,19 @@ export interface StartMediaJobInput { operation: MediaOperation; /** Validated by the operation's input schema before the job is created. */ input: unknown; - title: string; + /** Card label; the operation derives one from the input when omitted. */ + title?: string; spaceId: string; userId: string; + /** Author of the media card; a job started from a card keeps that card's bot. */ botId: string | null; groupId: string | null; threadId: string; runId: string | null; + /** Wake the starting bot with the job's summary when the job ends. */ + continueAfter?: boolean; + /** A replay with the same key returns the job created first (tool call idempotency). */ + operationKey?: string; } export interface MediaJobSnapshot { diff --git a/packages/adapter-kit/src/types.ts b/packages/adapter-kit/src/types.ts index d77cb5e..e320b18 100644 --- a/packages/adapter-kit/src/types.ts +++ b/packages/adapter-kit/src/types.ts @@ -524,6 +524,8 @@ export interface BackgroundJobPayloads { "cloud_agent.poll": { agentId: string }; /** Learn durable user facts from conversations since the profile's last consolidation. */ "profile.consolidate": { userId: string }; + /** Run one media job; scope, input and lease live on the job row. */ + "media.run": { jobId: string }; } export type BackgroundJobName = keyof BackgroundJobPayloads; @@ -534,6 +536,8 @@ export type BackgroundJob = { payload: BackgroundJobPayloads[Name]; availableAt?: Date; replaceKey?: string; + /** Jobs that share a queue run one at a time, in order, across every worker process. */ + queue?: string; }; }[BackgroundJobName]; diff --git a/packages/adapters/src/background-job-handlers.ts b/packages/adapters/src/background-job-handlers.ts index af869ac..1c45073 100644 --- a/packages/adapters/src/background-job-handlers.ts +++ b/packages/adapters/src/background-job-handlers.ts @@ -16,6 +16,7 @@ import { scheduleComputerSleep, sleepComputerIfIdle } from "./computer-idle.js"; import { performComputerUpdate } from "./computer-update.js"; import type { createRunExecutor } from "./executor.js"; import { compactHistory } from "./history-compaction.js"; +import type { MediaRuntimeHost } from "./media-runtime.js"; import type { MemoryProviderResolver } from "./memory-provider-factory.js"; import { deliverMessagingOutbound, mirrorMessagingOutbound } from "./messaging-delivery.js"; import { consolidateProfile } from "./profile-consolidation.js"; @@ -36,6 +37,7 @@ export function createBackgroundJobHandlers(deps: { deploymentModelKey?: string; messaging?: MessagingSurface; cloudAgent?: CloudAgentConnection | null; + media?: MediaRuntimeHost; }): BackgroundJobHandlers { const deliverMessaging = async (runId?: string) => { if (!deps.messaging) return; @@ -112,6 +114,9 @@ export function createBackgroundJobHandlers(deps: { payload.threadId, ); }, + "media.run": async (payload) => { + await deps.media?.runJob(payload.jobId); + }, "profile.consolidate": async (payload) => { await consolidateProfile( { diff --git a/packages/adapters/src/executor-types.ts b/packages/adapters/src/executor-types.ts index c32ea62..0215af5 100644 --- a/packages/adapters/src/executor-types.ts +++ b/packages/adapters/src/executor-types.ts @@ -6,6 +6,7 @@ import type { ConnectorProvider, JobPublisher, ManagedConnectorProvider, + MediaRuntime, MemoryStore, NotificationProvider, SandboxProvider, @@ -45,6 +46,8 @@ export interface ExecutorDeps { secretHttp?: RemoteTransportDependencies; /** Remote cloud coding agents. Null/omit means tools stay uninjected. */ cloudAgent?: CloudAgentConnection | null; + /** Media jobs (generation, edits, transcription). Omit means media tools stay hidden. */ + media?: MediaRuntime; /** Aborted when createApp stop() begins so in-flight continueRun boot waits exit promptly. */ shutdownSignal?: AbortSignal; } diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index 7b4ecd0..29306fd 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -76,6 +76,10 @@ export * from "./mcp-emulator.js"; export * from "./mcp-oauth.js"; export * from "./mcp-server-tool.js"; export * from "./mcp-transport.js"; +export * from "./media-operations.js"; +export * from "./media-runtime.js"; +export * from "./media-settings.js"; +export * from "./media-tools.js"; export * from "./memory-provider-factory.js"; export * from "./messaging-context.js"; export * from "./messaging-delivery.js"; diff --git a/packages/adapters/src/job-reconciler.ts b/packages/adapters/src/job-reconciler.ts index c5e5d07..b299aef 100644 --- a/packages/adapters/src/job-reconciler.ts +++ b/packages/adapters/src/job-reconciler.ts @@ -109,6 +109,7 @@ export function createJobReconciler( /** Enqueues daily profile consolidation; see createProfileConsolidationScheduler. */ reconcileProfileConsolidation?: () => Promise; reconcileDelegatedTasks?: () => Promise; + reconcileMediaJobs?: () => Promise; }, options: { intervalMs?: number; batchSize?: number } = {}, ) { @@ -132,6 +133,7 @@ export function createJobReconciler( deps.reconcileComputerUpdates, deps.reconcileProfileConsolidation, deps.reconcileDelegatedTasks, + deps.reconcileMediaJobs, ].map(async (reconcile) => reconcile?.()), ); for (const result of auxiliary) { diff --git a/packages/adapters/src/media-operations.ts b/packages/adapters/src/media-operations.ts new file mode 100644 index 0000000..98dbf7e --- /dev/null +++ b/packages/adapters/src/media-operations.ts @@ -0,0 +1,15 @@ +import type { MediaSettings } from "@rakazo/contracts"; +import type { MediaOperationRegistration } from "./media-runtime.js"; + +export interface MediaOperationDeps { + /** Current deployment-wide media backends; read per call so edits apply without a restart. */ + loadSettings(): Promise; +} + +/** + * Every media operation this deployment can run. The API and the worker build their runtime + * from this one list, so both processes run the same operations. + */ +export function createMediaOperations(_deps: MediaOperationDeps): MediaOperationRegistration[] { + return []; +} diff --git a/packages/adapters/src/media-runtime.ts b/packages/adapters/src/media-runtime.ts new file mode 100644 index 0000000..681f2e9 --- /dev/null +++ b/packages/adapters/src/media-runtime.ts @@ -0,0 +1,1041 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import type { + AdapterContext, + ArtifactStore, + GpuStatus, + JobPublisher, + MediaCapabilities, + MediaJobContext, + MediaJobInputFile, + MediaJobOutputFile, + MediaJobSnapshot, + MediaOperationHandler, + MediaRuntime, + StartMediaJobInput, +} from "@rakazo/adapter-kit"; +import { + MediaJobDeferredError, + mediaRunJob, + mediaRunJobKey, + runContinueJob, +} from "@rakazo/adapter-kit"; +import { + ARTIFACT_MAX_BYTES_BY_KIND, + attachmentKindOf, + MEDIA_JOB_ACTIVE_STATUSES, + type MediaJobBlock, + type MediaJobStatus, + type MediaOperation, + MediaOperationSchema, + type MediaOutput, + type MessageBlock, +} from "@rakazo/contracts"; +import { + attachmentExtensionForMimeType, + inferAttachmentMimeType, + validateAttachmentMimeType, +} from "@rakazo/core"; +import { + appendEventInTransaction, + createThreadMessageInTransaction, + type MediaJob, + Prisma, + type PrismaClient, + type ThreadEvents, + withTransactionRetry, +} from "@rakazo/db"; +import { getLogger } from "@rakazo/logging"; +import type { z } from "zod"; + +/** Run trigger of the turn that wakes a bot when its media job ends (`continue_after`). */ +export const MEDIA_WAKE_TRIGGER = "media"; + +/** + * One media operation (`generate_image`, `edit_video`, …): how to validate and label its input, + * when it is available, and the handler that does the work inside a media job. + */ +export interface MediaOperationRegistration { + operation: MediaOperation; + /** Validates tool and card input. The job stores the parsed value, so it must parse again. */ + inputSchema: z.ZodType; + /** Card label for a job, e.g. the prompt or what the edit does. */ + title(input: Input): string; + /** Whether the space can run the operation now (backend configured, binaries present). */ + available(context: { spaceId: string }): Promise; + handler: MediaOperationHandler; + /** GPU guard: the runner waits while the backend reports less free VRAM than this. */ + gpu?: { minVramMb: number; status(signal?: AbortSignal): Promise }; + /** Artifact ids the input reads, so starting a job refuses ones the user cannot read. */ + artifactIds?(input: Input): string[]; +} + +/** A media job that cannot start. The message is safe to show to the user and the model. */ +export class MediaJobRejectedError extends Error { + constructor(message: string) { + super(message); + this.name = "MediaJobRejectedError"; + } +} + +export interface MediaRuntimeDeps { + prisma: PrismaClient; + events: Pick; + jobs: JobPublisher; + artifacts: ArtifactStore; + registrations: readonly MediaOperationRegistration[]; +} + +export interface MediaRuntimeOptions { + /** Runner lease; a crashed runner's job is recovered after it expires. */ + leaseMs?: number; + /** Lease renewal and cancel polling interval while a job runs. */ + heartbeatMs?: number; + /** Minimum time between two progress writes to the card. */ + progressIntervalMs?: number; + /** Deferrals (GPU busy, backend not ready) before a job fails. */ + maxDeferrals?: number; + /** Longest wait between two attempts of a deferred job. */ + maxDeferMs?: number; + /** Restarts after a runner died mid-job before the job fails. */ + maxRecoveries?: number; + /** Hard cap on one attempt. */ + maxRunMs?: number; + /** How long a handler may ignore an abort before the runner settles without it. */ + abortGraceMs?: number; + /** Parent of the per-job temp directories. */ + tempRoot?: string; +} + +/** The runtime plus what the background job handler and the reconciler call. */ +export interface MediaRuntimeHost extends MediaRuntime { + /** Runs one job (`media.run`); safe to call for any id, any number of times. */ + runJob(jobId: string): Promise; + /** Re-enqueues jobs whose enqueue was lost, whose runner died, or whose wait is over. */ + reconcile(): Promise; +} + +const ACTIVE_STATUSES: MediaJobStatus[] = [...MEDIA_JOB_ACTIVE_STATUSES]; +const TITLE_MAX_LENGTH = 200; +const ERROR_MAX_LENGTH = 500; +const SUMMARY_MAX_LENGTH = 4_000; +const GPU_STATUS_TIMEOUT_MS = 10_000; +const GPU_RETRY_MS = 15_000; +/** Retry delay while another media job holds the GPU. */ +const BUSY_RETRY_MS = 5_000; +/** A due job not picked up by its queue in this time is re-enqueued outside the queue. */ +const STUCK_QUEUE_MS = 120_000; +/** Due jobs younger than this are left to the enqueue that created them. */ +const RECONCILE_GRACE_MS = 30_000; +const RECONCILE_BATCH_SIZE = 100; +// Shares the reconciliation lock namespace (job-reconciler.ts) with its own id. +const MEDIA_CLAIM_LOCK_NAMESPACE = 1_380_019_075; +const MEDIA_CLAIM_LOCK_ID = 2; + +/** Card errors. They are shown to the user, so they stay short and actionable. */ +const UNAVAILABLE_ERROR = "This media operation is not available right now."; +const INTERRUPTED_ERROR = "The job was interrupted. Try again."; +const BUSY_ERROR = "The media server stayed busy. Try again later."; +const TIMEOUT_ERROR = "The job took too long."; +const FAILED_ERROR = "The job failed."; +const MISSING_SOURCE_ERROR = "A source file is no longer available."; +const TOO_LARGE_ERROR = "The result is too large to keep."; + +type AbortReason = "cancel" | "lease" | "abandoned" | "timeout"; + +function clamp(text: string, max: number): string { + return text.length > max ? `${text.slice(0, max - 1)}…` : text; +} + +function issuesText(error: z.ZodError): string { + return error.issues + .slice(0, 5) + .map((issue) => + issue.path.length ? `${issue.path.join(".")}: ${issue.message}` : issue.message, + ) + .join("; "); +} + +function cardBlock(job: MediaJob, outputs: MediaOutput[]): MediaJobBlock { + return { + kind: "media", + jobId: job.id, + operation: job.operation as MediaOperation, + status: job.status as MediaJobStatus, + title: job.title, + ...(job.progress === null ? {} : { progress: job.progress }), + ...(job.error === null ? {} : { error: job.error }), + outputs, + }; +} + +function mediaBlockOf(blocks: unknown, jobId: string): MediaJobBlock | undefined { + if (!Array.isArray(blocks)) return undefined; + return (blocks as MessageBlock[]).find( + (block): block is MediaJobBlock => block.kind === "media" && block.jobId === jobId, + ); +} + +function positiveInt(value: number | undefined): number | undefined { + if (value === undefined || !Number.isFinite(value)) return undefined; + const rounded = Math.round(value); + return rounded > 0 ? rounded : undefined; +} + +function nonNegativeInt(value: number | undefined): number | undefined { + if (value === undefined || !Number.isFinite(value)) return undefined; + const rounded = Math.round(value); + return rounded >= 0 ? rounded : undefined; +} + +function insideDir(dir: string, file: string): string { + const resolved = path.resolve(dir, file); + if (!resolved.startsWith(`${path.resolve(dir)}${path.sep}`)) { + throw new Error("Media output must be a file in the job's temp directory"); + } + return resolved; +} + +/** The sentence a bot is woken with when its job ends; job text is data, not instructions. */ +export function mediaWakePrompt(job: { + id: string; + operation: string; + title: string; + status: string; + error: string | null; + summary: string; + outputs: MediaOutput[]; +}): string { + const label = `Media job ${job.id} (${job.operation}: ${JSON.stringify(job.title)})`; + if (job.status === "cancelled") { + return `${label} was cancelled. Do not start it again unless the user asks.`; + } + if (job.status === "failed") { + return `${label} failed: ${job.error ?? FAILED_ERROR} Continue your task if you can, otherwise tell the user.`; + } + const outputs = job.outputs.map((output) => `${output.name} (${output.artifactId})`).join(", "); + return [ + `${label} is done.`, + job.summary ? `Result: ${clamp(job.summary, SUMMARY_MAX_LENGTH)}` : "", + outputs ? `Outputs: ${outputs}.` : "", + "The media card already shows the result to the user; continue your task.", + ] + .filter(Boolean) + .join(" "); +} + +export function createMediaRuntime( + deps: MediaRuntimeDeps, + options: MediaRuntimeOptions = {}, +): MediaRuntimeHost { + const leaseMs = options.leaseMs ?? 90_000; + const heartbeatMs = options.heartbeatMs ?? 5_000; + const progressIntervalMs = options.progressIntervalMs ?? 2_000; + const maxDeferrals = options.maxDeferrals ?? 30; + const maxDeferMs = options.maxDeferMs ?? 60_000; + const maxRecoveries = options.maxRecoveries ?? 2; + const maxRunMs = options.maxRunMs ?? 2 * 60 * 60_000; + const abortGraceMs = options.abortGraceMs ?? 15_000; + const tempRoot = options.tempRoot ?? path.join(tmpdir(), "evoo-media"); + const registry = new Map(); + for (const registration of deps.registrations) { + if (registry.has(registration.operation)) { + throw new Error(`Media operation ${registration.operation} is registered twice`); + } + registry.set(registration.operation, registration); + } + /** Jobs this process runs, so a cancel here stops them without waiting for a heartbeat. */ + const running = new Map(); + + const isAvailable = async ( + registration: MediaOperationRegistration | undefined, + spaceId: string, + ): Promise => { + if (!registration) return false; + try { + return await registration.available({ spaceId }); + } catch (error) { + getLogger().warn("media operation availability check failed", { + operation: registration.operation, + error: error instanceof Error ? error.message : String(error), + }); + return false; + } + }; + + const notify = async (threadId: string, seq: number | undefined) => { + if (seq === undefined) return; + await deps.events.notify(threadId, seq).catch((error) => { + getLogger().error("media card realtime notification", error); + }); + }; + + const enqueue = async (jobId: string, availableAt?: Date) => { + await deps.jobs.enqueue(mediaRunJob(jobId, availableAt)).catch((error) => { + // The job row is durable; the reconciler enqueues it again. + getLogger().error("media job enqueue", error); + }); + }; + + /** Rewrites the job's media block; returns the event seq when the card changed. */ + const writeCard = async ( + tx: Prisma.TransactionClient, + job: MediaJob, + appended: MediaOutput[] = [], + ): Promise<{ exists: boolean; seq?: number; outputs: MediaOutput[] }> => { + if (!job.messageId) return { exists: false, outputs: appended }; + const message = await tx.message.findFirst({ + where: { id: job.messageId, threadId: job.threadId }, + select: { id: true, role: true, botId: true, blocks: true }, + }); + if (!message) return { exists: false, outputs: appended }; + const previous = mediaBlockOf(message.blocks, job.id); + const outputs = [...(previous?.outputs ?? []), ...appended]; + const next = cardBlock(job, outputs); + const current = Array.isArray(message.blocks) ? (message.blocks as MessageBlock[]) : []; + const blocks = previous + ? current.map((block) => (block === previous ? next : block)) + : [...current, next]; + if (JSON.stringify(blocks) === JSON.stringify(message.blocks)) { + return { exists: true, outputs }; + } + await tx.message.update({ + where: { id: message.id }, + data: { blocks: blocks as Prisma.InputJsonValue }, + }); + const event = await appendEventInTransaction(tx, { + spaceId: job.spaceId, + threadId: job.threadId, + botId: message.botId ?? job.botId ?? "", + type: "thread.message.updated", + payload: { messageId: message.id, role: message.role, blocks }, + }); + return { exists: true, seq: event.seq, outputs }; + }; + + /** + * Applies `data` to the job when the fence still matches and mirrors the job on its card in + * the same transaction. A null token fences on "not held by a live runner". + */ + const updateJob = async ( + jobId: string, + token: string | null, + data: Prisma.MediaJobUpdateManyMutationInput, + appended: MediaOutput[] = [], + ) => { + const committed = await withTransactionRetry(() => + deps.prisma.$transaction(async (tx) => { + const now = new Date(); + const fence: Prisma.MediaJobWhereInput = token + ? { id: jobId, leaseToken: token } + : { + id: jobId, + status: { in: ACTIVE_STATUSES }, + OR: [{ leaseToken: null }, { leaseExpiresAt: { lte: now } }], + }; + const before = await tx.mediaJob.findFirst({ where: fence, select: { threadId: true } }); + if (!before) return null; + // Thread row first, like clearing a chat, so a clear and a card write cannot deadlock. + await tx.thread.updateMany({ + where: { id: before.threadId }, + data: { nextEventSeq: { increment: 0 } }, + }); + const updated = await tx.mediaJob.updateMany({ where: fence, data }); + if (!updated.count) return null; + const job = await tx.mediaJob.findUniqueOrThrow({ where: { id: jobId } }); + const card = await writeCard(tx, job, appended); + return { job, card }; + }), + ); + if (committed) await notify(committed.job.threadId, committed.card.seq); + return committed; + }; + + /** Ends a job: terminal status on the row and the card, and the bot's wake when requested. */ + const settle = async ( + jobId: string, + token: string | null, + outcome: { status: "done" | "failed" | "cancelled"; error?: string; summary?: string }, + ): Promise => { + const committed = await withTransactionRetry(() => + deps.prisma.$transaction(async (tx) => { + const now = new Date(); + const fence: Prisma.MediaJobWhereInput = token + ? { id: jobId, leaseToken: token } + : { + id: jobId, + status: { in: ACTIVE_STATUSES }, + OR: [{ leaseToken: null }, { leaseExpiresAt: { lte: now } }], + }; + const before = await tx.mediaJob.findFirst({ where: fence, select: { threadId: true } }); + if (!before) return null; + await tx.thread.updateMany({ + where: { id: before.threadId }, + data: { nextEventSeq: { increment: 0 } }, + }); + const updated = await tx.mediaJob.updateMany({ + where: fence, + data: { + status: outcome.status, + error: + outcome.status === "failed" + ? clamp(outcome.error ?? FAILED_ERROR, ERROR_MAX_LENGTH) + : null, + ...(outcome.status === "done" ? { progress: 1 } : {}), + finishedAt: now, + leaseToken: null, + leaseExpiresAt: null, + nextAttemptAt: null, + }, + }); + if (!updated.count) return null; + const job = await tx.mediaJob.findUniqueOrThrow({ where: { id: jobId } }); + const card = await writeCard(tx, job); + let wakeRunId: string | undefined; + if (job.continueAfter && job.botId && card.exists) { + const [bot, member] = await Promise.all([ + tx.bot.findFirst({ + where: { id: job.botId, spaceId: job.spaceId, userId: job.userId, archivedAt: null }, + select: { id: true }, + }), + tx.spaceMember.findUnique({ + where: { spaceId_userId: { spaceId: job.spaceId, userId: job.userId } }, + select: { id: true }, + }), + ]); + if (bot && member) { + const prompt = mediaWakePrompt({ + ...job, + summary: outcome.summary ?? "", + outputs: card.outputs, + }); + const task = await tx.task.create({ + data: { + spaceId: job.spaceId, + botId: job.botId, + threadId: job.threadId, + userId: job.userId, + prompt, + status: "queued", + }, + }); + const run = await tx.run.create({ + data: { + spaceId: job.spaceId, + botId: job.botId, + threadId: job.threadId, + userId: job.userId, + taskId: task.id, + status: "queued", + trigger: MEDIA_WAKE_TRIGGER, + clientNonce: `media-wake:${job.id}`, + }, + select: { id: true }, + }); + wakeRunId = run.id; + } + } + return { job, seq: card.seq, wakeRunId }; + }), + ); + if (!committed) return false; + await notify(committed.job.threadId, committed.seq); + if (committed.wakeRunId) { + // A queued run is durable; the run reconciler repairs a missed wake. + await deps.jobs.enqueue(runContinueJob(committed.wakeRunId)).catch((error) => { + getLogger().error("media job wake enqueue", error); + }); + } + return true; + }; + + /** + * Takes the job for this process. At most one media job holds a live lease at a time, which + * keeps the GPU to one job even when the Graphile queue is bypassed or several in-memory + * drivers run. + */ + const claim = async (jobId: string, token: string): Promise<"claimed" | "busy" | "gone"> => + deps.prisma.$transaction(async (tx) => { + await tx.$executeRaw`SELECT pg_advisory_xact_lock(${MEDIA_CLAIM_LOCK_NAMESPACE}::integer, ${MEDIA_CLAIM_LOCK_ID}::integer)`; + const now = new Date(); + const claimable: Prisma.MediaJobWhereInput = { + id: jobId, + status: { in: ACTIVE_STATUSES }, + AND: [ + { OR: [{ leaseToken: null }, { leaseExpiresAt: { lte: now } }] }, + { OR: [{ nextAttemptAt: null }, { nextAttemptAt: { lte: now } }] }, + ], + }; + const candidate = await tx.mediaJob.findFirst({ where: claimable, select: { id: true } }); + if (!candidate) return "gone"; + const holder = await tx.mediaJob.findFirst({ + where: { + id: { not: jobId }, + status: { in: ACTIVE_STATUSES }, + leaseToken: { not: null }, + leaseExpiresAt: { gt: now }, + }, + select: { id: true }, + }); + if (holder) return "busy"; + const updated = await tx.mediaJob.updateMany({ + where: claimable, + data: { + status: "running", + leaseToken: token, + leaseExpiresAt: new Date(now.getTime() + leaseMs), + attempts: { increment: 1 }, + nextAttemptAt: null, + }, + }); + return updated.count ? "claimed" : "gone"; + }); + + const waitBehindHolder = async (job: MediaJob) => { + const nextAttemptAt = new Date(Date.now() + BUSY_RETRY_MS); + // A stale `running` job keeps its status so the next claim still counts the crash. + const committed = await updateJob(job.id, null, { + ...(job.status === "running" ? {} : { status: "waiting" }), + nextAttemptAt, + }); + if (committed) await enqueue(job.id, nextAttemptAt); + }; + + const storeArtifactFile = async ( + job: MediaJob, + file: { + path: string; + name: string; + mimeType: string; + width?: number; + height?: number; + durationMs?: number; + parentArtifactId?: string; + }, + ) => { + // Swap point for the streaming artifact store: until it lands, files go through put(). + const bytes = new Uint8Array(await readFile(file.path)); + const context = artifactContext(job, `media-output:${job.id}`); + const stored = await deps.artifacts.put( + { name: file.name, mimeType: file.mimeType, bytes }, + context, + ); + try { + return await deps.prisma.artifact.create({ + data: { + spaceId: job.spaceId, + userId: job.userId, + botId: job.botId, + groupId: job.groupId, + runId: job.runId, + name: file.name, + mimeType: file.mimeType, + size: bytes.byteLength, + hash: createHash("sha256").update(bytes).digest("hex"), + storageKey: stored.id, + width: file.width, + height: file.height, + durationMs: file.durationMs, + parentArtifactId: file.parentArtifactId, + }, + }); + } catch (error) { + await deps.artifacts.remove(stored.id, context).catch(() => undefined); + throw error; + } + }; + + const removeArtifacts = async (job: MediaJob, ids: string[]) => { + const rows = await deps.prisma.artifact.findMany({ + where: { id: { in: ids }, spaceId: job.spaceId }, + select: { id: true, storageKey: true }, + }); + await deps.prisma.artifact.deleteMany({ where: { id: { in: rows.map((row) => row.id) } } }); + const context = artifactContext(job, `media-output-cleanup:${job.id}`); + for (const row of rows) { + await deps.artifacts.remove(row.storageKey, context).catch(() => undefined); + } + }; + + const runClaimed = async (jobId: string, token: string, previous: MediaJob) => { + const job = await deps.prisma.mediaJob.findUniqueOrThrow({ where: { id: jobId } }); + const recoveries = previous.attempts - previous.deferrals; + if (previous.status === "running" && recoveries > maxRecoveries) { + await settle(jobId, token, { status: "failed", error: INTERRUPTED_ERROR }); + return; + } + if (job.cancelRequested) { + await settle(jobId, token, { status: "cancelled" }); + return; + } + const registration = registry.get(job.operation as MediaOperation); + if (!registration || !(await isAvailable(registration, job.spaceId))) { + await settle(jobId, token, { status: "failed", error: UNAVAILABLE_ERROR }); + return; + } + const member = await deps.prisma.spaceMember.findUnique({ + where: { spaceId_userId: { spaceId: job.spaceId, userId: job.userId } }, + select: { id: true }, + }); + const started = member + ? await updateJob(jobId, token, { startedAt: job.startedAt ?? new Date() }) + : null; + if (!started?.card.exists) { + // The conversation was cleared or the user left the space: nobody can see the result. + await settle(jobId, token, { status: "cancelled" }); + return; + } + + const controller = new AbortController(); + const abort = (reason: AbortReason) => { + if (!controller.signal.aborted) controller.abort(reason); + }; + running.set(jobId, controller); + const deadline = setTimeout(() => abort("timeout"), maxRunMs); + deadline.unref?.(); + let beating = false; + const heartbeat = setInterval(() => { + if (beating) return; + beating = true; + void deps.prisma.mediaJob + .updateMany({ + where: { id: jobId, leaseToken: token }, + data: { leaseExpiresAt: new Date(Date.now() + leaseMs) }, + }) + .then(async (renewed) => { + if (!renewed.count) return abort("lease"); + const current = await deps.prisma.mediaJob.findUnique({ + where: { id: jobId }, + select: { cancelRequested: true }, + }); + if (current?.cancelRequested) abort("cancel"); + }) + .catch(() => { + // A database blip keeps the job running until the lease is really gone. + }) + .finally(() => { + beating = false; + }); + }, heartbeatMs); + heartbeat.unref?.(); + + const tempDir = path.join(tempRoot, jobId); + const inputs = new Map(); + let lastProgress = { at: 0, value: job.progress ?? 0 }; + const signal = controller.signal; + + const context: MediaJobContext = { + jobId, + spaceId: job.spaceId, + userId: job.userId, + botId: job.botId, + input: job.input, + tempDir, + signal, + progress: async (fraction) => { + signal.throwIfAborted(); + const value = Math.min(1, Math.max(0, Number.isFinite(fraction) ? fraction : 0)); + const now = Date.now(); + if (Math.abs(value - lastProgress.value) < 0.01 && value < 1) return; + if (now - lastProgress.at < progressIntervalMs && value < 1) return; + lastProgress = { at: now, value }; + const written = await updateJob(jobId, token, { progress: value }); + if (!written) abort("lease"); + else if (!written.card.exists) abort("abandoned"); + }, + fetchArtifact: async (artifactId) => { + signal.throwIfAborted(); + const known = inputs.get(artifactId); + if (known) return known; + const row = await deps.prisma.artifact.findFirst({ + where: { id: artifactId, spaceId: job.spaceId, userId: job.userId }, + }); + if (!row) throw new MediaJobRejectedError(MISSING_SOURCE_ERROR); + const extension = + attachmentExtensionForMimeType(row.mimeType) || + (/^\.[a-z0-9]{1,8}$/i.test(path.extname(row.name)) ? path.extname(row.name) : ""); + const file = path.join(tempDir, `input-${row.id}${extension.toLowerCase()}`); + // Swap point for the streaming artifact store (localPath / createReadStream). + const bytes = await deps.artifacts.get( + row.storageKey, + artifactContext(job, `media-input:${job.id}`), + ); + await writeFile(file, bytes, { mode: 0o600, flag: "wx" }); + const input: MediaJobInputFile = { + artifactId: row.id, + path: file, + name: row.name, + mimeType: row.mimeType, + kind: attachmentKindOf(row.mimeType), + size: row.size, + ...(row.width ? { width: row.width } : {}), + ...(row.height ? { height: row.height } : {}), + ...(row.durationMs !== null ? { durationMs: row.durationMs } : {}), + }; + inputs.set(artifactId, input); + return input; + }, + addOutput: async (file: MediaJobOutputFile) => { + signal.throwIfAborted(); + validateAttachmentMimeType(file.mimeType); + const kind = attachmentKindOf(file.mimeType); + const filePath = insideDir(tempDir, file.path); + const size = (await stat(filePath)).size; + if (size > ARTIFACT_MAX_BYTES_BY_KIND[kind]) + throw new MediaJobRejectedError(TOO_LARGE_ERROR); + const parent = file.parentArtifactId + ? await deps.prisma.artifact.findFirst({ + where: { id: file.parentArtifactId, spaceId: job.spaceId, userId: job.userId }, + select: { id: true }, + }) + : null; + const name = + path.basename(file.name).slice(0, 200) || + `output${attachmentExtensionForMimeType(file.mimeType)}`; + const stored: string[] = []; + try { + let posterArtifactId: string | undefined; + if (file.posterPath) { + const posterPath = insideDir(tempDir, file.posterPath); + const posterName = `${path.parse(name).name}-poster${path.extname(posterPath).toLowerCase()}`; + const posterMime = inferAttachmentMimeType(posterName); + if (posterMime && attachmentKindOf(posterMime) === "image") { + const poster = await storeArtifactFile(job, { + path: posterPath, + name: posterName, + mimeType: posterMime, + width: positiveInt(file.width), + height: positiveInt(file.height), + }); + stored.push(poster.id); + posterArtifactId = poster.id; + } + } + const width = positiveInt(file.width); + const height = positiveInt(file.height); + const durationMs = nonNegativeInt(file.durationMs); + const artifact = await storeArtifactFile(job, { + path: filePath, + name, + mimeType: file.mimeType, + width, + height, + durationMs, + parentArtifactId: parent?.id, + }); + stored.push(artifact.id); + const output: MediaOutput = { + artifactId: artifact.id, + kind, + mimeType: file.mimeType, + name, + size: artifact.size, + ...(width ? { width } : {}), + ...(height ? { height } : {}), + ...(durationMs !== undefined ? { durationMs } : {}), + ...(posterArtifactId ? { posterArtifactId } : {}), + ...(parent ? { parentArtifactId: parent.id } : {}), + ...(file.preset ? { preset: file.preset } : {}), + }; + const written = await updateJob(jobId, token, { outputArtifactIds: { push: stored } }, [ + output, + ]); + if (!written) { + abort("lease"); + throw new Error("Media job lost its lease"); + } + if (!written.card.exists) abort("abandoned"); + return output; + } catch (error) { + await removeArtifacts(job, stored).catch(() => undefined); + throw error; + } + }, + }; + + try { + await rm(tempDir, { recursive: true, force: true }); + await mkdir(tempDir, { recursive: true, mode: 0o700 }); + if (registration.gpu) { + const gpu = await registration.gpu + .status(AbortSignal.any([signal, AbortSignal.timeout(GPU_STATUS_TIMEOUT_MS)])) + .catch(() => undefined); + if (gpu && gpu.vramFreeMb < registration.gpu.minVramMb) { + throw new MediaJobDeferredError("Waiting for free GPU memory", GPU_RETRY_MS); + } + } + const result = await Promise.race([ + registration.handler.run(context), + new Promise((_, reject) => { + const onAbort = () => { + const timer = setTimeout(() => reject(signal.reason), abortGraceMs); + timer.unref?.(); + }; + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); + }), + ]); + if (signal.aborted) throw signal.reason; + await settle(jobId, token, { status: "done", summary: result.summary }); + } catch (error) { + const reason = signal.aborted ? (signal.reason as AbortReason) : undefined; + if (reason === "lease") return; + if (reason === "cancel" || reason === "abandoned") { + await settle(jobId, token, { status: "cancelled" }); + } else if (reason === "timeout") { + await settle(jobId, token, { status: "failed", error: TIMEOUT_ERROR }); + } else if (error instanceof MediaJobDeferredError) { + await defer(job, token, error); + } else { + if (!(error instanceof MediaJobRejectedError)) { + getLogger().warn("media job failed", { + jobId, + operation: job.operation, + error: error instanceof Error ? error.message : String(error), + }); + } + const message = error instanceof Error ? error.message.split(tempDir).join("") : ""; + await settle(jobId, token, { status: "failed", error: message.trim() || FAILED_ERROR }); + } + } finally { + clearInterval(heartbeat); + clearTimeout(deadline); + running.delete(jobId); + await rm(tempDir, { recursive: true, force: true }).catch((error) => { + getLogger().error("media job temp cleanup", error); + }); + } + }; + + const defer = async (job: MediaJob, token: string, error: MediaJobDeferredError) => { + if (job.deferrals >= maxDeferrals) { + await settle(job.id, token, { status: "failed", error: BUSY_ERROR }); + return; + } + const delay = Math.min( + maxDeferMs, + Math.max(1_000, error.retryAfterMs) * 2 ** Math.min(job.deferrals, 5), + ); + const nextAttemptAt = new Date(Date.now() + delay); + const committed = await updateJob(job.id, token, { + status: "waiting", + deferrals: { increment: 1 }, + leaseToken: null, + leaseExpiresAt: null, + nextAttemptAt, + }); + if (committed) await enqueue(job.id, nextAttemptAt); + }; + + return { + async capabilities(spaceId) { + const entries = await Promise.all( + MediaOperationSchema.options.map( + async (operation) => + [operation, await isAvailable(registry.get(operation), spaceId)] as const, + ), + ); + return Object.fromEntries(entries) as MediaCapabilities; + }, + + async start(input: StartMediaJobInput) { + const registration = registry.get(input.operation); + if (!registration || !(await isAvailable(registration, input.spaceId))) { + throw new MediaJobRejectedError(`${input.operation} is not available.`); + } + if (!input.botId) throw new MediaJobRejectedError("A media job needs a bot to report to."); + const botId = input.botId; + const parsed = registration.inputSchema.safeParse(input.input); + if (!parsed.success) { + throw new MediaJobRejectedError( + `Invalid ${input.operation} input: ${issuesText(parsed.error)}`, + ); + } + const artifactIds = [...new Set(registration.artifactIds?.(parsed.data) ?? [])]; + if (artifactIds.length) { + const readable = await deps.prisma.artifact.count({ + where: { id: { in: artifactIds }, spaceId: input.spaceId, userId: input.userId }, + }); + if (readable !== artifactIds.length) throw new MediaJobRejectedError("Unknown artifact."); + } + const title = clamp( + (input.title?.trim() || registration.title(parsed.data)).replace(/\s+/g, " ").trim(), + TITLE_MAX_LENGTH, + ); + const replay = async () => { + if (!input.operationKey) return null; + const existing = await deps.prisma.mediaJob.findUnique({ + where: { operationKey: input.operationKey }, + }); + if (!existing) return null; + if ( + existing.spaceId !== input.spaceId || + existing.userId !== input.userId || + existing.threadId !== input.threadId || + !existing.messageId + ) { + throw new Error("Media job operation scope changed"); + } + return { jobId: existing.id, messageId: existing.messageId }; + }; + const replayed = await replay(); + if (replayed) return replayed; + let committed: { jobId: string; messageId: string; seq: number }; + try { + committed = await withTransactionRetry(() => + deps.prisma.$transaction(async (tx) => { + // Also locks the thread row, serializing with clearing the chat. + const thread = await tx.thread.updateMany({ + where: { id: input.threadId, spaceId: input.spaceId, userId: input.userId }, + data: { nextEventSeq: { increment: 0 } }, + }); + if (!thread.count) { + throw new MediaJobRejectedError("This conversation is no longer available."); + } + const job = await tx.mediaJob.create({ + data: { + spaceId: input.spaceId, + userId: input.userId, + botId, + groupId: input.groupId, + threadId: input.threadId, + runId: input.runId, + operation: input.operation, + title, + input: parsed.data as Prisma.InputJsonValue, + continueAfter: input.continueAfter === true, + operationKey: input.operationKey, + }, + }); + const blocks: MessageBlock[] = [cardBlock(job, [])]; + const message = await createThreadMessageInTransaction(tx, { + threadId: input.threadId, + role: "bot", + blocks, + botId, + runId: input.runId ?? undefined, + }); + await tx.mediaJob.update({ where: { id: job.id }, data: { messageId: message.id } }); + const event = await appendEventInTransaction(tx, { + spaceId: input.spaceId, + threadId: input.threadId, + botId, + runId: input.runId ?? undefined, + type: "thread.message.created", + payload: { messageId: message.id, role: "bot", blocks }, + }); + return { jobId: job.id, messageId: message.id, seq: event.seq }; + }), + ); + } catch (error) { + // A concurrent replay of the same tool call won the unique operation key. + if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") { + const existing = await replay(); + if (existing) return existing; + } + throw error; + } + await notify(input.threadId, committed.seq); + await enqueue(committed.jobId); + return { jobId: committed.jobId, messageId: committed.messageId }; + }, + + async get(jobId): Promise { + const job = await deps.prisma.mediaJob.findUnique({ where: { id: jobId } }); + if (!job) return null; + const message = job.messageId + ? await deps.prisma.message.findFirst({ + where: { id: job.messageId, threadId: job.threadId }, + select: { blocks: true }, + }) + : null; + return { + id: job.id, + operation: job.operation as MediaOperation, + status: job.status as MediaJobStatus, + title: job.title, + progress: job.progress, + error: job.error, + outputs: mediaBlockOf(message?.blocks, job.id)?.outputs ?? [], + }; + }, + + async cancel(jobId) { + const requested = await deps.prisma.mediaJob.updateMany({ + where: { id: jobId, status: { in: ACTIVE_STATUSES } }, + data: { cancelRequested: true }, + }); + if (!requested.count) return; + running.get(jobId)?.abort("cancel" satisfies AbortReason); + // Not held by a live runner: end it now. A running job stops at its next heartbeat. + if (await settle(jobId, null, { status: "cancelled" })) { + await deps.jobs.cancel(mediaRunJobKey(jobId)).catch(() => undefined); + } + }, + + async runJob(jobId) { + const previous = await deps.prisma.mediaJob.findUnique({ where: { id: jobId } }); + if (!previous || !ACTIVE_STATUSES.includes(previous.status as MediaJobStatus)) return; + const token = randomUUID(); + const claimed = await claim(jobId, token); + if (claimed === "busy") { + if (previous.cancelRequested) await settle(jobId, null, { status: "cancelled" }); + else await waitBehindHolder(previous); + return; + } + if (claimed === "gone") return; + await runClaimed(jobId, token, previous); + }, + + async reconcile() { + const now = Date.now(); + const graceAgo = new Date(now - RECONCILE_GRACE_MS); + const due = await deps.prisma.mediaJob.findMany({ + where: { + OR: [ + { status: "queued", leaseToken: null, updatedAt: { lte: graceAgo } }, + { status: "waiting", leaseToken: null, nextAttemptAt: { lte: graceAgo } }, + { + status: "waiting", + leaseToken: null, + nextAttemptAt: null, + updatedAt: { lte: graceAgo }, + }, + { status: { in: ACTIVE_STATUSES }, leaseExpiresAt: { lte: new Date(now) } }, + { status: { in: ACTIVE_STATUSES }, cancelRequested: true, leaseToken: null }, + ], + }, + orderBy: [{ createdAt: "asc" }, { id: "asc" }], + take: RECONCILE_BATCH_SIZE, + }); + for (const job of due) { + const dueAt = + job.status === "waiting" + ? (job.nextAttemptAt ?? job.updatedAt) + : job.leaseExpiresAt && job.leaseExpiresAt.getTime() <= now + ? job.leaseExpiresAt + : job.createdAt; + // The run_at keeps queue order; a job stuck behind a dead worker's queue lock (Graphile + // holds it for hours) runs outside the queue and relies on the claim's GPU lock. + const scheduled = mediaRunJob(job.id, dueAt); + const stuck = now - dueAt.getTime() > STUCK_QUEUE_MS; + await deps.jobs + .enqueue(stuck ? { ...scheduled, queue: undefined } : scheduled) + .catch((error) => getLogger().error("media job reconciliation", error)); + } + }, + }; +} + +function artifactContext(job: MediaJob, operationId: string): AdapterContext { + return { + operationId, + traceId: operationId, + spaceId: job.spaceId, + userId: job.userId, + botId: job.botId ?? undefined, + signal: new AbortController().signal, + }; +} diff --git a/packages/adapters/src/media-settings.ts b/packages/adapters/src/media-settings.ts new file mode 100644 index 0000000..d237a13 --- /dev/null +++ b/packages/adapters/src/media-settings.ts @@ -0,0 +1,132 @@ +import { + type MediaSettings, + MediaSettingsSchema, + type MediaSettingsUpdate, + MediaSettingsUpdateSchema, + type MediaSettingsView, +} from "@rakazo/contracts"; +import type { PrismaClient } from "@rakazo/db"; +import { + assertAllowedOpenAiCompatibleRequestUrl, + assertAllowedOpenAiCompatibleUrl, + assertHttpsForKeyedOpenAiCompatibleUrl, +} from "./openai-compatible-url.js"; +import type { EncryptedSecretStore } from "./secrets.js"; + +/** The IntegrationProviderConfig row that stores the media backends, encrypted as one record. */ +export const MEDIA_SETTINGS_ID = "media"; +const RECORD_ID = `integration-provider:${MEDIA_SETTINGS_ID}`; + +export interface MediaSettingsDeps { + prisma: Pick; + secrets: Pick; +} + +const NOT_CONFIGURED: MediaSettings = { comfyui: null, transcription: null }; + +// Decrypting derives a scrypt key; tool selection reads the settings on every run. +let cached: { ciphertext: string; settings: MediaSettings } | undefined; + +/** Deployment-wide media backends, read fresh from the database so every process sees edits. */ +export async function loadMediaSettings(deps: MediaSettingsDeps): Promise { + const row = await deps.prisma.integrationProviderConfig.findUnique({ + where: { id: MEDIA_SETTINGS_ID }, + }); + if (!row) return NOT_CONFIGURED; + if (cached?.ciphertext === row.ciphertext) return cached.settings; + const settings = MediaSettingsSchema.parse( + JSON.parse(deps.secrets.load(row.ciphertext, RECORD_ID)), + ); + cached = { ciphertext: row.ciphertext, settings }; + return settings; +} + +export function mediaSettingsView(settings: MediaSettings): MediaSettingsView { + return { + comfyui: settings.comfyui ? withoutKey(settings.comfyui) : null, + transcription: settings.transcription ? withoutKey(settings.transcription) : null, + }; +} + +function withoutKey({ apiKey, ...rest }: T) { + return { ...rest, hasApiKey: Boolean(apiKey) }; +} + +/** A ComfyUI root such as `http://127.0.0.1:8188`, without credentials, query or fragment. */ +export function normalizeComfyUiBaseUrl(raw: string): string { + let url: URL; + try { + url = new URL(raw.trim()); + } catch { + throw new Error("Server URL must be an absolute HTTP(S) URL"); + } + if (url.username || url.password) throw new Error("Server URL must not contain credentials"); + const normalized = `${url.origin}${url.pathname.replace(/\/+$/, "")}`; + // Same address policy as OpenAI-compatible servers: private hosts pass, metadata hosts never. + return assertAllowedOpenAiCompatibleRequestUrl(normalized).href.replace(/\/+$/, ""); +} + +/** An OpenAI-compatible API root ending in a version segment, such as `…/v1`. */ +export function normalizeTranscriptionBaseUrl(raw: string): string { + return assertAllowedOpenAiCompatibleUrl(raw).href.replace(/\/+$/, ""); +} + +function keptKey( + next: string | undefined, + previous: { baseUrl: string; apiKey?: string } | null, + baseUrl: string, +): string | undefined { + if (next !== undefined) return next || undefined; + // A saved key never follows the settings to another server. + if (!previous?.apiKey) return undefined; + return new URL(previous.baseUrl).origin === new URL(baseUrl).origin ? previous.apiKey : undefined; +} + +/** + * Validates and stores the media backends (deployment owner only; the caller authorizes). + * Throws an Error with a user-facing message when a URL is not allowed. + */ +export async function saveMediaSettings( + deps: MediaSettingsDeps, + update: MediaSettingsUpdate, +): Promise { + const parsed = MediaSettingsUpdateSchema.parse(update); + const previous = await loadMediaSettings(deps); + let comfyui: MediaSettings["comfyui"] = null; + if (parsed.comfyui) { + const baseUrl = normalizeComfyUiBaseUrl(parsed.comfyui.baseUrl); + const apiKey = keptKey(parsed.comfyui.apiKey, previous.comfyui, baseUrl); + assertHttpsForKeyedOpenAiCompatibleUrl(new URL(baseUrl), apiKey); + comfyui = { ...parsed.comfyui, baseUrl, apiKey }; + } + let transcription: MediaSettings["transcription"] = null; + if (parsed.transcription) { + const baseUrl = normalizeTranscriptionBaseUrl(parsed.transcription.baseUrl); + const apiKey = keptKey(parsed.transcription.apiKey, previous.transcription, baseUrl); + assertHttpsForKeyedOpenAiCompatibleUrl(new URL(baseUrl), apiKey); + transcription = { ...parsed.transcription, baseUrl, apiKey }; + } + const settings = MediaSettingsSchema.parse({ comfyui, transcription }); + if (!settings.comfyui && !settings.transcription) { + await deps.prisma.integrationProviderConfig.deleteMany({ where: { id: MEDIA_SETTINGS_ID } }); + return mediaSettingsView(settings); + } + const stored = await deps.secrets.put( + JSON.stringify(settings), + { + operationId: "media-settings", + traceId: "media-settings", + spaceId: "", + userId: "", + signal: new AbortController().signal, + }, + RECORD_ID, + ); + await deps.prisma.integrationProviderConfig.upsert({ + where: { id: MEDIA_SETTINGS_ID }, + create: { id: MEDIA_SETTINGS_ID, ciphertext: stored.ciphertext }, + update: { ciphertext: stored.ciphertext }, + }); + cached = { ciphertext: stored.ciphertext, settings }; + return mediaSettingsView(settings); +} diff --git a/packages/adapters/src/media-tools.ts b/packages/adapters/src/media-tools.ts new file mode 100644 index 0000000..7c0d219 --- /dev/null +++ b/packages/adapters/src/media-tools.ts @@ -0,0 +1,78 @@ +import type { ConnectorTool, MediaCapabilities, MediaRuntime } from "@rakazo/adapter-kit"; +import { type MediaOperation, MediaOperationSchema } from "@rakazo/contracts"; +import { MediaJobRejectedError } from "./media-runtime.js"; + +/** Media tools are named exactly like their operations and dispatched by one module. */ +const MEDIA_TOOL_NAMES: ReadonlySet = new Set(MediaOperationSchema.options); + +export function isMediaToolName(name: string): name is MediaOperation { + return MEDIA_TOOL_NAMES.has(name); +} + +/** The shared `continue_after` property every media tool definition accepts. */ +export const MEDIA_CONTINUE_AFTER_PROPERTY = { + continue_after: { + type: "boolean", + description: + "Wake me with the result when the job ends so I can continue a multi-step task. Default false: the media card shows the result to the user.", + }, +} as const; + +/** A media tool's JSON schema: its own properties plus `continue_after`. */ +export function mediaToolInputSchema(schema: { + properties: Record; + required?: string[]; +}): ConnectorTool["inputSchema"] { + return { + type: "object", + properties: { ...schema.properties, ...MEDIA_CONTINUE_AFTER_PROPERTY }, + ...(schema.required?.length ? { required: schema.required } : {}), + }; +} + +/** Drops media tools whose operation the space cannot run right now. */ +export function selectMediaTools( + tools: T[], + capabilities: MediaCapabilities | undefined, +): T[] { + return tools.filter((tool) => !isMediaToolName(tool.name) || capabilities?.[tool.name] === true); +} + +/** Starts the job behind a media tool call; the turn never waits for the job itself. */ +export async function executeMediaTool( + media: MediaRuntime | undefined, + scope: { + spaceId: string; + userId: string; + botId: string; + groupId: string | null; + threadId: string; + runId: string; + /** The tool call's effect key, so a replayed call returns the job it started. */ + operationKey: string; + }, + operation: MediaOperation, + args: Record, +) { + if (!media) return { error: `${operation} is not available.` }; + const { continue_after: continueAfter, ...input } = args; + try { + const started = await media.start({ + operation, + input, + ...scope, + continueAfter: continueAfter === true, + }); + return { + jobId: started.jobId, + status: "queued", + note: + continueAfter === true + ? "The job runs in the background. You are woken with its result when it ends; do not poll or wait." + : "The job runs in the background and its card shows progress and the result to the user. Do not poll or wait.", + }; + } catch (error) { + if (error instanceof MediaJobRejectedError) return { error: error.message }; + throw error; + } +} diff --git a/packages/adapters/src/wakeup-graphile-host.test.ts b/packages/adapters/src/wakeup-graphile-host.test.ts index b6cc385..7b955b9 100644 --- a/packages/adapters/src/wakeup-graphile-host.test.ts +++ b/packages/adapters/src/wakeup-graphile-host.test.ts @@ -24,6 +24,7 @@ function handlers(): BackgroundJobHandlers { "messaging.deliver": vi.fn(async () => undefined), "cloud_agent.poll": vi.fn(async () => undefined), "profile.consolidate": vi.fn(async () => undefined), + "media.run": vi.fn(async () => undefined), }; } diff --git a/packages/adapters/src/wakeup.postgres.test.ts b/packages/adapters/src/wakeup.postgres.test.ts index b164554..ad9d241 100644 --- a/packages/adapters/src/wakeup.postgres.test.ts +++ b/packages/adapters/src/wakeup.postgres.test.ts @@ -29,6 +29,7 @@ function handlers(overrides: Partial = {}): BackgroundJob "messaging.deliver": vi.fn(async () => undefined), "cloud_agent.poll": vi.fn(async () => undefined), "profile.consolidate": vi.fn(async () => undefined), + "media.run": vi.fn(async () => undefined), ...overrides, }; } @@ -141,6 +142,39 @@ describePostgres("Graphile background jobs (PostgreSQL contract)", () => { } }); + it("runs jobs that share a queue one at a time across workers", async () => { + const publisher = new GraphileJobPublisher(testPool()); + const first = new GraphileJobWorkerHost(testPool(), { concurrency: 2, pollInterval: 25 }); + const second = new GraphileJobWorkerHost(testPool(), { concurrency: 2, pollInterval: 25 }); + const queue = `contract-queue-${Date.now()}`; + let active = 0; + let maxActive = 0; + const done: string[] = []; + const target = handlers({ + "media.run": async ({ jobId }) => { + active += 1; + maxActive = Math.max(maxActive, active); + await new Promise((resolve) => setTimeout(resolve, 150)); + active -= 1; + done.push(jobId); + }, + }); + + try { + await first.start(target); + await second.start(target); + for (const jobId of ["q1", "q2", "q3"]) { + await publisher.enqueue({ name: "media.run", payload: { jobId }, queue }); + } + await waitFor(() => expect(done).toHaveLength(3)); + expect(maxActive).toBe(1); + } finally { + await first.stop(); + await second.stop(); + await publisher.close(); + } + }); + it("persists a handler failure and retries the job successfully", async () => { const publisher = new GraphileJobPublisher(testPool()); const host = new GraphileJobWorkerHost(testPool(), { concurrency: 1, pollInterval: 25 }); diff --git a/packages/adapters/src/wakeup.test.ts b/packages/adapters/src/wakeup.test.ts index 3856671..691f17a 100644 --- a/packages/adapters/src/wakeup.test.ts +++ b/packages/adapters/src/wakeup.test.ts @@ -14,6 +14,7 @@ function handlers(): BackgroundJobHandlers { "messaging.deliver": vi.fn(async () => undefined), "cloud_agent.poll": vi.fn(async () => undefined), "profile.consolidate": vi.fn(async () => undefined), + "media.run": vi.fn(async () => undefined), }; } @@ -198,6 +199,59 @@ describe("InMemoryJobQueue", () => { await expect(enqueue).rejects.toThrow("Background job publisher is stopped"); }); + it("runs jobs of one named queue one at a time, in order", async () => { + const queue = new InMemoryJobQueue(); + const order: string[] = []; + let active = 0; + let maxActive = 0; + const target = handlers(); + target["media.run"] = async ({ jobId }) => { + active += 1; + maxActive = Math.max(maxActive, active); + order.push(`start:${jobId}`); + await new Promise((resolve) => setTimeout(resolve, 20)); + order.push(`end:${jobId}`); + active -= 1; + }; + target["computer.sleep"] = async () => { + order.push("unqueued"); + }; + await queue.start(target); + + for (const jobId of ["a", "b", "c"]) { + await queue.enqueue({ name: "media.run", payload: { jobId }, queue: "media" }); + } + await queue.enqueue({ name: "computer.sleep", payload: { computerId: "computer-1" } }); + await queue.close(); + + expect(maxActive).toBe(1); + expect(order.filter((entry) => entry !== "unqueued")).toEqual([ + "start:a", + "end:a", + "start:b", + "end:b", + "start:c", + "end:c", + ]); + // Jobs outside the queue do not wait for it. + expect(order.indexOf("unqueued")).toBeLessThan(order.indexOf("end:a")); + }); + + it("keeps a named queue moving after a job fails", async () => { + const queue = new InMemoryJobQueue(); + const ran: string[] = []; + const target = handlers(); + target["media.run"] = async ({ jobId }) => { + ran.push(jobId); + if (jobId === "a") throw new Error("boom"); + }; + await queue.start(target); + await queue.enqueue({ name: "media.run", payload: { jobId: "a" }, queue: "media" }); + await queue.enqueue({ name: "media.run", payload: { jobId: "b" }, queue: "media" }); + await queue.close(); + expect(ran).toEqual(["a", "b"]); + }); + it("drains an accepted immediate timer during close", async () => { const queue = new InMemoryJobQueue(); const target = handlers(); diff --git a/packages/adapters/src/wakeup.ts b/packages/adapters/src/wakeup.ts index 74ee26d..60bb1d5 100644 --- a/packages/adapters/src/wakeup.ts +++ b/packages/adapters/src/wakeup.ts @@ -25,6 +25,7 @@ export class GraphileJobPublisher implements JobPublisher { await utils.addJob(job.name, wrapJobPayload(job.payload), { runAt: job.availableAt, jobKey: job.replaceKey, + queueName: job.queue, }); } @@ -180,6 +181,7 @@ interface QueuedJob { payload: unknown; availableAt?: Date; replaceKey?: string; + queue?: string; } function toQueuedJob(job: BackgroundJob): QueuedJob { @@ -188,6 +190,7 @@ function toQueuedJob(job: BackgroundJob): QueuedJob { payload: wrapJobPayload(job.payload), availableAt: job.availableAt, replaceKey: job.replaceKey, + queue: job.queue, }; } @@ -197,6 +200,8 @@ export class InMemoryJobQueue implements JobPublisher, JobWorkerHost { private readonly scheduled = new Map, QueuedJob>(); private readonly keyed = new Map>(); private readonly active = new Set>(); + /** Last job of each named queue; the next one starts after it settles, like Graphile queues. */ + private readonly queueTails = new Map>(); private readonly closingJobs: QueuedJob[] = []; private draining: Promise | undefined; private closed = false; @@ -260,12 +265,22 @@ export class InMemoryJobQueue implements JobPublisher, JobWorkerHost { private dispatch(handlers: BackgroundJobHandlers, job: QueuedJob): Promise { const unpacked = unwrapJobPayload(job.payload); - const active = runCorrelatedJob({ - name: job.name, - payload: unpacked.payload, - correlation: unpacked.correlation, - run: () => dispatchBackgroundJob(handlers, job.name, unpacked.payload), - }).catch(() => undefined); + const run = () => + runCorrelatedJob({ + name: job.name, + payload: unpacked.payload, + correlation: unpacked.correlation, + run: () => dispatchBackgroundJob(handlers, job.name, unpacked.payload), + }); + const queue = job.queue; + const tail = queue ? this.queueTails.get(queue) : undefined; + const active = (tail ? tail.then(run) : run()).catch(() => undefined); + if (queue) { + this.queueTails.set(queue, active); + void active.finally(() => { + if (this.queueTails.get(queue) === active) this.queueTails.delete(queue); + }); + } this.active.add(active); void active.finally(() => this.active.delete(active)); return active; diff --git a/packages/contracts/src/domain.ts b/packages/contracts/src/domain.ts index 18b4bcc..6cc3e4f 100644 --- a/packages/contracts/src/domain.ts +++ b/packages/contracts/src/domain.ts @@ -922,6 +922,7 @@ export const RunSchema = z.object({ "messaging", "cloud_agent", "delegation", + "media", ]), routineId: Id.nullable(), modelProvider: z.string().nullable(), diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index ba2c70b..48aa488 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -10,6 +10,7 @@ export * from "./integration-settings.js"; export * from "./local-settings.js"; export * from "./mcp.js"; export * from "./media.js"; +export * from "./media-settings.js"; export * from "./model-tiers.js"; export * from "./openai-compatible-ui.js"; export * from "./project-goals.js"; diff --git a/packages/contracts/src/media-settings.ts b/packages/contracts/src/media-settings.ts new file mode 100644 index 0000000..aa3189c --- /dev/null +++ b/packages/contracts/src/media-settings.ts @@ -0,0 +1,78 @@ +import * as z from "zod"; + +/** Cap for a custom ComfyUI workflow (API-format JSON). */ +export const MEDIA_WORKFLOW_MAX_LENGTH = 512 * 1024; + +/** Empty form fields mean "not set". */ +function optionalText(max: number) { + return z + .string() + .trim() + .max(max) + .optional() + .transform((value) => value || undefined); +} + +function isJsonObject(value: string | undefined): boolean { + if (value === undefined) return true; + try { + const parsed: unknown = JSON.parse(value); + return typeof parsed === "object" && parsed !== null && !Array.isArray(parsed); + } catch { + return false; + } +} + +const workflow = optionalText(MEDIA_WORKFLOW_MAX_LENGTH).refine(isJsonObject, { + message: "Workflow must be a JSON object", +}); + +/** Image and video generation on the deployment owner's ComfyUI server. */ +export const ComfyUiSettingsSchema = z.object({ + baseUrl: z.string().trim().min(1).max(2048), + apiKey: optionalText(4096), + /** Checkpoint name; the first installed checkpoint when unset. */ + imageModel: optionalText(512), + /** Custom API-format workflows that replace the built-in defaults. */ + imageWorkflow: workflow, + videoWorkflow: workflow, +}); +export type ComfyUiSettings = z.infer; + +/** Speech to text on an OpenAI-compatible server (`POST /v1/audio/transcriptions`). */ +export const TranscriptionSettingsSchema = z.object({ + baseUrl: z.string().trim().min(1).max(2048), + model: optionalText(512), + apiKey: optionalText(4096), +}); +export type TranscriptionSettings = z.infer; + +/** Deployment-wide media backends; null means the backend is not configured. */ +export const MediaSettingsSchema = z.object({ + comfyui: ComfyUiSettingsSchema.nullable(), + transcription: TranscriptionSettingsSchema.nullable(), +}); +export type MediaSettings = z.infer; + +/** What the settings screen reads: saved keys are never returned, only whether one exists. */ +export const MediaSettingsViewSchema = z.object({ + comfyui: ComfyUiSettingsSchema.omit({ apiKey: true }) + .extend({ hasApiKey: z.boolean() }) + .nullable(), + transcription: TranscriptionSettingsSchema.omit({ apiKey: true }) + .extend({ hasApiKey: z.boolean() }) + .nullable(), +}); +export type MediaSettingsView = z.infer; + +const keyUpdate = z.string().trim().max(4096).optional(); + +/** + * Saving replaces both backends. An omitted `apiKey` keeps the saved key while the server's + * origin stays the same; an empty one removes it. + */ +export const MediaSettingsUpdateSchema = z.object({ + comfyui: ComfyUiSettingsSchema.extend({ apiKey: keyUpdate }).nullable(), + transcription: TranscriptionSettingsSchema.extend({ apiKey: keyUpdate }).nullable(), +}); +export type MediaSettingsUpdate = z.input; diff --git a/packages/contracts/src/runs.ts b/packages/contracts/src/runs.ts index 9b94346..cfb2a92 100644 --- a/packages/contracts/src/runs.ts +++ b/packages/contracts/src/runs.ts @@ -22,6 +22,7 @@ export const RunActivityRowSchema = z.object({ "messaging", "cloud_agent", "delegation", + "media", ]), notificationsEnabled: z.boolean(), promptSnippet: z.string(), diff --git a/packages/db/prisma/migrations/20261003132000_media_job_leases/migration.sql b/packages/db/prisma/migrations/20261003132000_media_job_leases/migration.sql new file mode 100644 index 0000000..4aac2b1 --- /dev/null +++ b/packages/db/prisma/migrations/20261003132000_media_job_leases/migration.sql @@ -0,0 +1,12 @@ +-- Media jobs run on whichever process claims them first: a fenced lease keeps two processes from +-- running one job, deferrals bound how long a job waits for the GPU, an operation key makes a +-- replayed tool call return the job it started, and continueAfter wakes the bot when it ends. +ALTER TABLE "media_jobs" ADD COLUMN "deferrals" INTEGER NOT NULL DEFAULT 0; +ALTER TABLE "media_jobs" ADD COLUMN "continueAfter" BOOLEAN NOT NULL DEFAULT false; +ALTER TABLE "media_jobs" ADD COLUMN "operationKey" TEXT; +ALTER TABLE "media_jobs" ADD COLUMN "leaseToken" TEXT; +ALTER TABLE "media_jobs" ADD COLUMN "leaseExpiresAt" TIMESTAMP(3); +ALTER TABLE "media_jobs" ADD COLUMN "nextAttemptAt" TIMESTAMP(3); + +CREATE UNIQUE INDEX "media_jobs_operationKey_key" ON "media_jobs"("operationKey"); +CREATE INDEX "media_jobs_status_updatedAt_idx" ON "media_jobs"("status", "updatedAt"); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index b9773da..edbbd53 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -1102,14 +1102,27 @@ model MediaJob { progress Float? error String? outputArtifactIds String[] @default([]) + /// Times a runner claimed the job; with deferrals this bounds crash recovery. attempts Int @default(0) + /// Times the job waited for the GPU or its backend instead of running. + deferrals Int @default(0) cancelRequested Boolean @default(false) + /// Wake the starting bot with the job's summary when it ends. + continueAfter Boolean @default(false) + /// Tool-call effect key; a replay returns the job created first. + operationKey String? @unique + /// Fenced runner lease so two processes never run one job. + leaseToken String? + leaseExpiresAt DateTime? + /// When a waiting job may run again. + nextAttemptAt DateTime? startedAt DateTime? finishedAt DateTime? createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@index([spaceId, status]) + @@index([status, updatedAt]) @@index([threadId]) @@map("media_jobs") }