diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index b3495c47ed06..cca7daa0f97c 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts @@ -2,6 +2,7 @@ import { assert, describe, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { CheckpointId, + EnvironmentId, NodeId, ProviderInstanceId, ProviderSessionId, @@ -23,9 +24,10 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; -import { ChildProcessSpawner } from "effect/unstable/process"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { ServerConfig } from "../../config.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { IdAllocatorV2, layer as idAllocatorLayer } from "../IdAllocator.ts"; import { ProviderAdapterV2RuntimePolicy, @@ -75,6 +77,10 @@ interface FakePi { readonly queueEntries: (data: unknown) => void; /** Make the next `switch_session` ack report an extension veto. */ readonly vetoNextSwitch: () => void; + readonly lastSpawn: () => { + readonly args: ReadonlyArray; + readonly env: NodeJS.ProcessEnv; + }; } /** @@ -144,9 +150,19 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { } }); - const spawner = ChildProcessSpawner.make(() => - Effect.succeed( - ChildProcessSpawner.makeHandle({ + let lastSpawn: { readonly args: ReadonlyArray; readonly env: NodeJS.ProcessEnv } = { + args: [], + env: {}, + }; + const spawner = ChildProcessSpawner.make((command) => + Effect.sync(() => { + if (ChildProcess.isStandardCommand(command)) { + lastSpawn = { + args: command.args, + env: command.options.env ?? {}, + }; + } + return ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(FAKE_PID), exitCode: Effect.never, isRunning: Effect.succeed(true), @@ -158,8 +174,8 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { all: Stream.empty, getInputFd: () => Sink.drain, getOutputFd: () => Stream.empty, - }), - ), + }); + }), ); const takeRequest = (type: string): Effect.Effect => @@ -178,6 +194,7 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { vetoNextSwitch: () => { vetoSwitch = true; }, + lastSpawn: () => lastSpawn, } satisfies FakePi; }); @@ -288,10 +305,39 @@ describe("PiAdapterV2", () => { assert.isFalse(PiProviderCapabilitiesV2.turns.supportsSteeringByInterruptRestart); assert.equal(PiProviderCapabilitiesV2.turns.terminalStatusQuality, "strong"); assert.isFalse(PiProviderCapabilitiesV2.approvals.supportsCommandApproval); - assert.isFalse(PiProviderCapabilitiesV2.tools.supportsMcpTools); + assert.isTrue(PiProviderCapabilitiesV2.tools.supportsMcpTools); + assert.isTrue(PiProviderCapabilitiesV2.subagents.exposesSubagentThreadIds); assert.equal(PiProviderCapabilitiesV2.identity.nativeThreadIds, "strong"); }); + it.effect("injects the T3 MCP extension and bearer when a session exists", () => + Effect.gen(function* () { + McpProviderSession.setMcpProviderSession({ + environmentId: EnvironmentId.make("environment-pi-mcp"), + threadId: THREAD_ID, + providerSessionId: "mcp-session-pi", + providerInstanceId: PI_INSTANCE_ID, + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer secret-pi-token", + }); + const fake = yield* makeFakePi; + yield* openRuntime(fake); + const spawn = fake.lastSpawn(); + assert.isTrue(spawn.args.includes("--extension")); + const extensions = spawn.args.flatMap((arg, index) => + arg === "--extension" ? [spawn.args[index + 1]] : [], + ); + assert.isTrue(extensions.some((path) => path?.endsWith("pi-t3-subagent-extension.ts"))); + assert.isTrue(extensions.some((path) => path?.endsWith("pi-t3-mcp-extension.ts"))); + assert.equal(spawn.env.T3_MCP_URL, "http://127.0.0.1:43123/mcp"); + assert.equal(spawn.env.T3_MCP_BEARER_TOKEN, "secret-pi-token"); + }).pipe( + Effect.ensuring(Effect.sync(() => McpProviderSession.clearMcpProviderSession(THREAD_ID))), + Effect.scoped, + Effect.provide(testLayer), + ), + ); + it.effect("registers the thread from get_state and resumes via switch_session", () => Effect.gen(function* () { const fake = yield* makeFakePi; @@ -303,6 +349,13 @@ describe("PiAdapterV2", () => { }); assert.equal(providerThread.nativeThreadRef?.nativeId, FAKE_SESSION_FILE); assert.equal(providerThread.driver, PI_PROVIDER); + const spawn = fake.lastSpawn(); + assert.isTrue( + spawn.args.some( + (arg, index) => + arg === "--extension" && spawn.args[index + 1]?.endsWith("pi-t3-subagent-extension.ts"), + ), + ); yield* runtime.resumeThread({ providerThread }); const switchRequest = yield* fake.takeRequest("switch_session"); @@ -673,6 +726,7 @@ describe("PiAdapterV2", () => { task: "map the repo", exitCode: 0, stderr: "", + sessionFile: "/tmp/pi-children/scout.jsonl", messages: [ { role: "assistant", content: [{ type: "text", text: "scanning files" }] }, ], @@ -681,6 +735,23 @@ describe("PiAdapterV2", () => { }, }, }); + const childThread = yield* takeEvent((event) => event.type === "app_thread.created"); + if (childThread.type !== "app_thread.created") { + assert.fail("expected app_thread.created"); + return; + } + const childThreadId = childThread.appThread.id; + const childProviderThread = yield* takeEvent( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.appThreadId === childThreadId, + ); + assert.isTrue( + childProviderThread.type === "provider_thread.updated" && + childProviderThread.providerThread.nativeThreadRef?.nativeId === + "/tmp/pi-children/scout.jsonl" && + childProviderThread.providerThread.providerSessionId === null, + ); const running = yield* takeEvent( (event) => event.type === "subagent.updated" && event.subagent.status === "running", ); @@ -688,7 +759,8 @@ describe("PiAdapterV2", () => { running.type === "subagent.updated" && running.subagent.title === "scout" && running.subagent.prompt === "map the repo" && - running.subagent.progress === "scanning files", + running.subagent.progress === "scanning files" && + running.subagent.childThreadId === childThreadId, ); yield* fake.emit({ type: "tool_execution_end", @@ -706,6 +778,7 @@ describe("PiAdapterV2", () => { exitCode: 0, stopReason: "stop", stderr: "", + sessionFile: "/tmp/pi-children/scout.jsonl", messages: [ { role: "assistant", content: [{ type: "text", text: "repo has one file" }] }, ], @@ -725,7 +798,22 @@ describe("PiAdapterV2", () => { (event) => event.type === "subagent.updated" && event.subagent.status === "completed", ); assert.isTrue( - doneCard.type === "subagent.updated" && doneCard.subagent.result === "repo has one file", + doneCard.type === "subagent.updated" && + doneCard.subagent.result === "repo has one file" && + doneCard.subagent.childThreadId === childThreadId, + ); + // Completed turn_item is emitted immediately after the completed card; + // waiting for the failed card first would consume it. + const subagentItem = yield* takeEvent( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "subagent" && + event.turnItem.status === "completed", + ); + assert.isTrue( + subagentItem.type === "turn_item.updated" && + subagentItem.turnItem.type === "subagent" && + subagentItem.turnItem.childThreadId === childThreadId, ); const failedCard = yield* takeEvent( (event) => event.type === "subagent.updated" && event.subagent.status === "failed", @@ -733,12 +821,9 @@ describe("PiAdapterV2", () => { assert.isTrue( failedCard.type === "subagent.updated" && failedCard.subagent.title === "worker" && - failedCard.subagent.result === "boom", - ); - const subagentItem = yield* takeEvent( - (event) => event.type === "turn_item.updated" && event.turnItem.type === "subagent", + failedCard.subagent.result === "boom" && + failedCard.subagent.childThreadId === null, ); - assert.equal(subagentItem.type, "turn_item.updated"); }).pipe(Effect.scoped, Effect.provide(testLayer)), ); diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts index d454d4757d59..4227a0eb088d 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -23,7 +23,6 @@ * are dropped until a dedicated Pi panel exists. */ import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; -import { tokenizeCliArgs } from "@t3tools/shared/cliArgs"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { defaultInstanceIdForDriver, @@ -37,6 +36,7 @@ import { type OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, type OrchestrationV2ProviderTurn, + type ThreadId, type OrchestrationV2RuntimeRequest, type OrchestrationV2TurnItem, type OrchestrationV2UserInputQuestion, @@ -56,6 +56,8 @@ import { ChildProcessSpawner } from "effect/unstable/process"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import { t3OrchestrationPromptForFirstRun } from "../../provider/T3OrchestrationInstructions.ts"; import { mergeProviderInstanceEnvironment } from "../../provider/ProviderInstanceEnvironment.ts"; import { IdAllocatorV2 } from "../IdAllocator.ts"; import { @@ -89,12 +91,22 @@ import { } from "../ProviderAdapterDriver.ts"; import { makeProviderFailure } from "../ProviderFailure.ts"; import { turnScopedSelectionTransition } from "../ProviderSelectionTransition.ts"; +import { + makeSubagentChildThread, + makeSubagentConversationArtifacts, + subagentThreadTitle, +} from "../SubagentProjection.ts"; import { makePiRpcConnection, parsePiModelSlug, type PiRpcConnection, type PiRpcRecord, } from "./PiRpc.ts"; +import { + buildPiRpcLaunch, + materializePiT3McpExtension, + materializePiT3SubagentExtension, +} from "./piT3McpInjection.ts"; export const PI_PROVIDER = ProviderDriverKind.make("pi"); export const PI_DRIVER_KIND = PI_PROVIDER; @@ -151,7 +163,7 @@ export const PiProviderCapabilitiesV2 = { emitsToolStarted: true, emitsToolCompleted: true, emitsToolOutput: true, - supportsMcpTools: false, + supportsMcpTools: true, supportsDynamicToolCallbacks: false, }, approvals: { @@ -173,11 +185,11 @@ export const PiProviderCapabilitiesV2 = { planDeltasHaveItemIds: false, }, subagents: { - // Pi has no core subagents; the official subagent extension delegates to - // separate pi processes through a tool, and the adapter projects its - // per-task progress into native subagent lifecycle events when present. + // Pi has no core subagents; the T3-owned `subagent` override persists a + // session file per task so each child is a resumeable T3 thread. The + // official tool is omitted because a second `subagent` registration aborts Pi. supportsSubagents: true, - exposesSubagentThreadIds: false, + exposesSubagentThreadIds: true, emitsSubagentLifecycle: true, canWaitForSubagents: false, canCloseSubagents: false, @@ -285,6 +297,7 @@ interface ActivePiTurn { * tool keeps one start timestamp and reports a real duration. */ readonly toolStartedAt: Map; + readonly childSubagents: Map; interrupted: boolean; /** * Whether any agent run activity was observed. Command-only prompts (pure @@ -296,6 +309,16 @@ interface ActivePiTurn { failure: ReturnType | null; } +interface PiChildSubagent { + readonly nativeTaskId: string; + readonly sessionFile: string; + readonly childThreadId: ThreadId; + readonly childProviderThreadId: OrchestrationV2ProviderThread["id"]; + readonly childRootNodeId: OrchestrationV2ExecutionNode["id"]; + emittedUserPrompt: boolean; + emittedMessageCount: number; +} + interface PendingPiPrompt { readonly nativeRequestId: string; readonly method: "select" | "confirm" | "input" | "editor"; @@ -332,11 +355,41 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S ) { const scope = yield* Effect.scope; const cwd = input.runtimePolicy.cwd ?? options.serverConfig.cwd; + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); + const provideCacheFs = (effect: Effect.Effect) => + effect.pipe( + Effect.provideService(FileSystem.FileSystem, options.fileSystem), + Effect.mapError( + (cause) => + new ProviderAdapterOpenSessionError({ + driver: PI_PROVIDER, + providerSessionId: input.providerSessionId, + cause, + }), + ), + ); + const extensionPath = + mcpSession === undefined + ? undefined + : yield* provideCacheFs( + materializePiT3McpExtension(options.serverConfig.providerStatusCacheDir), + ); + const subagentExtensionPath = yield* provideCacheFs( + materializePiT3SubagentExtension(options.serverConfig.providerStatusCacheDir), + ); + const launch = buildPiRpcLaunch({ + launchArgs: options.settings.launchArgs, + environment: options.environment, + mcpSession, + extensionPath, + subagentExtensionPath, + }); + const hasT3Mcp = launch.hasT3Mcp; const connection: PiRpcConnection = yield* makePiRpcConnection({ command: options.settings.binaryPath || "pi", - args: ["--mode", "rpc", ...tokenizeCliArgs(options.settings.launchArgs)], + args: launch.args, cwd, - env: options.environment, + env: launch.env, }).pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, options.spawner), Effect.mapError( @@ -710,12 +763,10 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S }); /** - * Project the official pi subagent extension's per-task progress into - * v2's native subagent surface. The extension reports - * `details: { results: [{agent, task, exitCode, stopReason, messages, - * step?, model?}] }` on every tool update, so each delegated task - * becomes a first-class subagent card with live progress. Tolerant by - * design: any other tool named `subagent` without that shape is simply + * Project the T3-owned pi subagent override's per-task progress into + * v2's native subagent surface. Each result may include `sessionFile`; + * when present the adapter binds a resumeable child thread. Tolerant by + * design: any other tool named `subagent` without the results shape is * ignored. */ const emitSubagentTasks = Effect.fnUntraced(function* ( @@ -727,6 +778,10 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S const results = recordField(recordField(resultRecord, "details"), "results"); if (!Array.isArray(results)) return; const emittedAt = yield* DateTime.now; + const parentNodeId = idAllocator.derive.nodeFromProviderItem({ + driver: PI_PROVIDER, + nativeItemId: toolCallId, + }); for (const [index, result] of results.entries()) { const agent = recordString(result, "agent"); const task = recordString(result, "task"); @@ -740,11 +795,6 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S turn.toolStartedAt.set(nativeTaskId, startedAt); const stopReason = recordString(result, "stopReason"); const exitCode = recordNumber(result, "exitCode") ?? 0; - // A non-zero exit code is a failure whether or not the parent tool - // has ended, matching `piSubagentOutput`. Gating it on `completed` - // let a finished child report "completed" while its result text was - // the stderr of a failure. An aborted child (user Stop) presents as - // interrupted, matching the run and tool cards. const interrupted = stopReason === "aborted"; const failed = !interrupted && (exitCode !== 0 || stopReason === "error"); const finished = completed || interrupted || failed || stopReason !== undefined; @@ -756,7 +806,208 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S ? "completed" : "running"; const outputText = piSubagentOutput(result); - const title = recordString(result, "step") === undefined ? agent : `${agent}`; + const title = agent; + const sessionFile = recordString(result, "sessionFile"); + let child = turn.childSubagents.get(nativeTaskId); + if (sessionFile !== undefined && child === undefined) { + const childThreadId = idAllocator.derive.threadFromProviderThread({ + driver: PI_PROVIDER, + nativeThreadId: sessionFile, + }); + const childProviderThreadId = idAllocator.derive.providerThread({ + driver: PI_PROVIDER, + nativeThreadId: sessionFile, + }); + const childRootNodeId = idAllocator.derive.nodeFromProviderItem({ + driver: PI_PROVIDER, + nativeItemId: `${nativeTaskId}:child-root`, + }); + child = { + nativeTaskId, + sessionFile, + childThreadId, + childProviderThreadId, + childRootNodeId, + emittedUserPrompt: false, + emittedMessageCount: 0, + }; + turn.childSubagents.set(nativeTaskId, child); + const childModelSelection = { + ...turn.turnInput.modelSelection, + model: recordString(result, "model") ?? turn.turnInput.modelSelection.model, + }; + const childThread = makeSubagentChildThread({ + parentThread: turn.turnInput.appThread, + childThreadId, + parentNodeId, + activeProviderThreadId: childProviderThreadId, + providerInstanceId: options.instanceId, + modelSelection: childModelSelection, + title: subagentThreadTitle({ + parentTitle: turn.turnInput.appThread.title, + title, + prompt: task, + ordinal: index + 1, + }), + now: emittedAt, + createdBy: "agent", + creationSource: "provider", + }); + // Null session id so a later send allocates a fresh RPC. Pi cannot + // host two threads on the parent process; resume uses switch_session + // against nativeThreadRef on that new session. + const childProviderThread: OrchestrationV2ProviderThread = { + id: childProviderThreadId, + driver: PI_PROVIDER, + providerInstanceId: options.instanceId, + providerSessionId: null, + appThreadId: childThreadId, + ownerNodeId: parentNodeId, + nativeThreadRef: providerRef(sessionFile), + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: { + providerThreadId: turn.turnInput.providerThread.id, + providerTurnId: turn.providerTurn.id, + }, + pendingBackgroundTasks: [], + createdAt: emittedAt, + updatedAt: emittedAt, + }; + yield* emit({ + type: "app_thread.created", + driver: PI_PROVIDER, + appThread: childThread, + }); + yield* emit({ + type: "provider_thread.updated", + driver: PI_PROVIDER, + providerThread: childProviderThread, + }); + yield* emit({ + type: "node.updated", + driver: PI_PROVIDER, + node: { + id: childRootNodeId, + threadId: childThreadId, + runId: null, + parentNodeId: null, + rootNodeId: childRootNodeId, + kind: "root_turn", + status: "running", + countsForRun: false, + providerThreadId: childProviderThreadId, + providerTurnId: null, + nativeItemRef: providerRef(sessionFile), + runtimeRequestId: null, + checkpointScopeId: null, + startedAt, + completedAt: null, + }, + }); + } + if (child !== undefined && !child.emittedUserPrompt) { + child.emittedUserPrompt = true; + const promptArtifacts = makeSubagentConversationArtifacts({ + messageId: idAllocator.derive.messageFromProviderItem({ + driver: PI_PROVIDER, + nativeItemId: `${nativeTaskId}:prompt`, + }), + turnItemId: idAllocator.derive.turnItemFromProviderItem({ + driver: PI_PROVIDER, + nativeItemId: `${nativeTaskId}:prompt`, + }), + threadId: child.childThreadId, + rootNodeId: child.childRootNodeId, + providerThreadId: child.childProviderThreadId, + providerTurnId: null, + nativeItemRef: providerRef(`${nativeTaskId}:prompt`), + role: "user", + text: task, + ordinal: 100, + now: emittedAt, + }); + yield* emit({ + type: "message.updated", + driver: PI_PROVIDER, + message: promptArtifacts.message, + }); + yield* emit({ + type: "turn_item.updated", + driver: PI_PROVIDER, + turnItem: promptArtifacts.turnItem, + }); + } + if (child !== undefined) { + const messages = recordField(result, "messages"); + if (Array.isArray(messages)) { + for (let messageIndex = child.emittedMessageCount; messageIndex < messages.length; ) { + const message = messages[messageIndex]; + messageIndex += 1; + child.emittedMessageCount = messageIndex; + if (recordString(message, "role") !== "assistant") continue; + const text = contentText(recordField(message, "content")); + if (text.length === 0) continue; + const nativeMessageId = `${nativeTaskId}:assistant:${messageIndex}`; + const artifacts = makeSubagentConversationArtifacts({ + messageId: idAllocator.derive.messageFromProviderItem({ + driver: PI_PROVIDER, + nativeItemId: nativeMessageId, + }), + turnItemId: idAllocator.derive.turnItemFromProviderItem({ + driver: PI_PROVIDER, + nativeItemId: nativeMessageId, + }), + threadId: child.childThreadId, + rootNodeId: child.childRootNodeId, + providerThreadId: child.childProviderThreadId, + providerTurnId: null, + nativeItemRef: providerRef(nativeMessageId), + role: "assistant", + text, + ordinal: 100 + messageIndex, + now: emittedAt, + }); + yield* emit({ + type: "message.updated", + driver: PI_PROVIDER, + message: artifacts.message, + }); + yield* emit({ + type: "turn_item.updated", + driver: PI_PROVIDER, + turnItem: artifacts.turnItem, + }); + } + } + if (finished) { + yield* emit({ + type: "node.updated", + driver: PI_PROVIDER, + node: { + id: child.childRootNodeId, + threadId: child.childThreadId, + runId: null, + parentNodeId: null, + rootNodeId: child.childRootNodeId, + kind: "root_turn", + status, + countsForRun: false, + providerThreadId: child.childProviderThreadId, + providerTurnId: null, + nativeItemRef: providerRef(child.sessionFile), + runtimeRequestId: null, + checkpointScopeId: null, + startedAt, + completedAt: emittedAt, + }, + }); + } + } + const childThreadId = child?.childThreadId ?? null; yield* emit({ type: "subagent.updated", driver: PI_PROVIDER, @@ -764,16 +1015,13 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S id: subagentId, threadId: turn.turnInput.threadId, runId: turn.turnInput.runId, - parentNodeId: idAllocator.derive.nodeFromProviderItem({ - driver: PI_PROVIDER, - nativeItemId: toolCallId, - }), + parentNodeId, origin: "provider_native", createdBy: "agent", driver: PI_PROVIDER, providerInstanceId: options.instanceId, providerThreadId: turn.turnInput.providerThread.id, - childThreadId: null, + childThreadId, nativeTaskRef: providerRef(nativeTaskId), prompt: task, title, @@ -801,7 +1049,7 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S origin: "provider_native", driver: PI_PROVIDER, providerInstanceId: options.instanceId, - childThreadId: null, + childThreadId, prompt: task, ...(finished || outputText.length === 0 ? {} @@ -1627,7 +1875,11 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S // unreadable attachment) must not leave `activeTurn` set, which // would reject every later turn as already active. const payload = yield* resolvePromptPayload( - turnInput.message.text, + t3OrchestrationPromptForFirstRun({ + prompt: turnInput.message.text, + runOrdinal: turnInput.runOrdinal, + hasT3Mcp, + }), turnInput.message.attachments, ); const startedAt = yield* DateTime.now; @@ -1656,6 +1908,7 @@ export function makePiAdapterV2(options: PiAdapterV2Options): ProviderAdapterV2S streamItems: new Map(), toolArgs: new Map(), toolStartedAt: new Map(), + childSubagents: new Map(), interrupted: false, sawAgentActivity: false, failure: null, diff --git a/apps/server/src/orchestration-v2/Adapters/piT3McpExtensionSource.ts b/apps/server/src/orchestration-v2/Adapters/piT3McpExtensionSource.ts new file mode 100644 index 000000000000..8fc657dc09f3 --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/piT3McpExtensionSource.ts @@ -0,0 +1,239 @@ +/** + * Source for the T3-owned Pi extension that consumes T3's HTTP MCP server. + * + * Pi core has no MCP client. This file is TypeScript that Pi itself loads via + * `--extension`. It is written to a cache path at session open so packaged + * AppImage builds do not need a sibling .ts file next to the bundled server. + * + * Do not import t3code modules from the string body. The Pi process resolves + * `@earendil-works/pi-coding-agent` and `typebox` from the user's pi install. + */ +export const PI_T3_MCP_EXTENSION_FILENAME = "pi-t3-mcp-extension.ts"; + +export const T3_MCP_URL_ENV = "T3_MCP_URL"; +export const T3_MCP_BEARER_ENV = "T3_MCP_BEARER_TOKEN"; + +export const PI_T3_MCP_EXTENSION_SOURCE = `\ +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import { Type } from "typebox"; + +const URL_ENV = ${JSON.stringify(T3_MCP_URL_ENV)}; +const TOKEN_ENV = ${JSON.stringify(T3_MCP_BEARER_ENV)}; +const PROTOCOL = "2025-06-18"; + +type JsonRpcResponse = { + readonly id?: number | string; + readonly result?: unknown; + readonly error?: { readonly message?: string }; +}; + +type McpTool = { + readonly name: string; + readonly description?: string; + readonly inputSchema?: Record; +}; + +function env(name: string): string | undefined { + const value = process.env[name]; + return value && value.length > 0 ? value : undefined; +} + +function parseSseOrJson(body: string, contentType: string): JsonRpcResponse { + if (contentType.includes("text/event-stream")) { + for (const line of body.split("\\n")) { + const trimmed = line.startsWith("data:") ? line.slice(5).trim() : ""; + if (trimmed.length === 0) continue; + const parsed = JSON.parse(trimmed) as JsonRpcResponse; + if (parsed.id !== undefined || parsed.result !== undefined || parsed.error !== undefined) { + return parsed; + } + } + throw new Error("MCP SSE response had no JSON-RPC payload."); + } + return JSON.parse(body) as JsonRpcResponse; +} + +function jsonSchemaToTypebox(schema: Record | undefined) { + const unsafe = (Type as { Unsafe?: (value: unknown) => unknown }).Unsafe; + if (typeof unsafe === "function" && schema !== undefined) { + return unsafe(schema); + } + return Type.Object({}, { additionalProperties: true }); +} + +function formatMcpContent(result: unknown): string { + if (result === null || result === undefined) return ""; + if (typeof result !== "object") return String(result); + const record = result as { + readonly content?: ReadonlyArray<{ readonly type?: string; readonly text?: string }>; + readonly structuredContent?: unknown; + readonly isError?: boolean; + }; + const texts: string[] = []; + if (Array.isArray(record.content)) { + for (const part of record.content) { + if (part?.type === "text" && typeof part.text === "string") texts.push(part.text); + } + } + if (record.structuredContent !== undefined) { + texts.push(JSON.stringify(record.structuredContent)); + } + if (texts.length > 0) return texts.join("\\n"); + return JSON.stringify(result); +} + +function createMcpClient(endpoint: string, token: string) { + let nextId = 1; + let sessionId: string | undefined; + + const headers = (): Record => { + const next: Record = { + accept: "application/json, text/event-stream", + authorization: token.startsWith("Bearer ") ? token : \`Bearer \${token}\`, + "content-type": "application/json", + // Effect's HTTP MCP rejects post-initialize requests without this + // (400). The worktree client in McpHttpServer tests sends the same + // header; initialize itself does not require it. + "mcp-protocol-version": PROTOCOL, + }; + if (sessionId !== undefined) next["mcp-session-id"] = sessionId; + return next; + }; + + const request = async (method: string, params?: unknown, signal?: AbortSignal) => { + const id = nextId++; + const response = await fetch(endpoint, { + method: "POST", + headers: headers(), + body: JSON.stringify({ jsonrpc: "2.0", id, method, params }), + signal, + }); + const nextSession = response.headers.get("mcp-session-id"); + if (nextSession) sessionId = nextSession; + const body = await response.text(); + if (!response.ok) { + throw new Error(\`MCP \${method} failed (\${response.status}): \${body.slice(0, 400)}\`); + } + if (body.length === 0) return undefined; + const parsed = parseSseOrJson(body, response.headers.get("content-type") ?? ""); + if (parsed.error) { + throw new Error(parsed.error.message ?? \`MCP \${method} returned an error\`); + } + return parsed.result; + }; + + const notify = async (method: string, params?: unknown, signal?: AbortSignal) => { + await fetch(endpoint, { + method: "POST", + headers: headers(), + body: JSON.stringify({ jsonrpc: "2.0", method, params }), + signal, + }); + }; + + return { + async connect(signal?: AbortSignal) { + await request( + "initialize", + { + protocolVersion: PROTOCOL, + capabilities: {}, + clientInfo: { name: "t3-pi-mcp", version: "1.0.0" }, + }, + signal, + ); + await notify("notifications/initialized", {}, signal).catch(() => undefined); + }, + async listTools(signal?: AbortSignal) { + const tools: McpTool[] = []; + let cursor: string | undefined; + do { + const result = (await request( + "tools/list", + cursor === undefined ? {} : { cursor }, + signal, + )) as { tools?: McpTool[]; nextCursor?: string } | undefined; + tools.push(...(result?.tools ?? [])); + cursor = result?.nextCursor; + } while (cursor); + return tools; + }, + async callTool(name: string, args: Record, signal?: AbortSignal) { + return request("tools/call", { name, arguments: args }, signal); + }, + }; +} + +export default async function t3McpExtension(pi: ExtensionAPI) { + const endpoint = env(URL_ENV); + const token = env(TOKEN_ENV); + if (endpoint === undefined || token === undefined) { + pi.on("session_start", async (_event, ctx) => { + ctx.ui.notify( + "t3-code MCP unavailable: T3_MCP_URL or T3_MCP_BEARER_TOKEN is missing.", + "warning", + ); + }); + return; + } + + const client = createMcpClient(endpoint, token); + let started: Promise | undefined; + + const ensureStarted = () => { + started ??= (async () => { + const signal = AbortSignal.timeout(10_000); + await client.connect(signal); + const tools = await client.listTools(signal); + for (const tool of tools) { + const name = tool.name; + const description = tool.description ?? name; + pi.registerTool({ + name, + label: name, + description, + promptSnippet: description.split("\\n")[0] ?? name, + promptGuidelines: [ + \`Use \${name} from the t3-code MCP server when the user asks for T3 orchestration that this tool covers.\`, + ], + parameters: jsonSchemaToTypebox(tool.inputSchema), + async execute(_toolCallId, params, signal) { + const result = await client.callTool( + name, + (params ?? {}) as Record, + signal, + ); + const text = formatMcpContent(result); + return { + content: [{ type: "text", text }], + details: { server: "t3-code", tool: name }, + }; + }, + }); + } + })(); + return started; + }; + + // Await here so tools exist before session_start and the first prompt. + // session_start is a retry if the process later reloads the extension. + try { + await ensureStarted(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + pi.on("session_start", async (_event, ctx) => { + ctx.ui.notify(\`t3-code MCP unavailable: \${message}\`, "warning"); + }); + return; + } + + pi.on("session_start", async (_event, ctx) => { + try { + await ensureStarted(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + ctx.ui.notify(\`t3-code MCP unavailable: \${message}\`, "warning"); + } + }); +} +`; diff --git a/apps/server/src/orchestration-v2/Adapters/piT3McpInjection.test.ts b/apps/server/src/orchestration-v2/Adapters/piT3McpInjection.test.ts new file mode 100644 index 000000000000..f3e449714765 --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/piT3McpInjection.test.ts @@ -0,0 +1,166 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; + +import { + PI_T3_MCP_EXTENSION_FILENAME, + T3_MCP_BEARER_ENV, + T3_MCP_URL_ENV, +} from "./piT3McpExtensionSource.ts"; +import { + PI_T3_SUBAGENT_EXTENSION_FILENAME, + T3_PI_CHILD_SESSION_ROOT_ENV, +} from "./piT3SubagentExtensionSource.ts"; +import { + bearerTokenFromAuthorizationHeader, + buildPiRpcLaunch, + isConflictingPiSubagentExtensionPath, + materializePiT3McpExtension, + materializePiT3SubagentExtension, + piChildSessionRootFromLaunchArgs, + piT3McpExtensionDestPath, + piT3SubagentExtensionDestPath, +} from "./piT3McpInjection.ts"; + +const threadId = ThreadId.make("thread-pi-t3-mcp"); + +const mcpSession = { + environmentId: EnvironmentId.make("environment-pi-t3-mcp"), + threadId, + providerSessionId: "mcp-session-pi", + providerInstanceId: ProviderInstanceId.make("pi"), + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer secret-pi-token", +}; + +describe("pi T3 MCP injection", () => { + it("strips the Bearer prefix for the child env", () => { + assert.equal(bearerTokenFromAuthorizationHeader("Bearer secret-pi-token"), "secret-pi-token"); + assert.equal(bearerTokenFromAuthorizationHeader("secret-pi-token"), "secret-pi-token"); + }); + + it("leaves spawn args unchanged when no MCP session exists", () => { + const launch = buildPiRpcLaunch({ + launchArgs: "--session-dir /tmp/pi-sessions", + environment: { PATH: "/usr/bin" }, + mcpSession: undefined, + extensionPath: "/tmp/pi-t3-mcp-extension.ts", + }); + assert.isFalse(launch.hasT3Mcp); + assert.deepEqual(launch.args, ["--mode", "rpc", "--session-dir", "/tmp/pi-sessions"]); + assert.equal(launch.env.PATH, "/usr/bin"); + assert.isUndefined(launch.env[T3_MCP_URL_ENV]); + }); + + it("identifies official subagent paths and keeps the T3 override", () => { + assert.isTrue( + isConflictingPiSubagentExtensionPath("/opt/pi/examples/extensions/subagent/index.ts"), + ); + assert.isTrue( + isConflictingPiSubagentExtensionPath("/home/user/.pi/agent/extensions/subagent/index.ts"), + ); + assert.isFalse(isConflictingPiSubagentExtensionPath("/tmp/cache/pi-t3-subagent-extension.ts")); + }); + + it("prepends the subagent override and drops the official tool", () => { + const launch = buildPiRpcLaunch({ + launchArgs: + "--session-dir /tmp/pi-sessions --extension /opt/pi/examples/extensions/subagent/index.ts", + environment: { PATH: "/usr/bin" }, + mcpSession, + extensionPath: "/tmp/cache/pi-t3-mcp-extension.ts", + subagentExtensionPath: "/tmp/cache/pi-t3-subagent-extension.ts", + }); + assert.deepEqual(launch.args, [ + "--mode", + "rpc", + "--no-extensions", + "--extension", + "/tmp/cache/pi-t3-subagent-extension.ts", + "--session-dir", + "/tmp/pi-sessions", + "--extension", + "/tmp/cache/pi-t3-mcp-extension.ts", + ]); + assert.equal(launch.env[T3_PI_CHILD_SESSION_ROOT_ENV], "/tmp/pi-sessions/children"); + assert.equal(launch.env[T3_MCP_URL_ENV], "http://127.0.0.1:43123/mcp"); + assert.equal(launch.env[T3_MCP_BEARER_ENV], "secret-pi-token"); + }); + + it("appends --extension and scoped env when a session exists", () => { + const launch = buildPiRpcLaunch({ + launchArgs: "--session-dir /tmp/pi-sessions", + environment: { PATH: "/usr/bin" }, + mcpSession, + extensionPath: "/tmp/cache/pi-t3-mcp-extension.ts", + }); + assert.isTrue(launch.hasT3Mcp); + assert.deepEqual(launch.args, [ + "--mode", + "rpc", + "--session-dir", + "/tmp/pi-sessions", + "--extension", + "/tmp/cache/pi-t3-mcp-extension.ts", + ]); + assert.equal(launch.env[T3_MCP_URL_ENV], "http://127.0.0.1:43123/mcp"); + assert.equal(launch.env[T3_MCP_BEARER_ENV], "secret-pi-token"); + }); + + it("does not duplicate an already-present extension path", () => { + const launch = buildPiRpcLaunch({ + launchArgs: "--extension /tmp/cache/pi-t3-mcp-extension.ts", + environment: {}, + mcpSession, + extensionPath: "/tmp/cache/pi-t3-mcp-extension.ts", + }); + assert.deepEqual(launch.args, [ + "--mode", + "rpc", + "--extension", + "/tmp/cache/pi-t3-mcp-extension.ts", + ]); + }); + + it("derives the child session root from --session-dir", () => { + assert.equal( + piChildSessionRootFromLaunchArgs("--session-dir /tmp/pi-sessions --extension x.ts"), + "/tmp/pi-sessions/children", + ); + assert.isUndefined(piChildSessionRootFromLaunchArgs("")); + }); + + it.effect("writes the extension source to the cache directory", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const cacheDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pi-t3-mcp-" }); + const dest = yield* materializePiT3McpExtension(cacheDir); + assert.equal(dest, piT3McpExtensionDestPath(cacheDir)); + assert.isTrue(dest.endsWith(PI_T3_MCP_EXTENSION_FILENAME)); + const source = yield* fs.readFileString(dest); + assert.include(source, "export default async function t3McpExtension"); + assert.include(source, T3_MCP_URL_ENV); + assert.include(source, '"mcp-protocol-version"'); + assert.include(source, '"tools/call"'); + const again = yield* materializePiT3McpExtension(cacheDir); + assert.equal(again, dest); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("writes the subagent override source to the cache directory", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const cacheDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pi-t3-subagent-" }); + const dest = yield* materializePiT3SubagentExtension(cacheDir); + assert.equal(dest, piT3SubagentExtensionDestPath(cacheDir)); + assert.isTrue(dest.endsWith(PI_T3_SUBAGENT_EXTENSION_FILENAME)); + const source = yield* fs.readFileString(dest); + assert.include(source, "export default function t3SubagentExtension"); + assert.include(source, "--session"); + assert.include(source, T3_PI_CHILD_SESSION_ROOT_ENV); + assert.isFalse(source.includes("--no-session")); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/piT3McpInjection.ts b/apps/server/src/orchestration-v2/Adapters/piT3McpInjection.ts new file mode 100644 index 000000000000..0637de2ca4e5 --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/piT3McpInjection.ts @@ -0,0 +1,166 @@ +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import { tokenizeCliArgs } from "@t3tools/shared/cliArgs"; + +import type { McpProviderSessionConfig } from "../../mcp/McpProviderSession.ts"; +import { + PI_T3_MCP_EXTENSION_FILENAME, + PI_T3_MCP_EXTENSION_SOURCE, + T3_MCP_BEARER_ENV, + T3_MCP_URL_ENV, +} from "./piT3McpExtensionSource.ts"; +import { + PI_T3_SUBAGENT_EXTENSION_FILENAME, + PI_T3_SUBAGENT_EXTENSION_SOURCE, + T3_PI_CHILD_SESSION_ROOT_ENV, +} from "./piT3SubagentExtensionSource.ts"; + +export { + PI_T3_MCP_EXTENSION_FILENAME, + PI_T3_SUBAGENT_EXTENSION_FILENAME, + T3_MCP_BEARER_ENV, + T3_MCP_URL_ENV, + T3_PI_CHILD_SESSION_ROOT_ENV, +}; + +export function bearerTokenFromAuthorizationHeader(header: string): string { + return header.startsWith("Bearer ") ? header.slice("Bearer ".length) : header; +} + +export function piT3McpExtensionDestPath(cacheDir: string): string { + return `${cacheDir.replace(/\\/g, "/")}/${PI_T3_MCP_EXTENSION_FILENAME}`; +} + +export function piT3SubagentExtensionDestPath(cacheDir: string): string { + return `${cacheDir.replace(/\\/g, "/")}/${PI_T3_SUBAGENT_EXTENSION_FILENAME}`; +} + +export function piChildSessionRootFromLaunchArgs(launchArgs: string): string | undefined { + const args = tokenizeCliArgs(launchArgs); + const index = args.indexOf("--session-dir"); + const sessionDir = index >= 0 ? args[index + 1] : undefined; + if (sessionDir === undefined || sessionDir.length === 0) return undefined; + return `${sessionDir.replace(/\\/g, "/")}/children`; +} + +export const materializePiT3McpExtension = Effect.fn("materializePiT3McpExtension")(function* ( + cacheDir: string, +) { + const fs = yield* FileSystem.FileSystem; + yield* fs.makeDirectory(cacheDir, { recursive: true }); + const dest = piT3McpExtensionDestPath(cacheDir); + const existing = yield* fs.readFileString(dest).pipe(Effect.orElseSucceed(() => "")); + if (existing !== PI_T3_MCP_EXTENSION_SOURCE) { + yield* fs.writeFileString(dest, PI_T3_MCP_EXTENSION_SOURCE); + } + return dest; +}); + +export const materializePiT3SubagentExtension = Effect.fn("materializePiT3SubagentExtension")( + function* (cacheDir: string) { + const fs = yield* FileSystem.FileSystem; + yield* fs.makeDirectory(cacheDir, { recursive: true }); + const dest = piT3SubagentExtensionDestPath(cacheDir); + const existing = yield* fs.readFileString(dest).pipe(Effect.orElseSucceed(() => "")); + if (existing !== PI_T3_SUBAGENT_EXTENSION_SOURCE) { + yield* fs.writeFileString(dest, PI_T3_SUBAGENT_EXTENSION_SOURCE); + } + return dest; + }, +); + +function appendExtensionArg( + args: ReadonlyArray, + extensionPath: string | undefined, +): string[] { + if (extensionPath === undefined) return [...args]; + const alreadyHas = args.some( + (arg, index) => (arg === "--extension" || arg === "-e") && args[index + 1] === extensionPath, + ); + return alreadyHas ? [...args] : [...args, "--extension", extensionPath]; +} + +function normalizePiPath(value: string): string { + return value.replace(/\\/g, "/").replace(/\/+$/, ""); +} + +/** Official / user-installed `subagent` tool. Not the T3 override file. */ +export function isConflictingPiSubagentExtensionPath(extensionPath: string): boolean { + const normalized = normalizePiPath(extensionPath); + if (normalized.endsWith(`/${PI_T3_SUBAGENT_EXTENSION_FILENAME}`)) return false; + return ( + normalized.endsWith("/extensions/subagent/index.ts") || + normalized.endsWith("/extensions/subagent/index.js") || + normalized.endsWith("/examples/extensions/subagent/index.ts") || + normalized.endsWith("/examples/extensions/subagent/index.js") + ); +} + +export function stripConflictingPiSubagentExtensionArgs(args: ReadonlyArray): string[] { + const stripped: string[] = []; + for (let index = 0; index < args.length; index += 1) { + const arg = args[index]; + if (arg === undefined) continue; + const next = args[index + 1]; + if ( + (arg === "--extension" || arg === "-e") && + next !== undefined && + isConflictingPiSubagentExtensionPath(next) + ) { + index += 1; + continue; + } + stripped.push(arg); + } + return stripped; +} + +export function buildPiRpcLaunch(input: { + readonly launchArgs: string; + readonly environment: NodeJS.ProcessEnv; + readonly mcpSession: McpProviderSessionConfig | undefined; + readonly extensionPath: string | undefined; + readonly subagentExtensionPath?: string | undefined; +}): { + readonly args: ReadonlyArray; + readonly env: NodeJS.ProcessEnv; + readonly hasT3Mcp: boolean; +} { + const userArgs = tokenizeCliArgs(input.launchArgs); + const hasT3Mcp = input.mcpSession !== undefined && input.extensionPath !== undefined; + // Duplicate `subagent` registrations abort Pi. Disable discovery and drop + // the official tool from launchArgs so only the T3 override remains. + let args = ["--mode", "rpc"]; + if (input.subagentExtensionPath !== undefined) { + if (!userArgs.includes("--no-extensions") && !userArgs.includes("-ne")) { + args.push("--no-extensions"); + } + args = appendExtensionArg(args, input.subagentExtensionPath); + args = [...args, ...stripConflictingPiSubagentExtensionArgs(userArgs)]; + } else { + args = [...args, ...userArgs]; + } + if (hasT3Mcp && input.extensionPath !== undefined) { + args = appendExtensionArg(args, input.extensionPath); + } + + const childSessionRoot = piChildSessionRootFromLaunchArgs(input.launchArgs); + return { + args, + env: { + ...input.environment, + ...(input.subagentExtensionPath === undefined || childSessionRoot === undefined + ? {} + : { [T3_PI_CHILD_SESSION_ROOT_ENV]: childSessionRoot }), + ...(hasT3Mcp && input.mcpSession !== undefined + ? { + [T3_MCP_URL_ENV]: input.mcpSession.endpoint, + [T3_MCP_BEARER_ENV]: bearerTokenFromAuthorizationHeader( + input.mcpSession.authorizationHeader, + ), + } + : {}), + }, + hasT3Mcp, + }; +} diff --git a/apps/server/src/orchestration-v2/Adapters/piT3SubagentExtensionSource.ts b/apps/server/src/orchestration-v2/Adapters/piT3SubagentExtensionSource.ts new file mode 100644 index 000000000000..3b32dfd83533 --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/piT3SubagentExtensionSource.ts @@ -0,0 +1,561 @@ +/** + * T3-owned override of the official Pi `subagent` tool. + * + * Official spawn is `pi --mode json -p --no-session`, so there is no session + * file to bind as a T3 child thread. This copy persists `--session ` and + * reports `sessionFile` on each result so the adapter can resume it. + * + * Loaded via CLI `--extension`. Pi aborts if two extensions register + * `subagent`, so the launcher omits the official tool. + */ +export const PI_T3_SUBAGENT_EXTENSION_FILENAME = "pi-t3-subagent-extension.ts"; + +export const T3_PI_CHILD_SESSION_ROOT_ENV = "T3_PI_CHILD_SESSION_ROOT"; + +export const PI_T3_SUBAGENT_EXTENSION_SOURCE = `\ +import { spawn } from "node:child_process"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { randomBytes } from "node:crypto"; +import { StringEnum } from "@earendil-works/pi-ai"; +import { + CONFIG_DIR_NAME, + type ExtensionAPI, + getAgentDir, + parseFrontmatter, +} from "@earendil-works/pi-coding-agent"; +import { Type } from "typebox"; + +const SESSION_ROOT_ENV = ${JSON.stringify(T3_PI_CHILD_SESSION_ROOT_ENV)}; +const MAX_PARALLEL_TASKS = 8; +const MAX_CONCURRENCY = 4; + +type AgentScope = "user" | "project" | "both"; + +type AgentConfig = { + name: string; + description: string; + tools?: string[]; + model?: string; + systemPrompt: string; + source: "user" | "project"; +}; + +type SingleResult = { + agent: string; + agentSource: "user" | "project" | "unknown"; + task: string; + exitCode: number; + messages: unknown[]; + stderr: string; + usage: { + input: number; + output: number; + cacheRead: number; + cacheWrite: number; + cost: number; + contextTokens: number; + turns: number; + }; + model?: string; + stopReason?: string; + errorMessage?: string; + step?: number; + sessionFile?: string; +}; + +function parseToolList(value: unknown): string[] | undefined { + const raw = Array.isArray(value) ? value : typeof value === "string" ? value.split(",") : []; + const tools = raw + .filter((entry): entry is string => typeof entry === "string") + .map((entry) => entry.trim()) + .filter(Boolean); + return tools.length > 0 ? tools : undefined; +} + +function loadAgentsFromDir(dir: string, source: "user" | "project"): AgentConfig[] { + if (!fs.existsSync(dir)) return []; + let entries: fs.Dirent[]; + try { + entries = fs.readdirSync(dir, { withFileTypes: true }); + } catch { + return []; + } + const agents: AgentConfig[] = []; + for (const entry of entries) { + if (!entry.name.endsWith(".md")) continue; + if (!entry.isFile() && !entry.isSymbolicLink()) continue; + const filePath = path.join(dir, entry.name); + let content: string; + try { + content = fs.readFileSync(filePath, "utf-8"); + } catch { + continue; + } + const { frontmatter, body } = parseFrontmatter<{ + name?: unknown; + description?: unknown; + tools?: unknown; + model?: unknown; + }>(content); + if (typeof frontmatter.name !== "string" || typeof frontmatter.description !== "string") { + continue; + } + agents.push({ + name: frontmatter.name, + description: frontmatter.description, + tools: parseToolList(frontmatter.tools), + model: typeof frontmatter.model === "string" ? frontmatter.model : undefined, + systemPrompt: body, + source, + }); + } + return agents; +} + +function discoverAgents(cwd: string, scope: AgentScope) { + const userDir = path.join(getAgentDir(), "agents"); + let projectAgentsDir: string | null = null; + let currentDir = cwd; + while (true) { + const candidate = path.join(currentDir, CONFIG_DIR_NAME, "agents"); + try { + if (fs.statSync(candidate).isDirectory()) { + projectAgentsDir = candidate; + break; + } + } catch { + /* keep walking */ + } + const parentDir = path.dirname(currentDir); + if (parentDir === currentDir) break; + currentDir = parentDir; + } + const userAgents = scope === "project" ? [] : loadAgentsFromDir(userDir, "user"); + const projectAgents = + scope === "user" || projectAgentsDir === null + ? [] + : loadAgentsFromDir(projectAgentsDir, "project"); + const agentMap = new Map(); + if (scope === "both") { + for (const agent of userAgents) agentMap.set(agent.name, agent); + for (const agent of projectAgents) agentMap.set(agent.name, agent); + } else if (scope === "user") { + for (const agent of userAgents) agentMap.set(agent.name, agent); + } else { + for (const agent of projectAgents) agentMap.set(agent.name, agent); + } + return { agents: Array.from(agentMap.values()), projectAgentsDir }; +} + +function getFinalOutput(messages: unknown[]): string { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index] as { role?: string; content?: unknown }; + if (message?.role !== "assistant" || !Array.isArray(message.content)) continue; + for (const part of message.content) { + const record = part as { type?: string; text?: string }; + if (record.type === "text" && typeof record.text === "string" && record.text.length > 0) { + return record.text; + } + } + } + return ""; +} + +function isFailedResult(result: SingleResult): boolean { + return result.exitCode !== 0 || result.stopReason === "error" || result.stopReason === "aborted"; +} + +function getResultOutput(result: SingleResult): string { + if (isFailedResult(result)) { + return result.errorMessage || result.stderr || getFinalOutput(result.messages) || "(no output)"; + } + return getFinalOutput(result.messages) || "(no output)"; +} + +function getPiInvocation(args: string[]): { command: string; args: string[] } { + const currentScript = process.argv[1]; + const isBunVirtualScript = currentScript?.startsWith("/$bunfs/root/"); + if (currentScript && !isBunVirtualScript && fs.existsSync(currentScript)) { + return { command: process.execPath, args: [currentScript, ...args] }; + } + const execName = path.basename(process.execPath).toLowerCase(); + if (/^(node|bun)(\\.exe)?$/.test(execName)) return { command: "pi", args }; + return { command: process.execPath, args }; +} + +function childSessionFile(): string { + const root = + process.env[SESSION_ROOT_ENV] && process.env[SESSION_ROOT_ENV]!.length > 0 + ? process.env[SESSION_ROOT_ENV]! + : path.join(getAgentDir(), "sessions"); + fs.mkdirSync(root, { recursive: true }); + const id = randomBytes(6).toString("hex"); + return path.join(root, \`t3-subagent-\${Date.now()}-\${id}.jsonl\`); +} + +async function runSingleAgent( + defaultCwd: string, + defaults: { model?: string; thinkingLevel?: string }, + agents: AgentConfig[], + agentName: string, + task: string, + cwd: string | undefined, + step: number | undefined, + signal: AbortSignal | undefined, + onUpdate: ((partial: { content: { type: "text"; text: string }[]; details: { results: SingleResult[] } }) => void) | undefined, +): Promise { + const agent = agents.find((entry) => entry.name === agentName); + if (!agent) { + const available = agents.map((entry) => \`"\${entry.name}"\`).join(", ") || "none"; + return { + agent: agentName, + agentSource: "unknown", + task, + exitCode: 1, + messages: [], + stderr: \`Unknown agent: "\${agentName}". Available agents: \${available}.\`, + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 }, + step, + }; + } + + const sessionFile = childSessionFile(); + const args: string[] = [ + "--mode", + "json", + "-p", + "--session", + sessionFile, + "--name", + \`t3-subagent \${agentName}\`, + ]; + const model = agent.model ?? defaults.model; + if (model) args.push("--model", model); + if (!agent.model && defaults.thinkingLevel) args.push("--thinking", defaults.thinkingLevel); + if (agent.tools && agent.tools.length > 0) args.push("--tools", agent.tools.join(",")); + + let tmpPromptDir: string | null = null; + const currentResult: SingleResult = { + agent: agentName, + agentSource: agent.source, + task, + exitCode: 0, + messages: [], + stderr: "", + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 }, + model, + step, + sessionFile, + }; + + const emitUpdate = () => { + onUpdate?.({ + content: [{ type: "text", text: getFinalOutput(currentResult.messages) || "(running...)" }], + details: { results: [currentResult] }, + }); + }; + emitUpdate(); + + try { + if (agent.systemPrompt.trim()) { + tmpPromptDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "pi-t3-subagent-")); + const promptPath = path.join(tmpPromptDir, "prompt.md"); + await fs.promises.writeFile(promptPath, agent.systemPrompt, { encoding: "utf-8", mode: 0o600 }); + args.push("--append-system-prompt", promptPath); + } + args.push(\`Task: \${task}\`); + + const exitCode = await new Promise((resolve) => { + const invocation = getPiInvocation(args); + const proc = spawn(invocation.command, invocation.args, { + cwd: cwd ?? defaultCwd, + shell: false, + stdio: ["ignore", "pipe", "pipe"], + }); + let buffer = ""; + const processLine = (line: string) => { + if (!line.trim()) return; + let event: { type?: string; message?: unknown }; + try { + event = JSON.parse(line) as { type?: string; message?: unknown }; + } catch { + return; + } + if ((event.type === "message_end" || event.type === "tool_result_end") && event.message) { + currentResult.messages.push(event.message); + const message = event.message as { + role?: string; + usage?: Record; + model?: string; + stopReason?: string; + errorMessage?: string; + }; + if (message.role === "assistant") { + currentResult.usage.turns += 1; + if (message.usage) { + currentResult.usage.input += message.usage.input || 0; + currentResult.usage.output += message.usage.output || 0; + currentResult.usage.cacheRead += message.usage.cacheRead || 0; + currentResult.usage.cacheWrite += message.usage.cacheWrite || 0; + const cost = message.usage.cost as number | { total?: number } | undefined; + currentResult.usage.cost += + typeof cost === "number" ? cost : typeof cost?.total === "number" ? cost.total : 0; + currentResult.usage.contextTokens = message.usage.totalTokens || 0; + } + if (!currentResult.model && message.model) currentResult.model = message.model; + if (message.stopReason) currentResult.stopReason = message.stopReason; + if (message.errorMessage) currentResult.errorMessage = message.errorMessage; + } + emitUpdate(); + } + }; + proc.stdout?.setEncoding("utf8"); + proc.stdout?.on("data", (chunk: string) => { + buffer += chunk; + const lines = buffer.split("\\n"); + buffer = lines.pop() ?? ""; + for (const line of lines) processLine(line); + }); + proc.stderr?.setEncoding("utf8"); + proc.stderr?.on("data", (chunk: string) => { + currentResult.stderr += chunk; + }); + const onAbort = () => { + currentResult.stopReason = "aborted"; + proc.kill("SIGTERM"); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + proc.on("close", (code) => { + signal?.removeEventListener("abort", onAbort); + if (buffer.trim()) processLine(buffer); + resolve(code ?? 1); + }); + proc.on("error", (error) => { + currentResult.stderr += error.message; + resolve(1); + }); + }); + currentResult.exitCode = exitCode; + return currentResult; + } finally { + if (tmpPromptDir) { + await fs.promises.rm(tmpPromptDir, { recursive: true, force: true }).catch(() => undefined); + } + } +} + +const TaskItem = Type.Object({ + agent: Type.String(), + task: Type.String(), + cwd: Type.Optional(Type.String()), +}); + +export default function t3SubagentExtension(pi: ExtensionAPI) { + pi.registerTool({ + name: "subagent", + label: "Subagent", + description: + "Delegate tasks to specialized Pi subagents with isolated context. Each child persists a Pi session that T3 can open and continue.", + parameters: Type.Object({ + agent: Type.Optional(Type.String()), + task: Type.Optional(Type.String()), + tasks: Type.Optional(Type.Array(TaskItem)), + chain: Type.Optional(Type.Array(TaskItem)), + agentScope: Type.Optional(StringEnum(["user", "project", "both"] as const)), + confirmProjectAgents: Type.Optional(Type.Boolean()), + cwd: Type.Optional(Type.String()), + }), + async execute(_toolCallId, params, signal, onUpdate, ctx) { + const agentScope: AgentScope = params.agentScope ?? "user"; + const discovery = discoverAgents(ctx.cwd, agentScope); + const agents = discovery.agents; + const defaults = { + model: ctx.model ? \`\${ctx.model.provider}/\${ctx.model.id}\` : undefined, + thinkingLevel: ctx.thinkingLevel, + }; + const confirmProjectAgents = params.confirmProjectAgents ?? true; + const hasChain = (params.chain?.length ?? 0) > 0; + const hasTasks = (params.tasks?.length ?? 0) > 0; + const hasSingle = Boolean(params.agent && params.task); + const makeDetails = (results: SingleResult[]) => ({ + mode: hasChain ? "chain" : hasTasks ? "parallel" : "single", + agentScope, + projectAgentsDir: discovery.projectAgentsDir, + results, + }); + + if (Number(hasChain) + Number(hasTasks) + Number(hasSingle) !== 1) { + const available = agents.map((agent) => agent.name).join(", ") || "none"; + return { + content: [{ type: "text", text: \`Invalid parameters. Available agents: \${available}\` }], + details: makeDetails([]), + }; + } + + if ( + (agentScope === "project" || agentScope === "both") && + confirmProjectAgents && + ctx.hasUI + ) { + const names = new Set(); + if (params.chain) for (const step of params.chain) names.add(step.agent); + if (params.tasks) for (const item of params.tasks) names.add(item.agent); + if (params.agent) names.add(params.agent); + const projectRequested = agents.filter( + (agent) => agent.source === "project" && names.has(agent.name), + ); + if (projectRequested.length > 0) { + const ok = await ctx.ui.confirm( + "Run project-local agents?", + \`Agents: \${projectRequested.map((agent) => agent.name).join(", ")}\\nSource: \${discovery.projectAgentsDir ?? "(unknown)"}\`, + ); + if (!ok) { + return { + content: [{ type: "text", text: "Canceled: project-local agents not approved." }], + details: makeDetails([]), + }; + } + } + } + + if (params.chain && params.chain.length > 0) { + const results: SingleResult[] = []; + let previousOutput = ""; + for (let index = 0; index < params.chain.length; index += 1) { + const step = params.chain[index]; + const result = await runSingleAgent( + ctx.cwd, + defaults, + agents, + step.agent, + step.task.replace(/\\{previous\\}/g, previousOutput), + step.cwd, + index + 1, + signal, + onUpdate + ? (partial) => { + const current = partial.details?.results[0]; + if (current) onUpdate({ ...partial, details: makeDetails([...results, current]) }); + } + : undefined, + ); + results.push(result); + if (isFailedResult(result)) { + return { + content: [ + { + type: "text", + text: \`Chain stopped at step \${index + 1} (\${step.agent}): \${getResultOutput(result)}\`, + }, + ], + details: makeDetails(results), + isError: true, + }; + } + previousOutput = getFinalOutput(result.messages); + } + return { + content: [ + { + type: "text", + text: getFinalOutput(results[results.length - 1]?.messages ?? []) || "(no output)", + }, + ], + details: makeDetails(results), + }; + } + + if (params.tasks && params.tasks.length > 0) { + if (params.tasks.length > MAX_PARALLEL_TASKS) { + return { + content: [{ type: "text", text: \`Too many parallel tasks (\${params.tasks.length}).\` }], + details: makeDetails([]), + }; + } + const allResults: SingleResult[] = params.tasks.map((item) => ({ + agent: item.agent, + agentSource: "unknown", + task: item.task, + exitCode: -1, + messages: [], + stderr: "", + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 }, + })); + const emitParallel = () => { + onUpdate?.({ + content: [{ type: "text", text: "Parallel tasks running..." }], + details: makeDetails([...allResults]), + }); + }; + let nextIndex = 0; + const workers = Array.from({ length: Math.min(MAX_CONCURRENCY, params.tasks.length) }, async () => { + while (true) { + const index = nextIndex; + nextIndex += 1; + if (index >= params.tasks.length) return; + const item = params.tasks[index]; + const result = await runSingleAgent( + ctx.cwd, + defaults, + agents, + item.agent, + item.task, + item.cwd, + undefined, + signal, + (partial) => { + if (partial.details?.results[0]) { + allResults[index] = partial.details.results[0]; + emitParallel(); + } + }, + ); + allResults[index] = result; + emitParallel(); + } + }); + await Promise.all(workers); + const successCount = allResults.filter((result) => !isFailedResult(result)).length; + return { + content: [ + { + type: "text", + text: \`Parallel: \${successCount}/\${allResults.length} succeeded\`, + }, + ], + details: makeDetails(allResults), + }; + } + + const result = await runSingleAgent( + ctx.cwd, + defaults, + agents, + params.agent as string, + params.task as string, + params.cwd, + undefined, + signal, + onUpdate + ? (partial) => onUpdate({ ...partial, details: makeDetails(partial.details.results) }) + : undefined, + ); + const failed = isFailedResult(result); + return { + content: [ + { + type: "text", + text: failed + ? \`Agent \${result.stopReason || "failed"}: \${getResultOutput(result)}\` + : getFinalOutput(result.messages) || "(no output)", + }, + ], + details: makeDetails([result]), + ...(failed ? { isError: true } : {}), + }; + }, + }); +} +`; diff --git a/apps/server/src/provider/Layers/PiProvider.ts b/apps/server/src/provider/Layers/PiProvider.ts index ecdc9b276206..19cd59d0ac4c 100644 --- a/apps/server/src/provider/Layers/PiProvider.ts +++ b/apps/server/src/provider/Layers/PiProvider.ts @@ -9,7 +9,6 @@ * shows up in T3 without any hardcoded catalog. */ import { - type ModelCapabilities, type PiSettings, type ServerProvider, type ServerProviderModel, @@ -17,7 +16,6 @@ import { type ServerProviderSlashCommand, } from "@t3tools/contracts"; import { causeErrorTag } from "@t3tools/shared/observability"; -import { createModelCapabilities } from "@t3tools/shared/model"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; import { tokenizeCliArgs } from "@t3tools/shared/cliArgs"; import * as DateTime from "effect/DateTime"; @@ -41,6 +39,10 @@ import { enrichProviderSnapshotWithVersionAdvisory, type ProviderMaintenanceCapabilities, } from "../providerMaintenance.ts"; +import { + EMPTY_PI_MODEL_CAPABILITIES, + thinkingCapabilitiesForPiModel, +} from "./piThinkingCapabilities.ts"; const PI_PRESENTATION = { displayName: "Pi", @@ -52,43 +54,12 @@ const PI_PRESENTATION = { const VERSION_PROBE_TIMEOUT_MS = 4_000; const PI_RPC_DISCOVERY_TIMEOUT_MS = 15_000; -const EMPTY_CAPABILITIES: ModelCapabilities = createModelCapabilities({ - optionDescriptors: [], -}); - -/** - * Reasoning-capable Pi models expose Pi's thinking levels. "Inherit" leaves - * the user's settings.json `defaultThinkingLevel` untouched. - * - * Only the levels every reasoning model accepts are advertised. Pi exposes - * `xhigh` and `max` per model (see `get_available_thinking_levels`), and - * offering them globally makes `set_thinking_level` fail on models that lack - * them, which blocks the turn from starting. - */ -const THINKING_CAPABILITIES: ModelCapabilities = createModelCapabilities({ - optionDescriptors: [ - { - id: "thinking", - label: "Thinking", - type: "select", - options: [ - { id: "inherit", label: "Pi default", isDefault: true }, - { id: "off", label: "Off" }, - { id: "minimal", label: "Minimal" }, - { id: "low", label: "Low" }, - { id: "medium", label: "Medium" }, - { id: "high", label: "High" }, - ], - }, - ], -}); - /** Deferring to the user's own settings.json default model. */ const PI_DEFAULT_MODEL: ServerProviderModel = { slug: "default", name: "Pi default", isCustom: false, - capabilities: EMPTY_CAPABILITIES, + capabilities: EMPTY_PI_MODEL_CAPABILITIES, }; interface PiDiscovery { @@ -105,7 +76,7 @@ function piModelsFromSettings( return providerModelsFromSettings( [PI_DEFAULT_MODEL, ...discovered], customModels ?? [], - EMPTY_CAPABILITIES, + EMPTY_PI_MODEL_CAPABILITIES, ); } @@ -135,8 +106,7 @@ function parseDiscoveredModels(data: unknown): ReadonlyArray { + it("returns no levels when the model does not advertise reasoning", () => { + assert.deepEqual(supportedPiThinkingLevelsFromModel({ reasoning: false }), []); + assert.deepEqual(supportedPiThinkingLevelsFromModel({}), []); + }); + + it("advertises off through high without Extra High or Max when the map is absent", () => { + assert.deepEqual(supportedPiThinkingLevelsFromModel({ reasoning: true }), [ + "off", + "minimal", + "low", + "medium", + "high", + ]); + }); + + it("adds Extra High and Max only when the map has a non-null entry", () => { + assert.deepEqual( + supportedPiThinkingLevelsFromModel({ + reasoning: true, + thinkingLevelMap: { xhigh: "xhigh", max: "max" }, + }), + ["off", "minimal", "low", "medium", "high", "xhigh", "max"], + ); + }); + + it("hides a mapped-null level and keeps Extra High when only that entry exists", () => { + assert.deepEqual( + supportedPiThinkingLevelsFromModel({ + reasoning: true, + thinkingLevelMap: { off: null, xhigh: "extra_high" }, + }), + ["minimal", "low", "medium", "high", "xhigh"], + ); + }); + + it("does not treat a null Extra High or Max entry as supported", () => { + assert.deepEqual( + supportedPiThinkingLevelsFromModel({ + reasoning: true, + thinkingLevelMap: { xhigh: null, max: null }, + }), + ["off", "minimal", "low", "medium", "high"], + ); + }); +}); + +describe("thinkingCapabilitiesForPiModel", () => { + it("returns empty capabilities for a non-reasoning model", () => { + assert.deepEqual( + thinkingCapabilitiesForPiModel({ reasoning: false }), + EMPTY_PI_MODEL_CAPABILITIES, + ); + }); + + it("prepends inherit and labels Extra High for grok-4.6-shaped maps", () => { + const capabilities = thinkingCapabilitiesForPiModel({ + reasoning: true, + thinkingLevelMap: { xhigh: "xhigh" }, + }); + const descriptors = capabilities.optionDescriptors ?? []; + const thinking = descriptors[0]; + assert.equal(thinking?.id, "thinking"); + assert.equal(thinking?.type, "select"); + if (thinking?.type !== "select") return; + assert.deepEqual( + thinking.options.map((option) => [option.id, option.label, option.isDefault === true]), + [ + ["inherit", "Pi default", true], + ["off", "Off", false], + ["minimal", "Minimal", false], + ["low", "Low", false], + ["medium", "Medium", false], + ["high", "High", false], + ["xhigh", "Extra High", false], + ], + ); + }); +}); diff --git a/apps/server/src/provider/Layers/piThinkingCapabilities.ts b/apps/server/src/provider/Layers/piThinkingCapabilities.ts new file mode 100644 index 000000000000..f6ffed21ca90 --- /dev/null +++ b/apps/server/src/provider/Layers/piThinkingCapabilities.ts @@ -0,0 +1,88 @@ +import { type ModelCapabilities, type ProviderOptionChoice } from "@t3tools/contracts"; +import { createModelCapabilities } from "@t3tools/shared/model"; + +/** + * Pi's full thinking ladder. Extra High (`xhigh`) and Max are opt-in per + * model via `thinkingLevelMap`; advertising them globally makes + * `set_thinking_level` fail on models that lack them. + */ +export const PI_THINKING_LEVELS = [ + "off", + "minimal", + "low", + "medium", + "high", + "xhigh", + "max", +] as const; + +export type PiThinkingLevel = (typeof PI_THINKING_LEVELS)[number]; + +const PI_THINKING_LEVEL_LABELS: Record = { + off: "Off", + minimal: "Minimal", + low: "Low", + medium: "Medium", + high: "High", + xhigh: "Extra High", + max: "Max", +}; + +const INHERIT_CHOICE: ProviderOptionChoice = { + id: "inherit", + label: "Pi default", + isDefault: true, +}; + +export const EMPTY_PI_MODEL_CAPABILITIES: ModelCapabilities = createModelCapabilities({ + optionDescriptors: [], +}); + +export function thinkingCapabilitiesForPiModel(model: unknown): ModelCapabilities { + const levels = supportedPiThinkingLevelsFromModel(model); + if (levels.length === 0) return EMPTY_PI_MODEL_CAPABILITIES; + return createModelCapabilities({ + optionDescriptors: [ + { + id: "thinking", + label: "Thinking", + type: "select", + options: [ + INHERIT_CHOICE, + ...levels.map((level) => ({ + id: level, + label: PI_THINKING_LEVEL_LABELS[level], + })), + ], + }, + ], + }); +} + +/** + * Mirror of `@earendil-works/pi-ai` `getSupportedThinkingLevels`. + * + * A reasoning model always exposes off through high unless a map entry is + * `null`. Extra High and Max appear only when the map has a non-null entry. + */ +export function supportedPiThinkingLevelsFromModel(model: unknown): ReadonlyArray { + if (recordField(model, "reasoning") !== true) return []; + const thinkingLevelMap = thinkingLevelMapFromModel(model); + return PI_THINKING_LEVELS.filter((level) => { + const mapped = thinkingLevelMap?.[level]; + if (mapped === null) return false; + if (level === "xhigh" || level === "max") return mapped !== undefined; + return true; + }); +} + +function thinkingLevelMapFromModel(model: unknown): Record | undefined { + const value = recordField(model, "thinkingLevelMap"); + if (typeof value !== "object" || value === null || Array.isArray(value)) return undefined; + return value as Record; +} + +function recordField(input: unknown, key: string): unknown { + if (typeof input !== "object" || input === null) return undefined; + return (input as Record)[key]; +} diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 3aa9a6cb401a..dc7ba1c1abf0 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -130,10 +130,34 @@ forking uses portable context when native `session/fork` is unavailable, and subagents use orchestrator-owned child threads. Registry agents do not receive provider-specific extensions; those remain in flavors such as Grok. +### Pi V2 + +Pi core has no MCP client. When a provider session credential exists, the +adapter writes a T3-owned extension into the server cache and spawns +`pi --mode rpc --extension /pi-t3-mcp-extension.ts` with: + +```text +T3_MCP_URL=http://127.0.0.1:/mcp +T3_MCP_BEARER_TOKEN= +``` + +The extension connects to that HTTP endpoint, lists tools, and registers each +one with `pi.registerTool` under its original name (`delegate_task`, +`t3_thread_start`, and the rest). Follow-up HTTP requests send +`mcp-protocol-version: 2025-06-18`; Effect's MCP transport returns 400 +without it. User `launchArgs` are preserved. The first turn of a session +also receives the shared T3 orchestration instructions. + +A second T3-owned extension overrides the official `subagent` tool to +persist `--session` and report `sessionFile`. Duplicate `subagent` +registrations abort Pi, so the launcher disables extension discovery and +drops the official tool from `launchArgs`. The adapter binds each result +as a child thread that later sends resume through `switch_session`. + ### Initial Provider Support -The V2 provider adapters are Codex, Claude Agent SDK, Cursor Agent SDK, and -Grok plus generic registry agents over ACP. +The V2 provider adapters are Codex, Claude Agent SDK, Cursor Agent SDK, Grok +plus generic registry agents over ACP, OpenCode, OpenCode 2, and Pi. Capability discovery still reports other registered provider instances, but marks them unavailable for orchestration when no V2 adapter exists. This keeps provider selection model-visible without allowing a request that cannot run.