feat(core): let a bot share a file from its sandbox in its reply

packages/contracts/src/events.ts

@@ -2,6 +2,7 @@ import { Schema } from "effect";
 import { customerThreadTypeSchema } from "./chats.ts";
 import {
 	collaborationPartSchema,
+	filePartSchema,
 	messageReactionsSchema,
 	messageSchema,
 	messageStatusSchema,
@@ -64,6 +65,15 @@ export const eventPayloadSchemas = {
 		messageId,
 		reactions: messageReactionsSchema,
 	}),
+	/**
+	 * An agent attached a file to its reply; `files` is all the reply carries
+	 * now. A reply that starts again comes with `turn.started`, without them.
+	 */
+	"message.files_changed": Schema.Struct({
+		threadId,
+		messageId,
+		files: Schema.Array(filePartSchema),
+	}),
 	/** An agent's reply called a built-in tool; the part is `running`. */
 	"tool_call.started": Schema.Struct({ threadId, messageId, toolCall: toolCallPartSchema }),
 	/** The tool returned or failed; the part carries its output or error. */
@@ -140,6 +150,7 @@ export const durableEventTypeSchema = Schema.Literals([
 	"message.completed",
 	"message.failed",
 	"message.reactions_changed",
+	"message.files_changed",
 	"tool_call.started",
 	"tool_call.completed",
 	"tool_call.updated",
@@ -259,6 +270,7 @@ export const threadUpdateEventSchema = Schema.Union([
 	withEnvelope("message.completed", P["message.completed"]),
 	withEnvelope("message.failed", P["message.failed"]),
 	withEnvelope("message.reactions_changed", P["message.reactions_changed"]),
+	withEnvelope("message.files_changed", P["message.files_changed"]),
 	withEnvelope("tool_call.started", P["tool_call.started"]),
 	withEnvelope("tool_call.completed", P["tool_call.completed"]),
 	withEnvelope("tool_call.updated", P["tool_call.updated"]),

packages/core/src/conversations/chats/chat-files.ts

@@ -44,7 +44,10 @@ export interface Interface {
 		chatId: string;
 		fileId: string;
 	}) => Effect.Effect<void, ResourceHidden, CurrentActor.Service>;
-	/** A file attached to a message in the chat. An upload not yet sent is not found. */
+	/**
+	 * A file attached to a message in the chat. Any other file in its thread,
+	 * such as an upload not yet sent or a tool call's result, is not found.
+	 */
 	readonly read: (input: {
 		chatId: string;
 		fileId: string;
@@ -92,7 +95,7 @@ export const make = Effect.gen(function* () {
 			operation(
 				"read",
 				Effect.flatMap(visibility.chat(chatId), ({ mainThreadId }) =>
-					files.read({ threadId: mainThreadId, fileId }),
+					files.readMessageFile({ threadId: mainThreadId, fileId }),
 				),
 			),
 	});

packages/core/src/conversations/events.ts

@@ -1,5 +1,6 @@
 import type {
 	CollaborationPart,
+	FilePart,
 	Message,
 	MessageReaction,
 	PersonParticipant,
@@ -35,6 +36,15 @@ export type ConversationEvent = Data.TaggedEnum<{
 		readonly messageId: string;
 		readonly reactions: readonly MessageReaction[];
 	};
+	/**
+	 * An agent attached a file to its reply. `files` is all the reply carries
+	 * now; a reply that starts again starts without them, as its `TurnStarted` says.
+	 */
+	MessageFilesChanged: {
+		readonly threadId: string;
+		readonly messageId: string;
+		readonly files: readonly FilePart[];
+	};
 	/** Agents were brought into the thread to answer in it. */
 	AgentsJoined: { readonly threadId: string; readonly agentIds: readonly string[] };
 	/**

packages/core/src/conversations/thread-feed.ts

@@ -41,6 +41,12 @@ function streamEventsFor(event: ConversationEvent): PendingEvent[] {
 				}),
 			),
 		],
+		MessageFilesChanged: ({ threadId, messageId, files }) => [
+			onThread(
+				threadId,
+				streamEvent("message.files_changed", { threadId, messageId, files: [...files] }),
+			),
+		],
 		AgentsJoined: ({ threadId }) => [
 			onThread(threadId, streamEvent("thread.changed", { threadId })),
 		],

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

@@ -1,3 +1,4 @@
+import { MAX_ATTACHMENTS_PER_MESSAGE } from "@sugabots/contracts";
 import { eq } from "drizzle-orm";
 import { ConfigProvider, Duration, Effect, Layer, Stream } from "effect";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
@@ -261,6 +262,27 @@ describe.skipIf(!process.env.DATABASE_URL)("thread files, against Postgres", ()
 		expect(await textOf(read.stream)).toBe("draft");
 	});
 
+	it("reads as a message's file only one attached to a message", async () => {
+		const { threadId, userId } = await aThread();
+		const result = await files.write({ threadId, content: text("long result") });
+		const uploaded = await files.upload({
+			threadId,
+			uploaderId: userId,
+			name: "notes.txt",
+			maxSize: 1024,
+			maxUnsent: 20,
+			content: text("draft"),
+		});
+		const readMessageFile = (fileId: string) => files.readMessageFile({ threadId, fileId });
+
+		await expect(readMessageFile(result.id)).rejects.toBeInstanceOf(ThreadFiles.FileNotFound);
+		await expect(readMessageFile(uploaded.id)).rejects.toBeInstanceOf(ThreadFiles.FileNotFound);
+
+		await send(threadId, userId, [uploaded.id]);
+
+		expect(await textOf((await readMessageFile(uploaded.id)).stream)).toBe("draft");
+	});
+
 	it("refuses an upload over its own limit, keeping nothing", async () => {
 		const { threadId, userId } = await aThread();
 
@@ -369,6 +391,57 @@ describe.skipIf(!process.env.DATABASE_URL)("thread files, against Postgres", ()
 		);
 	});
 
+	it("attaches an agent's files to its reply in order, up to what a message carries, and removes them when it starts again", async () => {
+		const { threadId, agentId } = await aThread();
+		const [reply] = await onDatabase((db) =>
+			db
+				.insert(message)
+				.values({
+					threadId,
+					authorAgentId: agentId,
+					kind: "text",
+					status: "streaming",
+					parts: [],
+					content: "",
+				})
+				.returning({ id: message.id }),
+		);
+		if (!reply) throw new Error("Could not create reply");
+		const attach = (name: string) =>
+			runOnPostgres(
+				Effect.flatMap(ThreadFiles.Service, (service) =>
+					service.withReplyFile(
+						{
+							threadId,
+							messageId: reply.id,
+							name,
+							contentType: "text/csv",
+							content: new TextEncoder().encode(name),
+						},
+						service.attachToReply,
+					),
+				).pipe(Effect.provide(ThreadFiles.layer.pipe(Layer.orDie))),
+			);
+
+		const first = await attach("first.csv");
+		const second = await attach("second.csv");
+		expect(await files.replyAttachments(reply.id)).toEqual([first, second]);
+		expect(await textOf((await files.read({ threadId, fileId: second.id })).stream)).toBe(
+			"second.csv",
+		);
+
+		for (let attached = 2; attached < MAX_ATTACHMENTS_PER_MESSAGE; attached++) {
+			await attach(`more-${attached}.csv`);
+		}
+		await expect(attach("one-too-many.csv")).rejects.toBeInstanceOf(ThreadFiles.TooManyReplyFiles);
+
+		await files.removeReplyAttachments({ threadId, messageId: reply.id });
+		expect(await files.replyAttachments(reply.id)).toEqual([]);
+		await expect(files.read({ threadId, fileId: first.id })).rejects.toBeInstanceOf(
+			ThreadFiles.FileNotFound,
+		);
+	});
+
 	it("deletes a thread's blobs once a cascade has deleted the thread", async () => {
 		const { threadId, podId } = await aThread();
 		const written = await files.write({ threadId, content: text("kept until the pod goes") });