Skip to content

Commit 59e64ee

Browse files
darjssjuliusmarminge
darjss
and
juliusmarminge
authored
fix(server): t3_thread_read returns the newest items and fits in agent context (#33428)
Co-authored-by: Julius Marminge <[email protected]>
1 parent 4137412 commit 59e64ee

14 files changed

Lines changed: 318 additions & 73 deletions

‎apps/server/src/mcp/McpHttpServer.test.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -696,13 +696,13 @@ it.effect(
696696
...thread,
697697
limit: 100,
698698
runLimit: 50,
699-
maxCharsPerItem: 50_000,
699+
maxCharsPerItem: 20_000,
700700
});
701701
yield* callTool("t3_thread_read", { ...thread, limit: 7, runLimit: 1, maxCharsPerItem: 1 });
702702

703703
expect(calls).toEqual([
704-
{ ...thread, limit: 100, runLimit: 50, maxCharsPerItem: 50_000 },
705-
{ ...thread, limit: 100, runLimit: 50, maxCharsPerItem: 50_000 },
704+
{ ...thread, limit: 100, runLimit: 50, maxCharsPerItem: 20_000 },
705+
{ ...thread, limit: 100, runLimit: 50, maxCharsPerItem: 20_000 },
706706
{ ...thread, limit: 7, runLimit: 1, maxCharsPerItem: 1 },
707707
]);
708708
}).pipe(Effect.provide(layerRecordingTools(calls)));

‎apps/server/src/mcp/OrchestratorMcpService.activity.test.ts‎

Lines changed: 107 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -176,7 +176,8 @@ it("readThread preserves caller access errors and prefers activity-run status ov
176176
Layer.provide(
177177
Layer.mergeAll(
178178
Layer.mock(ThreadManagementService.ThreadManagementService)({
179-
getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }),
179+
getTimelinePage: () =>
180+
Effect.succeed({ items: [], totalItems: 0, hasMore: false, hasEarlier: false }),
180181
getThreadRecords: (threadId) =>
181182
callerUnavailable
182183
? Effect.fail(
@@ -328,6 +329,7 @@ it("readThread pages cut-off text without splitting an emoji across pages", asyn
328329
) as unknown as OrchestrationV2ThreadProjection["visibleTurnItems"],
329330
totalItems: rows.length,
330331
hasMore: remaining.length > options.limit,
332+
hasEarlier: false,
331333
});
332334
},
333335
getThreadRecords: () => Effect.succeed(projection),
@@ -390,6 +392,107 @@ it("readThread pages cut-off text without splitting an emoji across pages", asyn
390392
}).pipe(Effect.provide(layer), Effect.runPromise);
391393
});
392394

395+
it("readThread returns the newest items that fit the response budget", async () => {
396+
const projection = {
397+
thread: baseThread({
398+
threadId: parentThreadId,
399+
title: "Parent",
400+
instanceId: parentInstanceId,
401+
model: "gpt-5.4",
402+
}),
403+
runs: [makeRun({ id: activeRunId, ordinal: 1, status: "running" })],
404+
visibleTurnItems: [],
405+
providerTurns: [],
406+
runtimeRequests: [],
407+
messages: [],
408+
contextTransfers: [],
409+
subagents: [],
410+
updatedAt: now,
411+
} as unknown as OrchestrationV2ThreadProjection;
412+
// A long thread's newest 100 activity items, the read agents make most.
413+
const rows = Array.from({ length: 100 }, (_, position) => ({
414+
position,
415+
visibility: "local",
416+
sourceThreadId: parentThreadId,
417+
sourceItemId: TurnItemId.make(`turn-item-mcp-long-${position}`),
418+
item: {
419+
type: "reasoning",
420+
runId: activeRunId,
421+
status: "completed",
422+
title: null,
423+
text: "x".repeat(1_500),
424+
updatedAt: now,
425+
},
426+
}));
427+
const pageOptions: Array<unknown> = [];
428+
429+
const layer = layerOrchestratorService.pipe(
430+
Layer.provide(
431+
Layer.mergeAll(
432+
Layer.mock(ThreadManagementService.ThreadManagementService)({
433+
getTimelinePage: (_threadId, options) => {
434+
pageOptions.push(options);
435+
return Effect.succeed({
436+
items: rows as unknown as OrchestrationV2ThreadProjection["visibleTurnItems"],
437+
totalItems: 3_000,
438+
hasMore: false,
439+
hasEarlier: true,
440+
});
441+
},
442+
getThreadRecords: () => Effect.succeed(projection),
443+
getProjectThreadRecords: () => Effect.succeed(projection),
444+
getThreadShell: () =>
445+
Effect.succeed(projection.thread as unknown as OrchestrationV2ThreadShell),
446+
} satisfies Partial<ThreadManagementService.ThreadManagementService["Service"]>),
447+
ServerSettings.layerTest(),
448+
Layer.mock(ProviderRegistry.ProviderRegistry)({
449+
getProviders: Effect.succeed([]),
450+
} satisfies Partial<ProviderRegistry.ProviderRegistry["Service"]>),
451+
Layer.mock(ProjectService.ProjectService)({}),
452+
Layer.mock(SecretRequests.SecretRequests)({}),
453+
Layer.mock(RemoteDelegation.RemoteDelegation)({}),
454+
Layer.mock(ScheduledTaskService.ScheduledTaskService)({
455+
list: () => Effect.succeed({ tasks: [] }),
456+
} satisfies Partial<ScheduledTaskService.ScheduledTaskService["Service"]>),
457+
Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
458+
list: () => Effect.succeed([]),
459+
} satisfies Partial<ProviderAdapterRegistry.ProviderAdapterRegistryV2["Service"]>),
460+
NodeCrypto.layer,
461+
),
462+
),
463+
);
464+
465+
await Effect.gen(function* () {
466+
const service = yield* OrchestratorMcpService.OrchestratorMcpService;
467+
const read = {
468+
threadId: parentThreadId,
469+
view: "activity" as const,
470+
limit: 100,
471+
maxCharsPerItem: 1_500,
472+
};
473+
const result = yield* service.readThread(makeScope(), read);
474+
// Linked environments and the CLI read as clients without a cursor and follow only hasMore.
475+
const first = yield* service.readThread({ ...makeScope(), thread: undefined }, read);
476+
const fromStart = yield* service.readThread(makeScope(), { ...read, afterPosition: -1 });
477+
478+
expect(pageOptions).toEqual([
479+
{ beforePosition: Number.POSITIVE_INFINITY, limit: 100, view: "activity" },
480+
{ limit: 100, view: "activity" },
481+
{ afterPosition: -1, limit: 100, view: "activity" },
482+
]);
483+
// Claude Code saves results over ~50k chars of this JSON to a file instead of showing them.
484+
expect(JSON.stringify(result).length).toBeLessThan(31_000);
485+
expect(result.items.length).toBeLessThan(100);
486+
expect(result.items.at(-1)?.position).toBe(99);
487+
expect(result.earlierPosition).toBe(result.items[0]?.position);
488+
expect(result.hasMore).toBe(false);
489+
expect(first.items[0]?.position).toBe(0);
490+
expect(first.hasMore).toBe(true);
491+
expect(first.nextPosition).toBe(first.items.at(-1)?.position);
492+
expect(fromStart.items[0]?.position).toBe(0);
493+
}).pipe(Effect.provide(layer), Effect.runPromise);
494+
});
495+
393496
it("taskStatus returns task.providerInstanceId rather than the driver kind", async () => {
394497
const parentProjection = {
395498
thread: baseThread({
@@ -474,7 +577,8 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy
474577
Layer.provide(
475578
Layer.mergeAll(
476579
Layer.mock(ThreadManagementService.ThreadManagementService)({
477-
getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }),
580+
getTimelinePage: () =>
581+
Effect.succeed({ items: [], totalItems: 0, hasMore: false, hasEarlier: false }),
478582
getThreadRecords: (threadId) => {
479583
if (threadId === parentThreadId) return Effect.succeed(parentProjection);
480584
if (threadId === childThreadId) return Effect.succeed(childProjection);
@@ -595,6 +699,7 @@ it("readThread and sendToThread reach threads in other projects", async () => {
595699
items: foreignProjection(threadId).visibleTurnItems,
596700
totalItems: 1,
597701
hasMore: false,
702+
hasEarlier: false,
598703
}),
599704
getProjectThreadRecords: (input) =>
600705
input.threadId === foreignThreadId && input.projectId === foreignProjectId

‎apps/server/src/mcp/OrchestratorMcpService.ts‎

Lines changed: 52 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,10 @@ const DEFAULT_THREAD_LIST_LIMIT = 10;
146146
const DEFAULT_THREAD_READ_LIMIT = 10;
147147
const DEFAULT_THREAD_RUN_LIMIT = 10;
148148
const DEFAULT_THREAD_ITEM_MAX_CHARS = 500;
149+
// Claude Code saves MCP results over ~25k tokens to a file instead of showing them. Thread JSON
150+
// runs about 2 chars per token, so a read stops adding items here and reports the rest as more.
151+
const THREAD_READ_BUDGET_CHARS = 30_000;
152+
const encodeJsonText = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
149153

150154
interface ResolvedTarget {
151155
readonly modelSelection: ModelSelection;
@@ -3061,20 +3065,25 @@ const make = Effect.gen(function* () {
30613065
readThread: Effect.fn("OrchestratorMcpService.readThread")(function* (scope, input) {
30623066
const { parent, target, shell } = yield* loadReadableThread(scope, input.threadId);
30633067
const view = input.view ?? "messages";
3064-
const afterPosition = input.afterPosition ?? -1;
30653068
const limit = input.limit ?? DEFAULT_THREAD_READ_LIMIT;
30663069
const maxChars = input.maxCharsPerItem ?? DEFAULT_THREAD_ITEM_MAX_CHARS;
3070+
// Client callers (the CLI, older linked environments) page forward from a cursorless read.
3071+
const newestFirst =
3072+
input.beforePosition !== undefined ||
3073+
(input.afterPosition === undefined && scope.thread !== undefined);
30673074
const timeline = yield* threadManagement
30683075
.getTimelinePage(input.threadId, {
3069-
afterPosition,
3076+
...(input.afterPosition === undefined ? {} : { afterPosition: input.afterPosition }),
3077+
...(newestFirst
3078+
? { beforePosition: input.beforePosition ?? Number.POSITIVE_INFINITY }
3079+
: {}),
30703080
limit,
30713081
view,
30723082
...(input.itemId === undefined ? {} : { itemId: input.itemId }),
30733083
})
30743084
.pipe(Effect.mapError(threadManagementFailure));
3075-
const page = timeline.items;
30763085
const messageIdsByThread = new Map<ThreadId, Array<MessageId>>();
3077-
for (const row of page) {
3086+
for (const row of timeline.items) {
30783087
if (row.item.type !== "user_message" && row.item.type !== "assistant_message") continue;
30793088
const ids = messageIdsByThread.get(row.sourceThreadId) ?? [];
30803089
ids.push(row.item.messageId);
@@ -3090,6 +3099,39 @@ const make = Effect.gen(function* () {
30903099
{ concurrency: 1 },
30913100
);
30923101
const messagesByThreadId = new Map(sourceMessages);
3102+
const thread = threadDetail(
3103+
scope,
3104+
target,
3105+
timeline.totalItems,
3106+
shell,
3107+
yield* Clock.currentTimeMillis,
3108+
input.afterPosition === undefined &&
3109+
input.beforePosition === undefined &&
3110+
input.itemId === undefined,
3111+
);
3112+
const recentRuns = target.runs
3113+
.toSorted((left, right) => right.ordinal - left.ordinal)
3114+
.slice(0, input.runLimit ?? DEFAULT_THREAD_RUN_LIMIT)
3115+
.map((run) => threadRun(target, run));
3116+
const allItems = timeline.items.map((row) =>
3117+
timelineItem({
3118+
row,
3119+
maxChars,
3120+
messagesByThreadId,
3121+
...(input.itemId === undefined ? {} : { textOffset: input.textOffset ?? 0 }),
3122+
}),
3123+
);
3124+
// Keep the end of the page the caller is reading toward, and at least one item.
3125+
let size = encodeJsonText({ thread, recentRuns }).length;
3126+
let kept = 0;
3127+
for (const item of newestFirst ? allItems.toReversed() : allItems) {
3128+
size += encodeJsonText(item).length + 1;
3129+
if (kept > 0 && size > THREAD_READ_BUDGET_CHARS) break;
3130+
kept++;
3131+
}
3132+
const start = newestFirst ? allItems.length - kept : 0;
3133+
const items = allItems.slice(start, start + kept);
3134+
const page = timeline.items.slice(start, start + kept);
30933135
const returnedRunIds = new Set(
30943136
page.filter((row) => row.sourceThreadId === target.thread.id).map((row) => row.item.runId),
30953137
);
@@ -3141,28 +3183,12 @@ const make = Effect.gen(function* () {
31413183
}
31423184
}
31433185
return {
3144-
thread: threadDetail(
3145-
scope,
3146-
target,
3147-
timeline.totalItems,
3148-
shell,
3149-
yield* Clock.currentTimeMillis,
3150-
input.afterPosition === undefined && input.itemId === undefined,
3151-
),
3152-
recentRuns: target.runs
3153-
.toSorted((left, right) => right.ordinal - left.ordinal)
3154-
.slice(0, input.runLimit ?? DEFAULT_THREAD_RUN_LIMIT)
3155-
.map((run) => threadRun(target, run)),
3156-
items: page.map((row) =>
3157-
timelineItem({
3158-
row,
3159-
maxChars,
3160-
messagesByThreadId,
3161-
...(input.itemId === undefined ? {} : { textOffset: input.textOffset ?? 0 }),
3162-
}),
3163-
),
3164-
nextPosition: page.at(-1)?.position ?? null,
3165-
hasMore: timeline.hasMore,
3186+
thread,
3187+
recentRuns,
3188+
items,
3189+
nextPosition: items.at(-1)?.position ?? null,
3190+
hasMore: timeline.hasMore || (!newestFirst && kept < allItems.length),
3191+
earlierPosition: timeline.hasEarlier || start > 0 ? (items[0]?.position ?? null) : null,
31663192
} satisfies OrchestratorMcpThreadReadResult;
31673193
}),
31683194
sendToThread: Effect.fn("OrchestratorMcpService.sendToThread")(function* (scope, input) {

‎apps/server/src/mcp/toolkits/core.test.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1114,7 +1114,8 @@ it.effect("discovers workflow stop targets and requires orchestration capability
11141114
getThreadShell: (id) => Effect.succeed(McpToolAccessTestkit.liveThreadShell(id)),
11151115
getProjectThreadRecords: () =>
11161116
Effect.succeed(McpToolAccessTestkit.idleThreadProjection(shell)),
1117-
getTimelinePage: () => Effect.succeed({ items: [row], totalItems: 1, hasMore: false }),
1117+
getTimelinePage: () =>
1118+
Effect.succeed({ items: [row], totalItems: 1, hasMore: false, hasEarlier: false }),
11181119
stopWorkflow: (input) =>
11191120
Effect.sync(() => {
11201121
stopped.push(input);

‎apps/server/src/mcp/toolkits/orchestrator/tools.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ it("states t3_thread_read page size limits in the parameter descriptions", () =>
1111
const schema = Tool.getJsonSchema(OrchestratorToolkit.tools.t3_thread_read) as {
1212
readonly properties: Readonly<Record<string, { readonly description?: string }>>;
1313
};
14-
const limits = { limit: "100", runLimit: "50", maxCharsPerItem: "50000" };
14+
const limits = { limit: "100", runLimit: "50", maxCharsPerItem: "20000" };
1515
for (const [name, maximum] of Object.entries(limits)) {
1616
expect(schema.properties[name]?.description, name).toContain(maximum);
1717
}

‎apps/server/src/mcp/toolkits/orchestrator/tools.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -245,7 +245,7 @@ const ThreadListTool = Tool.make("t3_thread_list", {
245245

246246
const ThreadReadTool = Tool.make("t3_thread_read", {
247247
description:
248-
"Read any thread by id. `messages` (default): user and assistant messages and plans. `activity`: all timeline items. Keep reads small: low `limit` and `maxCharsPerItem`. Page with `afterPosition=nextPosition`. Cut-off text: `itemId` with `textOffset=nextTextOffset`. Reading a child's final result marks it delivered. In your reply, name it as its `link` verbatim.",
248+
"Read any thread by id. View `messages` (default) or `activity` (all items). Newest first; older: `beforePosition=earlierPosition`; from the start: `afterPosition=-1`, then `nextPosition`. Cut-off text: `itemId`, `textOffset=nextTextOffset`. Reading a child's final result marks it delivered. In your reply, name it as its `link` verbatim.",
249249
parameters: OrchestratorMcpThreadReadInput,
250250
success: OrchestratorMcpThreadReadResult,
251251
failure: OrchestratorMcpFailure,

‎apps/server/src/orchestration-v2/ContextHandoffDelivery.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
4444
(handoff) =>
4545
`Context handoff (${handoff.strategy === "fork_delta_summary" ? "merge_back / fork_delta_summary" : handoff.strategy}):\n${
4646
handoff.history?.coverage ??
47-
`From thread ${handoff.threadId}, runs ${handoff.coveredRunOrdinals.from}-${handoff.coveredRunOrdinals.to}. Recover history with t3_thread_read, view=activity; paginate with afterPosition, and use itemId/textOffset for long items.`
47+
`From thread ${handoff.threadId}, runs ${handoff.coveredRunOrdinals.from}-${handoff.coveredRunOrdinals.to}. Recover history with t3_thread_read, view=activity, afterPosition=-1; paginate with afterPosition=nextPosition, and use itemId/textOffset for long items.`
4848
}`,
4949
)
5050
.join("\n");

‎apps/server/src/orchestration-v2/ProjectionStore.test.ts‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6104,6 +6104,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
61046104
for (const view of ["messages", "activity"] as const) {
61056105
for (const afterPosition of [-1, Math.floor(totalItems / 3), totalItems - 5])
61066106
yield* windowed(timeline({ view, afterPosition, limit: 20 }));
6107+
for (const beforePosition of [
6108+
Number.POSITIVE_INFINITY,
6109+
Math.floor(totalItems / 3),
6110+
totalItems - 5,
6111+
])
6112+
yield* windowed(timeline({ view, beforePosition, limit: 20 }));
6113+
yield* windowed(timeline({ view, afterPosition: 3, beforePosition: 40, limit: 20 }));
61076114
}
61086115
for (const itemId of [newest.sourceItemId, ...tiedWithHidden])
61096116
yield* windowed(timeline({ itemId, limit: 1 }));
@@ -9834,6 +9841,23 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
98349841
});
98359842
assert.deepEqual(followingPage.items, targetAfterRollback.visibleTurnItems.slice(2));
98369843
assert.isFalse(followingPage.hasMore);
9844+
assert.isTrue(followingPage.hasEarlier);
9845+
const newestPage = yield* projectionStore.getTimelinePage(targetThreadId, {
9846+
view: "activity",
9847+
limit: 2,
9848+
beforePosition: Number.POSITIVE_INFINITY,
9849+
});
9850+
assert.deepEqual(newestPage.items, targetAfterRollback.visibleTurnItems.slice(-2));
9851+
assert.isFalse(newestPage.hasMore);
9852+
assert.isTrue(newestPage.hasEarlier);
9853+
const earlierPage = yield* projectionStore.getTimelinePage(targetThreadId, {
9854+
view: "activity",
9855+
limit: 10,
9856+
beforePosition: newestPage.items[0]!.position,
9857+
});
9858+
assert.deepEqual(earlierPage.items, targetAfterRollback.visibleTurnItems.slice(0, -2));
9859+
assert.isTrue(earlierPage.hasMore);
9860+
assert.isFalse(earlierPage.hasEarlier);
98379861
const tracedTimelinePage = (limit: number) =>
98389862
Effect.gen(function* () {
98399863
const { statements, tracer } = traceSqlStatements();

0 commit comments

Comments
 (0)