From 4e88a9dfe6bf83290d7160b6d44359d375b5bf45 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Oct 2026 14:07:53 +0000 Subject: [PATCH 1/2] fix(media): abandoned handlers cannot crash the worker, sweep stale temp dirs - A handler that ignores its abort is left behind after the grace period; a late rejection from it (or from the grace timer after the job settled) was unobserved, and the worker exits on unhandled rejections. Both sides of the race are now observed and the grace timer is cleared with the job. - Ending a job removes its temp directory when no runner in this process holds it (leftovers of a runner that died here), and reconciliation sweeps directories of jobs that no longer run. - An executor test runs a media tool end to end through applyTool: shown only for operations the space can run, started with the run's scope and effect key, continue_after split off, refused when not exposed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u --- .../src/executor-capabilities.test.ts | 52 +++++++++++++- .../src/media-runtime.postgres.test.ts | 18 ++++- packages/adapters/src/media-runtime.ts | 71 +++++++++++++++---- 3 files changed, 124 insertions(+), 17 deletions(-) diff --git a/packages/adapters/src/executor-capabilities.test.ts b/packages/adapters/src/executor-capabilities.test.ts index 57de9d8..d38f94c 100644 --- a/packages/adapters/src/executor-capabilities.test.ts +++ b/packages/adapters/src/executor-capabilities.test.ts @@ -45,7 +45,12 @@ function effectStore() { }; } -async function runBot(policy: BotPolicy, calls: Call[] = [], models: Record = {}) { +async function runBot( + policy: BotPolicy, + calls: Call[] = [], + models: Record = {}, + deps: Record = {}, +) { const run = { id: "run-1", botId: "bot-1", @@ -138,6 +143,7 @@ async function runBot(policy: BotPolicy, calls: Call[] = [], models: Record[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 () => { diff --git a/packages/adapters/src/media-runtime.postgres.test.ts b/packages/adapters/src/media-runtime.postgres.test.ts index 75c597e..2852dbe 100644 --- a/packages/adapters/src/media-runtime.postgres.test.ts +++ b/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" }); diff --git a/packages/adapters/src/media-runtime.ts b/packages/adapters/src/media-runtime.ts index f36b3c8..5e26917 100644 --- a/packages/adapters/src/media-runtime.ts +++ b/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 | undefined; + const onAbort = () => { + timer = setTimeout(() => giveUp(context.signal.reason), graceMs); + timer.unref?.(); + }; + let giveUp: (reason: unknown) => void = () => undefined; + const abandoned = new Promise((_, 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([ - 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 }); - }), - ]); + const result = await runHandler(registration.handler, context, abortGraceMs); 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) => { - getLogger().error("media job temp cleanup", error); - }); + await removeTempDir(jobId); } }; @@ -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({ From f407829524decbda751b07a26e98e7a7ce353e84 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Oct 2026 14:25:43 +0000 Subject: [PATCH 2/2] test(web): media settings E2E with the media-settings screenshot The Media settings section (added in the previous web commit) lets the deployment owner point media at their own servers: ComfyUI URL and optional key, with the image model and custom image/video workflows folded under Advanced, and a transcription URL with optional model and key. Saved keys are never shown again; the key field says one is saved and offers to remove it. Members who do not own the deployment do not see the section. The spec fills both servers, captures `media-settings`, saves and checks the exact update the server receives, then removes the saved key and the transcription server; a second spec checks a member has no Media section. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u --- apps/web/e2e/media-settings.spec.ts | 122 ++++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 apps/web/e2e/media-settings.spec.ts diff --git a/apps/web/e2e/media-settings.spec.ts b/apps/web/e2e/media-settings.spec.ts new file mode 100644 index 0000000..4f50ad7 --- /dev/null +++ b/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 = { 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); +});