authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-10-10 23:22:40+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 16:22:40-05:00
log38c187bfb7a0020db36ccc1479e7c3501b205fb8
treeb40d8076e0c403fb5c03fd1f3927398e4a2605bc
parent4f63275ac7885281d245022e65f1e9e8816b1382
signature Signed by PGP key B5690EEEBB952194

fix(pi): preserve tool images and structured results (#17851)

6 files changed, 659 insertions(+), 141 deletions(-)

packages/provider-pi/src/server/adapter.test.ts+294-7
......@@ -39,6 +39,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
3939import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
4040import { handoffBudget } from "@t3tools/provider-core/server/handoffBudget";
4141import * as HostProcess from "@t3tools/shared/HostProcess";
42import { toolOutputImages, compactDynamicToolOutput } from "@t3tools/shared/toolOutput";
4243import {
4344 makePiAdapterV2,
4445 PiAdapterV2Driver,
......@@ -46,6 +47,8 @@ import {
4647 type PiAdapterV2Options,
4748} from "./adapter.ts";
4849import { makePiRpcConnection, type PiRpcRecord } from "./rpc.ts";
50import { loadMcpBridge } from "./mcpBridge.testkit.ts";
51import { turnItemOutputText } from "../../../client-runtime/src/work-log/itemDetail.ts";
4952
5053const layerTest = Layer.mergeAll(
5154 NodeServices.layer,
......@@ -77,7 +80,7 @@ const modelSelection = (model: string): ModelSelection => ({
7780
7881interface FakePi {
7982 readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"];
80 readonly emit: (record: PiRpcRecord) => Effect.Effect<void>;
83 readonly emit: (record: PiRpcRecord, chunkBytes?: number) => Effect.Effect<void>;
8184 readonly takeRequest: (type: string) => Effect.Effect<PiRpcRecord>;
8285 /** Data returned by the next `get_entries` acks, consumed in order. */
8386 readonly queueEntries: (data: unknown) => void;
......@@ -153,10 +156,14 @@ const makeFakePi: Effect.Effect<FakePi> = Effect.gen(function* () {
153156 let models: ReadonlyArray<unknown> = [];
154157 let stdinBuffer = "";
155158
156 const emit = (record: PiRpcRecord) =>
157 Queue.offer(stdout, new TextEncoder().encode(`${encodeJsonLine(record)}\n`)).pipe(
158 Effect.asVoid,
159 );
159 const emit = (record: PiRpcRecord, chunkBytes?: number) =>
160 Effect.gen(function* () {
161 const bytes = new TextEncoder().encode(`${encodeJsonLine(record)}\n`);
162 const size = chunkBytes ?? bytes.byteLength;
163 for (let offset = 0; offset < bytes.byteLength; offset += size) {
164 yield* Queue.offer(stdout, bytes.subarray(offset, offset + size));
165 }
166 });
160167
161168 const respondTo = (record: PiRpcRecord): PiRpcRecord | null => {
162169 if (typeof record["id"] !== "string") return null;
......@@ -1852,6 +1859,284 @@ describe("PiAdapterV2", () => {
18521859 }).pipe(Effect.scoped, Effect.provide(layerTest)),
18531860 );
18541861
1862 it.effect(
1863 "keeps native and MCP images available to the shared asset reader with typed results",
1864 () =>
1865 Effect.gen(function* () {
1866 const fake = yield* makeFakePi;
1867 const { runtime, takeEvent } = yield* openRuntime(fake);
1868 const providerThread = yield* runtime.ensureThread({
1869 threadId: THREAD_ID,
1870 modelSelection: modelSelection("default"),
1871 runtimePolicy,
1872 });
1873 yield* startTurn(runtime, providerThread);
1874 yield* fake.takeRequest("prompt");
1875 yield* fake.emit({ type: "agent_start" });
1876 const image = {
1877 type: "image",
1878 mimeType: "image/png",
1879 data: Buffer.alloc(64 * 1024, 0x61).toString("base64"),
1880 };
1881 const note = "Read image file [image/png]";
1882 const content = [{ type: "text", text: note }, image];
1883 const structured = { count: 2, ready: true, paths: ["one.png", "two.png"] };
1884 const scriptResult = { content, structuredContent: { threadId: "child-thread" } };
1885 const cases = [
1886 {
1887 toolName: "read",
1888 result: { content, structuredContent: { ...image, note } },
1889 expected: { content },
1890 },
1891 {
1892 toolName: "image_generate",
1893 result: { content, structuredContent: structured },
1894 expected: { content, structuredContent: structured },
1895 },
1896 {
1897 toolName: "mcp__t3-code__preview_snapshot",
1898 result: {
1899 content,
1900 structuredContent: scriptResult,
1901 details: { server: "t3-code", tool: "preview_snapshot" },
1902 },
1903 expected: { content, structuredContent: { threadId: "child-thread" } },
1904 },
1905 {
1906 toolName: "structured_tool",
1907 result: { content: [], structuredContent: [false, 2, null, { ready: true }] },
1908 expected: { content: [], structuredContent: [false, 2, null, { ready: true }] },
1909 },
1910 {
1911 toolName: "mcp__custom__extension",
1912 result: { content, structuredContent: { content: ["domain content"], ready: true } },
1913 expected: { content, structuredContent: { content: ["domain content"], ready: true } },
1914 },
1915 {
1916 toolName: "mcp__t3-code__preview_snapshot",
1917 result: {
1918 content: [
1919 { type: "text", text: "first" },
1920 { type: "text", text: "second" },
1921 ],
1922 structuredContent: {
1923 content: [
1924 { type: "text", text: "first" },
1925 { type: "text", text: "second" },
1926 ],
1927 },
1928 details: { server: "t3-code", tool: "preview_snapshot" },
1929 },
1930 expected: {
1931 content: [
1932 { type: "text", text: "first" },
1933 { type: "text", text: "second" },
1934 ],
1935 },
1936 },
1937 ];
1938 for (const [index, test] of cases.entries()) {
1939 for (const phase of ["update", "end"] as const) {
1940 yield* fake.emit({
1941 type: `tool_execution_${phase}`,
1942 toolCallId: `image-${index}`,
1943 toolName: test.toolName,
1944 ...(phase === "end" ? { result: test.result } : { partialResult: test.result }),
1945 isError: false,
1946 });
1947 const event = yield* takeEvent(
1948 (event) =>
1949 event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
1950 );
1951 if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
1952 return yield* Effect.die("Expected an image tool item");
1953 assert.equal(event.turnItem.status, phase === "end" ? "completed" : "running");
1954 assert.deepEqual(event.turnItem.output, test.expected);
1955 assert.deepEqual(
1956 toolOutputImages(event.turnItem.output),
1957 test.expected.content.some((block) => block.type === "image")
1958 ? [{ mimeType: "image/png", data: image.data }]
1959 : [],
1960 );
1961 if (index === 0) assert.equal(turnItemOutputText(event.turnItem), note);
1962 if (index === 5) assert.equal(turnItemOutputText(event.turnItem), "first\nsecond");
1963 if (index === 2) {
1964 assert.deepEqual(compactDynamicToolOutput(event.turnItem.output), {
1965 threadId: "child-thread",
1966 });
1967 assert.equal(JSON.stringify(event.turnItem.output).split(image.data).length - 1, 1);
1968 }
1969 }
1970 }
1971 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1972 );
1973
1974 it.effect(
1975 "keeps large tool text readable, drops identical read mirrors, and bounds structured values",
1976 () =>
1977 Effect.gen(function* () {
1978 const fake = yield* makeFakePi;
1979 const { runtime, takeEvent } = yield* openRuntime(fake);
1980 const providerThread = yield* runtime.ensureThread({
1981 threadId: THREAD_ID,
1982 modelSelection: modelSelection("default"),
1983 runtimePolicy,
1984 });
1985 yield* startTurn(runtime, providerThread);
1986 yield* fake.takeRequest("prompt");
1987 yield* fake.emit({ type: "agent_start" });
1988 const text = "hello\nworld\n" + "🙂".repeat(40_000);
1989 const content = [{ type: "text", text }];
1990 const cases = [
1991 { content, structuredContent: text },
1992 { content, structuredContent: { oversized: "x".repeat(128 * 1024) } },
1993 { content, structuredContent: Array.from({ length: 10_000 }, () => 1) },
1994 {
1995 content,
1996 structuredContent: Array.from({ length: 40 }).reduce<unknown>(
1997 (nested) => ({ nested }),
1998 null,
1999 ),
2000 },
2001 ];
2002 for (const [index, result] of cases.entries()) {
2003 yield* fake.emit({
2004 type: "tool_execution_end",
2005 toolCallId: `structured-${index}`,
2006 toolName: "read",
2007 result,
2008 isError: true,
2009 });
2010 const event = yield* takeEvent(
2011 (event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2012 );
2013 if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2014 return yield* Effect.die("Expected a tool item");
2015 assert.equal(event.turnItem.status, "failed");
2016 assert.notProperty(event.turnItem.output, "structuredContent");
2017 const displayed = turnItemOutputText(event.turnItem);
2018 if (index === 0) {
2019 assert.equal(event.turnItem.output, text);
2020 assert.equal(displayed, text);
2021 } else {
2022 assert.equal(
2023 displayed,
2024 text + "\nStructured output omitted because it exceeds the stored result limit.",
2025 );
2026 assert.deepEqual(Object.keys(event.turnItem.output as object), ["content"]);
2027 }
2028 }
2029 // Distinct typed metadata remains stored even when readable text exists.
2030 yield* fake.emit({
2031 type: "tool_execution_end",
2032 toolCallId: "distinct-metadata",
2033 toolName: "read",
2034 result: { content, structuredContent: { path: "read.ts", ready: true } },
2035 });
2036 const event = yield* takeEvent(
2037 (event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2038 );
2039 if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2040 return yield* Effect.die("Expected a tool item");
2041 assert.deepEqual(event.turnItem.output, {
2042 content,
2043 structuredContent: { path: "read.ts", ready: true },
2044 });
2045 assert.equal(turnItemOutputText(event.turnItem), text);
2046 const note = "Read image file [image/png]";
2047 const image = { type: "image", data: "AAAA", mimeType: "image/png" };
2048 const imageContent = [{ type: "text", text: note }, image];
2049 // An extension can attach distinct typed metadata to a read result.
2050 for (const structuredContent of [
2051 { ...image, note, width: 1200 },
2052 { ...image, note: "A distinct note" },
2053 { ...image, note, data: "AQID" },
2054 ]) {
2055 yield* fake.emit({
2056 type: "tool_execution_end",
2057 toolCallId: "distinct-image-metadata",
2058 toolName: "read",
2059 result: { content: imageContent, structuredContent },
2060 });
2061 const event = yield* takeEvent(
2062 (event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2063 );
2064 if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2065 return yield* Effect.die("Expected a read image tool item");
2066 assert.deepEqual(event.turnItem.output, { content: imageContent, structuredContent });
2067 assert.equal(turnItemOutputText(event.turnItem), note);
2068 }
2069 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2070 );
2071
2072 it.effect(
2073 "delivers realistic MCP screenshots through the bridge, JSONL transport, and adapter",
2074 () =>
2075 Effect.gen(function* () {
2076 const fake = yield* makeFakePi;
2077 const { runtime, takeEvent } = yield* openRuntime(fake);
2078 const providerThread = yield* runtime.ensureThread({
2079 threadId: THREAD_ID,
2080 modelSelection: modelSelection("default"),
2081 runtimePolicy,
2082 });
2083 yield* startTurn(runtime, providerThread);
2084 yield* fake.takeRequest("prompt");
2085 yield* fake.emit({ type: "agent_start" });
2086 for (const size of [3.25 * 1024 * 1024, 10 * 1024 * 1024]) {
2087 const pixels = Buffer.alloc(size, 0x61);
2088 const data = pixels.toString("base64");
2089 const content = [
2090 { type: "text", text: "Screenshot captured" },
2091 { type: "image", data, mimeType: "image/png" },
2092 ];
2093 const metadata = { screenshot: { width: 1200, height: 800 }, ready: true };
2094 const bridge = yield* Effect.promise(() =>
2095 loadMcpBridge({ modern: true, result: { content, structuredContent: metadata } }),
2096 );
2097 const tool = bridge.tools.find((tool) => tool.name === "mcp__t3-code__preview_snapshot")!;
2098 const result = yield* Effect.promise(() => tool.execute("screenshot", {}));
2099 assert.deepEqual(result.structuredContent, { content, structuredContent: metadata });
2100 yield* fake.emit(
2101 {
2102 type: "tool_execution_end",
2103 toolCallId: "screenshot",
2104 toolName: tool.name,
2105 result,
2106 isError: false,
2107 },
2108 64 * 1024,
2109 );
2110 // A following small completion proves a dropped screenshot immediately,
2111 // without waiting for a timeout when framing loses the large event.
2112 yield* fake.emit({
2113 type: "tool_execution_end",
2114 toolCallId: "after-screenshot",
2115 toolName: "marker",
2116 result: { content: [{ type: "text", text: "after" }] },
2117 });
2118 const event = yield* takeEvent(
2119 (event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2120 );
2121 if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2122 return yield* Effect.die("Expected a screenshot tool item");
2123 assert.equal(event.turnItem.toolName, tool.name);
2124 assert.equal(event.turnItem.status, "completed");
2125 assert.deepEqual(event.turnItem.output, { content, structuredContent: metadata });
2126 assert.equal(turnItemOutputText(event.turnItem), "Screenshot captured");
2127 const image = toolOutputImages(event.turnItem.output)[0];
2128 assert.equal(image?.mimeType, "image/png");
2129 assert.deepEqual(Buffer.from(image!.data!, "base64"), pixels);
2130 yield* takeEvent(
2131 (event) =>
2132 event.type === "turn_item.updated" &&
2133 event.turnItem.type === "dynamic_tool" &&
2134 event.turnItem.toolName === "marker",
2135 );
2136 }
2137 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2138 );
2139
18552140 it.effect("observes official subagent results without inventing child threads", () =>
18562141 Effect.gen(function* () {
18572142 const fake = yield* makeFakePi;
......@@ -3512,8 +3797,10 @@ describe("PiRpc framing", () => {
35123797 yield* push('{"type":"agent_');
35133798 yield* push('start"}\r\n{"type":"agent_settled"}\nnot json\n{"type":"queue_update"}\n');
35143799
3515 yield* push("x".repeat(8 * 1024 * 1024));
3516 yield* push('x{"type":"must_not_emit"}\n{"type":"after_oversized"}\n');
3800 yield* push('{"type":"must_not_emit","text":"');
3801 const chunk = new TextEncoder().encode("x".repeat(8 * 1024 * 1024));
3802 for (let index = 0; index < 32; index++) yield* Queue.offer(stdout, chunk);
3803 yield* push('"}\n{"type":"after_oversized"}\n');
35173804
35183805 const first = yield* Queue.take(connection.events);
35193806 assert.equal(first["type"], "agent_start");
packages/provider-pi/src/server/adapter.ts+102-1
......@@ -26,6 +26,8 @@
2626import * as HostProcess from "@t3tools/shared/HostProcess";
2727import { AgentScope } from "@t3tools/shared/AgentScope";
2828import { getModelSelectionStringOptionValue } from "@t3tools/shared/model";
29import { readToolOutputImage } from "@t3tools/shared/toolOutput";
30import * as Predicate from "effect/Predicate";
2931import {
3032 defaultInstanceIdForDriver,
3133 ProviderDriverKind,
......@@ -247,6 +249,104 @@ function contentText(content: unknown): string {
247249 .join("");
248250}
249251
252const MAX_STRUCTURED_TOOL_OUTPUT_BYTES = 64 * 1024;
253
254/** Preserve complete structured JSON only when its size and traversal remain bounded. */
255function isBoundedStructuredToolValue(value: unknown): boolean {
256 let bytes = MAX_STRUCTURED_TOOL_OUTPUT_BYTES;
257 let nodes = 4_096;
258 const visit = (entry: unknown, depth: number): boolean => {
259 if (depth > 32 || --nodes < 0 || bytes < 0) return false;
260 if (Array.isArray(entry)) {
261 bytes -= 2 + entry.length;
262 return entry.every((item) => visit(item, depth + 1));
263 }
264 if (Predicate.isObject(entry)) {
265 bytes -= 2;
266 for (const key in entry) {
267 if (!Object.hasOwn(entry, key)) continue;
268 if (key.length > bytes) return false;
269 bytes -= Buffer.byteLength(JSON.stringify(key), "utf8") + 2;
270 if (!visit(entry[key], depth + 1)) return false;
271 }
272 return bytes >= 0;
273 }
274 if (typeof entry === "string" && entry.length > bytes) return false;
275 bytes -= Buffer.byteLength(JSON.stringify(entry) ?? "null", "utf8");
276 return bytes >= 0;
277 };
278 return visit(value, 0);
279}
280
281/** Keep model content and distinct typed values; server ingestion owns image byte policy. */
282function piToolOutput(result: unknown, toolName: string): unknown {
283 const rawContent = recordField(result, "content");
284 const content = Array.isArray(rawContent)
285 ? rawContent
286 : typeof rawContent === "string"
287 ? [{ type: "text", text: rawContent }]
288 : [];
289 let structured = recordField(result, "structuredContent");
290 // Pi MCP tools return the full CallToolResult to scripts. The model content
291 // is already captured, so store only its distinct structured payload.
292 const details = recordField(result, "details");
293 const isMcpResult =
294 toolName.startsWith("mcp__") &&
295 recordString(details, "server") !== undefined &&
296 recordString(details, "tool") !== undefined &&
297 Array.isArray(recordField(structured, "content"));
298 if (isMcpResult) {
299 structured = recordField(structured, "structuredContent");
300 }
301 if (typeof structured === "string" && structured === contentText(content)) {
302 structured = undefined;
303 }
304 // Native read repeats its image and first text note as structured output.
305 // Drop only that exact mirror; an extension's extra metadata stays distinct.
306 if (toolName === "read" && Predicate.isObject(structured)) {
307 const image = readToolOutputImage(structured);
308 const note =
309 recordString(
310 content.find((block) => recordField(block, "type") === "text"),
311 "text",
312 ) ?? "";
313 if (
314 image !== null &&
315 Object.keys(structured).every((key) => ["type", "data", "mimeType", "note"].includes(key)) &&
316 recordString(structured, "note") === note &&
317 content.some((block) => {
318 const match = readToolOutputImage(block);
319 return match !== null && match.mimeType === image.mimeType && match.data === image.data;
320 })
321 ) {
322 structured = undefined;
323 }
324 }
325 if (structured !== undefined && !isBoundedStructuredToolValue(structured)) {
326 return {
327 content: [
328 ...content,
329 {
330 type: "text",
331 text: "Structured output omitted because it exceeds the stored result limit.",
332 },
333 ],
334 };
335 }
336 if (
337 !isMcpResult &&
338 structured === undefined &&
339 content.every((block) => recordField(block, "type") === "text")
340 ) {
341 const text = contentText(content);
342 return text.length > 0 ? text : undefined;
343 }
344 return {
345 content,
346 ...(structured === undefined ? {} : { structuredContent: structured }),
347 };
348}
349
250350function providerRef(
251351 nativeId: string,
252352 strength: "strong" | "weak" = "strong",
......@@ -1047,6 +1147,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
10471147 return;
10481148 }
10491149 }
1150 const output = piToolOutput(resultRecord, toolName);
10501151 yield* emit({
10511152 type: "turn_item.updated",
10521153 driver: PI_PROVIDER,
......@@ -1057,7 +1158,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
10571158 ...mcpToolPresentation({ toolName }),
10581159 toolName,
10591160 input: args ?? {},
1060 ...(outputText.length > 0 ? { output: outputText } : {}),
1161 ...(output === undefined ? {} : { output }),
10611162 },
10621163 });
10631164 if (toolName === "subagent") {
packages/provider-pi/src/server/mcpBridge.testkit.ts created+140
......@@ -0,0 +1,140 @@
1// @effect-diagnostics nodeBuiltinImport:off -- Executes the shipped Pi extension at its native JavaScript boundary.
2import * as NodeModule from "node:module";
3import * as NodeVM from "node:vm";
4import { PI_T3_MCP_EXTENSION_SOURCE } from "./mcpExtensionSource.ts";
5
6interface RegisteredTool {
7 readonly name: string;
8 readonly description: string;
9 readonly parameters: unknown;
10 readonly outputSchema?: unknown;
11 readonly exposure?: string;
12 readonly promptSnippet?: string;
13 readonly promptGuidelines?: ReadonlyArray<string>;
14 readonly execute: (
15 id: string,
16 args: Record<string, unknown>,
17 signal?: AbortSignal,
18 ) => Promise<{
19 readonly content: ReadonlyArray<{
20 readonly type: string;
21 readonly text?: string;
22 readonly data?: string;
23 readonly mimeType?: string;
24 }>;
25 readonly structuredContent?: unknown;
26 readonly isError?: boolean;
27 }>;
28}
29
30type AgentStartHook = (
31 event: { systemPrompt: string },
32 ctx: { ui: { notify: (message: string, severity: string) => void } },
33) => Promise<{ systemPrompt: string }>;
34
35export async function loadMcpBridge(
36 options: {
37 readonly modern?: boolean;
38 readonly result?: unknown;
39 readonly toolSearchAvailable?: boolean;
40 readonly toolSearchDisabled?: boolean;
41 readonly allowsTool?: (name: string) => boolean;
42 } = {},
43) {
44 const handlers = new Map<string, AgentStartHook>();
45 const tools: RegisteredTool[] = [];
46 const requests: Array<{ readonly method: string; readonly params?: unknown }> = [];
47 let activeTools = ["read"];
48 const transports: Array<{
49 readonly url: string;
50 readonly authorization: string;
51 readonly signal: AbortSignal | undefined;
52 }> = [];
53 const servers: Array<{ readonly name: string; readonly config: Record<string, unknown> }> = [];
54 const catalog = [
55 { name: "orchestrator_capabilities", description: "Discover available providers and models." },
56 { name: "delegate_task", description: "Delegate work to another agent." },
57 { name: "task_status", description: "Check delegated work." },
58 { name: "preview_snapshot", description: "Inspect the collaborative browser." },
59 ].map((tool) => ({
60 ...tool,
61 inputSchema: { type: "object", properties: {} },
62 outputSchema: { type: "object", properties: { count: { type: "number" } } },
63 }));
64 const source = NodeModule.stripTypeScriptTypes(
65 PI_T3_MCP_EXTENSION_SOURCE.replace(/^import .*;$/gm, "").replace(
66 "export default async function",
67 "async function",
68 ),
69 );
70 await NodeVM.runInNewContext(`${source}\nt3McpExtension(pi)`, {
71 process: {
72 env: { T3_MCP_URL: "http://fixture.invalid/mcp", T3_MCP_BEARER_TOKEN: "fixture-token" },
73 },
74 AbortSignal,
75 Type: { Unsafe: (schema: unknown) => schema },
76 fetch: async (
77 url: string,
78 transport: { body: string; headers: Record<string, string>; signal?: AbortSignal },
79 ) => {
80 transport.signal?.throwIfAborted();
81 transports.push({
82 url,
83 authorization: transport.headers.authorization!,
84 signal: transport.signal,
85 });
86 const request = JSON.parse(transport.body) as {
87 id: number;
88 method: string;
89 params?: unknown;
90 };
91 requests.push(request);
92 const result =
93 request.method === "tools/list"
94 ? { tools: catalog }
95 : request.method === "tools/call"
96 ? (options.result ?? { content: [{ type: "text", text: "browser snapshot" }] })
97 : {};
98 return new Response(JSON.stringify({ jsonrpc: "2.0", id: request.id, result }), {
99 headers: { "content-type": "application/json" },
100 });
101 },
102 pi: {
103 on: (name: string, handler: AgentStartHook) => handlers.set(name, handler),
104 registerTool: (tool: RegisteredTool) => {
105 if (options.allowsTool && !options.allowsTool(tool.name)) return;
106 const index = tools.findIndex((current) => current.name === tool.name);
107 if (index === -1) tools.push(tool);
108 else tools[index] = tool;
109 },
110 getActiveTools: () => activeTools,
111 setActiveTools: (names: string[]) => {
112 activeTools = names.filter((name) => options.allowsTool?.(name) ?? true);
113 },
114 getAllTools: () =>
115 options.toolSearchAvailable &&
116 !options.toolSearchDisabled &&
117 (options.allowsTool?.("tool_search") ?? true)
118 ? [{ name: "tool_search", sourceInfo: { path: "builtin:tool-search" } }]
119 : [],
120 ...(options.modern
121 ? {
122 registerMcpServer: (name: string, config: Record<string, unknown>) =>
123 servers.push({ name, config }),
124 unregisterMcpServer: () => servers.splice(0),
125 }
126 : {}),
127 },
128 });
129 return {
130 handlers,
131 tools,
132 requests,
133 servers,
134 transports,
135 getActiveTools: () => activeTools,
136 restoreActiveTools: (names: string[]) => {
137 activeTools = names;
138 },
139 };
140}
packages/provider-pi/src/server/mcpExtensionSource.test.ts+60-115
......@@ -4,9 +4,10 @@ import * as NodePath from "node:path";
44import * as NodeOS from "node:os";
55import * as NodeModule from "node:module";
66import * as NodeVM from "node:vm";
7import { assert, describe, it } from "@effect/vitest";
7import { assert, describe, expect, it } from "@effect/vitest";
88
99import { PI_T3_MCP_EXTENSION_SOURCE } from "./mcpExtensionSource.ts";
10import { loadMcpBridge } from "./mcpBridge.testkit.ts";
1011
1112type RequestHook = (
1213 event: { payload: unknown },
......@@ -47,123 +48,67 @@ async function loadHooks(commands: ReadonlyArray<SkillCommand> = []) {
4748 return handlers;
4849}
4950
50interface RegisteredTool {
51 readonly name: string;
52 readonly description: string;
53 readonly parameters: unknown;
54 readonly exposure?: string;
55 readonly promptSnippet?: string;
56 readonly promptGuidelines?: ReadonlyArray<string>;
57 readonly execute: (
58 id: string,
59 args: Record<string, unknown>,
60 signal?: AbortSignal,
61 ) => Promise<{
62 readonly content: ReadonlyArray<{ readonly type: string; readonly text: string }>;
63 }>;
64}
65
66type AgentStartHook = (
67 event: { systemPrompt: string },
68 ctx: { ui: { notify: (message: string, severity: string) => void } },
69) => Promise<{ systemPrompt: string }>;
70
71async function loadMcpBridge(
72 options: {
73 readonly modern?: boolean;
74 readonly toolSearchAvailable?: boolean;
75 readonly toolSearchDisabled?: boolean;
76 readonly allowsTool?: (name: string) => boolean;
77 } = {},
78) {
79 const handlers = new Map<string, AgentStartHook>();
80 const tools: RegisteredTool[] = [];
81 const requests: Array<{ readonly method: string; readonly params?: unknown }> = [];
82 let activeTools = ["read"];
83 const transports: Array<{
84 readonly url: string;
85 readonly authorization: string;
86 readonly signal: AbortSignal | undefined;
87 }> = [];
88 const servers: Array<{ readonly name: string; readonly config: Record<string, unknown> }> = [];
89 const catalog = [
90 { name: "orchestrator_capabilities", description: "Discover available providers and models." },
91 { name: "delegate_task", description: "Delegate work to another agent." },
92 { name: "task_status", description: "Check delegated work." },
93 { name: "preview_snapshot", description: "Inspect the collaborative browser." },
94 ].map((tool) => ({ ...tool, inputSchema: { type: "object", properties: {} } }));
95 const source = NodeModule.stripTypeScriptTypes(
96 PI_T3_MCP_EXTENSION_SOURCE.replace(/^import .*;$/gm, "").replace(
97 "export default async function",
98 "async function",
99 ),
100 );
101 await NodeVM.runInNewContext(`${source}\nt3McpExtension(pi)`, {
102 process: {
103 env: { T3_MCP_URL: "http://fixture.invalid/mcp", T3_MCP_BEARER_TOKEN: "fixture-token" },
104 },
105 AbortSignal,
106 Type: { Unsafe: (schema: unknown) => schema },
107 fetch: async (
108 url: string,
109 options: { body: string; headers: Record<string, string>; signal?: AbortSignal },
110 ) => {
111 transports.push({
112 url,
113 authorization: options.headers.authorization!,
114 signal: options.signal,
115 });
116 const request = JSON.parse(options.body) as { id: number; method: string; params?: unknown };
117 requests.push(request);
118 const result =
119 request.method === "tools/list"
120 ? { tools: catalog }
121 : request.method === "tools/call"
122 ? { content: [{ type: "text", text: "browser snapshot" }] }
123 : {};
124 return new Response(JSON.stringify({ jsonrpc: "2.0", id: request.id, result }), {
125 headers: { "content-type": "application/json" },
126 });
127 },
128 pi: {
129 on: (name: string, handler: AgentStartHook) => handlers.set(name, handler),
130 registerTool: (tool: RegisteredTool) => {
131 if (options.allowsTool && !options.allowsTool(tool.name)) return;
132 const index = tools.findIndex((current) => current.name === tool.name);
133 if (index === -1) tools.push(tool);
134 else tools[index] = tool;
135 },
136 getActiveTools: () => activeTools,
137 setActiveTools: (names: string[]) => {
138 activeTools = names.filter((name) => options.allowsTool?.(name) ?? true);
51describe("Pi MCP tool results", () => {
52 it("returns mixed screenshot blocks and typed script output without repeating structured text", async () => {
53 const content = [
54 { type: "text", text: "Browser screenshot" },
55 { type: "image", data: "AAAA", mimeType: "image/png", _meta: { private: true } },
56 { type: "text", text: "Screenshot captured" },
57 ] as const;
58 const result = { content, structuredContent: { count: 2 }, _meta: { private: true } };
59 const bridge = await loadMcpBridge({ modern: true, result });
60 const tool = bridge.tools.find((tool) => tool.name === "mcp__t3-code__preview_snapshot")!;
61 const output = await tool.execute("snapshot", {});
62 assert.deepEqual(output.content, [
63 content[0],
64 { type: "image", data: "AAAA", mimeType: "image/png" },
65 content[2],
66 ]);
67 assert.deepEqual(output.structuredContent, { content, structuredContent: { count: 2 } });
68 assert.deepEqual(tool.outputSchema, {
69 type: "object",
70 properties: {
71 content: { type: "array", items: { type: "object" } },
72 structuredContent: { type: "object", properties: { count: { type: "number" } } },
73 isError: { type: "boolean" },
74 _meta: { type: "object" },
13975 },
140 getAllTools: () =>
141 options.toolSearchAvailable &&
142 !options.toolSearchDisabled &&
143 (options.allowsTool?.("tool_search") ?? true)
144 ? [{ name: "tool_search", sourceInfo: { path: "builtin:tool-search" } }]
145 : [],
146 ...(options.modern
147 ? {
148 registerMcpServer: (name: string, config: Record<string, unknown>) =>
149 servers.push({ name, config }),
150 unregisterMcpServer: () => servers.splice(0),
151 }
152 : {}),
153 },
76 required: ["content"],
77 });
15478 });
155 return {
156 handlers,
157 tools,
158 requests,
159 servers,
160 transports,
161 getActiveTools: () => activeTools,
162 restoreActiveTools: (names: string[]) => {
163 activeTools = names;
79
80 it.each([false, true])(
81 "preserves image-only results and error status, isError=%s",
82 async (isError) => {
83 const image = { type: "image", data: "AAAA", mimeType: "image/png" };
84 const result = { content: [image], isError };
85 const bridge = await loadMcpBridge({ result });
86 const output = await bridge.tools[0]!.execute("image", {});
87 assert.deepEqual(output.content[0], image);
88 assert.equal(output.content.length, isError ? 2 : 1);
89 if (isError) assert.include(output.content[1]?.text ?? "", "returned an error");
90 assert.equal(output.isError, isError ? true : undefined);
91 assert.deepEqual(output.structuredContent, result);
16492 },
165 };
166}
93 );
94
95 it("uses structured output as model text when content is empty", async () => {
96 const result = { content: [], structuredContent: { count: 2 } };
97 const bridge = await loadMcpBridge({ result });
98 const output = await bridge.tools[0]!.execute("structured", {});
99 assert.deepEqual(output.content, [{ type: "text", text: '{"count":2}' }]);
100 assert.deepEqual(output.structuredContent, result);
101 });
102
103 it("propagates cancellation without returning a successful result", async () => {
104 const bridge = await loadMcpBridge();
105 const controller = new AbortController();
106 controller.abort();
107 await expect(bridge.tools[0]!.execute("cancelled", {}, controller.signal)).rejects.toThrow(
108 "aborted",
109 );
110 });
111});
167112
168113describe("Pi MCP tool exposure", () => {
169114 it("keeps orchestration direct and optional bridge tools discoverable on modern Pi", async () => {
packages/provider-pi/src/server/mcpExtensionSource.ts+54-17
......@@ -38,9 +38,11 @@ const FILE_CHANGE_TOOLS = new Set(${JSON.stringify(PI_FILE_CHANGE_TOOLS)});
3838
3939type RuntimeMode = "approval-required" | "auto-accept-edits" | "auto" | "full-access";
4040
41type JsonValue = string | number | boolean | null | JsonValue[] | { [key: string]: JsonValue };
42
4143type JsonRpcResponse = {
4244 readonly id?: number | string;
43 readonly result?: unknown;
45 readonly result?: JsonValue;
4446 readonly error?: { readonly message?: string };
4547};
4648
......@@ -48,6 +50,7 @@ type McpTool = {
4850 readonly name: string;
4951 readonly description?: string;
5052 readonly inputSchema?: Record<string, unknown>;
53 readonly outputSchema?: Record<string, unknown>;
5154};
5255
5356function env(name: string): string | undefined {
......@@ -96,28 +99,57 @@ function jsonSchemaToTypebox(schema: Record<string, unknown> | undefined) {
9699 return Type.Object({}, { additionalProperties: true });
97100}
98101
99function formatMcpContent(result: unknown): string {
100 if (result === null || result === undefined) return "";
101 if (typeof result !== "object") return String(result);
102type ModelContent =
103 | { type: "text"; text: string }
104 | { type: "image"; data: string; mimeType: string };
105
106function mcpModelContent(result: unknown): ModelContent[] {
107 if (result === null || result === undefined) return [];
108 if (typeof result !== "object") return [{ type: "text", text: String(result) }];
102109 const record = result as {
103 readonly content?: ReadonlyArray<{ readonly type?: string; readonly text?: string }>;
110 readonly content?: ReadonlyArray<{
111 readonly type?: string;
112 readonly text?: string;
113 readonly data?: string;
114 readonly mimeType?: string;
115 }>;
104116 readonly structuredContent?: unknown;
105 readonly isError?: boolean;
106117 };
107 const texts: string[] = [];
118 const content: ModelContent[] = [];
108119 if (Array.isArray(record.content)) {
109120 for (const part of record.content) {
110 if (part?.type === "text" && typeof part.text === "string") texts.push(part.text);
121 if (part?.type === "text" && typeof part.text === "string") {
122 content.push({ type: "text", text: part.text });
123 } else if (part?.type === "image" && typeof part.data === "string" && typeof part.mimeType === "string") {
124 content.push({ type: "image", data: part.data, mimeType: part.mimeType });
125 }
111126 }
112127 }
113 // Most T3 tools mirror structuredContent in a text block. Repeating it would
114 // leave T3's own output parsing two JSON documents instead of one.
115 if (record.structuredContent !== undefined) {
116 const structured = JSON.stringify(record.structuredContent);
117 if (!texts.includes(structured)) texts.push(structured);
128 // Structured output belongs to scripts. Only use it as model text when
129 // the server supplied no model-facing blocks, matching Pi's native MCP tools.
130 if (content.length === 0) {
131 content.push({ type: "text", text: JSON.stringify(record.structuredContent ?? result) });
118132 }
119 if (texts.length > 0) return texts.join("\\n");
120 return JSON.stringify(result);
133 return content;
134}
135
136function mcpScriptResult(result: JsonValue | undefined) {
137 if (typeof result !== "object" || result === null || Array.isArray(result)) return result;
138 const { _meta: _ignored, ...scriptResult } = result;
139 return scriptResult;
140}
141
142function mcpOutputSchema(tool: McpTool) {
143 return jsonSchemaToTypebox({
144 type: "object",
145 properties: {
146 content: { type: "array", items: { type: "object" } },
147 ...(tool.outputSchema === undefined ? {} : { structuredContent: tool.outputSchema }),
148 isError: { type: "boolean" },
149 _meta: { type: "object" },
150 },
151 required: ["content"],
152 });
121153}
122154
123155function isMcpToolError(result: unknown): boolean {
......@@ -328,6 +360,7 @@ export default async function t3McpExtension(pi: ExtensionAPI) {
328360 label: name,
329361 description: tool.description ?? name,
330362 parameters: jsonSchemaToTypebox(tool.inputSchema),
363 ...(supportsExposure ? { outputSchema: mcpOutputSchema(tool) } : {}),
331364 ...(supportsExposure ? { exposure } : {}),
332365 async execute(_toolCallId, params, signal) {
333366 const result = await client.callTool(
......@@ -335,9 +368,13 @@ export default async function t3McpExtension(pi: ExtensionAPI) {
335368 (params ?? {}) as Record<string, unknown>,
336369 signal,
337370 );
338 const text = formatMcpContent(result);
371 const content = mcpModelContent(result);
372 if (isMcpToolError(result) && !content.some((part) => part.type === "text" && part.text.length > 0)) {
373 content.push({ type: "text", text: "MCP tool t3-code/" + name + " returned an error" });
374 }
339375 return {
340 content: [{ type: "text", text }],
376 content,
377 structuredContent: mcpScriptResult(result),
341378 details: { server: "t3-code", tool: name },
342379 ...(isMcpToolError(result) ? { isError: true } : {}),
343380 };
packages/provider-pi/src/server/rpc.ts+9-1
......@@ -30,6 +30,10 @@ import * as ChildProcess from "effect/process/ChildProcess";
3030import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
3131import * as HostProcess from "@t3tools/shared/HostProcess";
3232import { resolveSpawnCommand } from "@t3tools/shared/shell";
33import {
34 MAX_TOOL_OUTPUT_IMAGES,
35 MAX_TOOL_OUTPUT_IMAGE_BASE64_LENGTH,
36} from "@t3tools/shared/toolOutput";
3337
3438import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup";
3539
......@@ -125,7 +129,11 @@ interface PendingPiRequest {
125129 readonly deferred: Deferred.Deferred<unknown, PiRpcError>;
126130}
127131
128const MAX_PI_RECORD_CHARS = 8 * 1024 * 1024;
132// MCP completion events carry images in both model content and the script
133// CallToolResult. Allow the shared image budget twice, plus bounded JSON/text
134// overhead, before dropping an extension's oversized record.
135const MAX_PI_RECORD_CHARS =
136 2 * MAX_TOOL_OUTPUT_IMAGES * MAX_TOOL_OUTPUT_IMAGE_BASE64_LENGTH + 8 * 1024 * 1024;
129137
130138function makeJsonlFramer() {
131139 let buffer = "";