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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u

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