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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u

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();

apps/web/src/lib/thread-events.ts

@@ -33,6 +33,7 @@ const runTriggers = new Set<Run["trigger"]>([
   "messaging",
   "cloud_agent",
   "delegation",
+  "media",
 ]);
 
 function runFromStartedEvent(event: ProductEvent, previous: Run | undefined): Run {

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();
 

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"));
+  });
+});

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<BackgroundJobPayloads[Name]> };
 
 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 } : {}),
+  };
+}

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 {

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];
 

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(
         {

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;
 }

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";

packages/adapters/src/job-reconciler.ts

@@ -109,6 +109,7 @@ export function createJobReconciler(
     /** Enqueues daily profile consolidation; see createProfileConsolidationScheduler. */
     reconcileProfileConsolidation?: () => Promise<void>;
     reconcileDelegatedTasks?: () => Promise<void>;
+    reconcileMediaJobs?: () => Promise<void>;
   },
   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) {

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<MediaSettings>;
+}
+
+/**
+ * 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 [];
+}

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<PrismaClient, "integrationProviderConfig">;
+  secrets: Pick<EncryptedSecretStore, "load" | "put">;
+}
+
+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<MediaSettings> {
+  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<T extends { apiKey?: string }>({ 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<MediaSettingsView> {
+  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);
+}

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<string> = 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<string, unknown>;
+  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<T extends { name: string }>(
+  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<string, unknown>,
+) {
+  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;
+  }
+}

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),
   };
 }
 

packages/adapters/src/wakeup.postgres.test.ts

@@ -29,6 +29,7 @@ function handlers(overrides: Partial<BackgroundJobHandlers> = {}): 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 });

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();

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<ReturnType<typeof setTimeout>, QueuedJob>();
   private readonly keyed = new Map<string, ReturnType<typeof setTimeout>>();
   private readonly active = new Set<Promise<void>>();
+  /** Last job of each named queue; the next one starts after it settles, like Graphile queues. */
+  private readonly queueTails = new Map<string, Promise<void>>();
   private readonly closingJobs: QueuedJob[] = [];
   private draining: Promise<void> | undefined;
   private closed = false;
@@ -260,12 +265,22 @@ export class InMemoryJobQueue implements JobPublisher, JobWorkerHost {
 
   private dispatch(handlers: BackgroundJobHandlers, job: QueuedJob): Promise<void> {
     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;

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(),

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";

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<typeof ComfyUiSettingsSchema>;
+
+/** 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<typeof TranscriptionSettingsSchema>;
+
+/** 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<typeof MediaSettingsSchema>;
+
+/** 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<typeof MediaSettingsViewSchema>;
+
+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<typeof MediaSettingsUpdateSchema>;

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(),

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");

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")
 }