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