From e9266a64b1b5fbfb76f03b2f09cce0d72940881e Mon Sep 17 00:00:00 2001 From: David Moore Date: Fri, 16 Oct 2026 14:39:52 +1100 Subject: [PATCH 1/4] feat(core): let a bot share a file from its sandbox in its reply --- packages/contracts/src/events.ts | 12 ++ packages/core/src/conversations/events.ts | 10 ++ .../core/src/conversations/thread-feed.ts | 6 + .../thread-files/thread-files.test.ts | 45 ++++++++ .../thread-files/thread-files.ts | 104 ++++++++++++++++-- .../src/conversations/threads/file-parts.ts | 4 +- .../src/conversations/tools/sandbox.test.ts | 59 +++++++++- .../core/src/conversations/tools/sandbox.ts | 20 +++- .../tools/sandbox/attachments.ts | 10 +- .../src/conversations/tools/sandbox/tools.ts | 104 +++++++++++++++++- .../src/conversations/turns/context.test.ts | 25 +++++ .../core/src/conversations/turns/context.ts | 20 +++- .../conversations/turns/repository.test.ts | 30 ++++- .../src/conversations/turns/repository.ts | 85 +++++++++++++- .../core/src/conversations/turns/tools.ts | 10 +- .../conversations/turns/turn.steps.test.ts | 39 ++++++- .../src/conversations/turns/turn.steps.ts | 12 +- packages/web/src/lib/thread-events.ts | 20 +++- packages/web/src/screens/MessageFiles.tsx | 6 +- .../web/src/screens/ThreadConversation.tsx | 4 +- packages/web/src/screens/agent-chat.test.tsx | 57 ++++++++++ 21 files changed, 637 insertions(+), 45 deletions(-) diff --git a/packages/contracts/src/events.ts b/packages/contracts/src/events.ts index 8b5c9dd6..4c071d64 100644 --- a/packages/contracts/src/events.ts +++ b/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"]), diff --git a/packages/core/src/conversations/events.ts b/packages/core/src/conversations/events.ts index fa069d70..65a9635e 100644 --- a/packages/core/src/conversations/events.ts +++ b/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[] }; /** diff --git a/packages/core/src/conversations/thread-feed.ts b/packages/core/src/conversations/thread-feed.ts index 61f288c7..f5a70515 100644 --- a/packages/core/src/conversations/thread-feed.ts +++ b/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 })), ], 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 5aa06e7d..c081b246 100644 --- a/packages/core/src/conversations/thread-files/thread-files.test.ts +++ b/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"; @@ -369,6 +370,50 @@ 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) => + files.attachToReply({ + threadId, + messageId: reply.id, + name, + contentType: "text/csv", + content: new TextEncoder().encode(name), + }); + + 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") }); diff --git a/packages/core/src/conversations/thread-files/thread-files.ts b/packages/core/src/conversations/thread-files/thread-files.ts index 408decb1..36929c89 100644 --- a/packages/core/src/conversations/thread-files/thread-files.ts +++ b/packages/core/src/conversations/thread-files/thread-files.ts @@ -1,5 +1,6 @@ export * as ThreadFiles from "./thread-files.ts"; +import { type FilePart, MAX_ATTACHMENTS_PER_MESSAGE } from "@sugabots/contracts"; import { type DomainError, userText } from "@sugabots/errors"; import { and, count, eq, inArray, isNotNull, lt, not, sql } from "drizzle-orm"; import { @@ -15,11 +16,11 @@ import { Stream, } from "effect"; import { BlobStore } from "../../blob-store/blob-store.ts"; -import { query, serviceOperations, transaction } from "../../database/database.ts"; -import { threadFile } from "../../database/schema.ts"; +import { afterCommit, query, serviceOperations, transaction } from "../../database/database.ts"; +import { message, 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"; +import { type AttachedFileRow, loadFileParts } from "../threads/file-parts.ts"; /** * The only writer of `thread_file`, and its reader but for the files messages @@ -83,6 +84,28 @@ export interface Interface { }) => 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>; + /** + * Keeps `content` as a file an agent attached to its reply `messageId`, + * after the reply's other files. Fails with `TooManyReplyFiles` once the + * reply carries as many as a message can. + */ + readonly attachToReply: (file: { + threadId: string; + messageId: string; + name: string; + contentType: string; + content: Uint8Array; + }) => Effect.Effect; + /** The files an agent attached to its reply `messageId`, in the order attached. */ + readonly replyAttachments: (messageId: string) => Effect.Effect; + /** + * Removes the files an agent attached to its reply `messageId`, as when the + * reply starts again. Their blobs go once the enclosing transaction commits. + */ + readonly removeReplyAttachments: (reply: { + threadId: string; + messageId: string; + }) => Effect.Effect; /** * Removes `fileId` when it is one of `uploaderId`'s unsent uploads in the * thread, such as one taken off a message before it was sent. Anything @@ -145,13 +168,13 @@ export const make = Effect.gen(function* () { return { id, contentType, size }; }); - /** Removes an unsent upload's blob, once its row is gone; one that cannot be removed goes with the thread's folder. */ - const removeUploadBlob = (threadId: string, fileId: string) => + /** Removes a file's blob, once its row is gone; one that cannot be removed goes with the thread's folder. */ + const removeBlob = (threadId: string, fileId: string) => blobs .delete(fileKey(threadId, fileId)) .pipe( Effect.catchCause((cause) => - Effect.logWarning("Removing an unsent upload's blob failed", cause), + Effect.logWarning("Removing a thread file's blob failed", cause), ), ); @@ -268,6 +291,62 @@ export const make = Effect.gen(function* () { ); }), ), + attachToReply: ({ threadId, messageId, name, contentType, content }) => + operation( + "attachToReply", + transaction( + Effect.gen(function* () { + // Locked, so files attached at once each take the next place, and the count holds. + yield* query((db) => + db + .select({ id: message.id }) + .from(message) + .where(eq(message.id, messageId)) + .for("update"), + ); + const [attached] = yield* query((db) => + db + .select({ files: count() }) + .from(threadFile) + .where( + and(eq(threadFile.messageId, messageId), eq(threadFile.kind, "agent_attachment")), + ), + ); + const position = attached?.files ?? 0; + if (position >= MAX_ATTACHMENTS_PER_MESSAGE) return yield* new TooManyReplyFiles(); + const kept = yield* keep( + threadId, + { stream: Stream.succeed(content), contentType }, + { kind: "agent_attachment", name, messageId, position }, + ); + return { type: "file", name, ...kept } as const; + }), + ), + ), + replyAttachments: (messageId) => + operation( + "replyAttachments", + query((db) => loadFileParts(db, [messageId])).pipe( + Effect.map((byMessage) => byMessage.get(messageId) ?? []), + ), + ), + removeReplyAttachments: ({ threadId, messageId }) => + operation( + "removeReplyAttachments", + Effect.gen(function* () { + const removed = yield* query((db) => + db + .delete(threadFile) + .where( + and(eq(threadFile.messageId, messageId), eq(threadFile.kind, "agent_attachment")), + ) + .returning({ id: threadFile.id }), + ); + yield* afterCommit( + Effect.forEach(removed, ({ id }) => removeBlob(threadId, id), { discard: true }), + ); + }), + ), removeUnsentUpload: ({ threadId, uploaderId, fileId }) => operation( "removeUnsentUpload", @@ -288,7 +367,7 @@ export const make = Effect.gen(function* () { ) .returning({ id: threadFile.id }), ); - if (removed.length > 0) yield* removeUploadBlob(threadId, fileId); + if (removed.length > 0) yield* removeBlob(threadId, fileId); }), ), removeUnsentUploads: (unsentFor) => @@ -310,7 +389,7 @@ export const make = Effect.gen(function* () { ) .returning({ id: threadFile.id, threadId: threadFile.threadId }), ); - yield* Effect.forEach(removed, ({ id, threadId }) => removeUploadBlob(threadId, id), { + yield* Effect.forEach(removed, ({ id, threadId }) => removeBlob(threadId, id), { discard: true, }); return removed.length; @@ -355,7 +434,8 @@ const unsentUpload = sql`(${threadFile.kind} = 'upload' and ${threadFile.message /** Which kind of file a row is, with the columns that kind sets. */ type FileSource = | { kind: "tool_result"; toolCallId: string | undefined } - | { kind: "upload"; uploaderId: string; name: string }; + | { kind: "upload"; uploaderId: string; name: string } + | { kind: "agent_attachment"; name: string; messageId: string; position: number }; export class FileNotFound extends Data.TaggedError("ThreadFileNotFound")<{ @@ -412,6 +492,12 @@ export class FileTooLarge } } +export class TooManyReplyFiles extends Data.TaggedError("TooManyReplyFiles") implements UserFacing { + get userMessage() { + return userText`A reply can carry at most ${MAX_ATTACHMENTS_PER_MESSAGE} files`; + } +} + /** `bytes` as people read a file's size, such as 30 MB or 512 KB. */ function sizeForPeople(bytes: number): { amount: number; unit: "MB" | "KB" } { const mebibytes = bytes / (1024 * 1024); diff --git a/packages/core/src/conversations/threads/file-parts.ts b/packages/core/src/conversations/threads/file-parts.ts index 9da98d8d..d723277f 100644 --- a/packages/core/src/conversations/threads/file-parts.ts +++ b/packages/core/src/conversations/threads/file-parts.ts @@ -25,14 +25,14 @@ 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. + // The table's checks name every file on a message: an upload or an agent's attachment. 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. */ +/** The files on each of these messages, in their order on it, by message id. */ export const loadFileParts = Effect.fn("FileParts.loadFileParts")(function* ( db: Executor, messageIds: readonly string[], diff --git a/packages/core/src/conversations/tools/sandbox.test.ts b/packages/core/src/conversations/tools/sandbox.test.ts index 0e85ac88..e3d242b4 100644 --- a/packages/core/src/conversations/tools/sandbox.test.ts +++ b/packages/core/src/conversations/tools/sandbox.test.ts @@ -105,7 +105,12 @@ describe.skipIf(!configured)("sandbox tools, against Postgres and OpenSandbox", await closeDatabase(); }, SLOW); - const toolsFor = (turnId: string, threadId: string, agentId: string) => + const toolsFor = ( + turnId: string, + threadId: string, + agentId: string, + attach?: SandboxTools.Turn["attach"], + ) => runOnPostgres( Scope.provide(scope)( sandboxTools.forTurn({ @@ -115,6 +120,7 @@ describe.skipIf(!configured)("sandbox tools, against Postgres and OpenSandbox", threadId, agentId, acceptsImages: false, + ...(attach && { attach }), }), ), ).then((offered) => offered.tools); @@ -257,6 +263,57 @@ describe.skipIf(!configured)("sandbox tools, against Postgres and OpenSandbox", SLOW, ); + it( + "attaches a file from the sandbox to the reply, read as it is, and refuses one too large", + async () => { + const attached: { name: string; contentType: string; content: Uint8Array }[] = []; + const tools = await toolsFor("turn-12", "thread-h", "agent-7", (file) => { + attached.push(file); + return Effect.succeed({ + type: "file" as const, + id: `file-${attached.length}`, + name: file.name, + contentType: file.contentType, + size: file.content.byteLength, + }); + }); + // A PNG's signature, then bytes that aren't text. + await run( + tools, + String.raw`mkdir -p out && printf '\x89PNG\r\n\x1a\n\x00\xff' > out/chart.png`, + ); + await run(tools, String.raw`printf 'region,total\nnorth,4\n' > report.csv`); + await run(tools, "truncate -s 11M big.bin"); + + expect(await call(tools, "share_file", { path: "out/chart.png" })).toMatchObject({ + shared: "chart.png", + file: "file-1", + }); + expect( + await call(tools, "share_file", { path: "report.csv", name: "../Q3 totals.csv" }), + ).toMatchObject({ + shared: "_Q3 totals.csv", + }); + expect(attached.map(({ name, contentType }) => ({ name, contentType }))).toEqual([ + { name: "chart.png", contentType: "image/png" }, + { name: "_Q3 totals.csv", contentType: "text/csv; charset=utf-8" }, + ]); + expect([...(attached[0]?.content ?? [])]).toEqual([ + 0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0xff, + ]); + + expect(await call(tools, "share_file", { path: "big.bin" })).toMatchObject({ + status: "failed", + error: expect.stringContaining("files up to 10 MB"), + }); + expect(await call(tools, "share_file", { path: "nowhere.png" })).toMatchObject({ + status: "failed", + }); + expect(attached).toHaveLength(2); + }, + SLOW, + ); + it( "keeps an agent's home across its threads", async () => { diff --git a/packages/core/src/conversations/tools/sandbox.ts b/packages/core/src/conversations/tools/sandbox.ts index 019f3195..f6e34df2 100644 --- a/packages/core/src/conversations/tools/sandbox.ts +++ b/packages/core/src/conversations/tools/sandbox.ts @@ -13,6 +13,7 @@ import { ThreadFiles } from "../thread-files/thread-files.ts"; import { type BrowserSession, browserSession, browserTools } from "./browser/browser.ts"; import { REQUEST_NETWORK_ACCESS_TOOL, requestNetworkAccess } from "./network-access/tool.ts"; import { + type AttachToReply, allowedHostsNote, COPY_FILE_TO_SANDBOX_TOOL, copyFileToSandboxTool, @@ -24,14 +25,16 @@ import { RUN_COMMAND_TOOL, readFileTool, runCommandTool, + SHARE_FILE_TOOL, + shareFileTool, WRITE_FILE_TOOL, writeFileTool, } from "./sandbox/tools.ts"; /** * The tools that work in a pod's sandbox: `run_command`, `read_file`, - * `write_file` and `copy_file_to_sandbox`, and the `browser_` tools of a - * browser the agent drives there (see `browser/browser.ts`), and + * `write_file`, `copy_file_to_sandbox` and `share_file`, and the `browser_` + * tools of a browser the agent drives there (see `browser/browser.ts`), and * `request_network_access`, whose calls wait for a person, to reach a host * the sandbox may not. Offered to every turn, and usable while the agent uses * the sandbox and the workspace has an enabled sandbox provider, looked up on @@ -86,6 +89,7 @@ export const make = Effect.gen(function* () { browser, run, blocked, + attach, }: Behind) => ({ tools: { [RUN_COMMAND_TOOL]: runCommandTool(openSandbox, place), @@ -99,6 +103,7 @@ export const make = Effect.gen(function* () { ids, run, }), + ...(attach ? { [SHARE_FILE_TOOL]: shareFileTool(openSandbox, place, attach) } : {}), ...browserTools(browser, acceptsImages), }, requests: { @@ -106,7 +111,7 @@ export const make = Effect.gen(function* () { }, }); return Service.of({ - forTurn: ({ pod, turnId, threadId, agentId, acceptsImages, usesSandbox }) => + forTurn: ({ pod, turnId, threadId, agentId, acceptsImages, usesSandbox, attach }) => Effect.gen(function* (): Effect.fn.Return { const place = placeOf({ threadId, agentId }); const provider = usesSandbox ? yield* providers.enabled(pod.workspaceId) : undefined; @@ -117,6 +122,7 @@ export const make = Effect.gen(function* () { threadId, place, acceptsImages, + attach, openSandbox: NEVER_OPENED, browser: browserSession(NEVER_OPENED, place, { threadId, agentId }), run: NEVER_RUN, @@ -165,6 +171,7 @@ export const make = Effect.gen(function* () { threadId, place, acceptsImages, + attach, openSandbox, browser, run, @@ -199,6 +206,7 @@ interface Behind { readonly run: RunEffect; /** Hosts a request is refused for without asking anyone. */ readonly blocked: readonly string[]; + readonly attach: AttachToReply | undefined; } // Behind tools that aren't usable, whose calls are refused before they run. @@ -221,6 +229,12 @@ export interface Turn { readonly agentId: string; /** Whether the agent's model takes images, which results then may carry. */ readonly acceptsImages: boolean; + /** + * Attaches a file to the turn's reply, where people can be shown it: only + * in a chat's own thread, whose files the chat serves. Without it, the + * turn has no `share_file`. + */ + readonly attach?: AttachToReply; } /** No sandbox tools, for cases that offer none. */ diff --git a/packages/core/src/conversations/tools/sandbox/attachments.ts b/packages/core/src/conversations/tools/sandbox/attachments.ts index 00385ce2..c77dd4cc 100644 --- a/packages/core/src/conversations/tools/sandbox/attachments.ts +++ b/packages/core/src/conversations/tools/sandbox/attachments.ts @@ -39,12 +39,12 @@ export function sandboxPathOf(file: ThreadFile): string { } /** - * `name`, as a person's computer gave it, made safe as one file name: no - * folders, nothing hidden or read as a command's option, no characters that - * break a line or reorder what's shown, and short enough for Linux, keeping - * its extension. + * `name`, as a person's computer or an agent gave it, made safe as one file + * name: no folders, nothing hidden or read as a command's option, no + * characters that break a line or reorder what's shown, and short enough for + * Linux, keeping its extension. */ -function safeFileName(name: string | undefined): string { +export function safeFileName(name: string | undefined): string { const cleaned = (name ?? "") .replace(/[/\\\p{Cc}]/gu, "_") .replace(/\p{Cf}/gu, "") diff --git a/packages/core/src/conversations/tools/sandbox/tools.ts b/packages/core/src/conversations/tools/sandbox/tools.ts index 05806c74..f0c91483 100644 --- a/packages/core/src/conversations/tools/sandbox/tools.ts +++ b/packages/core/src/conversations/tools/sandbox/tools.ts @@ -1,22 +1,45 @@ import { posix } from "node:path"; +import { + attachmentNameSchema, + type FilePart, + MAX_ATTACHMENTS_PER_MESSAGE, +} from "@sugabots/contracts"; import { UserText, userText } from "@sugabots/errors"; import { tool } from "ai"; -import { Effect, Schema, Stream } from "effect"; +import { Effect, Schema, Semaphore, Stream } from "effect"; +import { contentType as mediaTypeForName } from "mime-types"; import type { BlobStore } from "../../../blob-store/blob-store.ts"; import type { RunEffect } from "../../../database/database.ts"; import type { Ids } from "../../../ids/ids.ts"; import type { PodSandboxes } from "../../../sandboxes/pod-sandboxes.ts"; import { Sandboxes } from "../../../sandboxes/sandboxes.ts"; +import type { UserFacing } from "../../../user-message.ts"; import type { ThreadFiles } from "../../thread-files/thread-files.ts"; import { MAX_IMAGE_BYTES, SHOWN_IMAGE_FORMATS, shownImageType, withImages } from "../images.ts"; import { REQUEST_NETWORK_ACCESS_TOOL } from "../network-access/tool.ts"; import { unreadableFileRefusal } from "../read-thread-file/tool.ts"; -import { type Copy, copyIntoSandbox } from "./attachments.ts"; +import { type Copy, copyIntoSandbox, safeFileName } from "./attachments.ts"; export const RUN_COMMAND_TOOL = "run_command"; export const READ_FILE_TOOL = "read_file"; export const WRITE_FILE_TOOL = "write_file"; export const COPY_FILE_TO_SANDBOX_TOOL = "copy_file_to_sandbox"; +export const SHARE_FILE_TOOL = "share_file"; + +/** + * The largest file an agent can share. It is read as the agents' user, + * through a command that prints it as base64, which holds the whole file in + * the server's memory on the way out. + */ +const MAX_SHARED_FILE_BYTES = 10 * 1024 * 1024; +const MAX_SHARED_FILE_MEGABYTES = MAX_SHARED_FILE_BYTES / (1024 * 1024); + +/** Attaches a file, read out of the sandbox, to the reply being written. */ +export type AttachToReply = (file: { + name: string; + contentType: string; + content: Uint8Array; +}) => Effect.Effect; const DEFAULT_TIMEOUT_SECONDS = 120; const MAX_TIMEOUT_SECONDS = 900; @@ -296,6 +319,83 @@ function refusedRead(failure: ThreadFiles.FileNotFound | BlobStore.Failure) { ); } +/** + * Shows people a file from the sandbox: it is attached to the agent's reply, + * where people see it as soon as it is. One file at a time, since each is + * held whole on its way out. + */ +export function shareFileTool(openSandbox: OpenSandbox, place: Place, attach: AttachToReply) { + const oneAtATime = Semaphore.makeUnsafe(1); + return tool({ + description: `Share a file from the pod's sandbox with the people in this chat, such as a chart, picture or spreadsheet you made: it is attached to your reply as soon as you call this, an image shown as a picture and anything else offered to download. Never paste a file's contents, base64 or a link to it into your reply instead. Files up to ${MAX_SHARED_FILE_MEGABYTES} MB, at most ${MAX_ATTACHMENTS_PER_MESSAGE} on one reply.`, + inputSchema: Schema.Struct({ + path: pathIn(place), + name: Schema.optional( + attachmentNameSchema.annotate({ + description: "The name people see, such as chart.png. The file's own name unless given.", + }), + ), + }).pipe(Schema.toStandardSchemaV1, Schema.toStandardJSONSchemaV1), + execute: ({ path, name }) => + inSandbox(openSandbox, (sandbox) => + oneAtATime.withPermits(1)(shareFrom(sandbox, absolute(place, path), name, attach)), + ), + }); +} + +/** What sharing a file tells the agent: the name people see it under, and its id. */ +interface Shared { + shared: string; + file: string; + note: UserText; +} + +/** Reads the file at `path` as the agents' user, then attaches it to the reply, refusing one too large. */ +function shareFrom( + sandbox: Sandboxes.Sandbox, + path: string, + name: string | undefined, + attach: AttachToReply, +): Effect.Effect { + const tooLarge = (sizeBytes: number): Failed => ({ + status: "failed", + error: userText`The file is ${Math.round(sizeBytes / 104_857.6) / 10} MB; files up to ${MAX_SHARED_FILE_MEGABYTES} MB can be shared.`, + }); + return Effect.gen(function* () { + // Its size alone first, so a file too large is never read. + const { sizeBytes } = yield* sandbox.readFile(path, 0); + if (sizeBytes > MAX_SHARED_FILE_BYTES) return tooLarge(sizeBytes); + const read = yield* sandbox.readFile(path, MAX_SHARED_FILE_BYTES); + if (read.sizeBytes > MAX_SHARED_FILE_BYTES) return tooLarge(read.sizeBytes); + const shownName = safeFileName(name ?? posix.basename(path)); + const contentType = + shownImageType(read.content) ?? (mediaTypeForName(shownName) || "application/octet-stream"); + return yield* attach({ name: shownName, contentType, content: read.content }).pipe( + Effect.map((file): Shared | Failed => ({ + shared: file.name, + file: file.id, + note: userText`People see it under your reply now; it needn't be described again.`, + })), + Effect.catchTags({ + BlobStoreUnavailable: (failure) => + notKept(failure, userText`The file couldn't be kept right now. Try again shortly.`), + BlobStoreRejected: (failure) => + notKept(failure, userText`The workspace's file storage refused the file.`), + }), + Effect.catch((failure) => + Effect.succeed({ status: "failed", error: failure.userMessage }), + ), + ); + }); +} + +/** Logs why the file store couldn't keep a shared file, and tells the agent `error`. */ +function notKept(failure: BlobStore.Failure, error: UserText) { + return Effect.logWarning("Keeping a file an agent shared failed", failure).pipe( + Effect.as({ status: "failed", error }), + ); +} + export function writeFileTool(openSandbox: OpenSandbox, place: Place) { return tool({ description: `Write a text file in the pod's sandbox, replacing it if it exists and making its directories.`, diff --git a/packages/core/src/conversations/turns/context.test.ts b/packages/core/src/conversations/turns/context.test.ts index e25eea0f..11bd0dac 100644 --- a/packages/core/src/conversations/turns/context.test.ts +++ b/packages/core/src/conversations/turns/context.test.ts @@ -399,6 +399,31 @@ ${line}`, }); }); + it("leaves an agent's own attached files to its tool calls, and describes another's", () => { + const input = context(); + const [, own, others] = input.messages; + if (!own || !others) throw new Error("Context fixture has no agents' messages"); + const chart = { + type: "file", + id: "0199a3a0-0000-7000-8000-0000000000f4", + name: "chart.png", + contentType: "image/png", + size: 4_096, + } as const; + input.agent = { ...input.agent, usesSandbox: true }; + input.messages[1] = { ...own, parts: [...own.parts, chart] }; + input.messages[2] = { ...others, parts: [...others.parts, chart] }; + + const history = modelPrompt(input, environment()).messages; + + expect(history[1]).toEqual({ role: "assistant", content: "I am checking it." }); + expect(history[2]).toMatchObject({ + role: "user", + content: `Reviewer (@reviewer) (agent): Migration is present. +[Attached chart.png (image/png, 4.0 KB) as file ${chart.id}, which read_thread_file reads, or copy_file_to_sandbox copies]`, + }); + }); + 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 ba0448b8..b7bf8558 100644 --- a/packages/core/src/conversations/turns/context.ts +++ b/packages/core/src/conversations/turns/context.ts @@ -165,9 +165,13 @@ interface Reader { usesSandbox: boolean; } +/** Whether the agent `agentId` wrote `message`. */ +function isOwnMessage(message: Message, agentId: string): boolean { + return message.author.kind === "agent" && message.author.id === agentId; +} + function whose(message: Message, currentAgentId: string): string { - const ownMessage = message.author.kind === "agent" && message.author.id === currentAgentId; - return ownMessage ? "your" : `${messageAuthorName(message.author)}'s`; + return isOwnMessage(message, currentAgentId) ? "your" : `${messageAuthorName(message.author)}'s`; } function promptMessagesFor( @@ -175,9 +179,13 @@ function promptMessagesFor( reader: Reader, resultFiles: ReadonlyMap, ): Models.PromptMessage[] { - const attachments = message.parts.flatMap((part) => - part.type === "file" ? [describeAttachment(part, reader)] : [], - ); + // An agent's own files are in its share_file calls: listed again in its + // own words, they would teach it to write such lines instead. + const attachments = isOwnMessage(message, reader.agentId) + ? [] + : message.parts.flatMap((part) => + part.type === "file" ? [describeAttachment(part, reader)] : [], + ); const authored = authoredMessage(message, reader.agentId, attachments); const records = message.parts.flatMap((part): string[] => { if (part.type === "text" || part.type === "file") return []; @@ -207,7 +215,7 @@ function authoredMessage( ): Models.PromptMessage | undefined { const said = [message.content, ...attachments].filter((line) => line !== "").join("\n"); if (!said) return undefined; - if (message.author.kind === "agent" && message.author.id === currentAgentId) { + if (isOwnMessage(message, currentAgentId)) { return { role: "assistant", content: said }; } if (message.author.kind === "routine_trigger") { diff --git a/packages/core/src/conversations/turns/repository.test.ts b/packages/core/src/conversations/turns/repository.test.ts index e10f676c..d0bf1e93 100644 --- a/packages/core/src/conversations/turns/repository.test.ts +++ b/packages/core/src/conversations/turns/repository.test.ts @@ -1,7 +1,8 @@ import { UserText, userText } from "@sugabots/errors"; import { eq } from "drizzle-orm"; -import { Context } from "effect"; +import { Context, Effect } from "effect"; import { afterAll, beforeEach, describe, expect, it } from "vitest"; +import { BlobStore } from "../../blob-store/blob-store.ts"; import type { CommittedEvent } from "../../database/events/outbox.ts"; import { agent, turn } from "../../database/schema.ts"; import { closeDatabase, onDatabase, onPostgres, runOnPostgres } from "../../database/testing.ts"; @@ -67,6 +68,21 @@ describe.skipIf(!process.env.DATABASE_URL)("turns, against Postgres", async () = }); it("runs a failed turn again, starting its reply over, while no change stands in the way", async () => { + const blobsInThread = () => + runOnPostgres( + Effect.flatMap(BlobStore.Service, (store) => + store.list(BlobStore.key("threads", prepared.context.thread.id)), + ), + ); + const shared = await turns.attachFile(replyTurnOf(prepared), { + name: "chart.png", + contentType: "image/png", + content: new Uint8Array([0x89, 0x50, 0x4e, 0x47]), + }); + expect(deliveredEvents()).toContainEqual( + expect.objectContaining({ type: "message.files_changed", files: [shared] }), + ); + expect(await blobsInThread()).toHaveLength(1); expect( await turns.fail(replyTurnOf(prepared), emptyReply, { userMessage: providerDown, @@ -82,6 +98,9 @@ describe.skipIf(!process.env.DATABASE_URL)("turns, against Postgres", async () = status: "streaming", }); expect(await storedTurn()).toMatchObject({ status: "running", runs: 2, error: null }); + // The reply starts over without the file its last run shared, blob and all. + expect(await turns.replyFiles(replyTurnOf(second))).toEqual([]); + expect(await blobsInThread()).toEqual([]); // Running again could act again, so the turn stops here for a person. expect( await turns.fail( @@ -109,6 +128,15 @@ describe.skipIf(!process.env.DATABASE_URL)("turns, against Postgres", async () = expect(deliveredEvents()).toContainEqual( expect.objectContaining({ type: "message.failed", willRetry: false, error: told }), ); + // A file shared by a call still running when the turn ended isn't shared. + await expect( + turns.attachFile(replyTurnOf(prepared), { + name: "late.csv", + contentType: "text/csv", + content: new Uint8Array([0x61]), + }), + ).rejects.toBeInstanceOf(TurnRepository.ReplyEnded); + expect(await turns.replyFiles(replyTurnOf(prepared))).toEqual([]); }); it("gives up on a turn that keeps stopping before it gets anywhere", async () => { diff --git a/packages/core/src/conversations/turns/repository.ts b/packages/core/src/conversations/turns/repository.ts index 033879b6..a9246f4d 100644 --- a/packages/core/src/conversations/turns/repository.ts +++ b/packages/core/src/conversations/turns/repository.ts @@ -1,10 +1,11 @@ export * as TurnRepository from "./repository.ts"; -import { type Message, messagePartsFor } from "@sugabots/contracts"; -import type { UserText } from "@sugabots/errors"; +import { type FilePart, type Message, messagePartsFor } from "@sugabots/contracts"; +import { type UserText, userText } from "@sugabots/errors"; import type { ModelMessage } from "ai"; import { and, eq, inArray, type SQL, sql } from "drizzle-orm"; -import { Context, DateTime, Effect, Layer, Schema } from "effect"; +import { Context, Data, DateTime, Effect, Layer, Schema } from "effect"; +import type { BlobStore } from "../../blob-store/blob-store.ts"; import { type Database, query, @@ -22,8 +23,10 @@ import { toolCall, turn, } from "../../database/schema.ts"; +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 { type ParticipantRow, toMessage } from "../threads/participants.ts"; import { type Ended, @@ -70,6 +73,21 @@ export interface Interface { ) => Effect.Effect; /** Persists the reply so far, so a crash loses at most a second of text. */ readonly saveReply: (turn: ReplyTurn, draft: ReplyDraft) => Effect.Effect; + /** + * Attaches `file` to the turn's reply, after the files already on it, and + * announces what the reply carries now. Fails with `ReplyEnded` once the + * turn is no longer running, and `TooManyReplyFiles` once the reply + * carries as many as a message can. + */ + readonly attachFile: ( + turn: ReplyTurn, + file: ReplyFile, + ) => Effect.Effect< + FilePart, + ReplyEnded | ThreadFiles.TooManyReplyFiles | ThreadFiles.FileTooLarge | BlobStore.Failure + >; + /** The files the turn's reply carries, in the order attached. */ + readonly replyFiles: (turn: ReplyTurn) => Effect.Effect; /** * Parks the turn with this model transcript until people decide the tool * approvals it asked for. `false` when the turn is no longer running, @@ -144,6 +162,7 @@ export const make = Effect.gen(function* () { const { emit } = yield* ConversationEvents.Service; const admission = yield* WorkAdmission.Service; const toolCalls = yield* ToolCallRepository.Service; + const files = yield* ThreadFiles.Service; /** Writes the state the lifecycle decided, plus what else the command records. */ const write = ( @@ -387,8 +406,13 @@ export const make = Effect.gen(function* () { if (!restarted) { return yield* Effect.die(new Error("A turn being run again has no reply message")); } - // The reply starts again, so the calls its last run made go with its parts. + // The reply starts again, so the calls its last run made go with its + // parts, and the files it attached with them. yield* toolCalls.forgetReply(restarted.id); + yield* files.removeReplyAttachments({ + threadId: restarted.threadId, + messageId: restarted.id, + }); const reply = toMessage(restarted, request.author); yield* announceStart(request, existing.id, reply); return { _tag: "Opened", turnId: existing.id, reply, checkpoint: undefined }; @@ -415,6 +439,39 @@ export const make = Effect.gen(function* () { ), saveReply: (reply, draft) => operation("saveReply", Effect.asVoid(writeReply(reply, draft))), + attachFile: (reply, file) => + operation( + "attachFile", + transaction( + Effect.gen(function* () { + // Locked, as ending the turn locks it: a call still running when the + // turn was cancelled finds it ended, and shares nothing. + const [running] = yield* query((db) => + db + .select({ id: turn.id }) + .from(turn) + .where(and(eq(turn.id, reply.turnId), eq(turn.status, "running"))) + .for("update"), + ); + if (!running) return yield* new ReplyEnded(); + const attached = yield* files.attachToReply({ + threadId: reply.threadId, + messageId: reply.messageId, + ...file, + }); + const carried = yield* files.replyAttachments(reply.messageId); + yield* emit([ + ConversationEvent.MessageFilesChanged({ + threadId: reply.threadId, + messageId: reply.messageId, + files: carried, + }), + ]); + return attached; + }), + ), + ), + replyFiles: (reply) => operation("replyFiles", files.replyAttachments(reply.messageId)), suspend: (reply, checkpoint, approvals) => operation( @@ -687,7 +744,25 @@ export const make = Effect.gen(function* () { export const layerNoDeps = Layer.effect(Service, make); -export const layer = layerNoDeps.pipe(Layer.provide(ToolCallRepository.layer)); +/** Ending a turn ends its tool calls, and restarting a reply removes the files it attached. */ +export const layer = layerNoDeps.pipe(Layer.provide([ToolCallRepository.layer, ThreadFiles.layer])); + +/** A file an agent attaches to its reply: its name, media type and bytes. */ +export interface ReplyFile { + readonly name: string; + readonly contentType: string; + readonly content: Uint8Array; +} + +/** A file was shared to a reply after its turn had ended, as when it was cancelled. */ +export class ReplyEnded extends Data.TaggedError("ReplyEnded") implements UserFacing { + override get message() { + return "A file was shared to a reply whose turn had ended"; + } + get userMessage() { + return userText`The reply has ended, so the file wasn't shared`; + } +} /** Where an agent's turn is: its thread and workspace, and the reply it writes. */ export interface ReplyTurn { diff --git a/packages/core/src/conversations/turns/tools.ts b/packages/core/src/conversations/turns/tools.ts index cd01de5a..d7abd15c 100644 --- a/packages/core/src/conversations/turns/tools.ts +++ b/packages/core/src/conversations/turns/tools.ts @@ -23,7 +23,7 @@ import { ROUTINES_KEY, routineTools, } from "../tools/routines/tool.ts"; -import { READ_FILE_TOOL } from "../tools/sandbox/tools.ts"; +import { READ_FILE_TOOL, SHARE_FILE_TOOL } from "../tools/sandbox/tools.ts"; import type { SandboxTools } from "../tools/sandbox.ts"; import { SAVE_INSTRUCTIONS_TOOL, saveInstructionsTool } from "../tools/save-instructions/tool.ts"; import { searchHistoryTool } from "../tools/search-history/tool.ts"; @@ -106,6 +106,12 @@ const NO_SANDBOX = userText`This bot has no sandbox to work in.`; const readDocs = readDocsTool(docPages); +/** + * The sandbox tools after which a failed turn can run again: they change + * nothing in the sandbox, and the files a reply shared go when it starts again. + */ +const RERUNNABLE_SANDBOX_TOOLS: ReadonlySet = new Set([READ_FILE_TOOL, SHARE_FILE_TOOL]); + /** The tools sent with the turn's requests, and the catalog `tool_search` finds the rest in. */ export interface TurnTools { tools: ToolSet; @@ -136,7 +142,7 @@ export function toolsForTurn(prepared: PreparedTurn, deps: ToolDependencies): Tu // a turn that fails afterwards is not run again. for (const [key, tool] of Object.entries(deps.sandbox.tools)) { tools[key] = deps.sandbox.usable - ? recorded(key, tool, { ...recording, mutating: key !== READ_FILE_TOOL }) + ? recorded(key, tool, { ...recording, mutating: !RERUNNABLE_SANDBOX_TOOLS.has(key) }) : refused(key, tool, NO_SANDBOX, recording); } // What a request does once allowed changes something, so a turn that fails diff --git a/packages/core/src/conversations/turns/turn.steps.test.ts b/packages/core/src/conversations/turns/turn.steps.test.ts index 1a1f38a8..8f02f220 100644 --- a/packages/core/src/conversations/turns/turn.steps.test.ts +++ b/packages/core/src/conversations/turns/turn.steps.test.ts @@ -556,6 +556,35 @@ describe("runSegment", () => { expect(turns.abandon).not.toHaveBeenCalled(); }); + it("completes a reply that says nothing but carries a file the agent attached", async () => { + const { execution, turns } = fakes(); + vi.mocked(turns.replyFiles).mockReturnValue( + Effect.succeed([ + { + type: "file", + id: "0199a3a0-0000-7000-8000-0000000000f1", + name: "chart.png", + contentType: "image/png", + size: 4_096, + }, + ]), + ); + + await runWithServices( + segmentWith({ + execution, + turns, + model: scriptedModel(""), + events: eventBus(), + collaborations: collaborations(), + toolCalls: toolCalls(), + }), + ); + + expect(turns.complete).toHaveBeenCalledWith(replyTurn, reply(""), expect.anything()); + expect(turns.fail).not.toHaveBeenCalled(); + }); + it("finishes a turn that may not run without streaming", async () => { const { execution, turns } = fakes(); vi.mocked(execution.prepare).mockReturnValueOnce( @@ -674,6 +703,7 @@ function fakes() { cancel: vi.fn(() => Effect.void), isCancellationRequested: vi.fn(() => Effect.succeed(false)), abandon: vi.fn(() => Effect.undefined), + replyFiles: vi.fn(() => Effect.succeed([])), }; return { execution, turns }; } @@ -683,7 +713,14 @@ interface Given { execution: Pick; turns: Pick< TurnRepository.Interface, - "saveReply" | "suspend" | "complete" | "fail" | "cancel" | "isCancellationRequested" | "abandon" + | "saveReply" + | "suspend" + | "complete" + | "fail" + | "cancel" + | "isCancellationRequested" + | "abandon" + | "replyFiles" >; toolCalls: Pick; collaborations: Partial; diff --git a/packages/core/src/conversations/turns/turn.steps.ts b/packages/core/src/conversations/turns/turn.steps.ts index ee38dbb5..b295e970 100644 --- a/packages/core/src/conversations/turns/turn.steps.ts +++ b/packages/core/src/conversations/turns/turn.steps.ts @@ -435,6 +435,10 @@ const streamReply = ( threadId: prepared.context.thread.id, agentId: prepared.context.agent.id, acceptsImages: prepared.context.agent.acceptsImages, + // A chat serves the files in its own thread, so only there can people be shown one. + ...(prepared.context.thread.type === "chat" && { + attach: (file) => turns.attachFile(replyTurnOf(prepared), file), + }), }); // The connections' sessions live as long as the turn. const connections = yield* Effect.acquireRelease( @@ -568,7 +572,13 @@ const streamReply = ( const contextTokens = prepared.checkpoint?.contextTokens ?? finished.contextTokens; if (finished.approvalRequests.length === 0) { const { content, collaborations } = yield* Ref.get(reply); - if (content.trim() === "" && collaborations.length === 0) { + const answered = content.trim() !== "" || collaborations.length > 0; + // Files the agent shared answer as words do. Looked up only when nothing else answered. + const carriesFiles = Effect.map( + turns.replyFiles(replyTurnOf(prepared)), + (files) => files.length > 0, + ); + if (!answered && !(yield* carriesFiles)) { return yield* new ReplyWithoutAnswer(); } return { kind: "completed" as const, contextTokens }; diff --git a/packages/web/src/lib/thread-events.ts b/packages/web/src/lib/thread-events.ts index 8f96accf..5257661b 100644 --- a/packages/web/src/lib/thread-events.ts +++ b/packages/web/src/lib/thread-events.ts @@ -296,6 +296,10 @@ async function applyThreadEvent( const { toolCall } = update; applyMessage = (message) => withPlacedPart(message, toolCall); refreshArtifactsAfter(queries, toolCall); + } else if (update.type === "message.files_changed") { + messageId = update.messageId; + const { files } = update; + applyMessage = (message) => ({ ...message, parts: [...partsWithoutFiles(message), ...files] }); } else if (update.type === "message.failed") { messageId = update.messageId; const { error } = update; @@ -366,9 +370,19 @@ function withPlacedPart(message: Message, part: PlacedPart): Message { return { ...message, parts: rebuiltParts(message, message.content, updated) }; } -/** The parts drawn afresh from the text and what was placed in it. */ -function rebuiltParts(_message: Message, content: string, placed: PlacedPart[]): MessagePart[] { - return messagePartsFor(content, placed); +/** + * The parts drawn afresh from the text and what was placed in it, with the + * message's files after them, which are never placed in its text. + */ +function rebuiltParts(message: Message, content: string, placed: PlacedPart[]): MessagePart[] { + return [ + ...messagePartsFor(content, placed), + ...message.parts.filter((part) => part.type === "file"), + ]; +} + +function partsWithoutFiles(message: Message): MessagePart[] { + return message.parts.filter((part) => part.type !== "file"); } /** `details` with `read` in place of what it had of the same person, unless it had them further on. */ diff --git a/packages/web/src/screens/MessageFiles.tsx b/packages/web/src/screens/MessageFiles.tsx index 34282678..cf1f298c 100644 --- a/packages/web/src/screens/MessageFiles.tsx +++ b/packages/web/src/screens/MessageFiles.tsx @@ -13,9 +13,9 @@ import { MiddleTruncated } from "@/ui/middle-truncated.tsx"; export type FileUrlOf = (fileId: string) => string | undefined; /** - * The files attached to a sent message, under its words: its images as - * pictures, each opening full size in a new tab, then every other file as a - * chip to download. A file not served yet shows without a link. + * The files on a message, a person's or a bot's reply, under its words: its + * images as pictures, each opening full size in a new tab, then every other + * file as a chip to download. A file not served yet shows without a link. */ export function MessageFiles({ files, urlOf }: { files: readonly FilePart[]; urlOf: FileUrlOf }) { const images = files.filter((file) => lookOf(file) === "image"); diff --git a/packages/web/src/screens/ThreadConversation.tsx b/packages/web/src/screens/ThreadConversation.tsx index a2a735c7..b3b02948 100644 --- a/packages/web/src/screens/ThreadConversation.tsx +++ b/packages/web/src/screens/ThreadConversation.tsx @@ -456,7 +456,9 @@ function MessageRow({ .flatMap((segment) => (segment.type === "text" && segment.text ? [segment.text] : [])) .join("\n\n"); const showsSomething = - calls.length > 0 || segments.some((segment) => segment.type !== "text" || segment.text); + calls.length > 0 || + files.length > 0 || + segments.some((segment) => segment.type !== "text" || segment.text); // A reply with nothing settled yet is only its typing line, under the last message. if (message.status === "streaming" && !showsSomething) return null; const finished = message.status === "complete" && !sending; diff --git a/packages/web/src/screens/agent-chat.test.tsx b/packages/web/src/screens/agent-chat.test.tsx index ce2c3668..1a237a2a 100644 --- a/packages/web/src/screens/agent-chat.test.tsx +++ b/packages/web/src/screens/agent-chat.test.tsx @@ -651,6 +651,63 @@ describe("ongoing agent Chat", () => { await waitFor(() => expect(messages.scrollTop).toBe(1_700)); }); + it("shows a file the agent attaches while it still answers, and keeps it as the answer goes on", async () => { + const updates = controlledEventStream(); + client.events.thread.mockReturnValue(updates.stream); + mount(`/suga/pods/suga-team/agents/${linear.handle}`); + await screen.findByRole("log", { name: "Chat messages" }); + await waitFor(() => expect(client.events.thread).toHaveBeenCalledWith(chat.mainThreadId)); + const streaming = { + ...agentMessage, + id: "0199a3a0-0000-7000-8000-0000000000fc", + status: "streaming" as const, + content: "Here it is", + parts: [{ type: "text" as const, text: "Here it is" }], + createdAt: "2026-09-18T09:20:00.000Z", + }; + updates.emit( + streamEvent("message.created", { threadId: chat.mainThreadId, message: streaming }), + ); + await screen.findByRole("status", { name: `${linear.name} is typing` }); + + // Before the reply's words are settled enough to show, its file already is. + updates.emit( + streamEvent("message.files_changed", { + threadId: chat.mainThreadId, + messageId: streaming.id, + files: [ + { + type: "file", + id: "0199a3a0-0000-7000-8000-0000000000fd", + name: "q3-report.csv", + contentType: "text/csv", + size: 2_048, + }, + ], + }), + ); + expect(await screen.findByRole("link", { name: "Download q3-report.csv" })).toBeDefined(); + + updates.emit( + streamEvent("message.delta", { + threadId: chat.mainThreadId, + messageId: streaming.id, + offset: streaming.content.length, + text: ", with the totals.", + }), + ); + updates.emit( + streamEvent("message.completed", { + threadId: chat.mainThreadId, + messageId: streaming.id, + content: "Here it is, with the totals.", + status: "complete", + }), + ); + await screen.findByText("Here it is, with the totals."); + expect(screen.getByRole("link", { name: "Download q3-report.csv" })).toBeDefined(); + }); + it("shows the reply that finished while you were elsewhere", async () => { const writing: Message = { ...agentMessage, From 76a309920128df95a871e88f91e4cb27f3e97775 Mon Sep 17 00:00:00 2001 From: Tim Holm Date: Mon, 19 Oct 2026 15:02:40 +1100 Subject: [PATCH 2/4] chore(core): serve only message files from a chat's file endpoint --- .../src/conversations/chats/chat-files.ts | 7 +- .../thread-files/thread-files.test.ts | 21 ++++++ .../thread-files/thread-files.ts | 65 ++++++++++--------- 3 files changed, 61 insertions(+), 32 deletions(-) diff --git a/packages/core/src/conversations/chats/chat-files.ts b/packages/core/src/conversations/chats/chat-files.ts index e28b6c91..41923721 100644 --- a/packages/core/src/conversations/chats/chat-files.ts +++ b/packages/core/src/conversations/chats/chat-files.ts @@ -44,7 +44,10 @@ export interface Interface { chatId: string; fileId: string; }) => Effect.Effect; - /** 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 }), ), ), }); 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 c081b246..80b860e1 100644 --- a/packages/core/src/conversations/thread-files/thread-files.test.ts +++ b/packages/core/src/conversations/thread-files/thread-files.test.ts @@ -262,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(); diff --git a/packages/core/src/conversations/thread-files/thread-files.ts b/packages/core/src/conversations/thread-files/thread-files.ts index 36929c89..93b4c3b8 100644 --- a/packages/core/src/conversations/thread-files/thread-files.ts +++ b/packages/core/src/conversations/thread-files/thread-files.ts @@ -2,7 +2,7 @@ export * as ThreadFiles from "./thread-files.ts"; import { type FilePart, MAX_ATTACHMENTS_PER_MESSAGE } from "@sugabots/contracts"; import { type DomainError, userText } from "@sugabots/errors"; -import { and, count, eq, inArray, isNotNull, lt, not, sql } from "drizzle-orm"; +import { and, count, eq, inArray, isNotNull, lt, type SQLWrapper, sql } from "drizzle-orm"; import { Config, Context, @@ -70,6 +70,14 @@ export interface Interface { threadId: string; fileId: string; }) => Effect.Effect; + /** + * A file attached to a message in the thread, as `read` gives it. Fails + * with `FileNotFound` for any other file, such as a tool call's result. + */ + readonly readMessageFile: (file: { + 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 @@ -178,6 +186,29 @@ export const make = Effect.gen(function* () { ), ); + /** The file `fileId` in the thread, when its row also matches `which`. */ + const readWhere = (threadId: string, fileId: string, which: SQLWrapper) => + Effect.gen(function* () { + // The id comes from the model or a URL, and Postgres rejects a uuid + // comparison against anything that is not one. + if (!isUuid(fileId)) return yield* new FileNotFound({ fileId }); + const [file] = yield* query((db) => + db + .select({ id: threadFile.id, name: threadFile.name }) + .from(threadFile) + .where(and(eq(threadFile.id, fileId), eq(threadFile.threadId, threadId), which)), + ); + if (!file) return yield* new FileNotFound({ fileId }); + // A row is never without its blob, as `keep` puts the blob first. + const stored = yield* blobs + .get(fileKey(threadId, file.id)) + .pipe(Effect.catchTag("BlobNotFound", Effect.die)); + const stream = stored.stream.pipe( + Stream.catchTag("BlobNotFound", () => Stream.fail(new FileNotFound({ fileId }))), + ); + return { id: file.id, name: file.name ?? undefined, ...stored, stream }; + }); + return Service.of({ write: ({ threadId, toolCallId, content }) => operation("write", keep(threadId, content, { kind: "tool_result", toolCallId })), @@ -208,35 +239,9 @@ export const make = Effect.gen(function* () { }), ), read: ({ threadId, fileId }) => - operation( - "read", - Effect.gen(function* () { - // The id comes from the model, and Postgres rejects a uuid - // comparison against anything that is not one. - if (!isUuid(fileId)) return yield* new FileNotFound({ fileId }); - const [file] = yield* query((db) => - db - .select({ id: threadFile.id, name: threadFile.name }) - .from(threadFile) - .where( - and( - eq(threadFile.id, fileId), - eq(threadFile.threadId, threadId), - not(unsentUpload), - ), - ), - ); - if (!file) return yield* new FileNotFound({ fileId }); - // A row is never without its blob, as `keep` puts the blob first. - const stored = yield* blobs - .get(fileKey(threadId, file.id)) - .pipe(Effect.catchTag("BlobNotFound", Effect.die)); - const stream = stored.stream.pipe( - Stream.catchTag("BlobNotFound", () => Stream.fail(new FileNotFound({ fileId }))), - ); - return { id: file.id, name: file.name ?? undefined, ...stored, stream }; - }), - ), + operation("read", readWhere(threadId, fileId, sql`not ${unsentUpload}`)), + readMessageFile: ({ threadId, fileId }) => + operation("readMessageFile", readWhere(threadId, fileId, isNotNull(threadFile.messageId))), attach: ({ threadId, messageId, uploaderId, fileIds }) => operation( "attach", From 95dc13220b1626be12180fa64618c36f0a3f0904 Mon Sep 17 00:00:00 2001 From: Tim Holm Date: Mon, 19 Oct 2026 15:02:40 +1100 Subject: [PATCH 3/4] chore(core): write a shared file's blob before locking the turn --- .../thread-files/thread-files.test.ts | 21 ++-- .../thread-files/thread-files.ts | 107 +++++++++++++----- .../conversations/turns/repository.test.ts | 7 ++ .../src/conversations/turns/repository.ts | 52 ++++----- 4 files changed, 125 insertions(+), 62 deletions(-) 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 80b860e1..467fd610 100644 --- a/packages/core/src/conversations/thread-files/thread-files.test.ts +++ b/packages/core/src/conversations/thread-files/thread-files.test.ts @@ -408,13 +408,20 @@ describe.skipIf(!process.env.DATABASE_URL)("thread files, against Postgres", () ); if (!reply) throw new Error("Could not create reply"); const attach = (name: string) => - files.attachToReply({ - threadId, - messageId: reply.id, - name, - contentType: "text/csv", - content: new TextEncoder().encode(name), - }); + 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"); diff --git a/packages/core/src/conversations/thread-files/thread-files.ts b/packages/core/src/conversations/thread-files/thread-files.ts index 93b4c3b8..29999687 100644 --- a/packages/core/src/conversations/thread-files/thread-files.ts +++ b/packages/core/src/conversations/thread-files/thread-files.ts @@ -93,17 +93,28 @@ export interface Interface { /** 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>; /** - * Keeps `content` as a file an agent attached to its reply `messageId`, - * after the reply's other files. Fails with `TooManyReplyFiles` once the - * reply carries as many as a message can. + * Puts `content` in the blob store as a file for an agent's reply + * `messageId`, then runs `use` with it, for `use` to attach with + * `attachToReply`. The blob is written before `use` runs, so nothing `use` + * locks waits on it, and removed if `use` fails. Run outside a transaction: + * one that rolls back after `use` succeeds would leave the blob until the + * thread goes. */ - readonly attachToReply: (file: { - threadId: string; - messageId: string; - name: string; - contentType: string; - content: Uint8Array; - }) => Effect.Effect; + readonly withReplyFile: ( + file: { + threadId: string; + messageId: string; + name: string; + contentType: string; + content: Uint8Array; + }, + use: (stored: StoredReplyFile) => Effect.Effect, + ) => Effect.Effect; + /** + * Attaches `stored` to its reply, after the reply's other files. Fails with + * `TooManyReplyFiles` once the reply carries as many as a message can. + */ + readonly attachToReply: (stored: StoredReplyFile) => Effect.Effect; /** The files an agent attached to its reply `messageId`, in the order attached. */ readonly replyAttachments: (messageId: string) => Effect.Effect; /** @@ -141,15 +152,12 @@ export const make = Effect.gen(function* () { const maxFileBytes = yield* Config.option(maxFileBytesConfig); /** - * Keeps `content` as a new file in the thread, its row saying where it came - * from. The blob first: a row is never without its blob. A row that then - * fails to insert would leave a blob nothing reads, so it is removed; if - * that fails too, it goes with the thread's folder. + * Puts `content` in the blob store as a new file in the thread, with no row + * yet: a row is never without its blob. */ - const keep = ( + const putBlob = ( threadId: string, content: BlobStore.Content, - source: FileSource, maxSize = Number.POSITIVE_INFINITY, ) => Effect.gen(function* () { @@ -163,20 +171,31 @@ export const make = Effect.gen(function* () { ? content.stream : BlobStore.atMost(content.stream, limit); const id = yield* ids.next; - const key = fileKey(threadId, id); - const { size } = yield* blobs.put(key, { stream, contentType }).pipe( + const { size } = yield* blobs.put(fileKey(threadId, id), { stream, contentType }).pipe( Effect.catchIf( (error) => error instanceof BlobStore.TooLarge, ({ maxSize }) => new FileTooLarge({ maxSize }), ), ); - yield* query((db) => - db.insert(threadFile).values({ id, threadId, contentType, size, ...source }), - ).pipe(Effect.onError(() => blobs.delete(key).pipe(Effect.ignore))); - return { id, contentType, size }; + return { id, contentType, size } satisfies FileDescription; }); - /** Removes a file's blob, once its row is gone; one that cannot be removed goes with the thread's folder. */ + /** Keeps `content` as a new file in the thread, its row saying where it came from. */ + const keep = ( + threadId: string, + content: BlobStore.Content, + source: FileSource, + maxSize = Number.POSITIVE_INFINITY, + ) => + Effect.gen(function* () { + const kept = yield* putBlob(threadId, content, maxSize); + yield* query((db) => db.insert(threadFile).values({ ...kept, threadId, ...source })).pipe( + Effect.onError(() => removeBlob(threadId, kept.id)), + ); + return kept; + }); + + /** Removes a file's blob, once its row is gone or was never kept; one that cannot be removed goes with the thread's folder. */ const removeBlob = (threadId: string, fileId: string) => blobs .delete(fileKey(threadId, fileId)) @@ -296,11 +315,21 @@ export const make = Effect.gen(function* () { ); }), ), - attachToReply: ({ threadId, messageId, name, contentType, content }) => + withReplyFile: ({ threadId, messageId, name, contentType, content }, use) => + operation( + "withReplyFile", + Effect.gen(function* () { + const kept = yield* putBlob(threadId, { stream: Stream.succeed(content), contentType }); + const stored = { ...kept, threadId, messageId, name } as StoredReplyFile; + return yield* use(stored).pipe(Effect.onError(() => removeBlob(threadId, kept.id))); + }), + ), + attachToReply: (stored) => operation( "attachToReply", transaction( Effect.gen(function* () { + const { id, threadId, messageId, name, contentType, size } = stored; // Locked, so files attached at once each take the next place, and the count holds. yield* query((db) => db @@ -319,12 +348,19 @@ export const make = Effect.gen(function* () { ); const position = attached?.files ?? 0; if (position >= MAX_ATTACHMENTS_PER_MESSAGE) return yield* new TooManyReplyFiles(); - const kept = yield* keep( - threadId, - { stream: Stream.succeed(content), contentType }, - { kind: "agent_attachment", name, messageId, position }, + yield* query((db) => + db.insert(threadFile).values({ + id, + threadId, + contentType, + size, + kind: "agent_attachment", + name, + messageId, + position, + }), ); - return { type: "file", name, ...kept } as const; + return { type: "file", id, name, contentType, size } as const; }), ), ), @@ -428,6 +464,19 @@ export interface StoredFile extends FileDescription { stream: Stream.Stream; } +declare const storedReplyFile: unique symbol; + +/** + * A file for an agent's reply whose blob `withReplyFile` has written, for + * `attachToReply` to attach. Only `withReplyFile` makes one. + */ +export interface StoredReplyFile extends FileDescription { + readonly [storedReplyFile]: true; + readonly threadId: string; + readonly messageId: string; + readonly name: string; +} + /** A file a person uploaded, with the name it had on their computer. */ export interface Upload extends FileDescription { name: string; diff --git a/packages/core/src/conversations/turns/repository.test.ts b/packages/core/src/conversations/turns/repository.test.ts index d0bf1e93..aab8b879 100644 --- a/packages/core/src/conversations/turns/repository.test.ts +++ b/packages/core/src/conversations/turns/repository.test.ts @@ -137,6 +137,13 @@ describe.skipIf(!process.env.DATABASE_URL)("turns, against Postgres", async () = }), ).rejects.toBeInstanceOf(TurnRepository.ReplyEnded); expect(await turns.replyFiles(replyTurnOf(prepared))).toEqual([]); + expect( + await runOnPostgres( + Effect.flatMap(BlobStore.Service, (store) => + store.list(BlobStore.key("threads", prepared.context.thread.id)), + ), + ), + ).toEqual([]); }); it("gives up on a turn that keeps stopping before it gets anywhere", async () => { diff --git a/packages/core/src/conversations/turns/repository.ts b/packages/core/src/conversations/turns/repository.ts index a9246f4d..936f8861 100644 --- a/packages/core/src/conversations/turns/repository.ts +++ b/packages/core/src/conversations/turns/repository.ts @@ -442,33 +442,33 @@ export const make = Effect.gen(function* () { attachFile: (reply, file) => operation( "attachFile", - transaction( - Effect.gen(function* () { - // Locked, as ending the turn locks it: a call still running when the - // turn was cancelled finds it ended, and shares nothing. - const [running] = yield* query((db) => - db - .select({ id: turn.id }) - .from(turn) - .where(and(eq(turn.id, reply.turnId), eq(turn.status, "running"))) - .for("update"), - ); - if (!running) return yield* new ReplyEnded(); - const attached = yield* files.attachToReply({ - threadId: reply.threadId, - messageId: reply.messageId, - ...file, - }); - const carried = yield* files.replyAttachments(reply.messageId); - yield* emit([ - ConversationEvent.MessageFilesChanged({ - threadId: reply.threadId, - messageId: reply.messageId, - files: carried, + files.withReplyFile( + { threadId: reply.threadId, messageId: reply.messageId, ...file }, + (stored) => + transaction( + Effect.gen(function* () { + // Locked, as ending the turn locks it: a call still running when the + // turn was cancelled finds it ended, and shares nothing. + const [running] = yield* query((db) => + db + .select({ id: turn.id }) + .from(turn) + .where(and(eq(turn.id, reply.turnId), eq(turn.status, "running"))) + .for("update"), + ); + if (!running) return yield* new ReplyEnded(); + const attached = yield* files.attachToReply(stored); + const carried = yield* files.replyAttachments(reply.messageId); + yield* emit([ + ConversationEvent.MessageFilesChanged({ + threadId: reply.threadId, + messageId: reply.messageId, + files: carried, + }), + ]); + return attached; }), - ]); - return attached; - }), + ), ), ), replyFiles: (reply) => operation("replyFiles", files.replyAttachments(reply.messageId)), From e4f795f8ff4ab9ab3375299bf828cb5eac4b399f Mon Sep 17 00:00:00 2001 From: Tim Holm Date: Mon, 19 Oct 2026 15:02:40 +1100 Subject: [PATCH 4/4] chore(web): keep back a reply's unfinished text after it shares a file --- packages/web/src/screens/ThreadConversation.tsx | 6 +++++- .../web/src/screens/thread-conversation.test.tsx | 13 +++++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/packages/web/src/screens/ThreadConversation.tsx b/packages/web/src/screens/ThreadConversation.tsx index b3b02948..7f0242f8 100644 --- a/packages/web/src/screens/ThreadConversation.tsx +++ b/packages/web/src/screens/ThreadConversation.tsx @@ -346,7 +346,11 @@ function segmentsOf(message: Message): Segment[] { } // 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() !== ""; + // Files go after all of a reply's text as they are attached, so one shared + // mid-reply does not mean the text before it is finished. + const followed = + part.text.trim() !== "" && + message.parts.slice(index + 1).some((later) => later.type !== "file"); const settled = finished || followed; if (settled && !isNarration(message.parts, index)) { segments.push({ type: "text", key: `text@${written}`, text: part.text }); diff --git a/packages/web/src/screens/thread-conversation.test.tsx b/packages/web/src/screens/thread-conversation.test.tsx index dd9c76a5..a9a26b7f 100644 --- a/packages/web/src/screens/thread-conversation.test.tsx +++ b/packages/web/src/screens/thread-conversation.test.tsx @@ -341,6 +341,19 @@ describe("a reply still being written", () => { expect(screen.getByRole("article").textContent).toContain("Here are the 2 issues in Cycle 33."); }); + it("keeps back the text a reply is still writing after it shares a file", () => { + const chart: MessagePart = { + type: "file", + id: "0199a3a0-0000-7000-8000-0000000000f1", + name: "chart.png", + contentType: "image/png", + size: 2048, + }; + show([reply([{ type: "text", text: "Here is the cha" }, chart], { status: "streaming" })]); + + expect(screen.queryByText(/Here is the cha/)).toBeNull(); + }); + it("says the bot is typing between one call and the next", () => { show([reply([toolCall("sentry__search_issues")], { status: "streaming" })]);