feat(core): add the read_thread_file tool

packages/core/src/conversations/tools/read-thread-file/tool.test.ts

@@ -0,0 +1,108 @@
+import { Effect, Stream } from "effect";
+import { describe, expect, it } from "vitest";
+import { BlobStore } from "../../../blob-store/blob-store.ts";
+import { ThreadFiles } from "../../thread-files/thread-files.ts";
+import { MAX_READ_CHARACTERS, type ReadThreadFileResult, readThreadFileTool } from "./tool.ts";
+
+const THREAD = "0199a3a0-0000-7000-8000-000000000001";
+const FILE = "0199a3a0-0000-7000-8000-0000000000f1";
+
+/** `read_thread_file` over one file in `THREAD`, called with `input`. */
+async function read(
+	text: string,
+	input: { file?: string; offset?: number; limit?: number; pattern?: string },
+	contentType = "text/plain; charset=utf-8",
+): Promise<ReadThreadFileResult> {
+	return readFrom(
+		{
+			read: ({ threadId, fileId }) =>
+				threadId === THREAD && fileId === FILE
+					? Effect.succeed({
+							id: FILE,
+							contentType,
+							size: text.length,
+							stream: Stream.succeed(new TextEncoder().encode(text)),
+						})
+					: new ThreadFiles.FileNotFound({ fileId }),
+		},
+		input,
+	);
+}
+
+async function readFrom(
+	files: Pick<ThreadFiles.Interface, "read">,
+	input: { file?: string; offset?: number; limit?: number; pattern?: string },
+): Promise<ReadThreadFileResult> {
+	const tool = readThreadFileTool({ threadId: THREAD, files, run: Effect.runPromise });
+	const options = { toolCallId: "call", messages: [] } as unknown as Parameters<
+		NonNullable<typeof tool.execute>
+	>[1];
+	return tool.execute?.({ file: FILE, ...input }, options) as Promise<ReadThreadFileResult>;
+}
+
+const numbered = (count: number) =>
+	Array.from({ length: count }, (_, index) => `line ${index + 1}`).join("\n");
+
+describe("read_thread_file", () => {
+	it("reads numbered lines from an offset, and says where to continue", async () => {
+		expect(await read(numbered(10), { offset: 3, limit: 2 })).toEqual({
+			lines: "3\tline 3\n4\tline 4",
+			totalLines: 10,
+			continueFrom: 5,
+		});
+	});
+
+	it("says nothing about continuing once it reaches the end", async () => {
+		expect(await read("one\r\ntwo\n", {})).toEqual({ lines: "1\tone\n2\ttwo", totalLines: 2 });
+	});
+
+	it("returns only lines containing the pattern, ignoring case", async () => {
+		expect(await read("alpha\nBeta\ngamma\nbetamax", { pattern: "BETA" })).toEqual({
+			lines: "2\tBeta\n4\tbetamax",
+			totalLines: 4,
+		});
+	});
+
+	it("stops short of the character limit as JSON escapes it, at a line it can continue from", async () => {
+		const line = '"\\\u001b'.repeat(250);
+		const text = Array.from({ length: 100 }, () => line).join("\n");
+
+		const result = await read(text, {});
+
+		if (!("lines" in result)) throw new Error("refused");
+		expect(JSON.stringify(result.lines).length).toBeLessThanOrEqual(MAX_READ_CHARACTERS);
+		expect(result.continueFrom).toBe(result.lines.split("\n").length + 1);
+	});
+
+	it("refuses a file from another thread, an offset past the end, and a file that isn't text", async () => {
+		expect(await read("text", { file: "0199a3a0-0000-7000-8000-0000000000f2" })).toHaveProperty(
+			"refused",
+		);
+		expect(await read(numbered(3), { offset: 4 })).toEqual({
+			refused: "The file has 3 lines, so there is no line 4.",
+		});
+		expect(await read("PNG", {}, "image/png")).toHaveProperty("refused");
+	});
+
+	it("refuses, so the model can try again, while the blob store can't be reached", async () => {
+		const unreachable: Pick<ThreadFiles.Interface, "read"> = {
+			read: () =>
+				new BlobStore.Unavailable({ operation: "get", provider: "postgres", cause: "offline" }),
+		};
+
+		expect(await readFrom(unreachable, {})).toEqual({
+			refused: "The file can't be read right now. Try again shortly.",
+		});
+	});
+
+	it("refuses without suggesting another try when the blob store rejects reading", async () => {
+		const rejecting: Pick<ThreadFiles.Interface, "read"> = {
+			read: () =>
+				new BlobStore.Rejected({ operation: "get", provider: "postgres", cause: "denied" }),
+		};
+
+		expect(await readFrom(rejecting, {})).toEqual({
+			refused: "Files can't be read in this installation.",
+		});
+	});
+});

packages/core/src/conversations/tools/read-thread-file/tool.ts

@@ -0,0 +1,156 @@
+import { tool } from "ai";
+import { Effect, Schema, Stream } from "effect";
+import type { RunEffect } from "../../../database/database.ts";
+import type { ThreadFiles } from "../../thread-files/thread-files.ts";
+
+export const READ_THREAD_FILE_TOOL = "read_thread_file";
+
+/** Lines returned when the model asks for no particular number. */
+const DEFAULT_LINES = 2_000;
+
+/** Longer lines are cut, so one minified line cannot fill a whole read. */
+const MAX_LINE_CHARACTERS = 2_000;
+
+/**
+ * Most characters of lines one call returns, as JSON escapes them. Below the
+ * size at which a tool's result, measured as JSON, is kept as a file instead,
+ * so a read is never itself turned into one.
+ */
+export const MAX_READ_CHARACTERS = 40_000;
+
+const MAX_PATTERN_CHARACTERS = 500;
+
+const NOT_FOUND = {
+	refused: "There is no file with that id in this thread.",
+} satisfies ReadThreadFileResult;
+
+const UNAVAILABLE = {
+	refused: "The file can't be read right now. Try again shortly.",
+} satisfies ReadThreadFileResult;
+
+const REJECTED = {
+	refused: "Files can't be read in this installation.",
+} satisfies ReadThreadFileResult;
+
+export type ReadThreadFileResult =
+	| {
+			/** The lines read, each after its number and a tab. */
+			lines: string;
+			totalLines: number;
+			/** Where to start the next call to read on, when there is more. */
+			continueFrom?: number;
+	  }
+	| { refused: string };
+
+/**
+ * The `read_thread_file` tool: reads part of a file in this thread, such as a tool
+ * result too large to have been given whole, by line or by searching it.
+ * Offered on every turn, so the tools a thread sends never change for it.
+ */
+export function readThreadFileTool({
+	threadId,
+	files,
+	run,
+}: {
+	threadId: string;
+	files: Pick<ThreadFiles.Interface, "read">;
+	run: RunEffect<never>;
+}) {
+	return tool({
+		description: `Read a file in this thread, such as a tool result too large to have been given to you whole, which named its file and how to read on. Returns up to ${DEFAULT_LINES} lines from \`offset\`, each after its line number, with the file's total lines and, when there is more, the line to continue from. Give \`pattern\` to get only the lines containing it instead.`,
+		inputSchema: Schema.Struct({
+			file: Schema.String.annotate({ description: "The file's id." }),
+			offset: Schema.optional(
+				Schema.Int.check(Schema.isGreaterThanOrEqualTo(1)).annotate({
+					description: "The line to start at, counting from 1. Defaults to the first.",
+				}),
+			),
+			limit: Schema.optional(
+				Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: DEFAULT_LINES })).annotate({
+					description: `The most lines to return. Defaults to ${DEFAULT_LINES}.`,
+				}),
+			),
+			pattern: Schema.optional(
+				Schema.String.check(
+					Schema.isMinLength(1),
+					Schema.isMaxLength(MAX_PATTERN_CHARACTERS),
+				).annotate({
+					description:
+						"Return only lines containing this text, ignoring case, searching from `offset`.",
+				}),
+			),
+		}).pipe(Schema.toStandardSchemaV1, Schema.toStandardJSONSchemaV1),
+		execute: ({ file, offset, limit, pattern }): Promise<ReadThreadFileResult> =>
+			run(
+				Effect.gen(function* () {
+					const found = yield* files.read({ threadId, fileId: file });
+					if (!isText(found.contentType)) {
+						return {
+							refused: `The file is ${found.contentType}, not text, so it has no lines to read.`,
+						};
+					}
+					const text = yield* Stream.mkString(Stream.decodeText(found.stream));
+					return readLines(text, { offset: offset ?? 1, limit: limit ?? DEFAULT_LINES, pattern });
+				}).pipe(
+					Effect.catchTags({
+						ThreadFileNotFound: () => Effect.succeed(NOT_FOUND),
+						BlobStoreUnavailable: () => Effect.succeed(UNAVAILABLE),
+						BlobStoreRejected: (rejected) =>
+							Effect.as(
+								Effect.logError("The blob store rejected reading a file", rejected),
+								REJECTED,
+							),
+					}),
+				),
+			),
+	});
+}
+
+function isText(contentType: string): boolean {
+	return contentType.startsWith("text/") || contentType.startsWith("application/json");
+}
+
+/**
+ * Up to `limit` of `text`'s lines from `offset`, numbered, or only those
+ * containing `pattern`, stopping short at `MAX_READ_CHARACTERS` of JSON.
+ */
+function readLines(
+	text: string,
+	{ offset, limit, pattern }: { offset: number; limit: number; pattern: string | undefined },
+): ReadThreadFileResult {
+	const lines = splitLines(text);
+	if (offset > lines.length) {
+		return { refused: `The file has ${lines.length} lines, so there is no line ${offset}.` };
+	}
+	const needle = pattern?.toLowerCase();
+	const read: string[] = [];
+	let characters = 0;
+	for (let index = offset - 1; index < lines.length; index++) {
+		const line = lines[index] ?? "";
+		if (needle && !line.toLowerCase().includes(needle)) continue;
+		const numbered = `${index + 1}\t${clip(line)}`;
+		// Escaping can double a line of quotes or backslashes. The quotes
+		// `JSON.stringify` adds count for the escaped newline that joins it.
+		const escapedCharacters = JSON.stringify(numbered).length;
+		const full = read.length === limit || characters + escapedCharacters > MAX_READ_CHARACTERS;
+		if (full) {
+			return { lines: read.join("\n"), totalLines: lines.length, continueFrom: index + 1 };
+		}
+		read.push(numbered);
+		characters += escapedCharacters;
+	}
+	return { lines: read.join("\n"), totalLines: lines.length };
+}
+
+/** Lines as an editor numbers them: a final newline ends the last line rather than starting another. */
+function splitLines(text: string): string[] {
+	if (text === "") return [];
+	const lines = text.split(/\r?\n/);
+	return text.endsWith("\n") ? lines.slice(0, -1) : lines;
+}
+
+function clip(line: string): string {
+	return line.length <= MAX_LINE_CHARACTERS
+		? line
+		: `${line.slice(0, MAX_LINE_CHARACTERS)}… (${line.length - MAX_LINE_CHARACTERS} more characters)`;
+}

packages/core/src/conversations/turns/testing.ts

@@ -14,6 +14,7 @@ import {
 	facilitateLane,
 } from "../floor/facilitate.workflow.ts";
 import { lanesForTests } from "../testing.ts";
+import { ThreadFiles } from "../thread-files/thread-files.ts";
 import { ApprovedToolCalls } from "./approvals/approved-calls.ts";
 import { type PreparedTurn, TurnExecution, type TurnRun, turnRunFor } from "./execution.ts";
 import { TurnRepository } from "./repository.ts";
@@ -38,6 +39,7 @@ export const turnInternalsForTests = Layer.mergeAll(
 	ToolCallRepository.layer,
 	ApprovedToolCalls.layer,
 	AgentRepository.layer,
+	ThreadFiles.layer,
 );
 
 /*

packages/core/src/conversations/turns/tools.ts

@@ -5,11 +5,13 @@ import type { RunEffect } from "../../database/database.ts";
 import type { EventBus } from "../../database/events/bus.ts";
 import { UserMessage } from "../../user-message.ts";
 import type { AgentRepository } from "../../workspaces/agents/agent-repository.ts";
+import type { ThreadFiles } from "../thread-files/thread-files.ts";
 import { SEARCH_HISTORY_TOOL } from "../threads/message-text.ts";
 import type { BuiltInTools } from "../tools/built-in.ts";
 import type { Collaborations } from "../tools/collaborate/collaborations.ts";
 import { collaborateTool } from "../tools/collaborate/tool.ts";
 import type { OfferedTool } from "../tools/connections.ts";
+import { READ_THREAD_FILE_TOOL, readThreadFileTool } from "../tools/read-thread-file/tool.ts";
 import { READ_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";
@@ -52,6 +54,8 @@ export interface ToolDependencies {
 	sandbox: SandboxTools.Offered;
 	/** The pod connections' tools, keyed `handle__tool`, reached through `tool_search` and `call_tool`. */
 	connections: Readonly<Record<string, OfferedTool>>;
+	/** The thread's files, which `read_thread_file` reads. */
+	files: Pick<ThreadFiles.Interface, "read">;
 	/** Where an interviewing agent's own instructions are saved. */
 	agents: Pick<AgentRepository.Interface, "finishInterview">;
 	/** For a tool that watches for something else to happen. */
@@ -163,6 +167,11 @@ export function toolsForTurn(prepared: PreparedTurn, deps: ToolDependencies): To
 		}),
 		recording,
 	);
+	tools[READ_THREAD_FILE_TOOL] = recorded(
+		READ_THREAD_FILE_TOOL,
+		readThreadFileTool({ threadId: prepared.context.thread.id, files: deps.files, run: deps.run }),
+		recording,
+	);
 	if (prepared.context.agent.interviewing) {
 		tools[SAVE_INSTRUCTIONS_TOOL] = recorded(
 			SAVE_INSTRUCTIONS_TOOL,

packages/core/src/conversations/turns/turn.steps.test.ts

@@ -11,11 +11,13 @@ import { Models } from "../../providers/models/models.ts";
 import { chunks, scriptedModel, streamed, unusedModel } from "../../providers/models/testing.ts";
 import { unimplemented } from "../../testing.ts";
 import { AgentRepository } from "../../workspaces/agents/agent-repository.ts";
+import { ThreadFiles } from "../thread-files/thread-files.ts";
 import { SEARCH_HISTORY_TOOL } from "../threads/message-text.ts";
 import { BuiltInTools } from "../tools/built-in.ts";
 import { Collaborations } from "../tools/collaborate/collaborations.ts";
 import { ConnectionTools } from "../tools/connections.ts";
 import { REQUEST_NETWORK_ACCESS_TOOL, requestNetworkAccess } from "../tools/network-access/tool.ts";
+import { READ_THREAD_FILE_TOOL } from "../tools/read-thread-file/tool.ts";
 import { SandboxTools } from "../tools/sandbox.ts";
 import { CALL_TOOL, TOOL_SEARCH } from "../tools/tool-search/tool.ts";
 import {
@@ -335,6 +337,7 @@ describe("runSegment", () => {
 			TOOL_SEARCH,
 			CALL_TOOL,
 			SEARCH_HISTORY_TOOL,
+			READ_THREAD_FILE_TOOL,
 			"collaborate",
 		]);
 		const approvals = received?.toolApproval as
@@ -637,6 +640,7 @@ function segmentWith(given: Given) {
 				unimplemented(ToolCallRepository.Service, given.toolCalls),
 				unimplemented(Collaborations.Service, given.collaborations),
 				unimplemented(AgentRepository.Service, {}),
+				unimplemented(ThreadFiles.Service, {}),
 				unimplemented(
 					ApprovedToolCalls.Service,
 					given.approvals ?? {

packages/core/src/conversations/turns/turn.steps.ts

@@ -22,6 +22,7 @@ import { type UserFacing, UserMessage } from "../../user-message.ts";
 import { AgentRepository } from "../../workspaces/agents/agent-repository.ts";
 import { ConversationEvents } from "../conversation-events.ts";
 import { ConversationEvent } from "../events.ts";
+import { ThreadFiles } from "../thread-files/thread-files.ts";
 import { BuiltInTools } from "../tools/built-in.ts";
 import { Collaborations } from "../tools/collaborate/collaborations.ts";
 import { ConnectionTools } from "../tools/connections.ts";
@@ -84,6 +85,7 @@ type SegmentServices =
 	| Collaborations.Service
 	| ApprovedToolCalls.Service
 	| AgentRepository.Service
+	| ThreadFiles.Service
 	| Ids.Service;
 
 /** The turn workflow's steps, which its activities reach through `TurnSteps`. */
@@ -375,6 +377,7 @@ const streamReply = (
 			const collaborations = yield* Collaborations.Service;
 			const approvals = yield* ApprovedToolCalls.Service;
 			const agents = yield* AgentRepository.Service;
+			const files = yield* ThreadFiles.Service;
 			const signal = yield* Effect.abortSignal;
 
 			// One writer at a time: the periodic save, the on-size save and a tool
@@ -453,6 +456,7 @@ const streamReply = (
 				builtIn,
 				sandbox,
 				connections: connections.tools,
+				files,
 				agents,
 				bus: events,
 				run: effectRunner({ runPromiseExit: Effect.runPromiseExitWith(context) }),