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