From ce691e572b71ce99811c5b14e4c559d39c8a6433 Mon Sep 17 00:00:00 2001 From: Tim Holm Date: Thu, 8 Oct 2026 12:43:21 +1100 Subject: [PATCH] feat(core): add the read_thread_file tool --- .../tools/read-thread-file/tool.test.ts | 108 ++++++++++++ .../tools/read-thread-file/tool.ts | 156 ++++++++++++++++++ .../core/src/conversations/turns/testing.ts | 2 + .../core/src/conversations/turns/tools.ts | 9 + .../conversations/turns/turn.steps.test.ts | 4 + .../src/conversations/turns/turn.steps.ts | 4 + .../core/src/conversations/turns/turns.ts | 8 +- 7 files changed, 290 insertions(+), 1 deletion(-) create mode 100644 packages/core/src/conversations/tools/read-thread-file/tool.test.ts create mode 100644 packages/core/src/conversations/tools/read-thread-file/tool.ts diff --git a/packages/core/src/conversations/tools/read-thread-file/tool.test.ts b/packages/core/src/conversations/tools/read-thread-file/tool.test.ts new file mode 100644 index 00000000..f31696e7 --- /dev/null +++ b/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 { + 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, + input: { file?: string; offset?: number; limit?: number; pattern?: string }, +): Promise { + const tool = readThreadFileTool({ threadId: THREAD, files, run: Effect.runPromise }); + const options = { toolCallId: "call", messages: [] } as unknown as Parameters< + NonNullable + >[1]; + return tool.execute?.({ file: FILE, ...input }, options) as Promise; +} + +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 = { + 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 = { + read: () => + new BlobStore.Rejected({ operation: "get", provider: "postgres", cause: "denied" }), + }; + + expect(await readFrom(rejecting, {})).toEqual({ + refused: "Files can't be read in this installation.", + }); + }); +}); diff --git a/packages/core/src/conversations/tools/read-thread-file/tool.ts b/packages/core/src/conversations/tools/read-thread-file/tool.ts new file mode 100644 index 00000000..3e2380a6 --- /dev/null +++ b/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; + run: RunEffect; +}) { + 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 => + 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)`; +} diff --git a/packages/core/src/conversations/turns/testing.ts b/packages/core/src/conversations/turns/testing.ts index fa4fc267..31907d9e 100644 --- a/packages/core/src/conversations/turns/testing.ts +++ b/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, ); /* diff --git a/packages/core/src/conversations/turns/tools.ts b/packages/core/src/conversations/turns/tools.ts index 57c1856f..cb1e04fa 100644 --- a/packages/core/src/conversations/turns/tools.ts +++ b/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>; + /** The thread's files, which `read_thread_file` reads. */ + files: Pick; /** Where an interviewing agent's own instructions are saved. */ agents: Pick; /** 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, diff --git a/packages/core/src/conversations/turns/turn.steps.test.ts b/packages/core/src/conversations/turns/turn.steps.test.ts index 623d3b6f..318fc376 100644 --- a/packages/core/src/conversations/turns/turn.steps.test.ts +++ b/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 ?? { diff --git a/packages/core/src/conversations/turns/turn.steps.ts b/packages/core/src/conversations/turns/turn.steps.ts index f4c15ef9..94271f92 100644 --- a/packages/core/src/conversations/turns/turn.steps.ts +++ b/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) }), diff --git a/packages/core/src/conversations/turns/turns.ts b/packages/core/src/conversations/turns/turns.ts index 2016c8b3..c2addb95 100644 --- a/packages/core/src/conversations/turns/turns.ts +++ b/packages/core/src/conversations/turns/turns.ts @@ -25,6 +25,7 @@ import { ACTIVE_TURN_STATUSES, turn } from "../../database/schema.ts"; import type { UserFacing } from "../../user-message.ts"; import { laneBusy, waitingSince } from "../../workflows/lanes.ts"; import { AgentRepository } from "../../workspaces/agents/agent-repository.ts"; +import { ThreadFiles } from "../thread-files/thread-files.ts"; import { ApprovedToolCalls } from "./approvals/approved-calls.ts"; import { makeControls, @@ -148,7 +149,12 @@ export const workflow = turnWorkflow; /** What the turn workflow's steps do: the rest of a turn's machinery. */ export const stepsLayer = turnStepsLayer.pipe( Layer.provide( - Layer.mergeAll(TurnExecution.layer, ApprovedToolCalls.layer, AgentRepository.layer), + Layer.mergeAll( + TurnExecution.layer, + ApprovedToolCalls.layer, + AgentRepository.layer, + ThreadFiles.layer, + ), ), Layer.provide(internals), );