Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
89 changes: 88 additions & 1 deletion apps/server/src/cli/thread.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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<string>) => {
const items = types.map((type, position) => ({ position, itemId: `item-${position}`, type }));
const served: Array<number> = [];
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<string>(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<string>(200_000).fill("assistant_message"));

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

assert.deepStrictEqual(
messages.map((item) => item.itemId),
latest(200_000),
);
}),
);
65 changes: 61 additions & 4 deletions apps/server/src/cli/thread.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrchestratorMcpThreadTimelineItem> = [];
// 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<OrchestratorMcpThreadTimelineItem> = [];
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,
Expand Down Expand Up @@ -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) => ({
Expand Down
Loading