From c60c693d13c29752b4802f026a9151f28a9eb508 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Oct 2026 13:31:53 +0000 Subject: [PATCH] feat(media): media job runtime, media queue and media settings Media operations (generation, edits, transcription) need a runner that works outside the agent turn, keeps a card in the conversation current and never lets two processes put work on one GPU. Operations land in later commits; this is the runtime they register with. - Background job `media.run { jobId }` on a named `media` queue. Jobs carry an optional queue: the Graphile publisher passes it as queueName and the in-memory driver runs each queue one job at a time, in order. - `createMediaRuntime` implements MediaRuntime (capabilities, start, get, cancel) over a list of `MediaOperationRegistration`s and runs jobs: start validates the input, creates the job and posts the bot's media card in one transaction (idempotent per tool call through an operation key), then enqueues it. The runner claims a fenced lease (an advisory lock keeps one live lease at a time, so the GPU stays single even when the queue is bypassed), renews it while polling for cancellation, writes progress and outputs into the card, defers with backoff when the GPU is short of memory or the operation asks to wait, bounds deferrals, crash restarts and run time, always cleans the job's temp directory and always ends the card in a terminal state. With continue_after the starting bot is woken in the same transaction that ends the job (new run trigger `media`). - The reconciler re-enqueues jobs whose enqueue was lost, whose runner died or whose wait is over; a job stuck behind a dead worker's queue lock runs outside the queue. - Media backends are deployment-wide settings stored encrypted as the `media` IntegrationProviderConfig row: `loadMediaSettings` and `saveMediaSettings` with the same address policy as OpenAI-compatible servers (no metadata hosts, public hosts only when allowed, HTTPS when a key is sent), a saved key never follows the settings to another server, and the view never returns keys. Contract in contracts/media-settings.ts. - `createMediaOperations` is the one list both the API and the worker build their runtime from (empty until the operations land), and media-tools.ts has the shared `continue_after` schema helper for the tool definitions. - Migration: lease, deferral, operation key and continueAfter columns. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u --- apps/api/src/app.ts | 15 + apps/web/src/lib/thread-events.ts | 1 + apps/worker/src/index.ts | 15 + .../adapter-kit/src/background-jobs.test.ts | 22 + packages/adapter-kit/src/background-jobs.ts | 18 + packages/adapter-kit/src/media.ts | 8 +- packages/adapter-kit/src/types.ts | 4 + .../adapters/src/background-job-handlers.ts | 5 + packages/adapters/src/executor-types.ts | 3 + packages/adapters/src/index.ts | 4 + packages/adapters/src/job-reconciler.ts | 2 + packages/adapters/src/media-operations.ts | 15 + packages/adapters/src/media-runtime.ts | 1041 +++++++++++++++++ packages/adapters/src/media-settings.ts | 132 +++ packages/adapters/src/media-tools.ts | 78 ++ .../adapters/src/wakeup-graphile-host.test.ts | 1 + packages/adapters/src/wakeup.postgres.test.ts | 34 + packages/adapters/src/wakeup.test.ts | 54 + packages/adapters/src/wakeup.ts | 27 +- packages/contracts/src/domain.ts | 1 + packages/contracts/src/index.ts | 1 + packages/contracts/src/media-settings.ts | 78 ++ packages/contracts/src/runs.ts | 1 + .../migration.sql | 12 + packages/db/prisma/schema.prisma | 13 + 25 files changed, 1578 insertions(+), 7 deletions(-) create mode 100644 packages/adapters/src/media-operations.ts create mode 100644 packages/adapters/src/media-runtime.ts create mode 100644 packages/adapters/src/media-settings.ts create mode 100644 packages/adapters/src/media-tools.ts create mode 100644 packages/contracts/src/media-settings.ts create mode 100644 packages/db/prisma/migrations/20261003132000_media_job_leases/migration.sql 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") }