diff --git a/docs/user/remote-access.md b/docs/user/remote-access.md index 95892b710234a..2a7f2af6982df 100644 --- a/docs/user/remote-access.md +++ b/docs/user/remote-access.md @@ -467,9 +467,14 @@ machine with `environmentId` and `environmentLabel`, and `environments` lists an was skipped (disconnected or without read access) or failed, so a missing chat is not mistaken for one that does not exist. Pass `environmentId` to list one machine. -A chat's `profileSnapshot` names the agent and revision it was created with (`profileId`, -`profileName`, `revision`), not the agent's instructions. Pass `includeSystemPrompt: true` to -`t3_list_agents` when you need each agent's current prompt, for example before `t3_update_agent`. +`t3_list_threads` returns the 20 most recently updated chats unless you pass `limit` (up to +100). When `hasMore` is true, pass `nextCursor` back as `cursor` with the same filters for the +next page. `updatedAfter`, `updatedBefore`, and `createdAfter` take ISO times to pick a window. +Each row is a summary; `t3_get_thread` returns the whole chat. + +A chat's `profileId` and `profileName` name the agent it was created with, not the agent's +instructions. Pass `includeSystemPrompt: true` to `t3_list_agents` when you need each agent's +current prompt, for example before `t3_update_agent`. ### Answer an agent's questions through MCP diff --git a/packages/mcp-gateway/src/catalog.ts b/packages/mcp-gateway/src/catalog.ts index 0a9cef216b0cc..d6caa61f647ab 100644 --- a/packages/mcp-gateway/src/catalog.ts +++ b/packages/mcp-gateway/src/catalog.ts @@ -9,6 +9,7 @@ import { GatewayError, GATEWAY_THREAD_EXECUTION_STATES } from "./port.ts"; import { callGatewayTool, handoffInputSchema, + threadListWindowFields, threadTaskInputFields, type GatewayInvocation, type GatewayToolContext, @@ -249,7 +250,7 @@ export const TOOL_SPECS = { { environmentId: discoveryEnvironmentId, ...optionalRequestContext }, ], t3_list_threads: [ - "List chats across every connected T3 environment (or one), optionally filtered by agent profileId, project, parentThreadId, active/settled state, and executionState. state=active means unsettled (it still includes completed or stopped chats); use executionState to select running or waiting-input/waiting-approval work explicitly. Finished chats remain active until manually settled or automatically settled after their idle age (3 days by default); new activity and un-settling reset the idle clock. Manually settling a parent chat also settles its child chats. hasPendingUserInput marks a chat waiting on a question; pass includeQuestions to attach each one's pendingQuestions (full text and options, as t3_get_pending_questions returns). profileSnapshot names the chat's agent; read its instructions with t3_list_agents includeSystemPrompt.", + "List chats across every connected T3 environment (or one), most recently updated first. Returns the latest 20 unless you pass limit (max 100). When hasMore is true, pass nextCursor back as cursor with the same filters for the next page. Narrow with agent profileId, project, parentThreadId, active/settled state, executionState, and the updatedAfter, updatedBefore, and createdAfter ISO times; use updatedBefore or cursor to reach older chats. state=active means unsettled (it still includes completed or stopped chats); use executionState to select running or waiting-input/waiting-approval work explicitly. Finished chats remain active until manually settled or automatically settled after their idle age (3 days by default); new activity and un-settling reset the idle clock. Manually settling a parent chat also settles its child chats. Rows are summaries: read one chat with t3_get_thread or its messages with t3_get_messages. hasPendingUserInput marks a chat waiting on a question; pass includeQuestions to attach each one's pendingQuestions (full text and options, as t3_get_pending_questions returns). profileId and profileName name the chat's agent; read its instructions with t3_list_agents includeSystemPrompt.", { environmentId: discoveryEnvironmentId, state: z.enum(["all", "active", "settled"]).optional(), @@ -272,6 +273,7 @@ export const TOOL_SPECS = { .boolean() .optional() .describe("Attach pendingQuestions to chats waiting on a question."), + ...threadListWindowFields, ...optionalRequestContext, }, ], @@ -396,7 +398,7 @@ export const TOOL_SPECS = { { environmentId, ...threadTaskInputFields.update }, ], t3_task_watch: [ - "Wait up to timeoutMs (max 60000) for changes to tasks you own or work on after afterCursor, returning as soon as one changes. Omit afterCursor for the current tasks and cursor. Wakes already arrive as messages; use this only when this turn must wait.", + "Wait up to timeoutMs (max 60000) for changes to tasks you own or work on after afterCursor, returning as soon as one changes. Omit afterCursor for the current tasks and cursor. Task changes never send you a message or wake you after your turn ends; this call only holds the current turn until one arrives.", { environmentId, ...threadTaskInputFields.watch }, ], t3_task_board: [ diff --git a/packages/mcp-gateway/src/server.test.ts b/packages/mcp-gateway/src/server.test.ts index 24e02fea51c9e..a61e37c831835 100644 --- a/packages/mcp-gateway/src/server.test.ts +++ b/packages/mcp-gateway/src/server.test.ts @@ -277,12 +277,12 @@ describe("MCP gateway server", () => { expect(result.structuredContent).toEqual({ schemaVersion: "3", serverTime: expect.any(String), - data: { items: [], snapshotAt: "snapshot-1" }, + data: { items: [], hasMore: false, nextCursor: null, snapshotAt: "snapshot-1" }, }); expect(JSON.parse((result.content as Array<{ text: string }>)[0]!.text)).toEqual({ schemaVersion: "3", serverTime: expect.any(String), - data: { snapshotAt: "snapshot-1" }, + data: { hasMore: false, snapshotAt: "snapshot-1" }, }); await client.close(); diff --git a/packages/mcp-gateway/src/tools.test.ts b/packages/mcp-gateway/src/tools.test.ts index cbc5271fd7515..3472b10624b46 100644 --- a/packages/mcp-gateway/src/tools.test.ts +++ b/packages/mcp-gateway/src/tools.test.ts @@ -2559,9 +2559,19 @@ describe("gateway coordinator operations", () => { port.listThreads = async () => ({ snapshotAt: "now", items: [ - { id: "running", projectId: "p", settledAt: null, status: "running" }, - { id: "waiting", projectId: "p", settledAt: null, status: "waiting-input" }, - { id: "completed", projectId: "p", settledAt: null, status: "completed" }, + { id: "running", settledAt: null, status: "running", updatedAt: "2026-09-07T03:00:00Z" }, + { + id: "waiting", + settledAt: null, + status: "waiting-input", + updatedAt: "2026-09-07T02:00:00Z", + }, + { + id: "completed", + settledAt: null, + status: "completed", + updatedAt: "2026-09-07T01:00:00Z", + }, { id: "settled-done", projectId: "p", @@ -3212,8 +3222,18 @@ describe("pending question tools", () => { port.listThreads = async () => ({ snapshotAt: "runtime", items: [ - { id: "chat", hasPendingUserInput: true, status: "waiting-input" }, - { id: "busy", hasPendingUserInput: false, status: "running" }, + { + id: "chat", + hasPendingUserInput: true, + status: "waiting-input", + updatedAt: "2026-09-07T02:00:00Z", + }, + { + id: "busy", + hasPendingUserInput: false, + status: "running", + updatedAt: "2026-09-07T01:00:00Z", + }, ], }); const listed = (await callGatewayTool({ port, grants }, "t3_list_threads", { @@ -3225,6 +3245,145 @@ describe("pending question tools", () => { }); }); +describe("bounded thread list", () => { + // 30 chats per machine, one per minute; both machines share every timestamp so ties cross pages. + const minute = (n: number) => `2026-09-07T00:${String(n).padStart(2, "0")}:00.000Z`; + const chatId = (environmentId: string, n: number) => + `${environmentId}-${String(n).padStart(2, "0")}`; + const newestFirst = (count: number) => + Array.from({ length: count }, (_, k) => [chatId("a", 29 - k), chatId("b", 29 - k)]).flat(); + const chats = (environmentId: string) => + Array.from({ length: 30 }, (_, n) => ({ + id: chatId(environmentId, n), + projectId: n % 2 === 0 ? "even" : "odd", + title: `Chat ${n}`, + status: n % 3 === 0 ? "running" : "completed", + settledAt: null, + hasPendingUserInput: n === 29, + profileSnapshot: { profileId: n % 5 === 0 ? "review" : "code", profileName: "Code" }, + modelSelection: { instanceId: "codex", model: "gpt-5", options: [{ id: "effort" }] }, + latestRunId: `run-${n}`, + createdAt: minute(n), + updatedAt: minute(n), + })); + function boundedContext() { + const getThread = vi.fn(async () => ({ pendingQuestions: [] })); + const port: GatewayRuntimePort = { + ...makePort(), + listEnvironments: async () => + ["a", "b"].map((environmentId) => ({ + environmentId, + label: environmentId.toUpperCase(), + targetKind: "relay" as const, + connectionState: "connected" as const, + })), + listThreads: async (environmentId) => ({ snapshotAt: "now", items: chats(environmentId) }), + getThread, + }; + return { port, getThread, grants: { a: ["read"], b: ["read"] } as const }; + } + type Listed = { + items: ReadonlyArray>; + hasMore: boolean; + nextCursor: string | null; + }; + const list = (context: ReturnType, input: Record = {}) => + callGatewayTool(context, "t3_list_threads", input) as Promise; + const ids = (page: Listed) => page.items.map((row) => row.id); + + it("returns the latest 20 compact rows by default and pages without overlap or gaps", async () => { + const context = boundedContext(); + const first = await list(context); + expect(ids(first)).toEqual(newestFirst(10)); + expect(first.hasMore).toBe(true); + expect(first.items[0]).toEqual({ + id: "a-29", + environmentId: "a", + environmentLabel: "A", + projectId: "odd", + title: "Chat 29", + status: "completed", + hasPendingUserInput: true, + hasPendingApprovals: undefined, + profileId: "code", + profileName: "Code", + providerInstanceId: "codex", + model: "gpt-5", + parentThreadId: null, + parentEnvironmentId: null, + settledAt: null, + createdAt: minute(29), + updatedAt: minute(29), + }); + + const seen: unknown[] = []; + let cursor: string | undefined; + let pages = 0; + do { + const page = await list(context, { limit: 7, ...(cursor ? { cursor } : {}) }); + seen.push(...ids(page)); + cursor = page.nextCursor ?? undefined; + expect(page.hasMore).toBe(cursor !== undefined); + pages += 1; + } while (cursor !== undefined); + expect(pages).toBe(9); + expect(seen).toEqual(newestFirst(30)); + + await expect(list(context, { limit: 101 })).rejects.toMatchObject({ code: "invalid_input" }); + await expect(list(context, { cursor: "not-a-cursor" })).rejects.toMatchObject({ + code: "invalid_input", + }); + }); + + it("applies time windows and combined filters across environments before truncating", async () => { + const context = boundedContext(); + expect(ids(await list(context, { updatedAfter: minute(28) }))).toEqual([ + "a-29", + "b-29", + "a-28", + "b-28", + ]); + expect(ids(await list(context, { updatedBefore: minute(2), limit: 100 }))).toEqual([ + "a-01", + "b-01", + "a-00", + "b-00", + ]); + expect( + ids(await list(context, { createdAfter: minute(27), updatedBefore: minute(29) })), + ).toEqual(["a-28", "b-28", "a-27", "b-27"]); + await expect(list(context, { updatedAfter: "yesterday" })).rejects.toMatchObject({ + code: "invalid_input", + }); + + // Matches sit at n = 0, 6, 12, 18, 24 on each machine; oldest ones still reach the page. + const combined = await list(context, { + projectId: "even", + executionState: "running", + profileId: "code", + limit: 5, + }); + expect(ids(combined)).toEqual(["a-24", "b-24", "a-18", "b-18", "a-12"]); + const rest = await list(context, { + projectId: "even", + executionState: "running", + profileId: "code", + cursor: combined.nextCursor, + }); + expect(ids(rest)).toEqual(["b-12", "a-06", "b-06"]); + expect(rest.hasMore).toBe(false); + }); + + it("reads pending questions only for chats on the returned page", async () => { + const context = boundedContext(); + const page = await list(context, { includeQuestions: true, updatedBefore: minute(29) }); + expect(page.items).toHaveLength(20); + expect(context.getThread).not.toHaveBeenCalled(); + await list(context, { includeQuestions: true, limit: 1 }); + expect(context.getThread).toHaveBeenCalledExactlyOnceWith("a", "a-29"); + }); +}); + describe("cross-environment discovery", () => { const profile = (profileId: string, environmentIds?: ReadonlyArray): GatewayProfile => ({ profileId, @@ -3237,7 +3396,13 @@ describe("cross-environment discovery", () => { }); const threadsByEnvironment: Record>> = { macbook: [ - { id: "mac-idle", projectId: "t3", status: "idle", profileSnapshot: { profileId: "code" } }, + { + id: "mac-idle", + projectId: "t3", + status: "idle", + profileSnapshot: { profileId: "code" }, + updatedAt: "2026-09-07T02:00:00Z", + }, ], "dev-box": [ { @@ -3245,6 +3410,7 @@ describe("cross-environment discovery", () => { projectId: "tf", status: "running", profileSnapshot: { profileId: "code" }, + updatedAt: "2026-09-07T01:00:00Z", }, ], }; @@ -3344,6 +3510,8 @@ describe("cross-environment discovery", () => { expect(listed).toEqual({ snapshotAt: "now", items: [expect.objectContaining({ id: "tensorfold", environmentLabel: "dev-box" })], + hasMore: false, + nextCursor: null, }); await expect( callGatewayTool(context, "t3_list_threads", { environmentId: "phone" }), diff --git a/packages/mcp-gateway/src/tools.ts b/packages/mcp-gateway/src/tools.ts index b96290380005b..99167f3db9f13 100644 --- a/packages/mcp-gateway/src/tools.ts +++ b/packages/mcp-gateway/src/tools.ts @@ -61,6 +61,32 @@ const taskContinuation = z.discriminatedUnion("kind", [ z.object({ kind: z.literal("task"), threadId: taskThreadId }), ]); +const THREAD_LIST_DEFAULT_LIMIT = 20; +const isoTime = z + .string() + .trim() + .refine((value) => !Number.isNaN(Date.parse(value)), "must be an ISO date or time"); + +/** t3_list_threads paging and time-window inputs. */ +export const threadListWindowFields = { + updatedAfter: isoTime.optional().describe("Only chats updated at or after this ISO time."), + updatedBefore: isoTime.optional().describe("Only chats updated before this ISO time."), + createdAfter: isoTime.optional().describe("Only chats created at or after this ISO time."), + limit: z + .number() + .int() + .min(1) + .max(100) + .optional() + .describe(`Chats per page, newest first. Default ${THREAD_LIST_DEFAULT_LIMIT}.`), + cursor: z + .string() + .trim() + .min(1) + .optional() + .describe("nextCursor from the previous page; keep the other filters the same."), +}; + /** Tool inputs for thread tasks, without environmentId. The caller is never an input. */ export const threadTaskInputFields = { assign: { @@ -476,52 +502,189 @@ function tagged(target: DiscoveryTarget, page: GatewayListPage): GatewayListPage return { ...page, items: page.items.map((item) => ({ ...item, ...target })) }; } +type ThreadListFilter = ReturnType; + +/** Rows sort newest update first; environment and id break ties so pages never overlap. */ +type ThreadListKey = readonly [updatedAt: number, environmentId: string, id: string]; + +function threadListKey(row: Record): ThreadListKey { + const updatedAt = typeof row.updatedAt === "string" ? Date.parse(row.updatedAt) : Number.NaN; + return [ + Number.isNaN(updatedAt) ? 0 : updatedAt, + String(row.environmentId ?? ""), + String(row.id ?? ""), + ]; +} + +function compareThreadListKeys(a: ThreadListKey, b: ThreadListKey): number { + if (a[0] !== b[0]) return b[0] - a[0]; + if (a[1] !== b[1]) return a[1] < b[1] ? -1 : 1; + return a[2] === b[2] ? 0 : a[2] < b[2] ? -1 : 1; +} + +function encodeThreadListCursor(key: ThreadListKey): string { + return Buffer.from(JSON.stringify(key)).toString("base64url"); +} + +function decodeThreadListCursor(cursor: string): ThreadListKey { + let key: unknown; + try { + key = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")); + } catch { + key = undefined; + } + if ( + !Array.isArray(key) || + key.length !== 3 || + typeof key[0] !== "number" || + typeof key[1] !== "string" || + typeof key[2] !== "string" + ) { + throw new GatewayError({ + code: "invalid_input", + message: "cursor must be a nextCursor returned by t3_list_threads.", + retryable: false, + }); + } + return [key[0], key[1], key[2]]; +} + +function parseThreadListFilter(input: Record) { + const parsed = z + .object({ + state: z.enum(["all", "active", "settled"]).default("all"), + projectId: z.string().optional(), + profileId: z.string().optional(), + parentThreadId: z.string().optional(), + parentEnvironmentId: z.string().optional(), + ...threadListWindowFields, + }) + .safeParse(input); + if (!parsed.success) + throw new GatewayError({ + code: "invalid_input", + message: parsed.error.message, + retryable: false, + }); + const { updatedAfter, updatedBefore, createdAfter, limit, cursor, ...rest } = parsed.data; + const time = (value: string | undefined) => (value === undefined ? undefined : Date.parse(value)); + return { + ...rest, + executionState: parseExecutionStateFilter(input), + updatedAfter: time(updatedAfter), + updatedBefore: time(updatedBefore), + createdAfter: time(createdAfter), + limit: limit ?? THREAD_LIST_DEFAULT_LIMIT, + after: cursor === undefined ? undefined : decodeThreadListCursor(cursor), + }; +} + +function threadMatches( + thread: Record, + environmentId: string, + filter: ThreadListFilter, +): boolean { + const snapshot = thread.profileSnapshot as { profileId?: string } | null | undefined; + const updatedAt = typeof thread.updatedAt === "string" ? Date.parse(thread.updatedAt) : NaN; + const createdAt = typeof thread.createdAt === "string" ? Date.parse(thread.createdAt) : NaN; + return ( + (filter.projectId === undefined || thread.projectId === filter.projectId) && + (filter.parentThreadId === undefined || + (thread.parentThreadId === filter.parentThreadId && + (filter.parentEnvironmentId === undefined || + (thread.parentEnvironmentId ?? environmentId) === filter.parentEnvironmentId))) && + (filter.profileId === undefined || snapshot?.profileId === filter.profileId) && + (filter.state === "all" || + (filter.state === "settled" ? thread.settledAt != null : thread.settledAt == null)) && + (filter.executionState === undefined || + readThreadExecutionState(thread) === filter.executionState) && + (filter.updatedAfter === undefined || updatedAt >= filter.updatedAfter) && + (filter.updatedBefore === undefined || updatedAt < filter.updatedBefore) && + (filter.createdAfter === undefined || createdAt >= filter.createdAfter) + ); +} + +/** A listing row: enough to pick a chat. t3_get_thread has the rest. */ +function threadListRow(thread: Record, target: DiscoveryTarget) { + const profile = thread.profileSnapshot as + | { profileId?: string | null; profileName?: string | null } + | null + | undefined; + const model = thread.modelSelection as { instanceId?: string; model?: string } | null | undefined; + return { + id: thread.id, + ...target, + projectId: thread.projectId, + title: thread.title, + status: thread.status, + hasPendingUserInput: thread.hasPendingUserInput, + hasPendingApprovals: thread.hasPendingApprovals, + profileId: profile?.profileId ?? null, + profileName: profile?.profileName ?? null, + providerInstanceId: model?.instanceId ?? null, + model: model?.model ?? null, + parentThreadId: thread.parentThreadId ?? null, + parentEnvironmentId: thread.parentEnvironmentId ?? null, + settledAt: thread.settledAt ?? null, + createdAt: thread.createdAt, + updatedAt: thread.updatedAt, + }; +} + +/** One environment's matching rows past the cursor, newest first, at most one page plus one. */ async function listGatewayThreads( context: GatewayToolContext, target: DiscoveryTarget, - input: Record, + filter: ThreadListFilter, ): Promise { - const { environmentId } = target; - const state = z.enum(["all", "active", "settled"]).default("all").parse(input.state); - const executionState = parseExecutionStateFilter(input); - const page = await context.port.listThreads(environmentId); - const projectId = typeof input.projectId === "string" ? input.projectId : undefined; - const profileId = typeof input.profileId === "string" ? input.profileId : undefined; - const parentThreadId = - typeof input.parentThreadId === "string" ? input.parentThreadId : undefined; - const parentEnvironmentId = - typeof input.parentEnvironmentId === "string" ? input.parentEnvironmentId : undefined; - const items = page.items.filter((thread) => { - const snapshot = thread.profileSnapshot as { profileId?: string } | undefined; - return ( - (projectId === undefined || thread.projectId === projectId) && - (parentThreadId === undefined || - (thread.parentThreadId === parentThreadId && - (parentEnvironmentId === undefined || - (thread.parentEnvironmentId ?? environmentId) === parentEnvironmentId))) && - (profileId === undefined || snapshot?.profileId === profileId) && - (state === "all" || - (state === "settled" ? thread.settledAt != null : thread.settledAt == null)) && - (executionState === undefined || readThreadExecutionState(thread) === executionState) - ); - }); - if (input.includeQuestions !== true) return tagged(target, { ...page, items }); - // Only chats waiting on a question need a detail read; the rest pass through. - return tagged(target, { - ...page, - items: await Promise.all( - items.map(async (thread) => - thread.hasPendingUserInput === true && typeof thread.id === "string" - ? { - ...thread, - pendingQuestions: pendingQuestionsOf( - await context.port.getThread(environmentId, thread.id), - ), - } - : thread, - ), - ), - }); + const page = await context.port.listThreads(target.environmentId); + const rows = page.items + .filter((thread) => threadMatches(thread, target.environmentId, filter)) + .map((thread) => threadListRow(thread, target)) + .filter( + (row) => + filter.after === undefined || compareThreadListKeys(threadListKey(row), filter.after) > 0, + ) + .sort((a, b) => compareThreadListKeys(threadListKey(a), threadListKey(b))); + return { items: rows.slice(0, filter.limit + 1), snapshotAt: page.snapshotAt }; +} + +async function listThreadsPage(context: GatewayToolContext, input: Record) { + const filter = parseThreadListFilter(input); + const discovered = await discover(context, input, (target) => + listGatewayThreads(context, target, filter), + ); + const merged = discovered.results + .flatMap(({ value }) => value.items) + .sort((a, b) => compareThreadListKeys(threadListKey(a), threadListKey(b))); + const rows = merged.slice(0, filter.limit); + const hasMore = merged.length > filter.limit; + const items = + input.includeQuestions !== true + ? rows + : await Promise.all( + // Only chats waiting on a question need a detail read; the rest pass through. + rows.map(async (row) => + row.hasPendingUserInput === true && + typeof row.id === "string" && + typeof row.environmentId === "string" + ? { + ...row, + pendingQuestions: pendingQuestionsOf( + await context.port.getThread(row.environmentId, row.id), + ), + } + : row, + ), + ); + return { + items, + hasMore, + nextCursor: hasMore ? encodeThreadListCursor(threadListKey(rows.at(-1)!)) : null, + ...(discovered.coverage === undefined + ? { snapshotAt: discovered.results[0]!.value.snapshotAt } + : { environments: discovered.coverage, snapshotAt: "runtime" }), + }; } async function agentsView( @@ -1494,14 +1657,8 @@ export async function callGatewayTool( ), ); } - case "t3_list_threads": { - // Validate filters once so a bad input fails the call instead of every environment. - z.enum(["all", "active", "settled"]).optional().parse(input.state); - parseExecutionStateFilter(input); - return discoveredPage( - await discover(context, input, (target) => listGatewayThreads(context, target, input)), - ); - } + case "t3_list_threads": + return listThreadsPage(context, input); case "t3_list_devices": { const environmentId = environmentWithScope(context, input, "read"); if (!context.port.listDevices)