Merge media job crash safety, the media settings section and its E2E

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u

apps/web/e2e/media-settings.spec.ts

@@ -0,0 +1,122 @@
+import type { Page } from "@playwright/test";
+import { expect, test } from "@playwright/test";
+import { captureScreenshot, completeOnboarding, openUserSettings, signup } from "./helpers";
+
+type Draft = {
+  comfyui: {
+    baseUrl: string;
+    apiKey?: string;
+    imageModel?: string;
+    imageWorkflow?: string;
+    videoWorkflow?: string;
+  } | null;
+  transcription: { baseUrl: string; model?: string; apiKey?: string } | null;
+};
+
+/** The server keeps keys; the settings screen only learns whether one is saved. */
+function view(draft: Draft, previous: { comfyui: boolean; transcription: boolean }) {
+  const hasKey = (key: string | undefined, had: boolean) => (key === undefined ? had : key !== "");
+  const text = (value: string | undefined) => (value ? value : undefined);
+  return {
+    comfyui: draft.comfyui
+      ? {
+          baseUrl: draft.comfyui.baseUrl,
+          hasApiKey: hasKey(draft.comfyui.apiKey, previous.comfyui),
+          imageModel: text(draft.comfyui.imageModel),
+          imageWorkflow: text(draft.comfyui.imageWorkflow),
+          videoWorkflow: text(draft.comfyui.videoWorkflow),
+        }
+      : null,
+    transcription: draft.transcription
+      ? {
+          baseUrl: draft.transcription.baseUrl,
+          model: text(draft.transcription.model),
+          hasApiKey: hasKey(draft.transcription.apiKey, previous.transcription),
+        }
+      : null,
+  };
+}
+
+async function asDeploymentOwner(page: Page, isDeploymentOwner: boolean) {
+  await page.route("**/rpc/bootstrap", async (route) => {
+    const response = await route.fetch();
+    const body = await response.json();
+    await route.fulfill({
+      response,
+      json: { json: { ...body.json, me: { ...body.json.me, isDeploymentOwner } } },
+    });
+  });
+}
+
+test("the deployment owner points media at their own servers", async ({ page }, testInfo) => {
+  await asDeploymentOwner(page, true);
+  let current: ReturnType<typeof view> = { comfyui: null, transcription: null };
+  const saved: Draft[] = [];
+  await page.route("**/rpc/media/settings/get", (route) =>
+    route.fulfill({ json: { json: current } }),
+  );
+  await page.route("**/rpc/media/settings/update", (route) => {
+    const draft = (route.request().postDataJSON() as { json: Draft }).json;
+    saved.push(draft);
+    current = view(draft, {
+      comfyui: Boolean(current.comfyui?.hasApiKey),
+      transcription: Boolean(current.transcription?.hasApiKey),
+    });
+    return route.fulfill({ json: { json: current } });
+  });
+
+  await signup(page, `media-settings-${Date.now()}@rakazo.test`, "password12", "Media Owner");
+  await completeOnboarding(page);
+  const settings = await openUserSettings(page);
+  await settings.getByTestId("settings-nav-media").click();
+  const panel = settings.getByTestId("media-settings");
+  await expect(panel).toBeVisible();
+
+  const comfy = panel.locator("section").filter({ hasText: "ComfyUI" });
+  const transcription = panel.locator("section").filter({ hasText: "Transcription" });
+  await comfy.getByLabel("Server URL", { exact: true }).fill("http://127.0.0.1:8188");
+  await comfy.getByLabel("API key", { exact: true }).fill("fake-comfy-key");
+  await transcription.getByLabel("Server URL", { exact: true }).fill("http://127.0.0.1:8080/v1");
+  await transcription.getByLabel("Model", { exact: true }).fill("whisper-1");
+  // The ComfyUI model and workflows stay folded away until asked for.
+  await expect(comfy.getByLabel("Image model", { exact: true })).toBeHidden();
+  await comfy.scrollIntoViewIfNeeded();
+  await captureScreenshot(page, testInfo, "media-settings");
+  await comfy.getByText("Advanced", { exact: true }).click();
+  await comfy.getByLabel("Image model", { exact: true }).fill("sd_xl_base_1.0.safetensors");
+
+  await panel.getByRole("button", { name: "Save", exact: true }).click();
+  await expect(panel.getByText("Saved", { exact: true })).toBeVisible();
+  expect(saved).toEqual([
+    {
+      comfyui: {
+        baseUrl: "http://127.0.0.1:8188",
+        apiKey: "fake-comfy-key",
+        imageModel: "sd_xl_base_1.0.safetensors",
+        imageWorkflow: "",
+        videoWorkflow: "",
+      },
+      transcription: { baseUrl: "http://127.0.0.1:8080/v1", model: "whisper-1" },
+    },
+  ]);
+  // The key never comes back; the field shows that one is saved and offers to remove it.
+  const comfyKey = comfy.getByLabel("API key", { exact: true });
+  await expect(comfyKey).toHaveValue("");
+  await expect(comfyKey).toHaveAttribute("placeholder", "Saved");
+
+  await comfy.getByRole("button", { name: "Remove", exact: true }).click();
+  await transcription.getByLabel("Server URL", { exact: true }).fill("");
+  await panel.getByRole("button", { name: "Save", exact: true }).click();
+  await expect.poll(() => saved.length).toBe(2);
+  expect(saved[1]).toMatchObject({ comfyui: { apiKey: "" }, transcription: null });
+  await expect(comfyKey).toHaveAttribute("placeholder", "Optional");
+});
+
+test("members who do not own the deployment have no media settings", async ({ page }) => {
+  await asDeploymentOwner(page, false);
+  await signup(page, `media-member-${Date.now()}@rakazo.test`, "password12", "Media Member");
+  await completeOnboarding(page);
+  const settings = await openUserSettings(page);
+  await expect(settings.getByTestId("settings-nav-voice")).toBeVisible();
+  await expect(settings.getByTestId("settings-nav-media")).toHaveCount(0);
+});

packages/adapters/src/executor-capabilities.test.ts

@@ -45,7 +45,12 @@ function effectStore() {
   };
 }
 
-async function runBot(policy: BotPolicy, calls: Call[] = [], models: Record<string, unknown> = {}) {
+async function runBot(
+  policy: BotPolicy,
+  calls: Call[] = [],
+  models: Record<string, unknown> = {},
+  deps: Record<string, unknown> = {},
+) {
   const run = {
     id: "run-1",
     botId: "bot-1",
@@ -138,6 +143,7 @@ async function runBot(policy: BotPolicy, calls: Call[] = [], models: Record<stri
     },
     jobs,
     secrets: [],
+    ...deps,
   } as unknown as Parameters<typeof createRunExecutor>[0]);
   await executor.continueRun(run.id, "worker-1");
   if (!request) throw new Error("the runtime was never started");
@@ -149,6 +155,50 @@ describe("bot capabilities in runs", () => {
   it("exposes every builtin to an unrestricted bot", async () => {
     const { names } = await runBot({});
     expect(names).toEqual(expect.arrayContaining(["write_file", "shell", "schedule_list"]));
+    // Media tools need a media runtime that can run their operation.
+    expect(names).not.toContain("generate_image");
+  });
+
+  it("exposes media tools the space can run and starts their jobs without waiting", async () => {
+    const start = vi.fn(async () => ({ jobId: "job-1", messageId: "message-1" }));
+    const media = {
+      capabilities: async () => ({
+        generate_image: true,
+        generate_video: false,
+        edit_image: false,
+        edit_video: false,
+        export_video: false,
+        transcribe: false,
+      }),
+      start,
+      get: async () => null,
+      cancel: async () => undefined,
+    };
+    const { names, results } = await runBot(
+      {},
+      [
+        { name: "generate_image", args: { prompt: "a red fox", continue_after: true } },
+        { name: "generate_video", args: { prompt: "a red fox" } },
+      ],
+      {},
+      { media },
+    );
+    expect(names).toContain("generate_image");
+    expect(names).not.toContain("generate_video");
+    expect(results[0]).toMatchObject({ jobId: "job-1", status: "queued" });
+    expect(start).toHaveBeenCalledExactlyOnceWith({
+      operation: "generate_image",
+      input: { prompt: "a red fox" },
+      spaceId: "space-1",
+      userId: "user-1",
+      botId: "bot-1",
+      groupId: null,
+      threadId: "thread-1",
+      runId: "run-1",
+      operationKey: expect.any(String),
+      continueAfter: true,
+    });
+    expect(results[1]).toEqual({ error: "Tool generate_video is not available in this run." });
   });
 
   it("exposes only the allowlisted builtins and refuses calls to any other", async () => {

packages/adapters/src/media-runtime.postgres.test.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from "node:crypto";
 import { existsSync } from "node:fs";
-import { mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
+import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import type {
@@ -546,14 +546,30 @@ describePostgres("media job runtime (PostgreSQL, fake operations)", () => {
         leaseExpiresAt: crashedAt,
       },
     });
+    // The dead runner's scratch files are still on disk.
+    await mkdir(path.join(ctx.tempRoot, doomed.jobId), { recursive: true });
+    await writeFile(path.join(ctx.tempRoot, doomed.jobId, "partial.mp4"), PNG);
     await ctx.runtime.runJob(doomed.jobId);
     expect(await ctx.job(doomed.jobId)).toMatchObject({
       status: "failed",
       error: "The job was interrupted. Try again.",
     });
+    expect(existsSync(path.join(ctx.tempRoot, doomed.jobId))).toBe(false);
     expect(ctx.handlerCalls).toEqual([jobId]);
   });
 
+  it("sweeps temp directories of jobs that no longer run", async () => {
+    const ctx = await setup();
+    const active = await ctx.start({ prompt: "active" });
+    const ended = await ctx.start({ prompt: "ended" });
+    await ctx.runtime.cancel(ended.jobId);
+    for (const name of [active.jobId, ended.jobId, "unknown-job"]) {
+      await mkdir(path.join(ctx.tempRoot, name), { recursive: true });
+    }
+    await ctx.runtime.reconcile();
+    expect((await readdir(ctx.tempRoot)).sort()).toEqual([active.jobId]);
+  });
+
   it("re-enqueues lost and overdue jobs, outside the queue once they look stuck", async () => {
     const ctx = await setup();
     const fresh = await ctx.start({ prompt: "fresh" });

packages/adapters/src/media-runtime.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from "node:crypto";
 import { constants, createWriteStream } from "node:fs";
-import { copyFile, mkdir, rm } from "node:fs/promises";
+import { copyFile, mkdir, readdir, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import { Readable } from "node:stream";
@@ -211,6 +211,37 @@ function insideDir(dir: string, file: string): string {
   return resolved;
 }
 
+/**
+ * Runs the operation, giving up `graceMs` after an abort if the handler ignores the signal. A
+ * late rejection of either side is observed, so an abandoned handler cannot crash the process.
+ */
+async function runHandler(
+  handler: MediaOperationHandler,
+  context: MediaJobContext,
+  graceMs: number,
+) {
+  const work = Promise.resolve().then(() => handler.run(context));
+  work.catch(() => undefined);
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const onAbort = () => {
+    timer = setTimeout(() => giveUp(context.signal.reason), graceMs);
+    timer.unref?.();
+  };
+  let giveUp: (reason: unknown) => void = () => undefined;
+  const abandoned = new Promise<never>((_, reject) => {
+    giveUp = reject;
+  });
+  abandoned.catch(() => undefined);
+  if (context.signal.aborted) onAbort();
+  else context.signal.addEventListener("abort", onAbort, { once: true });
+  try {
+    return await Promise.race([work, abandoned]);
+  } finally {
+    clearTimeout(timer);
+    context.signal.removeEventListener("abort", onAbort);
+  }
+}
+
 /** The sentence a bot is woken with when its job ends; job text is data, not instructions. */
 export function mediaWakePrompt(job: {
   id: string;
@@ -292,6 +323,25 @@ export function createMediaRuntime(
     });
   };
 
+  const removeTempDir = async (jobId: string) => {
+    await rm(path.join(tempRoot, jobId), { recursive: true, force: true }).catch((error) => {
+      getLogger().error("media job temp cleanup", error);
+    });
+  };
+
+  /** Removes temp directories of jobs that no longer run (their runner died on this host). */
+  const sweepTempDirs = async () => {
+    const names = await readdir(tempRoot).catch(() => [] as string[]);
+    const candidates = names.filter((name) => !running.has(name));
+    if (!candidates.length) return;
+    const active = await deps.prisma.mediaJob.findMany({
+      where: { id: { in: candidates }, status: { in: ACTIVE_STATUSES } },
+      select: { id: true },
+    });
+    const keep = new Set(active.map((job) => job.id));
+    for (const name of candidates) if (!keep.has(name)) await removeTempDir(name);
+  };
+
   /** Rewrites the job's media block; returns the event seq when the card changed. */
   const writeCard = async (
     tx: Prisma.TransactionClient,
@@ -454,6 +504,8 @@ export function createMediaRuntime(
       }),
     );
     if (!committed) return false;
+    // Leftovers of a runner that died on this host; a live runner's directory is its own.
+    if (!running.has(jobId)) await removeTempDir(jobId);
     await notify(committed.job.threadId, committed.seq);
     if (committed.wakeRunId) {
       // A queued run is durable; the run reconciler repairs a missed wake.
@@ -772,17 +824,7 @@ export function createMediaRuntime(
           throw new MediaJobDeferredError("Waiting for free GPU memory", GPU_RETRY_MS);
         }
       }
-      const result = await Promise.race([
+      const result = await runHandler(registration.handler, context, abortGraceMs);
-        registration.handler.run(context),
-        new Promise<never>((_, 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) {
@@ -809,9 +851,7 @@ export function createMediaRuntime(
       clearInterval(heartbeat);
       clearTimeout(deadline);
       running.delete(jobId);
-      await rm(tempDir, { recursive: true, force: true }).catch((error) => {
+      await removeTempDir(jobId);
-        getLogger().error("media job temp cleanup", error);
-      });
     }
   };
 
@@ -996,6 +1036,7 @@ export function createMediaRuntime(
     },
 
     async reconcile() {
+      await sweepTempDirs();
       const now = Date.now();
       const graceAgo = new Date(now - RECONCILE_GRACE_MS);
       const due = await deps.prisma.mediaJob.findMany({