diff --git a/apps/api/src/media-jobs.ts b/apps/api/src/media-jobs.ts index a6d3a98..760f6cc 100644 --- a/apps/api/src/media-jobs.ts +++ b/apps/api/src/media-jobs.ts @@ -11,6 +11,7 @@ import { import type { Actor, MediaJobStartInput, MediaSettingsUpdate } from "@rakazo/contracts"; import { MediaOperationSchema } from "@rakazo/contracts"; import type { PrismaClient } from "@rakazo/db"; +import { IsolationError, requireMembership } from "@rakazo/db"; import { resolveThreadTarget } from "./thread-target.js"; export interface MediaRouteDeps { @@ -51,18 +52,45 @@ export async function updateMediaSettings( } /** - * Starts an operation a user picked on a card (export, transcribe, edit) in the same conversation. - * The thread is authorized like every thread procedure; the runtime checks that the operation's - * source artifacts belong to the actor, and the job can only ever read the actor's artifacts. + * The actor in the space a card lives in. A card is acted on where it is shown, which need not be + * the space the client has selected (the main conversation lives in the home space), so the + * space comes from the card's own record: it must belong to the actor, and the actor must still + * be a member of that space. Anything else reads as not found. + */ +async function actorInSpace(deps: MediaRouteDeps, actor: Actor, spaceId: string): Promise { + if (spaceId === actor.spaceId) return actor; + try { + return await requireMembership(deps.prisma, actor.userId, spaceId); + } catch (error) { + if (error instanceof IsolationError) throw new ORPCError("NOT_FOUND"); + throw error; + } +} + +/** + * Starts an operation a user picked on a card (export, transcribe, edit) in the card's + * conversation. The conversation is authorized like every thread procedure, in the card's space; + * the runtime checks that the operation's source artifacts belong to the actor there, and the + * job can only ever read the actor's artifacts of that space. */ export async function startMediaJob(deps: MediaRouteDeps, actor: Actor, input: MediaJobStartInput) { const media = requireMedia(deps); - const target = await resolveThreadTarget(deps.prisma, actor, input); const source = await deps.prisma.message.findFirst({ - where: { id: input.messageId, threadId: target.threadId }, - select: { botId: true }, + where: { + id: input.messageId, + thread: { + userId: actor.userId, + ...(input.botId ? { botId: input.botId } : { groupId: input.groupId }), + }, + }, + select: { botId: true, thread: { select: { spaceId: true } } }, }); if (!source) throw new ORPCError("NOT_FOUND"); + const scoped = await actorInSpace(deps, actor, source.thread.spaceId); + const target = await resolveThreadTarget(deps.prisma, scoped, input).catch((error) => { + if (error instanceof IsolationError) throw new ORPCError("NOT_FOUND"); + throw error; + }); const bots = target.kind === "bot" ? [target.botId] @@ -78,8 +106,8 @@ export async function startMediaJob(deps: MediaRouteDeps, actor: Actor, input: M return await media.start({ operation: input.operation, input: input.input, - spaceId: actor.spaceId, - userId: actor.userId, + spaceId: scoped.spaceId, + userId: scoped.userId, botId, groupId: target.kind === "group" ? target.groupId : null, threadId: target.threadId, @@ -93,36 +121,43 @@ export async function startMediaJob(deps: MediaRouteDeps, actor: Actor, input: M } } -/** Booleans only: which operations the actor's space can run, never how they are configured. */ +/** + * Booleans only: which operations a space can run, never how they are configured. With a job, + * the space is the job's (a card's follow-ups); otherwise the actor's current space. + */ export async function getMediaCapabilities( deps: MediaRouteDeps, actor: Actor, + input: { jobId?: string } = {}, ): Promise { - if (deps.media) return deps.media.capabilities(actor.spaceId); + const spaceId = input.jobId ? (await ownedJob(deps, actor, input.jobId)).spaceId : actor.spaceId; + if (deps.media) return deps.media.capabilities(spaceId); return Object.fromEntries( MediaOperationSchema.options.map((operation) => [operation, false]), ) as MediaCapabilities; } -async function ownedJobId(deps: MediaRouteDeps, actor: Actor, jobId: string): Promise { +/** The actor's job, in whichever of the actor's spaces it runs. */ +async function ownedJob(deps: MediaRouteDeps, actor: Actor, jobId: string) { const job = await deps.prisma.mediaJob.findFirst({ - where: { id: jobId, spaceId: actor.spaceId, userId: actor.userId }, - select: { id: true }, + where: { id: jobId, userId: actor.userId }, + select: { id: true, spaceId: true }, }); if (!job) throw new ORPCError("NOT_FOUND"); - return job.id; + await actorInSpace(deps, actor, job.spaceId); + return job; } export async function getMediaJob(deps: MediaRouteDeps, actor: Actor, jobId: string) { const media = requireMedia(deps); - const snapshot = await media.get(await ownedJobId(deps, actor, jobId)); + const snapshot = await media.get((await ownedJob(deps, actor, jobId)).id); if (!snapshot) throw new ORPCError("NOT_FOUND"); return snapshot; } export async function cancelMediaJob(deps: MediaRouteDeps, actor: Actor, jobId: string) { const media = requireMedia(deps); - const id = await ownedJobId(deps, actor, jobId); + const { id } = await ownedJob(deps, actor, jobId); await media.cancel(id); const snapshot = await media.get(id); if (!snapshot) throw new ORPCError("NOT_FOUND"); diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index 37d03f8..fdc7705 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -3501,8 +3501,8 @@ export function createRouter(deps: RouterDeps) { updateMediaSettings(deps, context.actor, input), ), }, - capabilities: authed.media.capabilities.handler(({ context }) => - getMediaCapabilities(deps, context.actor), + capabilities: authed.media.capabilities.handler(({ context, input }) => + getMediaCapabilities(deps, context.actor, input), ), jobs: { start: authed.media.jobs.start.handler(({ context, input }) => diff --git a/apps/web/src/components/media/MediaCard.tsx b/apps/web/src/components/media/MediaCard.tsx index 14b6dff..80e7db4 100644 --- a/apps/web/src/components/media/MediaCard.tsx +++ b/apps/web/src/components/media/MediaCard.tsx @@ -21,7 +21,7 @@ import { Captions, ChevronDown, Download, Film, ImageIcon, Scissors, Sparkles } import { useEffect, useId, useState } from "react"; import type { ArtifactTarget } from "../../lib/artifact-open"; import { artifactUrl } from "../../lib/artifact-url"; -import { rpc, selectedSpaceId } from "../../lib/rpc"; +import { rpc } from "../../lib/rpc"; import { ArtifactImage } from "../../pages/shell/message-cards"; import { ArtifactFileCard } from "../ArtifactFileCard"; import { Shimmer } from "../ai/primitives"; @@ -305,22 +305,26 @@ export function MediaCardView({ const CAPABILITIES_TTL_MS = 60_000; let capabilities: - | { spaceId: string | null; at: number; value: Promise } + | { conversation: string; at: number; value: Promise } | undefined; -/** What the space can run, shared by every card on screen and refreshed after a minute. */ -function loadAvailability(): Promise { - const spaceId = selectedSpaceId(); +/** + * What the card's space can run, shared by every card of the conversation on screen and + * refreshed after a minute. Asked through a job, since the main conversation can live in another + * space than the one selected. + */ +function loadAvailability(target: ArtifactTarget, jobId: string): Promise { + const conversation = "botId" in target ? `bot:${target.botId}` : `group:${target.groupId}`; if ( !capabilities || - capabilities.spaceId !== spaceId || + capabilities.conversation !== conversation || Date.now() - capabilities.at > CAPABILITIES_TTL_MS ) { capabilities = { - spaceId, + conversation, at: Date.now(), value: rpc.media - .capabilities() + .capabilities({ jobId }) .then((value) => ({ export: value.export_video === true, transcribe: value.transcribe === true, @@ -331,18 +335,22 @@ function loadAvailability(): Promise { return capabilities.value; } -function useAvailability(enabled: boolean): MediaCardAvailability { +function useAvailability( + enabled: boolean, + target: ArtifactTarget, + jobId: string, +): MediaCardAvailability { const [available, setAvailable] = useState(NOTHING_AVAILABLE); useEffect(() => { if (!enabled) return; let current = true; - void loadAvailability().then((value) => { + void loadAvailability(target, jobId).then((value) => { if (current) setAvailable(value); }); return () => { current = false; }; - }, [enabled]); + }, [enabled, target, jobId]); return available; } @@ -361,6 +369,8 @@ export function MediaCard({ const available = useAvailability( !isActiveMediaStatus(block.status) && block.outputs.some((output) => output.kind === "video" || output.kind === "audio"), + target, + block.jobId, ); const onAction = async (action: MediaCardAction) => { if (action.kind === "cancel") { diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index b294b44..cef44c1 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -703,7 +703,10 @@ export const appContract = { get: oc.output(MediaSettingsViewSchema), update: oc.input(MediaSettingsUpdateSchema).output(MediaSettingsViewSchema), }, - capabilities: oc.output(MediaCapabilitiesSchema), + /** With a job id, what the job's space can run (a card's follow-ups). */ + capabilities: oc + .input(z.object({ jobId: Id.optional() }).optional()) + .output(MediaCapabilitiesSchema), jobs: { start: oc.input(MediaJobStartInputSchema).output(MediaJobStartOutputSchema), get: oc.input(z.object({ jobId: Id })).output(MediaJobSnapshotSchema), diff --git a/packages/testkit/src/media.postgres.test.ts b/packages/testkit/src/media.postgres.test.ts index ef06a03..98d6681 100644 --- a/packages/testkit/src/media.postgres.test.ts +++ b/packages/testkit/src/media.postgres.test.ts @@ -6,7 +6,7 @@ import type { MediaOperationRegistration } from "@rakazo/adapters"; import { ComposioEmulator } from "@rakazo/adapters"; import type { MediaJobBlock, MediaSettingsView, MessageBlock } from "@rakazo/contracts"; import { TranscribeInputSchema } from "@rakazo/contracts"; -import { createThreadMessage } from "@rakazo/db"; +import { createThreadMessage, ensureMainConversation, MAIN_BOT_SPAWN_KEY } from "@rakazo/db"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import type { createApp } from "../../../apps/api/src/app.ts"; import { sessionCookieHeader } from "./index.js"; @@ -344,6 +344,160 @@ describeWithDatabase("media RPCs (PostgreSQL)", () => { expect(["running", "cancelled"]).toContain(cancelled.status); await waitForStatus(slow.jobId, "cancelled"); }); + + it("acts on a card where it lives, whichever of the user's spaces is selected", async () => { + const cookie = await signup(app, `media-main-${stamp}@rakazo.test`, "media-main"); + const me = await rpc<{ userId: string; spaceId: string }>(app, cookie, "me"); + await rpc(app, cookie, "bots/create", { + name: "Chief", + title: "", + description: "", + instructions: "", + notifyOnFinish: false, + spawnKey: MAIN_BOT_SPAWN_KEY, + }); + const main = await ensureMainConversation(handles.prisma, me.userId); + expect(main.spaceId).toBe(me.spaceId); + const card = await createThreadMessage(handles.prisma, { + threadId: main.threadId, + role: "bot", + botId: main.leadBotId, + blocks: [{ kind: "text", text: "Here is your clip." }], + }); + const clip = await rpc<{ id: string }>(app, cookie, "artifacts/create", { + botId: main.leadBotId, + name: "clip.png", + mimeType: "image/png", + contentBase64: PNG_BASE64, + }); + // The user has moved on to another space; the main conversation stays in the home space. + const elsewhere = await rpc<{ id: string }>(app, cookie, "spaces/create", { + name: "Elsewhere", + }); + + const started = await rpc<{ jobId: string; messageId: string }>( + app, + cookie, + "media/jobs/start", + { + groupId: main.groupId, + messageId: card.id, + operation: "export_video", + input: { artifact_id: clip.id }, + }, + elsewhere.id, + ); + expect(await waitForStatus(started.jobId, "done")).toMatchObject({ + spaceId: main.spaceId, + threadId: main.threadId, + groupId: main.groupId, + botId: main.leadBotId, + }); + expect( + await rpc<{ status: string }>( + app, + cookie, + "media/jobs/get", + { jobId: started.jobId }, + elsewhere.id, + ), + ).toMatchObject({ status: "done" }); + expect( + await rpc(app, cookie, "media/capabilities", { jobId: started.jobId }, elsewhere.id), + ).toMatchObject({ export_video: true, transcribe: true, generate_image: false }); + const slow = await rpc<{ jobId: string }>( + app, + cookie, + "media/jobs/start", + { + groupId: main.groupId, + messageId: card.id, + operation: "transcribe", + input: { artifact_id: clip.id }, + }, + elsewhere.id, + ); + await waitForStatus(slow.jobId, "running"); + await rpc(app, cookie, "media/jobs/cancel", { jobId: slow.jobId }, elsewhere.id); + await waitForStatus(slow.jobId, "cancelled"); + + // Another user gets nothing, from any space of theirs. + const mallory = await user("media-main-mallory"); + for (const [proc, body] of [ + ["media/jobs/get", { jobId: started.jobId }], + ["media/jobs/cancel", { jobId: started.jobId }], + ["media/capabilities", { jobId: started.jobId }], + [ + "media/jobs/start", + { + groupId: main.groupId, + messageId: card.id, + operation: "export_video", + input: { artifact_id: clip.id }, + }, + ], + [ + "media/jobs/start", + { + botId: mallory.bot.id, + messageId: card.id, + operation: "export_video", + input: { artifact_id: mallory.artifact.id }, + }, + ], + ] as const) { + expect((await raw(app, mallory.cookie, proc, body)).status, proc).toBe(404); + } + + // A card in a space the user has left reads as not found too. + const bot = await rpc<{ id: string }>( + app, + cookie, + "bots/create", + { name: "Away", title: "", description: "", instructions: "", notifyOnFinish: false }, + elsewhere.id, + ); + const awayThread = await handles.prisma.thread.findUniqueOrThrow({ where: { botId: bot.id } }); + const awayCard = await createThreadMessage(handles.prisma, { + threadId: awayThread.id, + role: "bot", + botId: bot.id, + blocks: [{ kind: "text", text: "Here is another clip." }], + }); + const awayClip = await rpc<{ id: string }>( + app, + cookie, + "artifacts/create", + { botId: bot.id, name: "away.png", mimeType: "image/png", contentBase64: PNG_BASE64 }, + elsewhere.id, + ); + const away = await rpc<{ jobId: string }>(app, cookie, "media/jobs/start", { + botId: bot.id, + messageId: awayCard.id, + operation: "export_video", + input: { artifact_id: awayClip.id }, + }); + await waitForStatus(away.jobId, "done"); + await handles.prisma.spaceMember.deleteMany({ + where: { spaceId: elsewhere.id, userId: me.userId }, + }); + for (const [proc, body] of [ + ["media/jobs/get", { jobId: away.jobId }], + ["media/jobs/cancel", { jobId: away.jobId }], + ["media/capabilities", { jobId: away.jobId }], + [ + "media/jobs/start", + { + botId: bot.id, + messageId: awayCard.id, + operation: "export_video", + input: { artifact_id: awayClip.id }, + }, + ], + ] as const) { + expect((await raw(app, cookie, proc, body)).status, proc).toBe(404); + } + }); }); async function signup(app: App, email: string, name: string) { @@ -356,16 +510,27 @@ async function signup(app: App, email: string, name: string) { return sessionCookieHeader(res); } -function raw(app: App, cookie: string, proc: string, body: unknown = {}) { +function raw(app: App, cookie: string, proc: string, body: unknown = {}, spaceId?: string) { return app.request(`/rpc/${proc}`, { method: "POST", - headers: { "content-type": "application/json", cookie, origin: "http://127.0.0.1:5173" }, + headers: { + "content-type": "application/json", + cookie, + origin: "http://127.0.0.1:5173", + ...(spaceId ? { "x-rakazo-space-id": spaceId } : {}), + }, body: JSON.stringify({ json: body }), }); } -async function rpc(app: App, cookie: string, proc: string, body: unknown = {}): Promise { - const res = await raw(app, cookie, proc, body); +async function rpc( + app: App, + cookie: string, + proc: string, + body: unknown = {}, + spaceId?: string, +): Promise { + const res = await raw(app, cookie, proc, body, spaceId); const text = await res.text(); const parsed = JSON.parse(text) as { json?: T; error?: { message?: string } }; if (res.status >= 400 || parsed.error) {