diff --git a/packages/contracts/src/threads.ts b/packages/contracts/src/threads.ts index 7055f1e2..aae8d9dc 100644 --- a/packages/contracts/src/threads.ts +++ b/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; diff --git a/packages/core/src/conversations/chats/chats.test.ts b/packages/core/src/conversations/chats/chats.test.ts index 14ea69e9..44b00e78 100644 --- a/packages/core/src/conversations/chats/chats.test.ts +++ b/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, diff --git a/packages/core/src/conversations/chats/chats.ts b/packages/core/src/conversations/chats/chats.ts index de23e7af..ba19c17f 100644 --- a/packages/core/src/conversations/chats/chats.ts +++ b/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; /** - * 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 diff --git a/packages/core/src/conversations/thread-files/thread-files.test.ts b/packages/core/src/conversations/thread-files/thread-files.test.ts index 2a6b54cf..381ca53c 100644 --- a/packages/core/src/conversations/thread-files/thread-files.test.ts +++ b/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( diff --git a/packages/core/src/conversations/thread-files/thread-files.ts b/packages/core/src/conversations/thread-files/thread-files.ts index 642274fe..86b64919 100644 --- a/packages/core/src/conversations/thread-files/thread-files.ts +++ b/packages/core/src/conversations/thread-files/thread-files.ts @@ -1,16 +1,18 @@ export * as ThreadFiles from "./thread-files.ts"; import { userText } from "@sugabots/errors"; -import { and, count, eq, isNotNull, lt, not, sql } from "drizzle-orm"; +import { and, count, eq, inArray, isNotNull, lt, not, sql } from "drizzle-orm"; import { Config, Context, Data, Duration, Effect, Layer, Option, Schema, Stream } from "effect"; import { BlobStore } from "../../blob-store/blob-store.ts"; -import { query, serviceOperations } from "../../database/database.ts"; +import { query, serviceOperations, transaction } from "../../database/database.ts"; import { threadFile } from "../../database/schema.ts"; import { Ids, isUuid } from "../../ids/ids.ts"; import type { UserFacing } from "../../user-message.ts"; +import type { AttachedFileRow } from "../threads/file-parts.ts"; /** - * The only reader and writer of `thread_file`: files in a thread, which its + * The only writer of `thread_file`, and its reader but for the files messages + * show, which `threads/file-parts.ts` reads: files in a thread, which its * agents read, and people too when a chat serves them. A file goes when its * thread does. * @@ -56,6 +58,18 @@ export interface Interface { threadId: string; fileId: string; }) => Effect.Effect; + /** + * Attaches the uploads `fileIds` to `messageId` in the order listed, and + * returns them in that order. Each must be one of `uploaderId`'s unsent + * uploads in the thread; when any is not, including one removed for + * waiting too long, fails with `UploadsNotFound` and attaches none. + */ + readonly attach: (input: { + threadId: string; + messageId: string; + uploaderId: string; + fileIds: readonly string[]; + }) => Effect.Effect; /** The files in the thread kept as tool calls' results: each call's file id, by the call's id. */ readonly resultFiles: (threadId: string) => Effect.Effect>; /** @@ -169,6 +183,45 @@ export const make = Effect.gen(function* () { return { id: file.id, name: file.name ?? undefined, ...stored, stream }; }), ), + attach: ({ threadId, messageId, uploaderId, fileIds }) => + operation( + "attach", + transaction( + Effect.gen(function* () { + const wanted = [...new Set(fileIds)]; + // The ids come from the sender, and Postgres rejects a uuid + // comparison against anything that is not one. + if (!wanted.every(isUuid)) return yield* new UploadsNotFound(); + const listedPosition = sql`case ${threadFile.id} ${sql.join( + wanted.map((id, position) => sql`when ${id}::uuid then ${position}::integer`), + sql` `, + )} end`; + const attached = yield* query((db) => + db + .update(threadFile) + .set({ messageId, position: listedPosition }) + .where( + and( + inArray(threadFile.id, wanted), + eq(threadFile.threadId, threadId), + eq(threadFile.uploaderId, uploaderId), + unsentUpload, + ), + ) + .returning({ + id: threadFile.id, + name: threadFile.name, + contentType: threadFile.contentType, + size: threadFile.size, + }), + ); + // Failing rolls back the ones that were found. + if (attached.length !== wanted.length) return yield* new UploadsNotFound(); + const listed = new Map(attached.map((file) => [file.id, file])); + return wanted.flatMap((id) => listed.get(id) ?? []); + }), + ), + ), resultFiles: (threadId) => operation( "resultFiles", @@ -270,6 +323,12 @@ export class FileNotFound } } +export class UploadsNotFound extends Data.TaggedError("UploadsNotFound") implements UserFacing { + get userMessage() { + return userText`A file to send is no longer there. Remove it and attach it again`; + } +} + export class TooManyUnsentUploads extends Data.TaggedError("TooManyUnsentUploads")<{ /** How many unsent uploads one person may have in a thread. */ diff --git a/packages/core/src/conversations/threads/file-parts.ts b/packages/core/src/conversations/threads/file-parts.ts new file mode 100644 index 00000000..9da98d8d --- /dev/null +++ b/packages/core/src/conversations/threads/file-parts.ts @@ -0,0 +1,60 @@ +import type { FilePart } from "@sugabots/contracts"; +import { asc, inArray } from "drizzle-orm"; +import { Effect } from "effect"; +import type { Executor } from "../../database/database.ts"; +import { threadFile } from "../../database/schema.ts"; + +/** + * Reading the files attached to messages back into the parts the API hands + * out. A message does not list its files: each attached `thread_file` row + * names its message. Writing them is `thread-files/thread-files.ts`. + */ + +/** The columns a file part is read from, for a relational query. */ +export const fileColumns = { id: true, name: true, contentType: true, size: true } as const; + +/** An attached file, read with {@link fileColumns}. */ +export interface AttachedFileRow { + id: string; + name: string | null; + contentType: string; + size: number; +} + +export function toFilePart(row: AttachedFileRow): FilePart { + return { + type: "file", + id: row.id, + // The table's checks keep messages to uploads, and every upload is named. + name: row.name ?? "", + contentType: row.contentType, + size: row.size, + }; +} + +/** The files attached to each of these messages, in the order they were sent, by message id. */ +export const loadFileParts = Effect.fn("FileParts.loadFileParts")(function* ( + db: Executor, + messageIds: readonly string[], +) { + const byMessage = new Map(); + if (messageIds.length === 0) { + return byMessage; + } + const rows = yield* db + .select({ + messageId: threadFile.messageId, + id: threadFile.id, + name: threadFile.name, + contentType: threadFile.contentType, + size: threadFile.size, + }) + .from(threadFile) + .where(inArray(threadFile.messageId, [...messageIds])) + .orderBy(asc(threadFile.position)); + for (const { messageId, ...row } of rows) { + if (messageId === null) continue; + byMessage.set(messageId, [...(byMessage.get(messageId) ?? []), toFilePart(row)]); + } + return byMessage; +}); diff --git a/packages/core/src/conversations/threads/message-text.ts b/packages/core/src/conversations/threads/message-text.ts index e39cc896..0c89f18b 100644 --- a/packages/core/src/conversations/threads/message-text.ts +++ b/packages/core/src/conversations/threads/message-text.ts @@ -1,5 +1,6 @@ import { type CollaborationPart, + type FilePart, type Message, type MessageReplyTo, messageAuthorName, @@ -31,6 +32,7 @@ export function messageTextWithPlacedParts(message: Message): string { .map((part) => { if (part.type === "text") return part.text; if (part.type === "collaboration") return describeCollaboration(part); + if (part.type === "file") return describeAttachment(part); return describeToolCall(part); }) .join("\n\n") @@ -69,6 +71,21 @@ export function describeToolCall(call: ToolCallPart, resultFile?: string): strin } } +/** + * A file attached to a message as one bracketed line: the agent is given what + * it is and how to read it, not its content, which may be large or not text. + */ +export function describeAttachment(file: FilePart): string { + return `[Attached ${file.name} (${file.contentType}, ${sizeOf(file.size)}) as file ${file.id}, which ${READ_THREAD_FILE_TOOL} reads]`; +} + +/** A file's size as people write it, such as 2.1 MB. */ +function sizeOf(bytes: number): string { + if (bytes < 1024) return `${bytes} bytes`; + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + return `${(bytes / (1024 * 1024)).toFixed(1)} MB`; +} + const FoundToolNames = Schema.Struct({ tools: Schema.Array(Schema.Struct({ tool: Schema.String })), }); diff --git a/packages/core/src/conversations/threads/participants.ts b/packages/core/src/conversations/threads/participants.ts index 950566ea..3b3ec64c 100644 --- a/packages/core/src/conversations/threads/participants.ts +++ b/packages/core/src/conversations/threads/participants.ts @@ -14,6 +14,7 @@ import { Schema } from "effect"; import type * as schema from "../../database/schema.ts"; import { agent, message, user } from "../../database/schema.ts"; import { toCollaborationPart } from "./collaboration-parts.ts"; +import { type AttachedFileRow, fileColumns, toFilePart } from "./file-parts.ts"; import type { PlacedPartsOf } from "./placed-parts.ts"; import { toToolCallPart } from "./tool-calls.ts"; @@ -101,6 +102,8 @@ export function toMessage( : placed.toolCalls?.find((candidate) => candidate.id === part.toolCallId); return made ? [made] : []; }); + // Files are attached to the message rather than placed in its text, so they follow it. + parts.push(...(placed.files ?? [])); return { id: row.id, threadId: row.threadId, @@ -199,6 +202,7 @@ export const messageRelations = { with: { decidedBy: { columns: { name: true } } }, }, collaborations: { with: { collaborator: { columns: { name: true } } } }, + files: { columns: fileColumns, orderBy: { position: "asc" } }, routineExecution: { columns: { id: true, threadId: true, routineName: true } }, replyTo: quotedMessageColumns, reactions: reactionColumns, @@ -214,6 +218,7 @@ export interface MessageWithRelations extends schema.MessageRow { authorAgent: AgentIdentity | null; toolCalls: Array; collaborations: Array; + files: AttachedFileRow[]; routineExecution: Pick | null; replyTo: QuotedMessage | null; reactions: ReactionRow[]; @@ -233,6 +238,7 @@ export function messageFromRelations( collaborations: stored.collaborations.map((made) => toCollaborationPart(made, made.collaborator.name), ), + files: stored.files.map(toFilePart), }, error, ); diff --git a/packages/core/src/conversations/threads/placed-parts.ts b/packages/core/src/conversations/threads/placed-parts.ts index 3ea9407f..7d0bbcfc 100644 --- a/packages/core/src/conversations/threads/placed-parts.ts +++ b/packages/core/src/conversations/threads/placed-parts.ts @@ -1,16 +1,19 @@ -import type { CollaborationPart, ToolCallPart } from "@sugabots/contracts"; +import type { CollaborationPart, FilePart, ToolCallPart } from "@sugabots/contracts"; import { Effect } from "effect"; import type { Executor } from "../../database/database.ts"; import { loadCollaborationParts } from "./collaboration-parts.ts"; +import { loadFileParts } from "./file-parts.ts"; import { loadToolCallParts } from "./tool-calls.ts"; /** * The parts a message keeps by reference, loaded from their own tables so - * `toMessage` can put them back where the reply's text placed them. + * `toMessage` can put them back where the reply's text placed them, and the + * files attached to it, which follow its text. */ export interface PlacedPartsOf { collaborations?: readonly CollaborationPart[]; toolCalls?: readonly ToolCallPart[]; + files?: readonly FilePart[]; } /** The referenced parts of each of these messages, as a lookup by message id. */ @@ -20,8 +23,10 @@ export const loadPlacedParts = Effect.fn("PlacedParts.loadPlacedParts")(function ) { const collaborations = yield* loadCollaborationParts(db, messageIds); const toolCalls = yield* loadToolCallParts(db, messageIds); + const files = yield* loadFileParts(db, messageIds); return (messageId: string): PlacedPartsOf => ({ collaborations: collaborations.get(messageId), toolCalls: toolCalls.get(messageId), + files: files.get(messageId), }); }); diff --git a/packages/core/src/conversations/threads/repository.ts b/packages/core/src/conversations/threads/repository.ts index a98a9b35..59a10cfd 100644 --- a/packages/core/src/conversations/threads/repository.ts +++ b/packages/core/src/conversations/threads/repository.ts @@ -25,6 +25,8 @@ import { import type { UserFacing } from "../../user-message.ts"; import { ConversationEvents } from "../conversation-events.ts"; import { ConversationEvent } from "../events.ts"; +import { ThreadFiles } from "../thread-files/thread-files.ts"; +import { loadFileParts, toFilePart } from "./file-parts.ts"; import { authorRow, type PersonIdentity, @@ -60,20 +62,29 @@ export interface Interface { initiatorUserId: string | null; }) => Effect.Effect; /** - * Posts a person's message, making them a participant, and announces it. - * Posting a message again with the same `id` returns the one already - * posted, unchanged and unannounced; an `id` already used by a different - * message fails, and so does quoting a message that is not in the thread. + * Posts a person's message with the files they attach, making them a + * participant, and announces it. Posting a message again with the same + * `id` returns the one already posted, unchanged and unannounced; an `id` + * already used by a different message, or the same text with other files, + * fails, and so does quoting a message that is not in the thread. Fails + * with `UploadsNotFound`, posting nothing, when an attachment is not one + * of the author's unsent uploads in the thread. */ readonly post: (input: { /** The id the person's client chose, which makes sending it again safe to detect. */ id: string; threadId: string; author: PersonIdentity; + /** May be empty when the message carries files. */ content: string; + /** Ids of the author's uploads to attach. */ + attachments: readonly string[]; /** The id of the message in the thread that this one quotes and answers. */ replyTo?: string; - }) => Effect.Effect; + }) => Effect.Effect< + Posted | AlreadyPosted, + MessageIdConflict | ReplyTargetNotFound | ThreadFiles.UploadsNotFound + >; /** Brings agents into the thread, announcing the ones who were not there yet. */ readonly addAgents: (threadId: string, agentIds: readonly string[]) => Effect.Effect; /** @@ -164,6 +175,7 @@ export class Service extends Context.Service()( export const make = Effect.gen(function* () { const operation = yield* serviceOperations("ThreadRepository"); const { emit } = yield* ConversationEvents.Service; + const files = yield* ThreadFiles.Service; /** What the message has now, announced when `changed`. */ const reactionsAfter = (reaction: Reaction, changed: boolean) => @@ -262,18 +274,24 @@ export const make = Effect.gen(function* () { ); const author = personAuthor(input.author); if (existing) { + const attached = + (yield* query((db) => loadFileParts(db, [existing.id]))).get(existing.id) ?? []; if ( existing.threadId !== input.threadId || existing.authorUserId !== input.author.id || existing.content !== input.content || - (existing.replyToMessageId ?? undefined) !== input.replyTo + (existing.replyToMessageId ?? undefined) !== input.replyTo || + !sameIds( + attached.map(({ id }) => id), + input.attachments, + ) ) { return yield* new MessageIdConflict(); } const replyTo = yield* quotedIn(existing.threadId, existing.replyToMessageId); return { _tag: "AlreadyPosted", - message: withReplyTo(toMessage(existing, author), replyTo), + message: withReplyTo(toMessage(existing, author, { files: attached }), replyTo), } as const; } const replyTo = yield* quotedIn(input.threadId, input.replyTo); @@ -287,7 +305,7 @@ export const make = Effect.gen(function* () { authorUserId: input.author.id, kind: "text", status: "complete", - parts: [{ type: "text", text: input.content }], + parts: input.content === "" ? [] : [{ type: "text", text: input.content }], content: input.content, replyToMessageId: input.replyTo, }) @@ -307,7 +325,16 @@ export const make = Effect.gen(function* () { .values({ threadId: input.threadId, userId: input.author.id }) .onConflictDoNothing(), ); - const posted = withReplyTo(toMessage(created, author), replyTo); + const attached = + input.attachments.length === 0 + ? [] + : (yield* files.attach({ + threadId: input.threadId, + messageId: created.id, + uploaderId: input.author.id, + fileIds: input.attachments, + })).map(toFilePart); + const posted = withReplyTo(toMessage(created, author, { files: attached }), replyTo); yield* emit([ ConversationEvent.MessagePosted({ threadId: input.threadId, @@ -689,7 +716,8 @@ export const make = Effect.gen(function* () { }); }); -export const layer = Layer.effect(Service, make); +/** Posting a message ends its files' wait to be sent, so this provides `ThreadFiles`. */ +export const layer = Layer.effect(Service, make).pipe(Layer.provide(ThreadFiles.layer)); /** A message posted now, and announced. */ export interface Posted { @@ -697,7 +725,7 @@ export interface Posted { readonly message: Message; } -/** A message posted before with the same id and content, returned as it was. */ +/** A message posted before with the same id, content and files, returned as it was. */ export interface AlreadyPosted { readonly _tag: "AlreadyPosted"; readonly message: Message; @@ -747,6 +775,12 @@ export class ReplyTargetNotFound } } +/** Whether two lists hold the same ids, in any order. */ +function sameIds(left: readonly string[], right: readonly string[]): boolean { + const wanted = new Set(right); + return left.length === wanted.size && left.every((id) => wanted.has(id)); +} + export class MessageIdConflict extends Data.TaggedError("MessageIdConflict") implements UserFacing { override get message() { return "A message id was reused for a different message"; diff --git a/packages/core/src/conversations/turns/context.test.ts b/packages/core/src/conversations/turns/context.test.ts index cf334039..b4a5584e 100644 --- a/packages/core/src/conversations/turns/context.test.ts +++ b/packages/core/src/conversations/turns/context.test.ts @@ -348,6 +348,35 @@ describe("modelPrompt", () => { ]); }); + it("lists the files a person attached after what they wrote, even with nothing written", () => { + const input = context(); + const asked = input.messages[0]; + if (!asked) { + throw new Error("Context fixture has no person's message"); + } + const report = { + type: "file", + id: "0199a3a0-0000-7000-8000-0000000000f2", + name: "report.pdf", + contentType: "application/pdf", + size: 2_200_000, + } as const; + const line = `[Attached report.pdf (application/pdf, 2.1 MB) as file ${report.id}, which read_thread_file reads]`; + + input.messages[0] = { ...asked, parts: [...asked.parts, report] }; + expect(modelPrompt(input, environment()).messages[0]).toEqual({ + role: "user", + content: `Sam (@sam): Check the release +${line}`, + }); + + input.messages[0] = { ...asked, content: "", parts: [report] }; + expect(modelPrompt(input, environment()).messages[0]).toEqual({ + role: "user", + content: `Sam (@sam): ${line}`, + }); + }); + it("excludes incomplete assistant output", () => { const input = context(); const assistantMessage = input.messages[1]; diff --git a/packages/core/src/conversations/turns/context.ts b/packages/core/src/conversations/turns/context.ts index 67dce24d..0ee95b42 100644 --- a/packages/core/src/conversations/turns/context.ts +++ b/packages/core/src/conversations/turns/context.ts @@ -1,6 +1,7 @@ import { type Message, messageAuthorName, quotedPreview } from "@sugabots/contracts"; import type { Models } from "../../providers/models/models.ts"; import { + describeAttachment, describeCollaboration, describeReplyTo, describeToolCall, @@ -167,9 +168,12 @@ function promptMessagesFor( currentAgentId: string, resultFiles: ReadonlyMap, ): Models.PromptMessage[] { - const authored = authoredMessage(message, currentAgentId); + const attachments = message.parts.flatMap((part) => + part.type === "file" ? [describeAttachment(part)] : [], + ); + const authored = authoredMessage(message, currentAgentId, attachments); const records = message.parts.flatMap((part): string[] => { - if (part.type === "text") return []; + if (part.type === "text" || part.type === "file") return []; if (part.type === "collaboration") return [describeCollaboration(part)]; return [describeToolCall(part, resultFiles.get(part.id))]; }); @@ -185,27 +189,31 @@ function promptMessagesFor( return authored ? [authored, eventLog] : [eventLog]; } +/** + * What the message's author wrote, with a line for each file they attached, + * which is all a message that carries only files says. + */ function authoredMessage( message: Message, currentAgentId: string, + attachments: readonly string[], ): Models.PromptMessage | undefined { - if (!message.content) return undefined; + const said = [message.content, ...attachments].filter((line) => line !== "").join("\n"); + if (!said) return undefined; if (message.author.kind === "agent" && message.author.id === currentAgentId) { - return { role: "assistant", content: message.content }; + return { role: "assistant", content: said }; } if (message.author.kind === "routine_trigger") { return { role: "user", - content: `[Routine trigger: ${message.author.routineName}]\n${message.content}`, + content: `[Routine trigger: ${message.author.routineName}]\n${said}`, }; } const agentMarker = message.author.kind === "agent" ? " (agent)" : ""; - const said = message.replyTo - ? `${describeReplyTo(message.replyTo)}\n${message.content}` - : message.content; + const replying = message.replyTo ? `${describeReplyTo(message.replyTo)}\n${said}` : said; return { role: "user", - content: `${message.author.name} (@${message.author.handle})${agentMarker}: ${said}`, + content: `${message.author.name} (@${message.author.handle})${agentMarker}: ${replying}`, }; } diff --git a/packages/core/src/database/relations.ts b/packages/core/src/database/relations.ts index 9238e1e6..246cf87d 100644 --- a/packages/core/src/database/relations.ts +++ b/packages/core/src/database/relations.ts @@ -48,6 +48,7 @@ export const relations = defineRelations(schema, (r) => ({ authorAgent: r.one.agent({ from: r.message.authorAgentId, to: r.agent.id }), turn: r.one.turn({ from: r.message.turnId, to: r.turn.id }), toolCalls: r.many.toolCall({ from: r.message.id, to: r.toolCall.messageId }), + files: r.many.threadFile({ from: r.message.id, to: r.threadFile.messageId }), collaborations: r.many.collaboration({ from: r.message.id, to: r.collaboration.parentMessageId, diff --git a/packages/server/src/routes/chats/routes.ts b/packages/server/src/routes/chats/routes.ts index 9b7477fe..2e1574e2 100644 --- a/packages/server/src/routes/chats/routes.ts +++ b/packages/server/src/routes/chats/routes.ts @@ -52,6 +52,7 @@ export const chatRoutes = HttpApiBuilder.group(ServerApi, "chats", (handlers) => chatId: params.chatId, messageId: payload.id, content: payload.message, + attachments: payload.attachments, peopleOnly: payload.peopleOnly, replyTo: payload.replyTo, }) @@ -106,6 +107,7 @@ const chatErrors = { ReplyTargetNotFound: BadRequest, ChatAgentHasNoModel: Conflict, InvalidChatCursor: BadRequest, + UploadsNotFound: BadRequest, }; const fileErrors = { diff --git a/packages/web/src/screens/ThreadConversation.tsx b/packages/web/src/screens/ThreadConversation.tsx index dc370e38..f4330ab3 100644 --- a/packages/web/src/screens/ThreadConversation.tsx +++ b/packages/web/src/screens/ThreadConversation.tsx @@ -332,6 +332,8 @@ function segmentsOf(message: Message): Segment[] { if (wasPutToSomeone(part)) segments.push({ type: "tool_call", key: part.id, toolCall: part }); return; } + // Attached files are not part of the message's text. + if (part.type === "file") return; const followed = index < message.parts.length - 1 && part.text.trim() !== ""; const settled = finished || followed; if (settled && !isNarration(message.parts, index)) {