From 8f0caecf4344d80beb220107583c6b2c75aa6c18 Mon Sep 17 00:00:00 2001 From: Fagner Sales Date: Sun, 8 Nov 2026 17:55:16 -0300 Subject: [PATCH 1/4] perf(cli): read long threads from the end instead of paging through them t3_thread_read only paged forward, and every page builds the thread's whole visible index on the server. `t3 thread show` on old CLIs and the result read behind `send/start --wait` (and a linked environment's task result) paged through the entire thread, so a 20k-item thread took about a minute. t3_thread_read now takes `beforePosition` (the newest `limit` matches before it) and returns `previousPosition`. The CLI and RemoteDelegation read backward from the newest item and stop once they have what they need, falling back to the forward read when an older server rejects the field. The CLI now reports a server's invalid-params answer as a tool failure instead of "could not reach the server". Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cli/thread.test.ts | 162 +++++++++++++----- apps/server/src/cli/thread.ts | 96 +++++++++-- apps/server/src/mcp/AcpMcpStdioBridge.ts | 12 +- .../OrchestratorMcpService.activity.test.ts | 8 +- apps/server/src/mcp/OrchestratorMcpService.ts | 2 + apps/server/src/mcp/toolkits/core.test.ts | 3 +- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../orchestration-v2/ProjectionStore.test.ts | 16 ++ .../src/orchestration-v2/ProjectionStore.ts | 29 +++- apps/server/src/peer/RemoteDelegation.test.ts | 99 ++++++----- apps/server/src/peer/RemoteDelegation.ts | 48 ++++-- .../src/provider/agentContextBudget.json | 2 +- .../orchestrator-mcp-server.md | 6 +- packages/contracts/src/orchestratorMcp.ts | 8 + 14 files changed, 364 insertions(+), 129 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index f182f708c4cfa..1581e4787b5c4 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -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"; @@ -20,7 +21,9 @@ import { persistServerRuntimeState, } from "../serverRuntimeState.ts"; import { + makeThreadTools, readLatestMessages, + ThreadToolFailedError, threadCommand, type ThreadTools, titleFromMessage, @@ -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) => { +/** + * 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, 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 = []; 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; } @@ -234,57 +270,95 @@ const timelineTools = (types: ReadonlyArray) => { }; const tail = (tools: number) => [...Array(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("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(), - ]); +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); + const { messages } = yield* readLatestMessages(tools, threadId, 10); - assert.deepStrictEqual( - messages.map((item) => item.itemId), - latest(10), - ); - assert.lengthOf(messages, 5); - }), + 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")); +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(200_000).fill("assistant_message"), + server, + ); - const { messages } = yield* readLatestMessages(tools, threadId, 200_000); + const { messages } = yield* readLatestMessages(tools, threadId, 200_000); - assert.deepStrictEqual( - messages.map((item) => item.itemId), - latest(200_000), - ); + assert.deepStrictEqual( + messages.map((item) => item.itemId), + latest(200_000), + ); + }), +); + +it.effect("a server's rejected arguments are a tool failure, so show can fall back", () => + Effect.gen(function* () { + 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" })); + + assert.strictEqual(yield* failWith(McpBridge.JSON_RPC_INVALID_PARAMS), "ThreadToolFailedError"); + assert.strictEqual(yield* failWith(undefined), "ThreadServerUnreachableError"); }), ); diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 80dfad018e158..b25b729d38343 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -6,6 +6,7 @@ import { OrchestratorMcpThreadWaitResult, ProjectId, RunId, + THREAD_READ_NEWEST_POSITION, ThreadId, TrimmedNonEmptyString, type OrchestratorMcpThreadListItem, @@ -28,7 +29,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"; @@ -172,7 +177,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, @@ -186,7 +191,13 @@ const makeThreadTools = (session: McpToolSession, origin: string) => { ): Effect.fn.Return { 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) }); @@ -337,11 +348,46 @@ 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) => boolean, +) { + const read = (beforePosition: number) => + tools.call( + "t3_thread_read", + { threadId, view, limit: nextLimit(), runLimit: 1, beforePosition }, + OrchestratorMcpThreadReadResult, + ); + const newest = yield* read(THREAD_READ_NEWEST_POSITION).pipe( + Effect.catchTags({ ThreadToolFailedError: () => Effect.succeed(null) }), + ); + if (newest === null || newest.previousPosition === undefined) return undefined; + const pages = [newest.items]; + let previous = done(newest.items) ? null : newest.previousPosition; + while (previous !== null) { + const page = yield* read(previous); + pages.push(page.items); + previous = done(page.items) ? null : (page.previousPosition ?? null); + } + 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, @@ -377,9 +423,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; @@ -393,6 +437,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, @@ -422,12 +486,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); }); diff --git a/apps/server/src/mcp/AcpMcpStdioBridge.ts b/apps/server/src/mcp/AcpMcpStdioBridge.ts index 485f85e1837c1..e9e8277760ab8 100644 --- a/apps/server/src/mcp/AcpMcpStdioBridge.ts +++ b/apps/server/src/mcp/AcpMcpStdioBridge.ts @@ -13,9 +13,17 @@ import { discardResponseBody, responsePayloads } from "@t3tools/shared/mcpRespon export class AcpMcpBridgeError extends Schema.TaggedError()( "AcpMcpBridgeError", - { message: Schema.String, cause: Schema.optional(Schema.Defect()) }, + { + message: Schema.String, + cause: Schema.optional(Schema.Defect()), + /** The JSON-RPC error code, when the server answered with an error. */ + code: Schema.optional(Schema.Number), + }, ) {} +/** JSON-RPC's code for arguments the server's schema rejects. */ +export const JSON_RPC_INVALID_PARAMS = -32602; + const isAcpMcpBridgeError = Schema.is(AcpMcpBridgeError); const bridgeError = (cause: unknown): AcpMcpBridgeError => @@ -127,8 +135,10 @@ export const mcpToolSession = ( const responses = yield* send({ jsonrpc: "2.0", id, method, params }); const envelope = asEnvelope(responses.find((entry) => asEnvelope(entry)?.id === id)); if (envelope === null || envelope.error !== undefined) { + const code = (envelope?.error as { readonly code?: unknown } | null | undefined)?.code; return yield* new AcpMcpBridgeError({ message: `T3 Code MCP ${method} failed${envelope?.error === undefined ? "." : `: ${JSON.stringify(envelope.error)}`}`, + ...(typeof code === "number" ? { code } : {}), }); } return envelope; diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index c35b605dca4fb..784785480d2d3 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -147,7 +147,8 @@ it("readThread preserves caller access errors and prefers activity-run status ov Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ - getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }), + getTimelinePage: () => + Effect.succeed({ items: [], totalItems: 0, hasMore: false, hasEarlier: false }), getThreadRecords: (threadId) => callerUnavailable ? Effect.fail( @@ -252,6 +253,7 @@ it("readThread pages cut-off text without splitting an emoji across pages", asyn items: [row] as unknown as OrchestrationV2ThreadProjection["visibleTurnItems"], totalItems: 1, hasMore: false, + hasEarlier: false, }), getThreadRecords: () => Effect.succeed(projection), getProjectThreadRecords: () => Effect.succeed(projection), @@ -384,7 +386,8 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ - getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }), + getTimelinePage: () => + Effect.succeed({ items: [], totalItems: 0, hasMore: false, hasEarlier: false }), getThreadRecords: (threadId) => { if (threadId === parentThreadId) return Effect.succeed(parentProjection); if (threadId === childThreadId) return Effect.succeed(childProjection); @@ -503,6 +506,7 @@ it("readThread and sendToThread reach threads in other projects", async () => { items: foreignProjection(threadId).visibleTurnItems, totalItems: 1, hasMore: false, + hasEarlier: false, }), getProjectThreadRecords: (input) => input.threadId === foreignThreadId && input.projectId === foreignProjectId diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 9d15408483a3e..340db3f55efce 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -3041,6 +3041,7 @@ const make = Effect.gen(function* () { afterPosition, limit, view, + ...(input.beforePosition === undefined ? {} : { beforePosition: input.beforePosition }), ...(input.itemId === undefined ? {} : { itemId: input.itemId }), }) .pipe(Effect.mapError(threadManagementFailure)); @@ -3138,6 +3139,7 @@ const make = Effect.gen(function* () { ), nextPosition: page.at(-1)?.position ?? null, hasMore: timeline.hasMore, + previousPosition: timeline.hasEarlier ? (page[0]?.position ?? null) : null, } satisfies OrchestratorMcpThreadReadResult; }), sendToThread: Effect.fn("OrchestratorMcpService.sendToThread")(function* (scope, input) { diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index fd07107b08bc6..e874f8d6ee714 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -968,7 +968,8 @@ it.effect("discovers workflow stop targets and requires orchestration capability getThreadShell: (id) => Effect.succeed(McpToolAccessTestkit.liveThreadShell(id)), getProjectThreadRecords: () => Effect.succeed(McpToolAccessTestkit.idleThreadProjection(shell)), - getTimelinePage: () => Effect.succeed({ items: [row], totalItems: 1, hasMore: false }), + getTimelinePage: () => + Effect.succeed({ items: [row], totalItems: 1, hasMore: false, hasEarlier: false }), stopWorkflow: (input) => Effect.sync(() => { stopped.push(input); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index f0affbf6dec8c..1a0aeaaf843de 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -238,7 +238,7 @@ const ThreadListTool = Tool.make("t3_thread_list", { const ThreadReadTool = Tool.make("t3_thread_read", { description: - "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.", + "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`, or back from the newest with `beforePosition=previousPosition`. 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.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 1ca7eb67036fa..307ae95d840d2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -7594,6 +7594,22 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); assert.deepEqual(followingPage.items, targetAfterRollback.visibleTurnItems.slice(2)); assert.isFalse(followingPage.hasMore); + assert.isTrue(followingPage.hasEarlier); + const newestPage = yield* projectionStore.getTimelinePage(targetThreadId, { + view: "activity", + limit: 2, + beforePosition: Number.MAX_SAFE_INTEGER, + }); + assert.deepEqual(newestPage.items, targetAfterRollback.visibleTurnItems.slice(-2)); + assert.isFalse(newestPage.hasMore); + assert.isTrue(newestPage.hasEarlier); + const earlierPage = yield* projectionStore.getTimelinePage(targetThreadId, { + view: "activity", + limit: 10, + beforePosition: newestPage.items[0]!.position, + }); + assert.deepEqual(earlierPage.items, targetAfterRollback.visibleTurnItems.slice(0, -2)); + assert.isFalse(earlierPage.hasEarlier); const tracedTimelinePage = (limit: number) => Effect.gen(function* () { const { statements, tracer } = traceSqlStatements(); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ba87697d60f03..65319269e8f07 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -397,6 +397,8 @@ export type ProjectionRecords = Pick< export interface ProjectionTimelinePageOptions { readonly afterPosition?: number; + /** Selects the last `limit` matches before this position instead of the first. */ + readonly beforePosition?: number; readonly itemId?: TurnItemId; readonly attachmentId?: string; readonly view?: "messages" | "activity"; @@ -405,7 +407,9 @@ export interface ProjectionTimelinePageOptions { export interface ProjectionTimelinePage { readonly items: ReadonlyArray; readonly totalItems: number; + /** Matches in the view after and before the page, beyond any position bounds. */ readonly hasMore: boolean; + readonly hasEarlier: boolean; } export interface ShellSnapshotOptions { @@ -5997,21 +6001,29 @@ export const layer: Layer.Layer = (row) => `${row.thread_id}:${row.turn_item_id}`, ), ); - const matching = index.visible + const inView = index.visible .map((row, position) => ({ ...row, position })) .filter((row) => options.itemId === undefined - ? row.position > (options.afterPosition ?? -1) && - (options.view === "activity" || - ["user_message", "assistant_message", "proposed_plan"].includes( - row.item.type, - )) + ? options.view === "activity" || + ["user_message", "assistant_message", "proposed_plan"].includes(row.item.type) : row.sourceItemId === options.itemId, ) .filter( (row) => attachmentOwners?.has(`${row.sourceThreadId}:${row.sourceItemId}`) ?? true, ); - const page = matching.slice(0, options.limit); + const matching = + options.itemId === undefined + ? inView.filter( + (row) => + row.position > (options.afterPosition ?? -1) && + row.position < (options.beforePosition ?? Number.POSITIVE_INFINITY), + ) + : inView; + const page = + options.beforePosition === undefined + ? matching.slice(0, options.limit) + : matching.slice(-options.limit); const sources = yield* encodeTimelineSources( page .filter((row) => row.synthetic === undefined) @@ -6063,7 +6075,8 @@ export const layer: Layer.Layer = return { items, totalItems: index.visible.length, - hasMore: matching.length > page.length, + hasMore: (page.at(-1)?.position ?? Infinity) < (inView.at(-1)?.position ?? -1), + hasEarlier: (page[0]?.position ?? -1) > (inView[0]?.position ?? Infinity), }; }), ) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 00b7b5d6f6963..f79070ec31e2c 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -120,6 +120,8 @@ interface BoxThread { readonly finished: Deferred.Deferred; /** While set, reading the thread there waits: completed once a read starts, released by `releaseRead`. */ readonly holdRead: Ref.Ref; + /** Reads there refuse `beforePosition`, as a box from before it does. */ + readonly forwardReadsOnly: Ref.Ref; readonly readStarted: Deferred.Deferred; readonly releaseRead: Deferred.Deferred; readonly holdFirstStop: Ref.Ref; @@ -156,6 +158,7 @@ const makeBoxThread = Effect.gen(function* () { interruptedOnce: yield* Deferred.make(), finished: yield* Deferred.make(), holdRead: yield* Ref.make(false), + forwardReadsOnly: yield* Ref.make(false), readStarted: yield* Deferred.make(), releaseRead: yield* Deferred.make(), holdFirstStop: yield* Ref.make(false), @@ -306,6 +309,9 @@ const boxToolkitLayer = (thread: BoxThread) => { yield* Deferred.succeed(thread.readStarted, undefined); yield* Deferred.await(thread.releaseRead); } + if (options.beforePosition !== undefined && (yield* Ref.get(thread.forwardReadsOnly))) { + return yield* Effect.die("Unknown key beforePosition"); + } const reply = yield* Ref.get(thread.reply); const later = yield* Ref.get(thread.laterReply); const error = yield* Ref.get(thread.error); @@ -352,9 +358,13 @@ const boxToolkitLayer = (thread: BoxThread) => { (entry) => options.view === "activity" || entry.type === "assistant_message", ) : replies.filter((entry) => entry.id === options.itemId); - const matching = visible - .filter((entry) => entry.position > (options.afterPosition ?? -1)) - .slice(0, 1); + const inRange = visible.filter( + (entry) => + entry.position > (options.afterPosition ?? -1) && + entry.position < (options.beforePosition ?? Number.POSITIVE_INFINITY), + ); + const matching = + options.beforePosition === undefined ? inRange.slice(0, 1) : inRange.slice(-1); return { items: matching.map(({ id, runId: itemRunId, text, position, type }) => ({ position, @@ -391,6 +401,8 @@ const boxToolkitLayer = (thread: BoxThread) => { options.itemId === undefined && matching.length > 0 && visible.some((entry) => entry.position > matching.at(-1)!.position), + hasEarlier: + matching.length > 0 && visible.some((entry) => entry.position < matching[0]!.position), }; }), interruptThread: (input) => @@ -1234,47 +1246,50 @@ it.effect("fails the task when the box revokes the link, and cancel interrupts i ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); -it.effect("a task's result is its own run's whole last reply, not a later run's", () => - Effect.scoped( - Effect.gen(function* () { - const thread = yield* makeBoxThread; - const b = yield* serveBox(thread); - const offers = yield* SubscriptionRef.make< - ReadonlyArray - >([]); - const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); - const a = yield* makeLaptop(database, offers); - yield* linkTo(a.links, b, "full-access"); - yield* seedParent(a); +it.effect.each([false, true])( + "a task's result is its own run's whole last reply, not a later run's (forward reads only: %s)", + (forwardReadsOnly) => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + yield* Ref.set(thread.forwardReadsOnly, forwardReadsOnly); + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const a = yield* makeLaptop(database, offers); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); - const result = yield* a.call("delegate_task", { - task: "Review the diff.", - role: "review", - target: { environmentId: box.environmentId }, - clientRequestId: "review-on-box", - }); - expect(result.isError, JSON.stringify(result.content)).toBe(false); - const { taskId } = result.structuredContent as { taskId: NodeId }; - // The role reaches the thread there, as it does a child here. - expect(yield* Ref.get(thread.message)).toBe( - "Act as the review sub-agent for this task.\n\nReview the diff.", - ); + const result = yield* a.call("delegate_task", { + task: "Review the diff.", + role: "review", + target: { environmentId: box.environmentId }, + clientRequestId: "review-on-box", + }); + expect(result.isError, JSON.stringify(result.content)).toBe(false); + const { taskId } = result.structuredContent as { taskId: NodeId }; + // The role reaches the thread there, as it does a child here. + expect(yield* Ref.get(thread.message)).toBe( + "Act as the review sub-agent for this task.\n\nReview the diff.", + ); - // A reply longer than one read, then a later run's reply on the same thread. - const long = "x".repeat(45_000); - yield* Ref.set(thread.reply, long); - yield* Ref.set(thread.laterReply, "Someone else asked a follow-up."); - yield* Ref.set(thread.status, "completed"); - yield* Deferred.succeed(thread.finished, undefined); - const task = yield* taskResult(a, taskId); - expect(task?.status).toBe("completed"); - expect(task?.result).toBe(long); - // Every wait was for the run the launch started. - const waits = yield* Ref.get(thread.waitedFor); - expect(waits.length).toBeGreaterThan(0); - expect(waits.every((runId) => runId === boxRunId)).toBe(true); - }), - ).pipe(Effect.provide(NodeHttpServer.layerTest)), + // A reply longer than one read, then a later run's reply on the same thread. + const long = "x".repeat(45_000); + yield* Ref.set(thread.reply, long); + yield* Ref.set(thread.laterReply, "Someone else asked a follow-up."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + const task = yield* taskResult(a, taskId); + expect(task?.status).toBe("completed"); + expect(task?.result).toBe(long); + // Every wait was for the run the launch started. + const waits = yield* Ref.get(thread.waitedFor); + expect(waits.length).toBeGreaterThan(0); + expect(waits.every((runId) => runId === boxRunId)).toBe(true); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); it.effect("a task whose thread there waits on its own task ends with the answer after it", () => diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 89917026ff8f7..0a796780fcee9 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -12,6 +12,7 @@ import { type ProviderInstanceId, ProviderDriverKind, type RunId, + THREAD_READ_NEWEST_POSITION, type ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -444,26 +445,43 @@ const make = Effect.gen(function* () { ) { let reply: OrchestratorMcpThreadTimelineItem | undefined; let error: OrchestratorMcpThreadTimelineItem | undefined; - let afterPosition: number | undefined; - while (true) { - const read = yield* forwarding.readThread(scope, remote.environmentId, { + const isReply = (item: OrchestratorMcpThreadTimelineItem) => + item.type === "assistant_message" && (runId === null || item.runId === runId); + const isError = (item: OrchestratorMcpThreadTimelineItem) => + failed && item.type === "error" && (runId === null || item.runId === runId); + const read = (position: { + readonly afterPosition?: number; + readonly beforePosition?: number; + }) => + forwarding.readThread(scope, remote.environmentId, { threadId: remote.threadId, view: failed ? "activity" : "messages", limit: 100, - ...(afterPosition === undefined ? {} : { afterPosition }), + ...position, }); - reply = - read.items.findLast( - (item) => item.type === "assistant_message" && (runId === null || item.runId === runId), - ) ?? reply; - if (failed) { - error = - read.items.findLast( - (item) => item.type === "error" && (runId === null || item.runId === runId), - ) ?? error; + // A peer that predates `beforePosition` rejects it and is read forward instead; + // a forward read fails again on any other cause, as it did before. + const newest = yield* read({ beforePosition: THREAD_READ_NEWEST_POSITION }).pipe( + Effect.orElseSucceed(() => null), + ); + if (newest !== null && newest.previousPosition !== undefined) { + let page = newest; + while (true) { + reply ??= page.items.findLast(isReply); + error ??= page.items.findLast(isError); + const previous = page.previousPosition ?? null; + if ((failed ? error : reply) !== undefined || previous === null) break; + page = yield* read({ beforePosition: previous }); + } + } else { + let afterPosition: number | undefined; + while (true) { + const page = yield* read(afterPosition === undefined ? {} : { afterPosition }); + reply = page.items.findLast(isReply) ?? reply; + error = page.items.findLast(isError) ?? error; + if (!page.hasMore || page.nextPosition === null) break; + afterPosition = page.nextPosition; } - if (!read.hasMore || read.nextPosition === null) break; - afterPosition = read.nextPosition; } reply = error ?? reply; if (reply === undefined) return undefined; diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 6cbe089533d76..cff372343ae45 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5350, "codexAdditionalContextBytes": 7080, - "mcpToolListJsonBytes": 48116 + "mcpToolListJsonBytes": 48324 } diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 2f6ba5363703a..e399bf59f867e 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -394,7 +394,11 @@ default `messages` view returns user messages, assistant messages, and proposed plans. The `activity` view also returns summarized tool, reasoning, checkpoint, handoff, and runtime-request items. Large item text is bounded and reports whether it was truncated. `afterPosition` and -`nextPosition` support incremental reads. +`nextPosition` support incremental reads; `beforePosition` and +`previousPosition` read back from the newest items, so a caller that wants the +end of a long thread does not page through all of it. Every page still builds +the thread's whole visible index, so each call costs time in proportion to the +thread's length: read only the pages you need. Thread and message results include required `createdBy` and `creationSource` provenance. MCP-created threads and user-role messages use `createdBy: "agent"` diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index c73bbd48e26f5..dc02afb2e3129 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -410,6 +410,9 @@ export const OrchestratorMcpThreadReadInput = Schema.Struct({ textOffset: Schema.optional(NonNegativeInt), view: Schema.optional(Schema.Literals(["messages", "activity"])), afterPosition: Schema.optional(NonNegativeInt), + beforePosition: Schema.optional(NonNegativeInt).annotate({ + description: "Read the `limit` items just before this position; `itemCount` reads the latest.", + }), limit: Schema.optional(clampedPositiveInt(100)).annotate({ description: "Max items, 1 to 100.", }), @@ -422,6 +425,9 @@ export const OrchestratorMcpThreadReadInput = Schema.Struct({ }); export type OrchestratorMcpThreadReadInput = typeof OrchestratorMcpThreadReadInput.Type; +/** A `beforePosition` past every item, so the read starts at the newest. */ +export const THREAD_READ_NEWEST_POSITION = Number.MAX_SAFE_INTEGER; + export const OrchestratorMcpThreadDetail = Schema.Struct({ threadId: ThreadId, projectId: ProjectId, @@ -495,6 +501,8 @@ export const OrchestratorMcpThreadReadResult = Schema.Struct({ items: Schema.Array(OrchestratorMcpThreadTimelineItem), nextPosition: Schema.NullOr(NonNegativeInt), hasMore: Schema.Boolean, + /** Pass as `beforePosition` for the page before this one; null when none is left. Older servers omit it. */ + previousPosition: Schema.optionalKey(Schema.NullOr(NonNegativeInt)), }); export type OrchestratorMcpThreadReadResult = typeof OrchestratorMcpThreadReadResult.Type; From 5780879ab1b3d8665dd5d49aa1303bb95af2db9b Mon Sep 17 00:00:00 2001 From: Fagner Sales Date: Sun, 8 Nov 2026 18:11:10 -0300 Subject: [PATCH 2/4] perf(cli): leave linked-environment result reads to their own PR The RemoteDelegation.lastReply change overlaps pingdotgg/t3code#26664, which already reworks that read, so this PR keeps to the CLI's reads. The newest-position constant moves back into the CLI, its only user. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cli/thread.ts | 5 +- apps/server/src/peer/RemoteDelegation.test.ts | 97 ++++++++----------- apps/server/src/peer/RemoteDelegation.ts | 48 +++------ .../orchestrator-mcp-server.md | 4 +- packages/contracts/src/orchestratorMcp.ts | 3 - 5 files changed, 61 insertions(+), 96 deletions(-) diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index b25b729d38343..339d2aaef5663 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -6,7 +6,6 @@ import { OrchestratorMcpThreadWaitResult, ProjectId, RunId, - THREAD_READ_NEWEST_POSITION, ThreadId, TrimmedNonEmptyString, type OrchestratorMcpThreadListItem, @@ -44,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()( @@ -365,7 +366,7 @@ const readBackward = Effect.fn("cli.thread.readBackward")(function* ( { threadId, view, limit: nextLimit(), runLimit: 1, beforePosition }, OrchestratorMcpThreadReadResult, ); - const newest = yield* read(THREAD_READ_NEWEST_POSITION).pipe( + const newest = yield* read(NEWEST_POSITION).pipe( Effect.catchTags({ ThreadToolFailedError: () => Effect.succeed(null) }), ); if (newest === null || newest.previousPosition === undefined) return undefined; diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index f79070ec31e2c..f751ab0cbc892 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -120,8 +120,6 @@ interface BoxThread { readonly finished: Deferred.Deferred; /** While set, reading the thread there waits: completed once a read starts, released by `releaseRead`. */ readonly holdRead: Ref.Ref; - /** Reads there refuse `beforePosition`, as a box from before it does. */ - readonly forwardReadsOnly: Ref.Ref; readonly readStarted: Deferred.Deferred; readonly releaseRead: Deferred.Deferred; readonly holdFirstStop: Ref.Ref; @@ -158,7 +156,6 @@ const makeBoxThread = Effect.gen(function* () { interruptedOnce: yield* Deferred.make(), finished: yield* Deferred.make(), holdRead: yield* Ref.make(false), - forwardReadsOnly: yield* Ref.make(false), readStarted: yield* Deferred.make(), releaseRead: yield* Deferred.make(), holdFirstStop: yield* Ref.make(false), @@ -309,9 +306,6 @@ const boxToolkitLayer = (thread: BoxThread) => { yield* Deferred.succeed(thread.readStarted, undefined); yield* Deferred.await(thread.releaseRead); } - if (options.beforePosition !== undefined && (yield* Ref.get(thread.forwardReadsOnly))) { - return yield* Effect.die("Unknown key beforePosition"); - } const reply = yield* Ref.get(thread.reply); const later = yield* Ref.get(thread.laterReply); const error = yield* Ref.get(thread.error); @@ -358,13 +352,9 @@ const boxToolkitLayer = (thread: BoxThread) => { (entry) => options.view === "activity" || entry.type === "assistant_message", ) : replies.filter((entry) => entry.id === options.itemId); - const inRange = visible.filter( - (entry) => - entry.position > (options.afterPosition ?? -1) && - entry.position < (options.beforePosition ?? Number.POSITIVE_INFINITY), - ); - const matching = - options.beforePosition === undefined ? inRange.slice(0, 1) : inRange.slice(-1); + const matching = visible + .filter((entry) => entry.position > (options.afterPosition ?? -1)) + .slice(0, 1); return { items: matching.map(({ id, runId: itemRunId, text, position, type }) => ({ position, @@ -1246,50 +1236,47 @@ it.effect("fails the task when the box revokes the link, and cancel interrupts i ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); -it.effect.each([false, true])( - "a task's result is its own run's whole last reply, not a later run's (forward reads only: %s)", - (forwardReadsOnly) => - Effect.scoped( - Effect.gen(function* () { - const thread = yield* makeBoxThread; - yield* Ref.set(thread.forwardReadsOnly, forwardReadsOnly); - const b = yield* serveBox(thread); - const offers = yield* SubscriptionRef.make< - ReadonlyArray - >([]); - const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); - const a = yield* makeLaptop(database, offers); - yield* linkTo(a.links, b, "full-access"); - yield* seedParent(a); +it.effect("a task's result is its own run's whole last reply, not a later run's", () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const a = yield* makeLaptop(database, offers); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); - const result = yield* a.call("delegate_task", { - task: "Review the diff.", - role: "review", - target: { environmentId: box.environmentId }, - clientRequestId: "review-on-box", - }); - expect(result.isError, JSON.stringify(result.content)).toBe(false); - const { taskId } = result.structuredContent as { taskId: NodeId }; - // The role reaches the thread there, as it does a child here. - expect(yield* Ref.get(thread.message)).toBe( - "Act as the review sub-agent for this task.\n\nReview the diff.", - ); + const result = yield* a.call("delegate_task", { + task: "Review the diff.", + role: "review", + target: { environmentId: box.environmentId }, + clientRequestId: "review-on-box", + }); + expect(result.isError, JSON.stringify(result.content)).toBe(false); + const { taskId } = result.structuredContent as { taskId: NodeId }; + // The role reaches the thread there, as it does a child here. + expect(yield* Ref.get(thread.message)).toBe( + "Act as the review sub-agent for this task.\n\nReview the diff.", + ); - // A reply longer than one read, then a later run's reply on the same thread. - const long = "x".repeat(45_000); - yield* Ref.set(thread.reply, long); - yield* Ref.set(thread.laterReply, "Someone else asked a follow-up."); - yield* Ref.set(thread.status, "completed"); - yield* Deferred.succeed(thread.finished, undefined); - const task = yield* taskResult(a, taskId); - expect(task?.status).toBe("completed"); - expect(task?.result).toBe(long); - // Every wait was for the run the launch started. - const waits = yield* Ref.get(thread.waitedFor); - expect(waits.length).toBeGreaterThan(0); - expect(waits.every((runId) => runId === boxRunId)).toBe(true); - }), - ).pipe(Effect.provide(NodeHttpServer.layerTest)), + // A reply longer than one read, then a later run's reply on the same thread. + const long = "x".repeat(45_000); + yield* Ref.set(thread.reply, long); + yield* Ref.set(thread.laterReply, "Someone else asked a follow-up."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + const task = yield* taskResult(a, taskId); + expect(task?.status).toBe("completed"); + expect(task?.result).toBe(long); + // Every wait was for the run the launch started. + const waits = yield* Ref.get(thread.waitedFor); + expect(waits.length).toBeGreaterThan(0); + expect(waits.every((runId) => runId === boxRunId)).toBe(true); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); it.effect("a task whose thread there waits on its own task ends with the answer after it", () => diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 0a796780fcee9..89917026ff8f7 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -12,7 +12,6 @@ import { type ProviderInstanceId, ProviderDriverKind, type RunId, - THREAD_READ_NEWEST_POSITION, type ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -445,43 +444,26 @@ const make = Effect.gen(function* () { ) { let reply: OrchestratorMcpThreadTimelineItem | undefined; let error: OrchestratorMcpThreadTimelineItem | undefined; - const isReply = (item: OrchestratorMcpThreadTimelineItem) => - item.type === "assistant_message" && (runId === null || item.runId === runId); - const isError = (item: OrchestratorMcpThreadTimelineItem) => - failed && item.type === "error" && (runId === null || item.runId === runId); - const read = (position: { - readonly afterPosition?: number; - readonly beforePosition?: number; - }) => - forwarding.readThread(scope, remote.environmentId, { + let afterPosition: number | undefined; + while (true) { + const read = yield* forwarding.readThread(scope, remote.environmentId, { threadId: remote.threadId, view: failed ? "activity" : "messages", limit: 100, - ...position, + ...(afterPosition === undefined ? {} : { afterPosition }), }); - // A peer that predates `beforePosition` rejects it and is read forward instead; - // a forward read fails again on any other cause, as it did before. - const newest = yield* read({ beforePosition: THREAD_READ_NEWEST_POSITION }).pipe( - Effect.orElseSucceed(() => null), - ); - if (newest !== null && newest.previousPosition !== undefined) { - let page = newest; - while (true) { - reply ??= page.items.findLast(isReply); - error ??= page.items.findLast(isError); - const previous = page.previousPosition ?? null; - if ((failed ? error : reply) !== undefined || previous === null) break; - page = yield* read({ beforePosition: previous }); - } - } else { - let afterPosition: number | undefined; - while (true) { - const page = yield* read(afterPosition === undefined ? {} : { afterPosition }); - reply = page.items.findLast(isReply) ?? reply; - error = page.items.findLast(isError) ?? error; - if (!page.hasMore || page.nextPosition === null) break; - afterPosition = page.nextPosition; + reply = + read.items.findLast( + (item) => item.type === "assistant_message" && (runId === null || item.runId === runId), + ) ?? reply; + if (failed) { + error = + read.items.findLast( + (item) => item.type === "error" && (runId === null || item.runId === runId), + ) ?? error; } + if (!read.hasMore || read.nextPosition === null) break; + afterPosition = read.nextPosition; } reply = error ?? reply; if (reply === undefined) return undefined; diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index e399bf59f867e..6bc1b561dc5b3 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -396,9 +396,7 @@ reasoning, checkpoint, handoff, and runtime-request items. Large item text is bounded and reports whether it was truncated. `afterPosition` and `nextPosition` support incremental reads; `beforePosition` and `previousPosition` read back from the newest items, so a caller that wants the -end of a long thread does not page through all of it. Every page still builds -the thread's whole visible index, so each call costs time in proportion to the -thread's length: read only the pages you need. +end of a long thread does not page through all of it. Thread and message results include required `createdBy` and `creationSource` provenance. MCP-created threads and user-role messages use `createdBy: "agent"` diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index dc02afb2e3129..475f73004a572 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -425,9 +425,6 @@ export const OrchestratorMcpThreadReadInput = Schema.Struct({ }); export type OrchestratorMcpThreadReadInput = typeof OrchestratorMcpThreadReadInput.Type; -/** A `beforePosition` past every item, so the read starts at the newest. */ -export const THREAD_READ_NEWEST_POSITION = Number.MAX_SAFE_INTEGER; - export const OrchestratorMcpThreadDetail = Schema.Struct({ threadId: ThreadId, projectId: ProjectId, From 2cdf5086d252d638514b5fb35e1c38cbbae82686 Mon Sep 17 00:00:00 2001 From: Fagner Sales Date: Sun, 8 Nov 2026 18:38:40 -0300 Subject: [PATCH 3/4] fix(cli): stop a backward read whose cursor does not move back A server answering with a previousPosition at or past the requested beforePosition would make readBackward read the same page forever. Also shortens the new t3_thread_read wording (+171 bytes instead of +208). Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cli/thread.test.ts | 17 +++++++++++++++++ apps/server/src/cli/thread.ts | 4 +++- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../server/src/provider/agentContextBudget.json | 2 +- packages/contracts/src/orchestratorMcp.ts | 2 +- 5 files changed, 23 insertions(+), 4 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index 1581e4787b5c4..3bcb49c68eceb 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -339,6 +339,23 @@ it.effect.each(SERVERS)( }), ); +it.effect("show stops when a server's previousPosition does not move back", () => + Effect.gen(function* () { + 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] }); + + yield* readLatestMessages(tools, threadId, 10); + + assert.lengthOf(calls, 2); + }), +); + it.effect("a server's rejected arguments are a tool failure, so show can fall back", () => Effect.gen(function* () { const failWith = (code: number | undefined) => diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 339d2aaef5663..a0a162c663dca 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -375,7 +375,9 @@ const readBackward = Effect.fn("cli.thread.readBackward")(function* ( while (previous !== null) { const page = yield* read(previous); pages.push(page.items); - previous = done(page.items) ? null : (page.previousPosition ?? null); + const next = page.previousPosition ?? null; + // A cursor that does not move back would read the same page forever. + previous = done(page.items) || next === null || next >= previous ? null : next; } return { thread: newest.thread, pages }; }); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 1a0aeaaf843de..1c650ea67e8b2 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -238,7 +238,7 @@ const ThreadListTool = Tool.make("t3_thread_list", { const ThreadReadTool = Tool.make("t3_thread_read", { description: - "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`, or back from the newest with `beforePosition=previousPosition`. 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.", + "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`, or back with `beforePosition=previousPosition`. 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.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index cff372343ae45..63191807e904d 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5350, "codexAdditionalContextBytes": 7080, - "mcpToolListJsonBytes": 48324 + "mcpToolListJsonBytes": 48287 } diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 475f73004a572..29fec9de59694 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -411,7 +411,7 @@ export const OrchestratorMcpThreadReadInput = Schema.Struct({ view: Schema.optional(Schema.Literals(["messages", "activity"])), afterPosition: Schema.optional(NonNegativeInt), beforePosition: Schema.optional(NonNegativeInt).annotate({ - description: "Read the `limit` items just before this position; `itemCount` reads the latest.", + description: "The `limit` items before it. `itemCount` reads the newest.", }), limit: Schema.optional(clampedPositiveInt(100)).annotate({ description: "Max items, 1 to 100.", From 69a7f526181638ca6b0fdc3cdda8976d68ed44a2 Mon Sep 17 00:00:00 2001 From: Fagner Sales Date: Sun, 8 Nov 2026 18:48:22 -0300 Subject: [PATCH 4/4] fix(cli): drop a repeated page's items from a backward read A page past the position requested is a repeat, so its items no longer reach thread show's output. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cli/thread.test.ts | 6 +++++- apps/server/src/cli/thread.ts | 10 ++++++---- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index 3bcb49c68eceb..3d3482b29a1de 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -350,9 +350,13 @@ it.effect("show stops when a server's previousPosition does not move back", () = }; const { tools, calls } = scriptedTools({ t3_thread_read: [page, page] }); - yield* readLatestMessages(tools, threadId, 10); + const { messages } = yield* readLatestMessages(tools, threadId, 10); assert.lengthOf(calls, 2); + assert.deepStrictEqual( + messages.map((item) => item.itemId), + ["item-5"], + ); }), ); diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index a0a162c663dca..9e2a130f96e9d 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -373,11 +373,13 @@ const readBackward = Effect.fn("cli.thread.readBackward")(function* ( const pages = [newest.items]; let previous = done(newest.items) ? null : newest.previousPosition; while (previous !== null) { - const page = yield* read(previous); - pages.push(page.items); + 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; - // A cursor that does not move back would read the same page forever. - previous = done(page.items) || next === null || next >= previous ? null : next; + previous = done(items) || next === null || next >= before ? null : next; } return { thread: newest.thread, pages }; });