diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 7e2959c767d0..5218d0aa2576 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -50,6 +50,7 @@ const TERMINAL_RUN_STATUSES = new Set([ "rolled_back", ]); const READ_PAGE_LIMIT = 100; +const READ_ITEM_MAX_CHARS = 20_000; const TITLE_MAX_CHARS = 80; export class ThreadServerNotRunningError extends Schema.TaggedError()( @@ -394,6 +395,7 @@ const readTimeline = Effect.fn("cli.thread.readTimeline")(function* ( threadId, view, limit: READ_PAGE_LIMIT, + maxCharsPerItem: READ_ITEM_MAX_CHARS, runLimit: 1, ...(afterPosition === undefined ? {} : { afterPosition }), }, @@ -439,6 +441,7 @@ export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(fun threadId, view: "messages", limit: READ_PAGE_LIMIT, + maxCharsPerItem: READ_ITEM_MAX_CHARS, runLimit: 1, ...(afterPosition < 0 ? {} : { afterPosition }), }, @@ -473,7 +476,13 @@ const readFullText = Effect.fn("cli.thread.readFullText")(function* ( while (nextTextOffset !== null) { const page = yield* tools.call( "t3_thread_read", - { threadId, itemId: item.itemId, textOffset: nextTextOffset, runLimit: 1 }, + { + threadId, + itemId: item.itemId, + textOffset: nextTextOffset, + runLimit: 1, + maxCharsPerItem: READ_ITEM_MAX_CHARS, + }, OrchestratorMcpThreadReadResult, ); const chunk = page.items.find((candidate) => candidate.itemId === item.itemId); diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index 123b5e6fcb8d..e54a759aac28 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -285,17 +285,35 @@ it("readThread pages cut-off text without splitting an emoji across pages", asyn updatedAt: now, }, }; + const rows = [ + row, + ...Array.from({ length: 11 }, (_, index) => ({ + ...row, + position: index + 1, + sourceItemId: TurnItemId.make(`turn-item-mcp-${index}`), + item: { ...row.item, text: "x".repeat(600) }, + })), + ]; const layer = layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ - getTimelinePage: () => - Effect.succeed({ - items: [row] as unknown as OrchestrationV2ThreadProjection["visibleTurnItems"], - totalItems: 1, - hasMore: false, - }), + getTimelinePage: (_threadId, options) => { + const remaining = rows.filter( + (candidate) => + candidate.position > (options.afterPosition ?? -1) && + (options.itemId === undefined || candidate.sourceItemId === options.itemId), + ); + return Effect.succeed({ + items: remaining.slice( + 0, + options.limit, + ) as unknown as OrchestrationV2ThreadProjection["visibleTurnItems"], + totalItems: rows.length, + hasMore: remaining.length > options.limit, + }); + }, getThreadRecords: () => Effect.succeed(projection), getProjectThreadRecords: () => Effect.succeed(projection), getThreadShell: () => @@ -342,6 +360,17 @@ it("readThread pages cut-off text without splitting an emoji across pages", asyn }); } } + const first = yield* service.readThread(makeScope(), { + threadId: parentThreadId, + view: "activity", + }); + expect(first.items[1]).toMatchObject({ + text: "x".repeat(500), + textTruncated: true, + nextTextOffset: 500, + }); + expect(first.items).toHaveLength(10); + expect(first).toMatchObject({ nextPosition: 9, hasMore: true }); }).pipe(Effect.provide(layer), Effect.runPromise); }); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 7c55c6a4ba90..3dcb57664b45 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -142,10 +142,10 @@ const TASK_WAKE_EVENTS = [ { thread: "child", eventType: "runtime-request.updated" }, ] as const; const SECRET_REQUEST_POLL_INTERVAL_MS = 500; -const DEFAULT_THREAD_LIST_LIMIT = 50; -const DEFAULT_THREAD_READ_LIMIT = 50; +const DEFAULT_THREAD_LIST_LIMIT = 10; +const DEFAULT_THREAD_READ_LIMIT = 10; const DEFAULT_THREAD_RUN_LIMIT = 10; -const DEFAULT_THREAD_ITEM_MAX_CHARS = 20_000; +const DEFAULT_THREAD_ITEM_MAX_CHARS = 500; interface ResolvedTarget { readonly modelSelection: ModelSelection; @@ -638,6 +638,7 @@ function pageIncludesTerminalTaskResult(input: { "thread" | "runs" | "contextTransfers" | "messages" | "turnItems" | "providerTurns" >; readonly maxChars: number; + readonly textOffset: number; }): boolean { const transfer = taskResultTransfer(input.parentThreadId, input.target, input.task); if (transfer === undefined) return false; @@ -660,7 +661,11 @@ function pageIncludesTerminalTaskResult(input: { if (!matchesResult) return false; const text = turnItemText(row.item); - return text !== null && text.length <= textPageEnd(text, 0, input.maxChars); + return ( + text !== null && + (input.textOffset === 0 || input.textOffset < text.length) && + text.length <= textPageEnd(text, input.textOffset, input.maxChars) + ); }); } @@ -3082,12 +3087,7 @@ const make = Effect.gen(function* () { target, returnedRunIds, ); - if ( - parent !== undefined && - scope.thread !== undefined && - tasks.length > 0 && - (input.textOffset ?? 0) === 0 - ) { + if (parent !== undefined && scope.thread !== undefined && tasks.length > 0) { for (const task of tasks) { const transfer = taskResultTransfer(parent.thread.id, target, task); const resultRunId = transfer?.sourcePoint.runId ?? delegatedTaskRun(target, task)?.id; @@ -3107,6 +3107,7 @@ const make = Effect.gen(function* () { task, target: { ...target, ...resultRecords }, maxChars, + textOffset: input.itemId === undefined ? 0 : (input.textOffset ?? 0), }) ) { yield* readTask( diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index bbeac142dfeb..3439ed4121d1 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -11,6 +11,7 @@ import { OrchestratorMcpDelegateTaskResult, OrchestratorMcpThreadInterruptResult, OrchestratorMcpThreadSendResult, + OrchestratorMcpThreadReadResult, ProjectId, ProviderDriverKind, ProviderInstanceId, @@ -79,6 +80,7 @@ const delegatedResult = "Delegated API boundary inspected."; const cancellationPrompt = "Remain active until the parent cancels this delegated task."; const queuedFollowupPrompt = "Complete the queued follow-up and return the final result."; const queuedFollowupResult = "Queued delegated follow-up completed."; +const decodeThreadRead = Schema.decodeUnknownEffect(OrchestratorMcpThreadReadResult); const decodeDelegateTaskResult = Schema.decodeUnknownEffect(OrchestratorMcpDelegateTaskResult); const decodeThreadInterruptResult = Schema.decodeUnknownEffect( @@ -199,6 +201,11 @@ describe("orchestrator MCP toolkit", () => { { label: "distinct", result: queuedFollowupResult, latestSummary: queuedFollowupResult }, { label: "identical", result: delegatedResult, latestSummary: null }, { label: "emoji", result: "😀", latestSummary: "😀" }, + { + label: "paginated", + result: "git safety ".repeat(120), + latestSummary: "git safety ".repeat(120), + }, ])("reports $label child results", ({ result, latestSummary }) => Effect.scoped( Effect.gen(function* () { @@ -724,11 +731,40 @@ describe("orchestrator MCP toolkit", () => { const read = yield* invoke("t3_thread_read", { threadId: delegated.childThreadId, limit: 100, - maxCharsPerItem: result === "😀" ? 1 : 20_000, + ...(result === "😀" ? { maxCharsPerItem: 1 } : {}), }).pipe(Effect.withTracer(tracer)); expect(read.isError).not.toBe(true); expect(transferReads.length).toBeGreaterThan(0); - expect(read.structuredContent).toEqual(parentlessRead.structuredContent); + const firstPage = yield* decodeThreadRead(read.structuredContent); + const finalItem = firstPage.items.find((item) => item.text === result.slice(0, 500)); + expect(finalItem).toBeDefined(); + if (result.length > 500) { + yield* invoke("t3_thread_read", { + threadId: delegated.childThreadId, + itemId: finalItem!.itemId, + textOffset: result.length, + }); + } else { + expect(read.structuredContent).toEqual(parentlessRead.structuredContent); + } + let text = finalItem!.text ?? ""; + let nextTextOffset = finalItem?.nextTextOffset ?? null; + while (nextTextOffset !== null) { + const beforeContinuation = yield* orchestrator.getThreadProjection(parentThreadId); + expect( + beforeContinuation.subagents.find((task) => task.id === queuedFollowup.taskId) + ?.completionDelivery?.state, + ).toBe("claimed"); + const continuation = yield* invoke("t3_thread_read", { + threadId: delegated.childThreadId, + itemId: finalItem!.itemId, + textOffset: nextTextOffset, + }).pipe(Effect.withTracer(tracer)); + const nextPage = yield* decodeThreadRead(continuation.structuredContent); + text += nextPage.items[0]?.text ?? ""; + nextTextOffset = nextPage.items[0]?.nextTextOffset ?? null; + } + expect(text).toBe(result); // The acknowledgement reads the task and its spawning run by id, not the parent's history. expect(acknowledgementReads.length).toBeGreaterThan(0); expect( diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index a5790332ebc6..016e825cefa8 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -613,6 +613,7 @@ const make = Effect.gen(function* () { view: failed ? "activity" : "messages", limit: 100, runLimit: 50, + maxCharsPerItem: 20_000, ...(afterPosition === undefined ? {} : { afterPosition }), }); run ??= read.recentRuns.find((candidate) => candidate.runId === runId);