feat(core): send uploaded files with a chat message

packages/contracts/src/threads.ts

@@ -2,6 +2,11 @@ import { Effect, Schema, SchemaGetter } from "effect";
 import { fromMarkdown } from "mdast-util-from-markdown";
 import { toString as mdastToString } from "mdast-util-to-string";
 import { agentColorSchema, agentFaceSchema } from "./agents.ts";
+import {
+	attachmentNameSchema,
+	MAX_ATTACHMENTS_PER_MESSAGE,
+	mediaTypeSchema,
+} from "./attachments.ts";
 import { emailSchema } from "./email.ts";
 import {
 	routineExecutionSchema,
@@ -258,10 +263,23 @@ export const toolCallPartSchema = Schema.Struct({
 
 export type ToolCallPart = typeof toolCallPartSchema.Type;
 
+/** A file attached to the message, served by the chat it was sent in. */
+export const filePartSchema = Schema.Struct({
+	type: Schema.Literal("file"),
+	id: uuidSchema,
+	name: attachmentNameSchema,
+	contentType: mediaTypeSchema,
+	/** In bytes. */
+	size: Schema.Int,
+});
+
+export type FilePart = typeof filePartSchema.Type;
+
 export const messagePartSchema = Schema.Union([
 	textPartSchema,
 	collaborationPartSchema,
 	toolCallPartSchema,
+	filePartSchema,
 ]);
 
 export type MessagePart = typeof messagePartSchema.Type;
@@ -550,11 +568,22 @@ export type ThreadActivity = typeof threadActivitySchema.Type;
 
 export const newMessageSchema = Schema.Struct({
 	id: uuidSchema,
-	message: Schema.Trim.check(Schema.isMinLength(1), Schema.isMaxLength(20_000)),
+	/** May be empty when the message carries files. */
+	message: Schema.Trim.check(Schema.isMaxLength(20_000)),
+	/** Files the sender uploaded to the chat and has not sent yet. They show in the order listed. */
+	attachments: Schema.optional(
+		Schema.Array(uuidSchema).check(Schema.isMaxLength(MAX_ATTACHMENTS_PER_MESSAGE)),
+	),
 	/** For the people in the thread only: the bots read it, but none replies. */
 	peopleOnly: Schema.optional(Schema.Boolean),
 	/** The id of the message in the same thread that this one quotes and answers. */
 	replyTo: Schema.optional(uuidSchema),
-});
+}).check(
+	Schema.makeFilter(({ message, attachments }) =>
+		message !== "" || (attachments?.length ?? 0) > 0
+			? undefined
+			: { issue: "A message needs text or a file", path: ["message"] },
+	),
+);
 
 export type NewMessage = typeof newMessageSchema.Type;

packages/core/src/conversations/chats/chats.test.ts

@@ -1,5 +1,10 @@
-import { type ChatMessageItem, handleFromName, threadChannel } from "@sugabots/contracts";
-import { and, eq, like } from "drizzle-orm";
+import {
+	type ChatMessageItem,
+	type FilePart,
+	handleFromName,
+	threadChannel,
+} from "@sugabots/contracts";
+import { and, desc, eq, like } from "drizzle-orm";
 import { Context, Effect, Stream } from "effect";
 import { afterAll, beforeEach, describe, expect, it } from "vitest";
 import { Visibility } from "../../authorization/visibility.ts";
@@ -29,7 +34,9 @@ import { onPostgresAs } from "../../workspaces/testing.ts";
 import { Reactions } from "../reactions/reactions.ts";
 import { Routines } from "../routines/routines.ts";
 import { conversationsForTests } from "../testing.ts";
+import { ThreadFiles } from "../thread-files/thread-files.ts";
 import { ThreadView } from "../thread-view.ts";
+import { ThreadRepository } from "../threads/repository.ts";
 import { queueFacilitationForTests, runningTurns } from "../turns/testing.ts";
 import { Turns } from "../turns/turns.ts";
 import { ChatFiles } from "./chat-files.ts";
@@ -1465,6 +1472,105 @@ describe.skipIf(!process.env.DATABASE_URL)("chats, against Postgres", async () =
 		});
 	});
 
+	describe("sending files", () => {
+		/** The chat, with the member's uploads to it. */
+		async function aChatWithFiles() {
+			const { id: chatId } = await chats.open({
+				workspace: workspaceId,
+				podId,
+				hostAgentId: agentId,
+			});
+			const files = onPostgresAs(userId)(Context.get(conversations, ChatFiles.Service));
+			const upload = (name: string) =>
+				files.upload({
+					chatId,
+					name,
+					type: "text/plain",
+					stream: Stream.succeed(new TextEncoder().encode(name)),
+				});
+			return { chatId, files, upload };
+		}
+		const asPart = ({ id, name, contentType, size }: ThreadFiles.Upload): FilePart => ({
+			type: "file",
+			id,
+			name,
+			contentType,
+			size,
+		});
+		const shownMessage = async (chatId: string, messageId: string) =>
+			(await view.messages(chatId)).items.find(
+				(item) => item.kind === "message" && item.message.id === messageId,
+			);
+
+		it("sends the files with the message, in the order listed, for its readers to open", async () => {
+			const { chatId, files, upload } = await aChatWithFiles();
+			const first = await upload("a.txt");
+			const second = await upload("b.txt");
+
+			const sent = await chats.post({
+				chatId,
+				messageId: crypto.randomUUID(),
+				content: "",
+				attachments: [second.id, first.id],
+			});
+
+			expect(sent).toMatchObject({ content: "", parts: [asPart(second), asPart(first)] });
+			expect(await shownMessage(chatId, sent.id)).toMatchObject({
+				message: { parts: [asPart(second), asPart(first)] },
+			});
+			// Watching clients add the announced message as it is, so it must carry its files.
+			const [announced] = await onDatabase((db) =>
+				db
+					.select({ payload: event.payload })
+					.from(event)
+					.where(
+						and(eq(event.channel, threadChannel(sent.threadId)), eq(event.type, "message.created")),
+					)
+					.orderBy(desc(event.seq))
+					.limit(1),
+			);
+			expect(announced?.payload).toMatchObject({
+				message: { id: sent.id, parts: [asPart(second), asPart(first)] },
+			});
+			expect(await files.read({ chatId, fileId: first.id })).toMatchObject({ name: "a.txt" });
+		});
+
+		it("posts nothing when a file is not one of the sender's unsent uploads", async () => {
+			const { chatId, upload } = await aChatWithFiles();
+			const sentBefore = await upload("sent.txt");
+			await chats.post({
+				chatId,
+				messageId: crypto.randomUUID(),
+				content: "first",
+				attachments: [sentBefore.id],
+			});
+			const fresh = await upload("fresh.txt");
+			const messageId = crypto.randomUUID();
+
+			await expect(
+				chats.post({ chatId, messageId, content: "again", attachments: [fresh.id, sentBefore.id] }),
+			).rejects.toBeInstanceOf(ThreadFiles.UploadsNotFound);
+			expect(await shownMessage(chatId, messageId)).toBeUndefined();
+			// The fresh upload is still the sender's to send.
+			await expect(
+				chats.post({ chatId, messageId, content: "again", attachments: [fresh.id] }),
+			).resolves.toMatchObject({ parts: [{ type: "text" }, asPart(fresh)] });
+		});
+
+		it("returns a message sent again with its files, and refuses its id for other files", async () => {
+			const { chatId, upload } = await aChatWithFiles();
+			const attached = await upload("a.txt");
+			const other = await upload("b.txt");
+			const messageId = crypto.randomUUID();
+			const send = (attachments: string[]) =>
+				chats.post({ chatId, messageId, content: "files", attachments });
+			const sent = await send([attached.id]);
+
+			await expect(send([attached.id])).resolves.toEqual(sent);
+			await expect(send([other.id])).rejects.toBeInstanceOf(ThreadRepository.MessageIdConflict);
+		});
+	});
+
 	it("keeps an uploaded file as the type it was given, or else the type its name suggests", async () => {
 		const { id: chatId } = await chats.open({
 			workspace: workspaceId,

packages/core/src/conversations/chats/chats.ts

@@ -12,6 +12,7 @@ import { query, serviceOperations, transaction } from "../../database/database.t
 import { agent, pod } from "../../database/schema.ts";
 import type { UserFacing } from "../../user-message.ts";
 import { FloorControl } from "../floor/floor-control.ts";
+import type { ThreadFiles } from "../thread-files/thread-files.ts";
 import { crewOf, personColumns } from "../threads/participants.ts";
 import { ThreadRepository } from "../threads/repository.ts";
 import { toChat } from "./chat.ts";
@@ -33,13 +34,17 @@ export interface Interface {
 		hostAgentId: string;
 	}) => Effect.Effect<Chat, AuthorizationDenied | ChatPlacementRejected, CurrentActor.Service>;
 	/**
-	 * Posts the actor's message into the chat's main thread and gives the
-	 * floor. Posting the same message again returns it unchanged.
+	 * Posts the actor's message into the chat's main thread, with the files
+	 * they uploaded to it, and gives the floor. Posting the same message again
+	 * returns it unchanged.
 	 */
 	readonly post: (input: {
 		chatId: string;
 		messageId: string;
+		/** May be empty when the message carries files. */
 		content: string;
+		/** Ids of the actor's uploads to the chat, not yet sent. None unless set. */
+		attachments?: readonly string[];
 		/** Written for the other people, so no bot replies. Off unless set. */
 		peopleOnly?: boolean;
 		/** The id of the message in the chat's main thread that this one quotes and answers. */
@@ -49,6 +54,7 @@ export interface Interface {
 		| ResourceHidden
 		| ThreadRepository.MessageIdConflict
 		| ThreadRepository.ReplyTargetNotFound
+		| ThreadFiles.UploadsNotFound
 		| ChatAgentHasNoModel,
 		CurrentActor.Service
 	>;
@@ -98,6 +104,7 @@ export const make = Effect.gen(function* () {
 							author: sender,
 							content: input.content,
 							replyTo: input.replyTo,
+							attachments: input.attachments ?? [],
 						});
 						if (posted._tag === "AlreadyPosted") return posted.message;
 						// Checked once the message is known to be new, so sending it again

packages/core/src/conversations/thread-files/thread-files.test.ts

@@ -194,10 +194,44 @@ describe.skipIf(!process.env.DATABASE_URL)("thread files, against Postgres", ()
 		return sent.id;
 	}
 
-	const attach = (fileId: string, messageId: string) =>
-		onDatabase((db) =>
-			db.update(threadFile).set({ messageId, position: 0 }).where(eq(threadFile.id, fileId)),
+	/** Sends `fileIds`, uploaded by `userId`, in a new message of theirs. */
+	const send = async (threadId: string, userId: string, fileIds: string[]) =>
+		files.attach({
+			threadId,
+			messageId: await aMessage(threadId, userId),
+			uploaderId: userId,
+			fileIds,
+		});
+
+	it("attaches only the uploader's own unsent uploads, all of them or none", async () => {
+		const { threadId, userId } = await aThread();
+		const other = await aRoutineOwner();
+		const upload = (uploaderId: string, name: string) =>
+			files.upload({
+				threadId,
+				uploaderId,
+				name,
+				maxSize: 1024,
+				maxUnsent: 20,
+				content: text(name),
+			});
+		const mine = await upload(userId, "mine.txt");
+		const theirs = await upload(other.userId, "theirs.txt");
+
+		await expect(send(threadId, userId, [mine.id, theirs.id])).rejects.toBeInstanceOf(
+			ThreadFiles.UploadsNotFound,
 		);
+		// Refusing one attached none.
+		await expect(files.read({ threadId, fileId: mine.id })).rejects.toBeInstanceOf(
+			ThreadFiles.FileNotFound,
+		);
+
+		await expect(send(threadId, userId, [mine.id])).resolves.toEqual([mine]);
+		// Sent once, it is no longer an upload to send.
+		await expect(send(threadId, userId, [mine.id])).rejects.toBeInstanceOf(
+			ThreadFiles.UploadsNotFound,
+		);
+	});
 
 	it("keeps an upload unreadable until it is attached to a message", async () => {
 		const { threadId, userId } = await aThread();
@@ -214,7 +248,7 @@ describe.skipIf(!process.env.DATABASE_URL)("thread files, against Postgres", ()
 			ThreadFiles.FileNotFound,
 		);
 
-		await attach(uploaded.id, await aMessage(threadId, userId));
+		await send(threadId, userId, [uploaded.id]);
 		const read = await files.read({ threadId, fileId: uploaded.id });
 
 		expect(uploaded).toEqual({
@@ -268,7 +302,7 @@ describe.skipIf(!process.env.DATABASE_URL)("thread files, against Postgres", ()
 		// Someone else's unsent uploads are theirs to count.
 		await expect(upload(other.userId, "theirs.txt")).resolves.toMatchObject({ name: "theirs.txt" });
 		// Sending one frees its place.
-		await attach(first.id, await aMessage(threadId, userId));
+		await send(threadId, userId, [first.id]);
 		await expect(upload(userId, "three.txt")).resolves.toMatchObject({ name: "three.txt" });
 	});
 
@@ -285,7 +319,7 @@ describe.skipIf(!process.env.DATABASE_URL)("thread files, against Postgres", ()
 			});
 		await upload("abandoned.txt");
 		const sent = await upload("sent.txt");
-		await attach(sent.id, await aMessage(threadId, userId));
+		await send(threadId, userId, [sent.id]);
 		const result = await files.write({ threadId, content: text("a tool's result") });
 		const blobsUnder = () =>
 			runOnPostgres(