diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index f182f708c4cfa..3d3482b29a1de 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,116 @@ 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.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(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(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"); }), ); diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 80dfad018e158..9e2a130f96e9d 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -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"; @@ -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()( @@ -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, @@ -186,7 +192,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 +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) => 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; + 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, @@ -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; @@ -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, @@ -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); }); 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 dfaebe9228d52..8f4776d1fd18d 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -148,7 +148,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( @@ -254,6 +255,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), @@ -388,7 +390,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); @@ -509,6 +512,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 4750e9a335f49..11e526b841945 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -3067,6 +3067,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)); @@ -3164,6 +3165,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 2ba8a5f2dd5fc..af6c4c9393301 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 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 3d33f76b7a494..8db47fdcb2c74 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 { @@ -5998,21 +6002,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) @@ -6064,7 +6076,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 317a11251aff4..504d159c09af3 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -402,6 +402,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) => diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 0d46d49479a95..5a9dd424c5294 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5549, "codexAdditionalContextBytes": 7279, - "mcpToolListJsonBytes": 48233 + "mcpToolListJsonBytes": 48404 } diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 2f6ba5363703a..6bc1b561dc5b3 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -394,7 +394,9 @@ 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. 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 d1bb9961a0c68..ccf9eddd0f7ee 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -418,6 +418,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: "The `limit` items before it. `itemCount` reads the newest.", + }), limit: Schema.optional(clampedPositiveInt(100)).annotate({ description: "Max items, 1 to 100.", }), @@ -504,6 +507,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;