Skip to content

perf(cli): read long threads from the end instead of paging through them #33079

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fagnersales wants to merge 5 commits into
base: main
Choose a base branch
from
+291 −66
Open

perf(cli): read long threads from the end instead of paging through them #33079

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8f0caec
perf(cli): read long threads from the end instead of paging through them
fagnersales Nov 8, 2026
5780879
perf(cli): leave linked-environment result reads to their own PR
fagnersales Nov 8, 2026
2cdf508
fix(cli): stop a backward read whose cursor does not move back
fagnersales Nov 8, 2026
69a7f52
fix(cli): drop a repeated page's items from a backward read
fagnersales Nov 8, 2026
f0531e7
Merge remote-tracking branch 'origin/main' into perf/thread-read-tail
fagnersales Nov 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
171 changes: 133 additions & 38 deletions apps/server/src/cli/thread.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Path from "effect/Path";
import * as Schema from "effect/Schema";
import * as TestConsole from "effect/testing/TestConsole";
import { Command } from "effect/cli";
import { vi } from "vite-plus/test";
Expand All @@ -20,7 +21,9 @@ import {
persistServerRuntimeState,
} from "../serverRuntimeState.ts";
import {
makeThreadTools,
readLatestMessages,
ThreadToolFailedError,
threadCommand,
type ThreadTools,
titleFromMessage,
Expand Down Expand Up @@ -201,17 +204,50 @@ 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<string>) => {
/**
* Serves t3_thread_read pages the way the server does: positions count every
* item. A legacy server's strict schema rejects `beforePosition`.
*/
const timelineTools = (types: ReadonlyArray<string>, server: "current" | "legacy" = "current") => {
const items = types.map((type, position) => ({ position, itemId: `item-${position}`, type }));
const isMatch = (item: (typeof items)[number]) =>
["user_message", "assistant_message", "proposed_plan"].includes(item.type);
const served: Array<number> = [];
const call = ((_tool: string, input: unknown) => {
const args = input as { readonly afterPosition?: number; readonly limit: number };
const args = input as {
readonly afterPosition?: number;
readonly beforePosition?: number;
readonly limit: number;
};
if (args.beforePosition !== undefined) {
if (server === "legacy") {
return Effect.fail(
new ThreadToolFailedError({
tool: "t3_thread_read",
detail: "Unknown key beforePosition",
}),
);
}
const newestFirst: typeof items = [];
const end = Math.min(args.beforePosition, items.length);
for (let position = end - 1; position >= 0 && newestFirst.length <= args.limit; position--) {
if (isMatch(items[position]!)) newestFirst.push(items[position]!);
}
const page = newestFirst.slice(0, args.limit).reverse();
served.push(page.length);
return Effect.succeed({
thread: { itemCount: items.length },
items: page,
nextPosition: page.at(-1)?.position ?? null,
hasMore: false,
previousPosition: newestFirst.length > page.length ? page[0]!.position : null,
});
}
// 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)) {
if (isMatch(item)) {
matching.push(item);
if (matching.length > args.limit) break;
}
Expand All @@ -234,57 +270,116 @@ const timelineTools = (types: ReadonlyArray<string>) => {
};

const tail = (tools: number) => [...Array<string>(tools).fill("tool_call"), "assistant_message"];
const SERVERS = ["current", "legacy"] as const;

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",
]);
it.effect.each(SERVERS)(
"show reads the end of a long thread, not its whole history (%s)",
(server) =>
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",
],
server,
);

const { messages } = yield* readLatestMessages(tools, threadId, 10);
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,
);
}),
assert.deepStrictEqual(
messages.map((item) => item.itemId),
latest(10),
);
assert.isBelow(
served.reduce((sum, count) => sum + count, 0),
100,
);
// Each call builds the whole thread's index on the server, so calls are the cost.
if (server === "current") assert.lengthOf(served, 2);
}),
);

it.effect.each(SERVERS)(
"show widens back to the first item when the thread has fewer messages (%s)",
(server) =>
Effect.gen(function* () {
const { tools, latest } = timelineTools(
["user_message", ...Array.from({ length: 4 }, () => tail(150)).flat()],
server,
);

const { messages } = yield* readLatestMessages(tools, threadId, 10);

assert.deepStrictEqual(
messages.map((item) => item.itemId),
latest(10),
);
assert.lengthOf(messages, 5);
}),
);

it.effect("show widens back to the first item when the thread has fewer messages", () =>
it.effect.each(SERVERS)(
"show returns a --messages count larger than one call can spread (%s)",
(server) =>
Effect.gen(function* () {
const { tools, latest } = timelineTools(
Array<string>(200_000).fill("assistant_message"),
server,
);

const { messages } = yield* readLatestMessages(tools, threadId, 200_000);

assert.deepStrictEqual(
messages.map((item) => item.itemId),
latest(200_000),
);
}),
);

it.effect("show stops when a server's previousPosition does not move back", () =>
Effect.gen(function* () {
const { tools, latest } = timelineTools([
"user_message",
...Array.from({ length: 4 }, () => tail(150)).flat(),
]);
const page = {
thread: { itemCount: 10 },
items: [{ position: 5, itemId: "item-5", type: "assistant_message" }],
nextPosition: 5,
hasMore: false,
previousPosition: 5,
};
const { tools, calls } = scriptedTools({ t3_thread_read: [page, page] });

const { messages } = yield* readLatestMessages(tools, threadId, 10);

assert.lengthOf(calls, 2);
assert.deepStrictEqual(
messages.map((item) => item.itemId),
latest(10),
["item-5"],
);
assert.lengthOf(messages, 5);
}),
);

it.effect("show returns a --messages count larger than one call can spread", () =>
it.effect("a server's rejected arguments are a tool failure, so show can fall back", () =>
Effect.gen(function* () {
const { tools, latest } = timelineTools(Array<string>(200_000).fill("assistant_message"));
const failWith = (code: number | undefined) =>
makeThreadTools(
{
call: () =>
Effect.fail(
new McpBridge.AcpMcpBridgeError({
message: "T3 Code MCP tools/call failed",
...(code === undefined ? {} : { code }),
}),
),
},
"http://127.0.0.1:3773",
)
.call("t3_thread_read", { threadId, beforePosition: 1 }, Schema.Unknown)
.pipe(Effect.match({ onFailure: (error) => error._tag, onSuccess: () => "succeeded" }));

const { messages } = yield* readLatestMessages(tools, threadId, 200_000);

assert.deepStrictEqual(
messages.map((item) => item.itemId),
latest(200_000),
);
assert.strictEqual(yield* failWith(McpBridge.JSON_RPC_INVALID_PARAMS), "ThreadToolFailedError");
assert.strictEqual(yield* failWith(undefined), "ThreadServerUnreachableError");
}),
);

Expand Down
101 changes: 88 additions & 13 deletions apps/server/src/cli/thread.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,11 @@ import { Argument, Command, Flag, GlobalFlag } from "effect/cli";

import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
import * as ServerConfig from "../config.ts";
import { mcpToolSession, type McpToolSession } from "../mcp/AcpMcpStdioBridge.ts";
import {
JSON_RPC_INVALID_PARAMS,
mcpToolSession,
type McpToolSession,
} from "../mcp/AcpMcpStdioBridge.ts";
import { expandHomePath } from "../os-jank.ts";
import { readPersistedServerRuntimeState } from "../serverRuntimeState.ts";
import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts";
Expand All @@ -39,6 +43,8 @@ import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts";
*/
const WAIT_SLICE_MS = 15_000;
const READ_PAGE_LIMIT = 100;
/** A `beforePosition` past every item, so the read starts at the newest. */
const NEWEST_POSITION = Number.MAX_SAFE_INTEGER;
const TITLE_MAX_CHARS = 80;

export class ThreadServerNotRunningError extends Schema.TaggedError<ThreadServerNotRunningError>()(
Expand Down Expand Up @@ -172,7 +178,7 @@ const toolFailureDetail = (result: typeof ToolResult.Type) => {
});
};

const makeThreadTools = (session: McpToolSession, origin: string) => {
export const makeThreadTools = (session: McpToolSession, origin: string) => {
const unexpected = (tool: string) =>
new ThreadToolFailedError({
tool,
Expand All @@ -186,7 +192,13 @@ const makeThreadTools = (session: McpToolSession, origin: string) => {
): Effect.fn.Return<S["Type"], ThreadServerUnreachableError | ThreadToolFailedError> {
const raw = yield* session
.call(tool, args)
.pipe(Effect.mapError((cause) => new ThreadServerUnreachableError({ origin, cause })));
.pipe(
Effect.mapError((cause) =>
cause.code === JSON_RPC_INVALID_PARAMS
? new ThreadToolFailedError({ tool, detail: cause.message })
: new ThreadServerUnreachableError({ origin, cause }),
),
);
const result = yield* decodeToolResult(raw).pipe(Effect.mapError(() => unexpected(tool)));
if (result.isError === true) {
return yield* new ThreadToolFailedError({ tool, detail: toolFailureDetail(result) });
Expand Down Expand Up @@ -337,11 +349,50 @@ const readTimeline = Effect.fn("cli.thread.readTimeline")(function* (
});

/**
* 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.
* Pages back from the newest item until `done` accepts a page, newest page
* first. Undefined when the server predates `beforePosition`: its strict
* schema rejects the field, and a lenient one answers without `previousPosition`.
*/
export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(function* (
const readBackward = Effect.fn("cli.thread.readBackward")(function* (
tools: ThreadTools,
threadId: ThreadId,
view: "messages" | "activity",
nextLimit: () => number,
done: (page: ReadonlyArray<OrchestratorMcpThreadTimelineItem>) => boolean,
) {
const read = (beforePosition: number) =>
tools.call(
"t3_thread_read",
{ threadId, view, limit: nextLimit(), runLimit: 1, beforePosition },
OrchestratorMcpThreadReadResult,
);
const newest = yield* read(NEWEST_POSITION).pipe(
Effect.catchTags({ ThreadToolFailedError: () => Effect.succeed(null) }),
);
if (newest === null || newest.previousPosition === undefined) return undefined;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
const pages = [newest.items];
let previous = done(newest.items) ? null : newest.previousPosition;
while (previous !== null) {
const before = previous;
const page = yield* read(before);
// A server that repeats a page, or a cursor that does not move back, would read forever.
const items = page.items.filter((item) => item.position < before);
pages.push(items);
const next = page.previousPosition ?? null;
previous = done(items) || next === null || next >= before ? null : next;
}
return { thread: newest.thread, pages };
});

const isConversationMessage = (item: OrchestratorMcpThreadTimelineItem) =>
item.type === "user_message" || item.type === "assistant_message";

/**
* `readLatestMessages` for servers without `beforePosition`. 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.
*/
const readLatestMessagesForward = Effect.fn("cli.thread.readLatestMessagesForward")(function* (
tools: ThreadTools,
threadId: ThreadId,
count: number,
Expand Down Expand Up @@ -377,9 +428,7 @@ export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(fun
);
window.push(
...page.items.filter(
(item) =>
(item.type === "user_message" || item.type === "assistant_message") &&
(end === undefined || item.position < end),
(item) => isConversationMessage(item) && (end === undefined || item.position < end),
),
);
afterPosition = page.nextPosition ?? Number.POSITIVE_INFINITY;
Expand All @@ -393,6 +442,26 @@ export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(fun
return { thread, messages: messages.slice(Math.max(0, messages.length - count)) };
});

/** The latest `count` user and assistant messages, oldest first. */
export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(function* (
tools: ThreadTools,
threadId: ThreadId,
count: number,
) {
let found = 0;
const read = yield* readBackward(
tools,
threadId,
"messages",
// Items can be long, so a page asks only for the messages still missing.
() => Math.min(READ_PAGE_LIMIT, Math.max(1, count - found)),
(page) => (found += page.filter(isConversationMessage).length) >= count,
);
if (read === undefined) return yield* readLatestMessagesForward(tools, threadId, count);
const messages = read.pages.toReversed().flat().filter(isConversationMessage);
return { thread: read.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,
Expand Down Expand Up @@ -422,12 +491,18 @@ const runResultText = Effect.fn("cli.thread.runResultText")(function* (
runId: RunId,
type: "assistant_message" | "error",
) {
const { items } = yield* readTimeline(
const view = type === "assistant_message" ? "messages" : "activity";
const matches = (candidate: OrchestratorMcpThreadTimelineItem) =>
candidate.runId === runId && candidate.type === type;
const read = yield* readBackward(
tools,
threadId,
type === "assistant_message" ? "messages" : "activity",
view,
() => READ_PAGE_LIMIT,
(page) => page.some(matches),
);
const item = items.findLast((candidate) => candidate.runId === runId && candidate.type === type);
const pages = read?.pages ?? [(yield* readTimeline(tools, threadId, view)).items];
const item = pages.map((page) => page.findLast(matches)).find((match) => match !== undefined);
return item === undefined ? null : yield* readFullText(tools, threadId, item);
});

Expand Down
Loading
Loading