Merge the media job runtime into the media integration branch

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,
@@ -376,6 +379,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,
@@ -412,6 +424,7 @@ export async function createApp(
     messaging: messaging ? createMessagingContextLoader(prisma) : undefined,
     web: createWebProvider(),
     cloudAgent,
+    media,
     shutdownSignal: shutdown.signal,
   });
 
@@ -429,6 +442,7 @@ export async function createApp(
     deploymentModelKey: env.deploymentModelKey,
     messaging,
     cloudAgent,
+    media,
   });
   if (inMemoryJobs) {
     await inMemoryJobs.start(jobHandlers);
@@ -447,6 +461,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

@@ -77,6 +77,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")
 }