diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index c2e4bc5c8168c..e09f3be00312b 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -2,7 +2,7 @@ import { assert, it } from "@effect/vitest"; import { RunId, ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; -import { type ThreadTools, waitForRun } from "./thread.ts"; +import { readLatestMessages, type ThreadTools, waitForRun } from "./thread.ts"; const threadId = ThreadId.make("thread-1"); const runId = RunId.make("run-1"); @@ -73,3 +73,90 @@ it.effect("--wait reports its run's result when a later run is the one waiting", }); }), ); + +/** Serves t3_thread_read pages the way the server does: positions count every item. */ +const timelineTools = (types: ReadonlyArray) => { + const items = types.map((type, position) => ({ position, itemId: `item-${position}`, type })); + const served: Array = []; + const call = ((_tool: string, input: unknown) => { + const args = input as { readonly afterPosition?: number; readonly limit: number }; + // Scanning one match past the page keeps reads of huge fixtures linear. + const matching: typeof items = []; + for (let position = (args.afterPosition ?? -1) + 1; position < items.length; position++) { + const item = items[position]!; + if (["user_message", "assistant_message", "proposed_plan"].includes(item.type)) { + matching.push(item); + if (matching.length > args.limit) break; + } + } + const page = matching.slice(0, args.limit); + served.push(page.length); + return Effect.succeed({ + thread: { itemCount: items.length }, + items: page, + nextPosition: page.at(-1)?.position ?? null, + hasMore: matching.length > page.length, + }); + }) as ThreadTools["call"]; + const latest = (count: number) => + items + .filter((item) => item.type === "user_message" || item.type === "assistant_message") + .slice(-count) + .map((item) => item.itemId); + return { tools: { call } satisfies ThreadTools, served, latest }; +}; + +const tail = (tools: number) => [...Array(tools).fill("tool_call"), "assistant_message"]; + +it.effect("show reads the end of a long thread, not its whole history", () => + Effect.gen(function* () { + const { tools, served, latest } = timelineTools([ + ...Array.from({ length: 1000 }, (_, index) => + index % 2 === 0 ? "user_message" : "assistant_message", + ), + ...Array.from({ length: 10 }, () => tail(30)).flat(), + "proposed_plan", + ]); + + const { messages } = yield* readLatestMessages(tools, threadId, 10); + + assert.deepStrictEqual( + messages.map((item) => item.itemId), + latest(10), + ); + assert.isBelow( + served.reduce((sum, count) => sum + count, 0), + 100, + ); + }), +); + +it.effect("show widens back to the first item when the thread has fewer messages", () => + Effect.gen(function* () { + const { tools, latest } = timelineTools([ + "user_message", + ...Array.from({ length: 4 }, () => tail(150)).flat(), + ]); + + const { messages } = yield* readLatestMessages(tools, threadId, 10); + + assert.deepStrictEqual( + messages.map((item) => item.itemId), + latest(10), + ); + assert.lengthOf(messages, 5); + }), +); + +it.effect("show returns a --messages count larger than one call can spread", () => + Effect.gen(function* () { + const { tools, latest } = timelineTools(Array(200_000).fill("assistant_message")); + + const { messages } = yield* readLatestMessages(tools, threadId, 200_000); + + assert.deepStrictEqual( + messages.map((item) => item.itemId), + latest(200_000), + ); + }), +); diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 989028afccb9d..6aa28a2c18810 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -332,6 +332,63 @@ const readTimeline = Effect.fn("cli.thread.readTimeline")(function* ( return { thread: page.thread, items }; }); +/** + * The latest `count` user and assistant messages, oldest first. Positions + * count every timeline item, so the read starts near the end and widens + * backward only while tool and plan items leave too few messages. + */ +export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(function* ( + tools: ThreadTools, + threadId: ThreadId, + count: number, +) { + let { thread } = yield* tools.call( + "t3_thread_read", + { threadId, limit: 1, runLimit: 1 }, + OrchestratorMcpThreadReadResult, + ); + let messages: Array = []; + // The newest window has no upper bound, so items added since the first + // read are still shown. + let before: number | undefined; + let start = thread.itemCount; + let span = Math.max(count, READ_PAGE_LIMIT); + do { + const end = before; + start = Math.max(0, start - span); + const window: Array = []; + let afterPosition = start - 1; + let page: OrchestratorMcpThreadReadResult; + do { + page = yield* tools.call( + "t3_thread_read", + { + threadId, + view: "messages", + limit: READ_PAGE_LIMIT, + runLimit: 1, + ...(afterPosition < 0 ? {} : { afterPosition }), + }, + OrchestratorMcpThreadReadResult, + ); + window.push( + ...page.items.filter( + (item) => + (item.type === "user_message" || item.type === "assistant_message") && + (end === undefined || item.position < end), + ), + ); + afterPosition = page.nextPosition ?? Number.POSITIVE_INFINITY; + } while (page.hasMore && afterPosition < (end ?? Number.POSITIVE_INFINITY) - 1); + thread = page.thread; + // A window can hold the whole --messages count, too many to spread as arguments. + messages = window.concat(messages); + before = start; + span *= 2; + } while (messages.length < count && start > 0); + return { thread, messages: messages.slice(Math.max(0, messages.length - count)) }; +}); + /** Item text past the read's per-item cap. */ const readFullText = Effect.fn("cli.thread.readFullText")(function* ( tools: ThreadTools, @@ -661,11 +718,11 @@ const threadShowCommand = Command.make("show", { yield* withThreadTools( flags, Effect.fn("cli.thread.show")(function* (tools) { - const { thread, items } = yield* readTimeline(tools, threadId, "messages"); - const conversation = items.filter( - (item) => item.type === "user_message" || item.type === "assistant_message", + const { thread, messages: latest } = yield* readLatestMessages( + tools, + threadId, + flags.messages, ); - const latest = conversation.slice(Math.max(0, conversation.length - flags.messages)); const messages = yield* Effect.forEach(latest, (item) => readFullText(tools, threadId, item).pipe( Effect.map((text) => ({