From e89c983eaf6fa874abb43e40dd88f5d96b6e8915 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Thu, 22 Oct 2026 16:49:11 -0500 Subject: [PATCH 1/2] perf(cli): read only recent history for thread show `t3 thread show` paged through every timeline item before keeping the last --messages. Read the thread's item count first, then read a window at the end and widen it backward only while tool and plan items leave too few messages. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cli/thread.test.ts | 72 +++++++++++++++++++++++++++++- apps/server/src/cli/thread.ts | 64 ++++++++++++++++++++++++-- 2 files changed, 131 insertions(+), 5 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index c2e4bc5c8168c..39932387ce80f 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,73 @@ 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 }; + const matching = items.filter( + (item) => + item.position > (args.afterPosition ?? -1) && + ["user_message", "assistant_message", "proposed_plan"].includes(item.type), + ); + 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); + }), +); diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 989028afccb9d..d7023d61b2979 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -332,6 +332,62 @@ 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, + ); + const 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; + messages.unshift(...window); + 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 +717,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) => ({ From a3256e5ae51b587c9e16f90dc4aa3e971522f789 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Fri, 23 Oct 2026 04:30:46 -0500 Subject: [PATCH 2/2] fix(cli): thread show returns very large --messages counts The first window can hold the whole --messages count, and spreading it into unshift threw RangeError once it passed the engine's argument limit. Concatenate the window instead, and let the test fixture page huge threads in linear time. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cli/thread.test.ts | 27 ++++++++++++++++++++++----- apps/server/src/cli/thread.ts | 5 +++-- 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index 39932387ce80f..e09f3be00312b 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -80,11 +80,15 @@ const timelineTools = (types: ReadonlyArray) => { const served: Array = []; const call = ((_tool: string, input: unknown) => { const args = input as { readonly afterPosition?: number; readonly limit: number }; - const matching = items.filter( - (item) => - item.position > (args.afterPosition ?? -1) && - ["user_message", "assistant_message", "proposed_plan"].includes(item.type), - ); + // 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({ @@ -143,3 +147,16 @@ it.effect("show widens back to the first item when the thread has fewer messages 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 d7023d61b2979..6aa28a2c18810 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -347,7 +347,7 @@ export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(fun { threadId, limit: 1, runLimit: 1 }, OrchestratorMcpThreadReadResult, ); - const messages: Array = []; + let messages: Array = []; // The newest window has no upper bound, so items added since the first // read are still shown. let before: number | undefined; @@ -381,7 +381,8 @@ export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(fun afterPosition = page.nextPosition ?? Number.POSITIVE_INFINITY; } while (page.hasMore && afterPosition < (end ?? Number.POSITIVE_INFINITY) - 1); thread = page.thread; - messages.unshift(...window); + // 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);