diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 417f0a95e8552..dc04aa9e06db3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -523,13 +523,13 @@ jobs: - name: Check Rust formatting run: | - for crate in resource-monitor kde-snap-shot hyprland-snap-shot; do + for crate in resource-monitor kde-snap-shot hyprland-snap-shot t3-mcp-plugin; do cargo fmt --manifest-path "native/$crate/Cargo.toml" -- --check done - name: Test Rust crates run: | - for crate in resource-monitor kde-snap-shot hyprland-snap-shot; do + for crate in resource-monitor kde-snap-shot hyprland-snap-shot t3-mcp-plugin; do cargo test --locked --manifest-path "native/$crate/Cargo.toml" done diff --git a/apps/server/src/mcp/mcpPlugin.test.ts b/apps/server/src/mcp/mcpPlugin.test.ts new file mode 100644 index 0000000000000..c56405f1d79b5 --- /dev/null +++ b/apps/server/src/mcp/mcpPlugin.test.ts @@ -0,0 +1,47 @@ +// @effect-diagnostics nodeBuiltinImport:off - Reads the plugin's files from native/. +import * as NodeFS from "node:fs"; +import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { claudeMcpQueryOverrides } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; +import { codexThreadRuntimeParams } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; + +// The wrappers in native/t3-mcp-plugin parse these servers, and its plugin copies their settings. +// A failure here means T3 Code's handoff changed: update the wrappers, plugin, and fixtures. +const native = new URL("../../../../native/t3-mcp-plugin/", import.meta.url); +const read = (path: string) => JSON.parse(NodeFS.readFileSync(new URL(path, native), "utf8")); +const without = (value: {} | null, ...keys: ReadonlyArray) => + Object.fromEntries(Object.entries(value ?? {}).filter(([key]) => !keys.includes(key))); + +const mcpSession = { + environmentId: EnvironmentId.make("environment"), + threadId: ThreadId.make("thread"), + providerSessionId: "session", + providerInstanceId: ProviderInstanceId.make("provider"), + endpoint: "http://127.0.0.1:3773/mcp", + authorizationHeader: "Bearer token", + browserToolsAvailable: true, +}; + +describe("T3 Code's handoff to the MCP plugin", () => { + it("gives Codex the servers t3-codex parses, with the settings the plugin copies", () => { + const { config } = codexThreadRuntimeParams({ threadId: mcpSession.threadId, mcpSession }); + const servers = Object.entries(config).filter(([key]) => key.startsWith("mcp_servers")); + expect(Object.fromEntries(servers)).toEqual(read("fixtures/codex-mcp-servers.json")); + const plugin = read("plugin/plugins/t3-code/codex.mcp.json").mcpServers; + for (const [key, server] of servers) { + expect( + without(plugin[key.replace("mcp_servers.", "")], "command", "args", "env_vars"), + ).toEqual(without(server, "url", "http_headers")); + } + }); + + it("gives Claude the server t3-claude parses, with the settings the plugin copies", () => { + const servers = claudeMcpQueryOverrides({ mcpSession, readOnlySandbox: false }).mcpServers; + expect(servers).toEqual(read("fixtures/claude-mcp-servers.json")); + const plugin = read("plugin/plugins/t3-code/.mcp.json").mcpServers; + expect(without(plugin["t3-code"], "command", "args")).toEqual( + without(servers?.["t3-code"] ?? {}, "type", "url", "headers"), + ); + }); +}); diff --git a/apps/server/src/mcp/mcpUnixSocket.test.ts b/apps/server/src/mcp/mcpUnixSocket.test.ts new file mode 100644 index 0000000000000..9f02e23faf2df --- /dev/null +++ b/apps/server/src/mcp/mcpUnixSocket.test.ts @@ -0,0 +1,90 @@ +// @effect-diagnostics nodeBuiltinImport:off - Real Unix sockets, and a child that leaves one behind. +import * as NodeChildProcess from "node:child_process"; +import * as NodeHttp from "node:http"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as ConfigProvider from "effect/ConfigProvider"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import { HttpServer, HttpServerRequest, HttpServerResponse } from "effect/http"; + +import { layerNodeHttpServer } from "./mcpUnixSocket.ts"; + +const request = (socketPath: string, path: string) => + Effect.promise( + () => + new Promise<{ status: number; body: string }>((resolve, reject) => { + const sent = NodeHttp.request( + { socketPath, path, method: "POST", agent: false }, + (response) => { + let body = ""; + response.on("data", (chunk) => (body += chunk)); + response.on("end", () => resolve({ status: response.statusCode ?? 0, body })); + }, + ); + sent.on("error", reject); + sent.end("{}"); + }), + ); + +// Through the real layer with real listen options, as server.ts builds it. +const serve = Effect.fn("mcpUnixSocket.test.serve")(function* (path: string) { + const context = yield* Layer.build( + layerNodeHttpServer(() => NodeHttp.createServer(), { host: "127.0.0.1", port: 0 }).pipe( + Layer.provide( + ConfigProvider.layer(ConfigProvider.fromEnv({ env: { T3CODE_MCP_SOCKET: path } })), + ), + ), + ); + yield* Context.get(context, HttpServer.HttpServer).serve( + Effect.map(HttpServerRequest.HttpServerRequest, (req) => + HttpServerResponse.text(`app saw ${req.url}`), + ), + ); +}); + +it.live("serves only /mcp from the main app, owner-only, and removes the socket on close", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = `${yield* fs.makeTempDirectoryScoped({ prefix: "t3-mcp-socket-" })}/mcp.sock`; + yield* Effect.scoped( + Effect.gen(function* () { + yield* serve(path); + expect(((yield* fs.stat(path)).mode & 0o777).toString(8)).toBe("600"); + expect(yield* request(path, "/mcp")).toEqual({ status: 200, body: "app saw /mcp" }); + expect(yield* request(path, "/mcp?x=1")).toEqual({ status: 200, body: "app saw /mcp?x=1" }); + expect((yield* request(path, "/api/settings")).status).toBe(404); + expect((yield* request(path, "/mcpx")).status).toBe(404); + }), + ); + expect(yield* fs.exists(path)).toBe(false); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.live("replaces a stale socket, but never a live one or another file", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const dir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-mcp-socket-" }); + const path = `${dir}/mcp.sock`; + // A server killed before cleanup leaves its socket file behind. + NodeChildProcess.spawnSync(process.execPath, [ + "-e", + `require("node:net").createServer().listen(${JSON.stringify(path)}, () => process.kill(process.pid, "SIGKILL"))`, + ]); + expect((yield* fs.stat(path)).type).toBe("Socket"); + yield* serve(path); + expect((yield* request(path, "/mcp")).status).toBe(200); + + expect(yield* Effect.flip(Effect.scoped(serve(path)))).toMatchObject({ reason: "in-use" }); + expect((yield* request(path, "/mcp")).status).toBe(200); + + const file = `${dir}/not-a-socket`; + yield* fs.writeFileString(file, "data"); + expect(yield* Effect.flip(Effect.scoped(serve(file)))).toMatchObject({ + reason: "not-a-socket", + }); + expect(yield* fs.readFileString(file)).toBe("data"); + }).pipe(Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/mcp/mcpUnixSocket.ts b/apps/server/src/mcp/mcpUnixSocket.ts new file mode 100644 index 0000000000000..da19a4a881e79 --- /dev/null +++ b/apps/server/src/mcp/mcpUnixSocket.ts @@ -0,0 +1,115 @@ +// @effect-diagnostics nodeBuiltinImport:off - Effect cannot tell whether a Unix socket is live. +import * as NodeHttp from "node:http"; +import * as NodeNet from "node:net"; +import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; +import * as Config from "effect/Config"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import type { LazyArg } from "effect/Function"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import type * as Scope from "effect/Scope"; +import { HttpMiddleware, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/http"; + +import { + guardHttpResponseWriteErrors, + rejectRequestsUntilServing, +} from "../httpResponseErrorGuard.ts"; + +// Opt-in, for hosts that forbid MCP over loopback TCP. +const MCP_SOCKET_ENV = "T3CODE_MCP_SOCKET"; + +class McpSocketError extends Schema.TaggedError()("McpSocketError", { + path: Schema.String, + reason: Schema.Literals(["not-a-socket", "in-use"]), +}) { + override get message(): string { + return this.reason === "in-use" + ? `${MCP_SOCKET_ENV} names ${this.path}, which another server is listening on.` + : `${MCP_SOCKET_ENV} names ${this.path}, which exists and is not a socket.`; + } +} + +// `/mcp`, `/mcp/...` and `/mcp?...` only. Without the boundary `/mcpx` would match, and the +// socket would expose a route other than MCP. +const isMcpUrl = (url: string) => /^\/mcp(?:[/?]|$)/.test(url); + +const onlyMcp = (app: Effect.Effect) => + Effect.flatMap(HttpServerRequest.HttpServerRequest, (request) => + isMcpUrl(request.url) ? app : Effect.succeed(HttpServerResponse.empty({ status: 404 })), + ); + +// Refused means nobody is listening: the file was left by a server that died. +const isListening = (path: string) => + Effect.callback((resume) => { + const probe = NodeNet.connect(path); + probe.once("connect", () => { + probe.destroy(); + resume(Effect.succeed(true)); + }); + probe.once("error", (cause: NodeJS.ErrnoException) => + resume(cause.code === "ECONNREFUSED" ? Effect.succeed(false) : Effect.die(cause)), + ); + return Effect.sync(() => probe.destroy()); + }); + +const removeStaleSocket = Effect.fn("McpUnixSocket.removeStaleSocket")(function* (path: string) { + const fs = yield* FileSystem.FileSystem; + const info = yield* fs + .stat(path) + .pipe(Effect.catchReason("PlatformError", "NotFound", () => Effect.undefined)); + if (!info) return; + if (info.type !== "Socket") return yield* new McpSocketError({ path, reason: "not-a-socket" }); + if (yield* isListening(path)) return yield* new McpSocketError({ path, reason: "in-use" }); + yield* fs.remove(path); +}); + +// For a caller that may or may not pass middleware, as `HttpServer.make` allows. +const serveWith = ( + server: HttpServer.HttpServer["Service"], + app: Effect.Effect, + middleware: HttpMiddleware.HttpMiddleware | undefined, +): Effect.Effect | Scope.Scope> => + // @effect-diagnostics-next-line anyUnknownInErrorContext:off - HttpMiddleware's types are `any` + middleware ? server.serve(app, middleware) : server.serve(app); + +/** + * `NodeHttpServer.layer`, plus the same app on the Unix socket `T3CODE_MCP_SOCKET` names, + * restricted to `/mcp`. The socket is chmodded to 0600 just after it is created, so put it in an + * owner-only directory. Node removes it on close. + */ +export const layerNodeHttpServer = ( + evaluate: LazyArg, + options: NodeHttpServer.Options, +) => + Layer.merge( + Layer.effect( + HttpServer.HttpServer, + Effect.gen(function* () { + const main = yield* NodeHttpServer.make(evaluate, options); + const path = yield* Config.option(Config.String(MCP_SOCKET_ENV)); + if (Option.isNone(path)) return main; + const fs = yield* FileSystem.FileSystem; + yield* removeStaleSocket(path.value); + const socket = yield* NodeHttpServer.make( + () => rejectRequestsUntilServing(guardHttpResponseWriteErrors(NodeHttp.createServer())), + { gracefulShutdownTimeout: options.gracefulShutdownTimeout, path: path.value }, + ); + yield* fs.chmod(path.value, 0o600); + yield* Effect.logInfo("Serving /mcp on a Unix socket", { path: path.value }); + return HttpServer.make({ + address: main.address, + serve: ( + app: Effect.Effect, + middleware?: HttpMiddleware.HttpMiddleware, + ) => + Effect.andThen( + serveWith(main, app, middleware), + serveWith(socket, onlyMcp(app), middleware), + ), + }); + }), + ), + NodeHttpServer.layerHttpServices, + ); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 07c4e4e72eca7..b82ff8955c35d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1395,7 +1395,7 @@ const CODEX_T3_MCP_TOOL_TIMEOUT_SEC = 65 * 60; const CODEX_T3_DIRECT_MCP_SERVER = "t3-code-direct"; const CODEX_DIRECT_ONLY_T3_TOOL = "t3_thread_wait"; -function codexThreadRuntimeParams(input: { +export function codexThreadRuntimeParams(input: { readonly threadId: ThreadId | null; readonly mcpSession: McpProviderSession.McpProviderSessionConfig | undefined; readonly modelSelection?: { readonly model: string }; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 5c9ab9c433ffc..291dc1b70322e 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -12,7 +12,6 @@ import * as DelegatedTaskQuestionReactor from "./orchestration-v2/DelegatedTaskQ // @effect-diagnostics nodeBuiltinImport:off import * as NodeHttp from "node:http"; -import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { EnvironmentHttpApi, type RepositoryIdentity } from "@t3tools/contracts"; @@ -77,6 +76,7 @@ import * as ProviderInstanceRegistryHydration from "./provider/ProviderInstanceR import * as ProjectSettingsOverrideSweep from "./provider/ProjectSettingsOverrideSweep.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; +import * as McpUnixSocket from "./mcp/mcpUnixSocket.ts"; import * as PeerForwarding from "./peer/PeerForwarding.ts"; import * as RemoteDelegation from "./peer/RemoteDelegation.ts"; import * as PeerLinkRequests from "./peer/PeerLinkRequests.ts"; @@ -350,7 +350,7 @@ const layerHttpServer = Layer.unwrap( privateNetworkOnly: config.privateNetworkOnly, }); yield* ManagedLimits.closeTailscaleServeTo(policy, { host: bindHost, port: config.port }); - return NodeHttpServer.layer( + return McpUnixSocket.layerNodeHttpServer( () => { const server = rejectRequestsUntilServing( guardHttpResponseWriteErrors(NodeHttp.createServer()), diff --git a/native/t3-mcp-plugin/Cargo.lock b/native/t3-mcp-plugin/Cargo.lock new file mode 100644 index 0000000000000..42d32493eaecf --- /dev/null +++ b/native/t3-mcp-plugin/Cargo.lock @@ -0,0 +1,350 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "hyper" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c3e324da4c95177d6291d4c8730197c0d1822f8a9766814a4a44fa5ab797c9c" +dependencies = [ + "atomic-waker", + "bytes", + "futures-core", + "http", + "http-body", + "httparse", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-util" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.191" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" + +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "rustix" +version = "1.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5e76742e5294153d4dde1207dbec5bdd81eef6fabdc0d2118babadabfd858aa" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.154" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "smallvec" +version = "1.16.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "149309498a5f7334b12ddae0cad8aba81ce785fe91755068867e017cf98fee46" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "syn" +version = "3.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "t3-mcp-plugin" +version = "0.1.0" +dependencies = [ + "bytes", + "http-body-util", + "hyper", + "hyper-util", + "rustix", + "serde", + "serde_json", + "tokio", +] + +[[package]] +name = "tokio" +version = "1.54.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b52fd3255258d26fac7264ab5a76b53c13367136dc011d3d6ad4f8ed070396a3" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + +[[package]] +name = "unicode-ident" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" + +[[package]] +name = "want" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec4cdd0dd910afe868b7ef477227d8d538b46b3075031afee8a9f2acb0a2ed0b" +dependencies = [ + "try-lock", +] + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/native/t3-mcp-plugin/Cargo.toml b/native/t3-mcp-plugin/Cargo.toml new file mode 100644 index 0000000000000..44d5f8b50d43b --- /dev/null +++ b/native/t3-mcp-plugin/Cargo.toml @@ -0,0 +1,42 @@ +[package] +name = "t3-mcp-plugin" +version = "0.1.0" +edition = "2024" +license = "MIT" +publish = false + +[[bin]] +name = "t3-codex" +path = "src/bin/t3-codex.rs" + +[[bin]] +name = "t3-claude" +path = "src/bin/t3-claude.rs" + +[dependencies] +bytes = "1.12.1" +http-body-util = "0.1.5" +hyper = { version = "1.12.0", features = ["client", "http1"] } +hyper-util = { version = "0.1.21", features = ["tokio"] } +serde = { version = "1.0.228", features = ["derive"] } +serde_json = { version = "1.0.150", features = ["raw_value"] } +tokio = { version = "1.54.1", features = [ + "io-std", + "io-util", + "macros", + "net", + "process", + "rt", + "signal", + "sync", + "time", +] } + +[target.'cfg(unix)'.dependencies] +rustix = { version = "1.1.2", features = ["process"] } + +[profile.release] +codegen-units = 1 +lto = "thin" +panic = "abort" +strip = true diff --git a/native/t3-mcp-plugin/README.md b/native/t3-mcp-plugin/README.md new file mode 100644 index 0000000000000..db0eddb5f8567 --- /dev/null +++ b/native/t3-mcp-plugin/README.md @@ -0,0 +1,137 @@ +# t3-mcp-plugin + +Gives Codex and Claude Code sessions started by T3 Code access to T3 Code's MCP tools on machines +where an admin policy allows MCP servers only from approved plugins. T3 Code does not change; an +optional transport (below) is the only server addition. + +## Pieces + +```text +src/ Rust; builds `t3-codex` and `t3-claude` +plugin/.agents/plugins/marketplace.json Codex marketplace "t3" +plugin/.claude-plugin/marketplace.json Claude Code marketplace "t3" +plugin/plugins/t3-code/codex.mcp.json Codex servers: `t3-codex mcp-bridge` +plugin/plugins/t3-code/.mcp.json Claude Code server: `t3-claude mcp-bridge` +plugin/policy/codex/requirements.toml Example Codex admin policy +plugin/policy/claude/managed-settings.json Example Claude Code admin policy +``` + +Each binary has two roles. `t3-codex app-server` and `t3-claude` (as the Agent SDK runs it) stand +in for the real CLI; ` mcp-bridge` is the plugin's MCP server. Any other invocation, such +as `--version`, runs the real CLI directly. + +## How it works + +T3 Code hands each CLI its MCP server with that thread's own credential. A plugin-only policy +refuses servers added that way, so the wrapper moves them into the plugin: + +1. T3 Code's **binary path** setting for the provider points at the wrapper, which starts the + real CLI and relays its stdio. +2. The wrapper removes T3 Code's servers and records the credential: + - Codex: `t3-code` and `t3-code-direct` from the config of `thread/start`, `thread/resume`, + and `thread/fork`. + - Claude Code: `t3-code` from the Agent SDK's `mcp_set_servers` control request, which T3 + Code also sends when it rotates the credential. +3. The CLI starts the plugin's bridge, which only pipes stdio to the wrapper over a private + channel named by `T3_MCP_BRIDGE` (a Unix socket, or a named pipe on Windows). The wrapper + answers as the MCP server and sends each tool call to T3 Code with the calling thread's + credential, one MCP session per credential. +4. Identity: + - Codex hosts many threads per app-server; `_meta.threadId` names the caller, and a subagent + T3 Code never saw is traced to its parent with `thread/read`. + - Claude Code runs one process per thread, and subagents share it, so every call is that + thread's. + +T3 Code sees exactly the per-thread credentials it issues today, so scopes, capabilities, idle +expiry, rotation, and revocation work unchanged. + +Claude Code names plugin tools `mcp__plugin_t3-code_t3-code__`, while T3 Code expects +`mcp__t3-code__` in its allowed tools, approvals, and tool rows. `t3-claude` maps the names +in both directions: in Claude's arguments and input, and in its output. + +## Plugin server settings + +The MCP files are plain JSON: Codex silently loads no servers from a file with comments. Their +settings copy what T3 Code sends each thread today, in `codexThreadRuntimeParams` and +`claudeMcpQueryOverrides`. `apps/server/src/mcp/mcpPlugin.test.ts` fails when they drift apart. + +| Setting | Why | +| ---------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `command`, `args` | The bridge. Each policy allows the server only when these match exactly. | +| `env_vars: ["T3_MCP_BRIDGE"]` (Codex) | Codex starts MCP servers with a reduced environment. Without this the bridge cannot find the wrapper and serves no tools. | +| `tool_timeout_sec: 3900` (Codex), `timeout: 3900000` (Claude Code, ms) | 65 minutes. `t3_thread_wait` can block for an hour. | +| `t3-code` and `t3-code-direct` (Codex) | The same bridge registered twice. Code Mode would run the blocking `t3_thread_wait` as a nested call while the model keeps working, so `t3-code` disables it and `t3-code-direct` carries only that tool, hidden from Code Mode and deferred loading. | + +## Setup on a managed machine + +1. Build and install the binaries: + + ```sh + cargo build --locked --release --manifest-path native/t3-mcp-plugin/Cargo.toml + ``` + + Each wrapper puts its own directory first on the CLI's `PATH`, so the CLI finds it when the + plugin runs ` mcp-bridge`. + +2. Install the plugin from `native/t3-mcp-plugin/plugin`: + + ```sh + codex plugin marketplace add + codex plugin add t3-code@t3 + claude plugin marketplace add + claude plugin install t3-code@t3 + ``` + +3. In T3 Code, set **Settings → Providers → Codex → Binary path** to `t3-codex` and + **Settings → Providers → Claude → Binary path** to `t3-claude`. Each finds the real CLI on + `PATH`, skipping itself; set `T3_CODEX_REAL` or `T3_CLAUDE_REAL` to choose another. + +4. Deploy the policies: Codex's as `/etc/codex/requirements.toml`, Claude Code's as + `/etc/claude-code/managed-settings.json` (Linux) or the macOS managed settings location. Set + `strictKnownMarketplaces` to the approved source of this plugin. + +## Transport: TCP or Unix socket + +By default the wrappers reach T3 Code's `/mcp` over loopback TCP, at the URL T3 Code gave. For +machines where MCP over loopback TCP is not allowed, start T3 Code with `T3CODE_MCP_SOCKET=`. +T3 Code then also serves `/mcp`, and nothing else, on that Unix socket (mode 0600; put it in an +owner-only directory such as `$XDG_RUNTIME_DIR`). The wrappers inherit the variable and use the +socket. See `apps/server/src/mcp/mcpUnixSocket.ts`. + +## Credential handling + +- The bridge channel lives in a new owner-only directory under `XDG_RUNTIME_DIR` (or the temp + directory). The wrapper removes it on exit and forwards SIGTERM, SIGINT, and SIGHUP to the CLI. + A directory left by SIGKILL is removed by the next wrapper start. +- Codex agent shells do not see `T3_MCP_BRIDGE`: `t3-codex` adds + `shell_environment_policy.filters."T3_MCP_*" = "exclude"` to each thread it rewrites. +- Claude Code passes its environment to Bash and has no per-variable filter, so Bash in a Claude + thread can see `T3_MCP_BRIDGE`. It reaches only that thread's own T3 Code access, which the + agent already has. +- A process that finds a Codex wrapper's channel can make tool calls as any thread on that + app-server. Its name is in the app-server's environment, which unsandboxed processes running + as the same user can read. Those processes can already read T3 Code's own data and secrets. + +## Outside T3 Code + +The plugin is on for every session. A session T3 Code did not start has no channel, so the bridge, +if its binary is on `PATH`, answers as an MCP server with no tools. Otherwise the CLI reports that +the server failed to start; that does not block the session. + +## Stable contract + +Managed installs pin these: the marketplace name `t3`, the plugin ID `t3-code@t3`, the server +names `t3-code` and `t3-code-direct`, the commands `t3-codex mcp-bridge` and `t3-claude +mcp-bridge`, and the variable `T3_MCP_BRIDGE`. + +The wrappers also depend on how T3 Code hands over its server: `codexThreadRuntimeParams` and +`claudeMcpQueryOverrides` in the provider adapters (a server named `t3-code` with `url` and an +`Authorization` header). `apps/server/src/mcp/mcpPlugin.test.ts` checks T3 Code's output against +`fixtures/`, which the Rust tests parse, so a change there fails CI. A wrapper older than the +T3 Code it runs under logs that it does not understand the fields; tool calls then fail instead +of running as another thread. + +## Requirements + +- Codex 0.167.0 or later; Claude Code 2.1.319 or later, through Agent SDK 0.3.316. +- Tested on Linux. macOS and Windows builds type-check but have not been run. diff --git a/native/t3-mcp-plugin/fixtures/claude-mcp-servers.json b/native/t3-mcp-plugin/fixtures/claude-mcp-servers.json new file mode 100644 index 0000000000000..3b09663590ca8 --- /dev/null +++ b/native/t3-mcp-plugin/fixtures/claude-mcp-servers.json @@ -0,0 +1,8 @@ +{ + "t3-code": { + "type": "http", + "url": "http://127.0.0.1:3773/mcp", + "headers": { "Authorization": "Bearer token" }, + "timeout": 3900000 + } +} diff --git a/native/t3-mcp-plugin/fixtures/codex-mcp-servers.json b/native/t3-mcp-plugin/fixtures/codex-mcp-servers.json new file mode 100644 index 0000000000000..dc0334745faf9 --- /dev/null +++ b/native/t3-mcp-plugin/fixtures/codex-mcp-servers.json @@ -0,0 +1,15 @@ +{ + "mcp_servers.t3-code": { + "url": "http://127.0.0.1:3773/mcp", + "tool_timeout_sec": 3900, + "http_headers": { "Authorization": "Bearer token" }, + "disabled_tools": ["t3_thread_wait"] + }, + "mcp_servers.t3-code-direct": { + "url": "http://127.0.0.1:3773/mcp", + "tool_timeout_sec": 3900, + "http_headers": { "Authorization": "Bearer token" }, + "enabled_tools": ["t3_thread_wait"], + "omit_tools_from": ["code_mode", "deferred"] + } +} diff --git a/native/t3-mcp-plugin/plugin/.agents/plugins/marketplace.json b/native/t3-mcp-plugin/plugin/.agents/plugins/marketplace.json new file mode 100644 index 0000000000000..f53405b7a6431 --- /dev/null +++ b/native/t3-mcp-plugin/plugin/.agents/plugins/marketplace.json @@ -0,0 +1,9 @@ +{ + "name": "t3", + "plugins": [ + { + "name": "t3-code", + "source": { "source": "local", "path": "./plugins/t3-code" } + } + ] +} diff --git a/native/t3-mcp-plugin/plugin/.claude-plugin/marketplace.json b/native/t3-mcp-plugin/plugin/.claude-plugin/marketplace.json new file mode 100644 index 0000000000000..33910bece380c --- /dev/null +++ b/native/t3-mcp-plugin/plugin/.claude-plugin/marketplace.json @@ -0,0 +1,11 @@ +{ + "name": "t3", + "owner": { "name": "T3 Tools" }, + "plugins": [ + { + "name": "t3-code", + "source": "./plugins/t3-code", + "description": "Connects Claude Code sessions started by T3 Code to T3 Code's tools." + } + ] +} diff --git a/native/t3-mcp-plugin/plugin/plugins/t3-code/.claude-plugin/plugin.json b/native/t3-mcp-plugin/plugin/plugins/t3-code/.claude-plugin/plugin.json new file mode 100644 index 0000000000000..1ef77ad7bada8 --- /dev/null +++ b/native/t3-mcp-plugin/plugin/plugins/t3-code/.claude-plugin/plugin.json @@ -0,0 +1,8 @@ +{ + "name": "t3-code", + "version": "0.1.0", + "description": "Connects Claude Code sessions started by T3 Code to T3 Code's tools.", + "author": { "name": "T3 Tools" }, + "homepage": "https://t3.codes", + "keywords": ["t3-code", "orchestration"] +} diff --git a/native/t3-mcp-plugin/plugin/plugins/t3-code/.codex-plugin/plugin.json b/native/t3-mcp-plugin/plugin/plugins/t3-code/.codex-plugin/plugin.json new file mode 100644 index 0000000000000..f079390d68e1d --- /dev/null +++ b/native/t3-mcp-plugin/plugin/plugins/t3-code/.codex-plugin/plugin.json @@ -0,0 +1,14 @@ +{ + "name": "t3-code", + "version": "0.1.0", + "description": "Connects Codex threads started by T3 Code to T3 Code's tools.", + "keywords": ["t3-code", "orchestration"], + "mcpServers": "./codex.mcp.json", + "interface": { + "displayName": "T3 Code", + "shortDescription": "T3 Code's tools for threads T3 Code starts", + "developerName": "T3 Tools", + "category": "Developer Tools", + "websiteUrl": "https://t3.codes" + } +} diff --git a/native/t3-mcp-plugin/plugin/plugins/t3-code/.mcp.json b/native/t3-mcp-plugin/plugin/plugins/t3-code/.mcp.json new file mode 100644 index 0000000000000..e9c1b31a92042 --- /dev/null +++ b/native/t3-mcp-plugin/plugin/plugins/t3-code/.mcp.json @@ -0,0 +1,9 @@ +{ + "mcpServers": { + "t3-code": { + "command": "t3-claude", + "args": ["mcp-bridge"], + "timeout": 3900000 + } + } +} diff --git a/native/t3-mcp-plugin/plugin/plugins/t3-code/codex.mcp.json b/native/t3-mcp-plugin/plugin/plugins/t3-code/codex.mcp.json new file mode 100644 index 0000000000000..f2f3eaab79e77 --- /dev/null +++ b/native/t3-mcp-plugin/plugin/plugins/t3-code/codex.mcp.json @@ -0,0 +1,19 @@ +{ + "mcpServers": { + "t3-code": { + "command": "t3-codex", + "args": ["mcp-bridge"], + "env_vars": ["T3_MCP_BRIDGE"], + "tool_timeout_sec": 3900, + "disabled_tools": ["t3_thread_wait"] + }, + "t3-code-direct": { + "command": "t3-codex", + "args": ["mcp-bridge"], + "env_vars": ["T3_MCP_BRIDGE"], + "tool_timeout_sec": 3900, + "enabled_tools": ["t3_thread_wait"], + "omit_tools_from": ["code_mode", "deferred"] + } + } +} diff --git a/native/t3-mcp-plugin/plugin/policy/claude/managed-settings.json b/native/t3-mcp-plugin/plugin/policy/claude/managed-settings.json new file mode 100644 index 0000000000000..53363910d46f0 --- /dev/null +++ b/native/t3-mcp-plugin/plugin/policy/claude/managed-settings.json @@ -0,0 +1,6 @@ +{ + "strictPluginOnlyCustomization": ["mcp"], + "strictKnownMarketplaces": [{ "source": "github", "repo": "/" }], + "allowManagedMcpServersOnly": true, + "allowedMcpServers": [{ "serverCommand": ["t3-claude", "mcp-bridge"] }] +} diff --git a/native/t3-mcp-plugin/plugin/policy/codex/requirements.toml b/native/t3-mcp-plugin/plugin/policy/codex/requirements.toml new file mode 100644 index 0000000000000..83a45b1dc59be --- /dev/null +++ b/native/t3-mcp-plugin/plugin/policy/codex/requirements.toml @@ -0,0 +1,13 @@ +# Example Codex admin policy: MCP servers only from approved plugins. +# Deploy as /etc/codex/requirements.toml (Linux) or through your device management tool. + +# No MCP server may come from config.toml, launch arguments, or per-thread config. +[mcp_servers.none-allowed-from-config.identity] +command = "/nonexistent" + +# The T3 Code plugin's two servers, pinned to exactly `t3-codex mcp-bridge`. +[plugins."t3-code@t3".mcp_servers.t3-code.identity] +command = { executable = "t3-codex", args = [{ match = "exact", value = "mcp-bridge" }] } + +[plugins."t3-code@t3".mcp_servers.t3-code-direct.identity] +command = { executable = "t3-codex", args = [{ match = "exact", value = "mcp-bridge" }] } diff --git a/native/t3-mcp-plugin/src/bin/t3-claude.rs b/native/t3-mcp-plugin/src/bin/t3-claude.rs new file mode 100644 index 0000000000000..2de2435753625 --- /dev/null +++ b/native/t3-mcp-plugin/src/bin/t3-claude.rs @@ -0,0 +1,3 @@ +fn main() { + t3_mcp_plugin::main(t3_mcp_plugin::claude::run); +} diff --git a/native/t3-mcp-plugin/src/bin/t3-codex.rs b/native/t3-mcp-plugin/src/bin/t3-codex.rs new file mode 100644 index 0000000000000..76eea5973c047 --- /dev/null +++ b/native/t3-mcp-plugin/src/bin/t3-codex.rs @@ -0,0 +1,3 @@ +fn main() { + t3_mcp_plugin::main(t3_mcp_plugin::codex::run); +} diff --git a/native/t3-mcp-plugin/src/bridge.rs b/native/t3-mcp-plugin/src/bridge.rs new file mode 100644 index 0000000000000..642ea3c7aa8a6 --- /dev/null +++ b/native/t3-mcp-plugin/src/bridge.rs @@ -0,0 +1,116 @@ +use crate::credentials::RequestId; +use crate::ipc; +use serde::Deserialize; +use serde_json::{Value, json}; +use std::ffi::OsStr; +use std::io::{BufRead, Read, Write}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::mpsc; + +const DEFAULT_PROTOCOL_VERSION: &str = "2025-06-18"; + +/// Pipes stdio to the wrapper that started this Codex. Without one, T3 Code did not start this +/// session, so it answers as an MCP server with no tools. +pub fn run() -> crate::Result { + let Some(name) = std::env::var_os(ipc::ENV).filter(|name| !name.is_empty()) else { + offline(); + return Ok(0); + }; + crate::runtime()?.block_on(pipe(&name))?; + Ok(0) +} + +/// Runs until either side closes: Codex closing stdin asks the server to exit (MCP stdio). +async fn pipe(name: &OsStr) -> crate::Result<()> { + let stream = ipc::connect(name) + .await + .map_err(|error| format!("cannot reach t3-codex: {error}"))?; + let (mut from_wrapper, mut to_wrapper) = tokio::io::split(stream); + // A plain thread, not tokio's stdin: that reads on a blocking thread the runtime cannot + // cancel, so the runtime could never shut down while Codex keeps stdin open. + let (stdin_chunks, mut requests) = mpsc::unbounded_channel::>(); + std::thread::spawn(move || { + let mut stdin = std::io::stdin().lock(); + let mut buffer = [0; 16 * 1024]; + while let Ok(read @ 1..) = stdin.read(&mut buffer) { + if stdin_chunks.send(buffer[..read].to_vec()).is_err() { + return; + } + } + }); + let forward_requests = async { + while let Some(chunk) = requests.recv().await { + if to_wrapper.write_all(&chunk).await.is_err() { + return; + } + } + }; + let forward_replies = async { + let mut stdout = tokio::io::stdout(); + let mut buffer = [0; 16 * 1024]; + while let Ok(read @ 1..) = from_wrapper.read(&mut buffer).await { + if stdout.write_all(&buffer[..read]).await.is_err() || stdout.flush().await.is_err() { + return; + } + } + }; + tokio::select! { + () = forward_requests => {} + () = forward_replies => {} + } + Ok(()) +} + +fn offline() { + #[derive(Deserialize)] + struct Request { + id: Option, + method: Option, + #[serde(default)] + params: Value, + } + let mut stdin = std::io::stdin().lock(); + let mut stdout = std::io::stdout(); + let mut line = Vec::new(); + loop { + line.clear(); + match stdin.read_until(b'\n', &mut line) { + Ok(0) | Err(_) => return, + Ok(_) => {} + } + // Notifications, replies and unreadable lines need no answer. + let Ok(Request { + id: Some(id), + method: Some(method), + params, + }) = serde_json::from_slice(line.trim_ascii()) + else { + continue; + }; + let reply = offline_reply(&id, &method, ¶ms); + if writeln!(stdout, "{reply}") + .and_then(|()| stdout.flush()) + .is_err() + { + return; + } + } +} + +pub fn offline_reply(id: &RequestId, method: &str, params: &Value) -> String { + let result = match method { + "initialize" => json!({ + "protocolVersion": params.get("protocolVersion").and_then(Value::as_str).unwrap_or(DEFAULT_PROTOCOL_VERSION), + "capabilities": { "tools": {} }, + "serverInfo": { "name": "t3-code", "version": "offline" }, + "instructions": "T3 Code did not start this session, so T3 Code's tools are unavailable.", + }), + "tools/list" => json!({ "tools": [] }), + "ping" => json!({}), + _ => { + return json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32601, "message": "T3 Code did not start this session." } }) + .to_string(); + } + }; + json!({ "jsonrpc": "2.0", "id": id, "result": result }).to_string() +} diff --git a/native/t3-mcp-plugin/src/child.rs b/native/t3-mcp-plugin/src/child.rs new file mode 100644 index 0000000000000..becfa049c3d73 --- /dev/null +++ b/native/t3-mcp-plugin/src/child.rs @@ -0,0 +1,223 @@ +//! Runs the real CLI between T3 Code and it, and serves the plugin's bridges meanwhile. + +use crate::credentials::Credentials; +use crate::ipc; +use crate::lock; +use crate::proxy::{self, Identity}; +use std::borrow::Cow; +use std::ffi::OsString; +use std::io::{self, BufRead}; +use std::path::{Path, PathBuf}; +use std::process::{ExitStatus, Stdio}; +use std::sync::{Arc, Mutex}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; + +/// T3 Code's opt-in Unix socket (or pipe) for `/mcp`; when set, sessions use it instead of TCP. +const T3CODE_MCP_SOCKET_ENV: &str = "T3CODE_MCP_SOCKET"; + +/// `real_env`, else the first `name` on PATH that is not this program. +pub fn real_binary(name: &str, real_env: &str) -> crate::Result { + if let Some(path) = std::env::var_os(real_env).filter(|path| !path.is_empty()) { + return Ok(PathBuf::from(path)); + } + let own = std::env::current_exe().and_then(std::fs::canonicalize).ok(); + let names: Vec = if cfg!(windows) { + vec![format!("{name}.exe"), format!("{name}.cmd")] + } else { + vec![name.to_owned()] + }; + std::env::var_os("PATH") + .iter() + .flat_map(std::env::split_paths) + .flat_map(|dir| names.iter().map(move |name| dir.join(name))) + .find(|candidate| { + candidate.is_file() + && std::fs::canonicalize(candidate) + .ok() + .is_some_and(|path| Some(path) != own) + }) + .ok_or_else(|| format!("no {name} on PATH; set {real_env}").into()) +} + +pub fn t3_socket() -> Option { + std::env::var_os(T3CODE_MCP_SOCKET_ENV) + .filter(|path| !path.is_empty()) + .map(PathBuf::from) +} + +#[cfg(unix)] +pub fn passthrough(real: &Path, args: &[OsString]) -> crate::Result { + use std::os::unix::process::CommandExt; + let error = std::process::Command::new(real).args(args).exec(); + Err(format!("cannot run {}: {error}", real.display()).into()) +} + +#[cfg(windows)] +pub fn passthrough(real: &Path, args: &[OsString]) -> crate::Result { + let status = std::process::Command::new(real).args(args).status(); + let status = status.map_err(|error| format!("cannot run {}: {error}", real.display()))?; + Ok(exit_code(status)) +} + +/// Starts the real CLI with the bridges' endpoint in its environment. The plugin runs +/// ` mcp-bridge`, so this program's directory goes first on PATH. +pub fn spawn(real: &Path, args: &[OsString]) -> io::Result<(Child, ipc::Listener)> { + let listener = ipc::Listener::bind()?; + let own_dir = std::env::current_exe() + .ok() + .and_then(|exe| exe.parent().map(Path::to_path_buf)); + let path = std::env::var_os("PATH").unwrap_or_default(); + let path = std::env::join_paths(own_dir.into_iter().chain(std::env::split_paths(&path))) + .unwrap_or(path); + let child = Command::new(real) + .args(args) + .env(ipc::ENV, &listener.name) + .env("PATH", path) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .kill_on_drop(true) + .spawn()?; + Ok((child, listener)) +} + +/// The CLI's stdin, written in order from anywhere. +pub struct Stdin(Mutex>>>); + +impl Stdin { + pub fn start(mut stdin: ChildStdin) -> Self { + let (sender, mut lines) = mpsc::unbounded_channel::>(); + tokio::spawn(async move { + while let Some(line) = lines.recv().await { + if stdin.write_all(&line).await.is_err() || stdin.flush().await.is_err() { + break; + } + } + }); + Self(Mutex::new(Some(sender))) + } + + pub fn send(&self, line: Vec) { + if let Some(sender) = lock(&self.0).as_ref() { + let _ = sender.send(line); + } + } + + /// Closing T3 Code's side closes the CLI's stdin, which shuts it down. + pub fn close(&self) { + lock(&self.0).take(); + } +} + +/// Reads T3 Code's lines on a plain thread, not tokio's stdin: that reads on a blocking thread +/// the runtime cannot cancel, so it could never shut down while T3 Code keeps stdin open. +pub fn read_from_t3( + stdin: Arc, + mut rewrite: impl FnMut(Vec) -> Vec + Send + 'static, +) { + std::thread::spawn(move || { + let mut input = io::stdin().lock(); + let mut line = Vec::new(); + loop { + line.clear(); + match input.read_until(b'\n', &mut line) { + Ok(0) | Err(_) => break, + Ok(_) => stdin.send(rewrite(line.clone())), + } + } + stdin.close(); + }); +} + +/// `filter` returns `None` for a line that must not reach T3 Code. +pub fn relay_to_t3( + stdout: ChildStdout, + mut filter: impl FnMut(&[u8]) -> Option> + Send + 'static, +) -> JoinHandle<()> { + tokio::spawn(async move { + let mut lines = BufReader::new(stdout); + let mut out = tokio::io::stdout(); + let mut line = Vec::new(); + loop { + line.clear(); + match lines.read_until(b'\n', &mut line).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + let Some(line) = filter(&line) else { + continue; + }; + if out.write_all(&line).await.is_err() || out.flush().await.is_err() { + break; + } + } + }) +} + +/// Serves bridges until the CLI exits, and returns its exit code. +pub async fn serve( + mut child: Child, + mut listener: ipc::Listener, + credentials: Arc, + identity: Identity, + to_t3: JoinHandle<()>, +) -> io::Result { + if let Some(pid) = child.id() { + tokio::spawn(forward_signals(pid)); + } + let bridges = async { + while let Ok(stream) = listener.accept().await { + tokio::spawn(proxy::serve(stream, Arc::clone(&credentials), identity)); + } + }; + let status = tokio::select! { + status = child.wait() => status?, + () = bridges => child.wait().await?, + }; + let _ = to_t3.await; + Ok(exit_code(status)) +} + +fn exit_code(status: ExitStatus) -> i32 { + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + if let Some(signal) = status.signal() { + return 128 + signal; + } + } + status.code().unwrap_or(1) +} + +/// Passes termination on to the CLI; the wrapper exits once it does. +#[cfg(unix)] +async fn forward_signals(child: u32) { + use rustix::process::{Pid, Signal, kill_process}; + use tokio::signal::unix::{SignalKind, signal}; + let Some(child) = i32::try_from(child).ok().and_then(Pid::from_raw) else { + return; + }; + let (Ok(mut term), Ok(mut int), Ok(mut hup)) = ( + signal(SignalKind::terminate()), + signal(SignalKind::interrupt()), + signal(SignalKind::hangup()), + ) else { + return; + }; + loop { + let signal = tokio::select! { + _ = term.recv() => Signal::TERM, + _ = int.recv() => Signal::INT, + _ = hup.recv() => Signal::HUP, + }; + // Our own child, kept unreaped by `kill_on_drop`, so the PID cannot be reused yet. + let _ = kill_process(child, signal); + } +} + +/// Console control events already reach the CLI, which shares the console. +#[cfg(windows)] +async fn forward_signals(_child: u32) {} diff --git a/native/t3-mcp-plugin/src/claude.rs b/native/t3-mcp-plugin/src/claude.rs new file mode 100644 index 0000000000000..d08495f406dc9 --- /dev/null +++ b/native/t3-mcp-plugin/src/claude.rs @@ -0,0 +1,181 @@ +use crate::child::{self, Stdin}; +use crate::credentials::{Credential, Credentials}; +use crate::proxy::Identity; +use serde_json::Value; +use std::borrow::Cow; +use std::ffi::OsString; +use std::path::Path; +use std::sync::Arc; + +const REAL_CLAUDE_ENV: &str = "T3_CLAUDE_REAL"; +const T3_SERVER: &str = "t3-code"; +// Claude names a plugin's server `plugin::` and its tools +// `mcp__plugin____`; T3 Code knows them as `t3-code` and `mcp__t3-code__*`. +const T3_TOOLS: &[u8] = b"mcp__t3-code__"; +const PLUGIN_TOOLS: &[u8] = b"mcp__plugin_t3-code_t3-code__"; +const T3_SERVER_NAME: &[u8] = b"\"t3-code\""; +const PLUGIN_SERVER_NAME: &[u8] = b"\"plugin:t3-code:t3-code\""; + +pub fn run(args: Vec) -> crate::Result { + let real = child::real_binary("claude", REAL_CLAUDE_ENV)?; + if !is_sdk_session(&args) { + return child::passthrough(&real, &args); + } + Ok(crate::runtime()?.block_on(session(&real, &args))?) +} + +/// The Agent SDK runs Claude with stream-json input; anything else (`--version`) is not a session. +fn is_sdk_session(args: &[OsString]) -> bool { + args.windows(2) + .any(|pair| pair[0] == "--input-format" && pair[1] == "stream-json") +} + +async fn session(real: &Path, args: &[OsString]) -> std::io::Result { + // T3 Code pre-approves its tools by name (`--allowedTools`), so the names follow the plugin. + let args: Vec = args.iter().map(to_plugin_names_arg).collect(); + let (mut process, listener) = child::spawn(real, &args)?; + let stdin = Arc::new(Stdin::start(process.stdin.take().expect("piped stdin"))); + let credentials = Arc::new(Credentials::without_parents()); + + let t3_socket = child::t3_socket(); + let from_t3 = Arc::clone(&credentials); + child::read_from_t3(stdin, move |line| { + let line = take_t3_server_from(&line, &from_t3, t3_socket.as_deref()).unwrap_or(line); + replace(&line, T3_TOOLS, PLUGIN_TOOLS).into_owned() + }); + let stdout = process.stdout.take().expect("piped stdout"); + let to_t3 = child::relay_to_t3(stdout, |line| Some(to_t3_names(line))); + child::serve(process, listener, credentials, Identity::Process, to_t3).await +} + +/// T3 Code sends its MCP server, with this thread's credential, in an `mcp_set_servers` control +/// request (the Agent SDK's `setMcpServers`), again whenever it rotates the credential. The admin +/// policy refuses servers added that way, so it moves into the plugin. `None` leaves the line. +fn take_t3_server_from( + line: &[u8], + credentials: &Credentials, + t3_socket: Option<&Path>, +) -> Option> { + if !contains(line, b"mcp_set_servers") { + return None; + } + let mut message: Value = serde_json::from_slice(line).ok()?; + let is_set_servers = + message["type"] == "control_request" && message["request"]["subtype"] == "mcp_set_servers"; + let servers = message + .pointer_mut("/request/servers") + .filter(|_| is_set_servers)? + .as_object_mut()?; + let server = servers.get(T3_SERVER)?; + let authorization = server["headers"].as_object().and_then(|headers| { + headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case("authorization")) + .and_then(|(_, value)| value.as_str()) + }); + let (Some(url), Some(authorization)) = (server["url"].as_str(), authorization) else { + // Passed on unchanged, so the policy's refusal reaches T3 Code as it would without this. + eprintln!( + "t3-claude: T3 Code sent its MCP server in a form t3-claude does not understand \ + (expected `url` and an `Authorization` header). Update t3-claude to match this \ + T3 Code version." + ); + return None; + }; + credentials.set_newest(Credential { + url: url.to_owned(), + authorization: authorization.to_owned(), + socket: t3_socket.map(Path::to_path_buf), + }); + servers.remove(T3_SERVER); + let mut rewritten = message.to_string().into_bytes(); + rewritten.push(b'\n'); + Some(rewritten) +} + +fn to_plugin_names_arg(arg: &OsString) -> OsString { + let Some(text) = arg.to_str() else { + return arg.clone(); + }; + match replace(text.as_bytes(), T3_TOOLS, PLUGIN_TOOLS) { + Cow::Borrowed(_) => arg.clone(), + Cow::Owned(bytes) => String::from_utf8(bytes).map_or_else(|_| arg.clone(), OsString::from), + } +} + +/// T3 Code recognizes its own tools and server by name in Claude's stream. +fn to_t3_names(line: &[u8]) -> Cow<'_, [u8]> { + match replace(line, PLUGIN_TOOLS, T3_TOOLS) { + Cow::Borrowed(line) => replace(line, PLUGIN_SERVER_NAME, T3_SERVER_NAME), + Cow::Owned(line) => { + Cow::Owned(replace(&line, PLUGIN_SERVER_NAME, T3_SERVER_NAME).into_owned()) + } + } +} + +fn contains(haystack: &[u8], needle: &[u8]) -> bool { + haystack + .windows(needle.len()) + .any(|window| window == needle) +} + +/// The names are plain ASCII, which JSON never escapes, so a byte replace is exact. +fn replace<'a>(line: &'a [u8], from: &[u8], to: &[u8]) -> Cow<'a, [u8]> { + if !contains(line, from) { + return Cow::Borrowed(line); + } + let mut out = Vec::with_capacity(line.len() + 32); + let mut rest = line; + while let Some(at) = rest.windows(from.len()).position(|window| window == from) { + out.extend_from_slice(&rest[..at]); + out.extend_from_slice(to); + rest = &rest[at + from.len()..]; + } + out.extend_from_slice(rest); + Cow::Owned(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn moves_t3_code_out_of_set_servers() { + let credentials = Credentials::without_parents(); + let mut servers: Value = + serde_json::from_str(include_str!("../fixtures/claude-mcp-servers.json")).unwrap(); + servers["other"] = json!({ "type": "stdio", "command": "x" }); + let set = json!({ "type": "control_request", "request_id": "r1", + "request": { "subtype": "mcp_set_servers", "servers": servers } }); + let rewritten = take_t3_server_from(set.to_string().as_bytes(), &credentials, None); + let rewritten: Value = serde_json::from_slice(&rewritten.unwrap()).unwrap(); + assert_eq!( + rewritten["request"]["servers"], + json!({ "other": { "type": "stdio", "command": "x" } }) + ); + assert_eq!( + credentials.newest(), + Some(Credential { + url: "http://127.0.0.1:3773/mcp".into(), + authorization: "Bearer token".into(), + socket: None, + }) + ); + let prompt = br#"{"type":"user","message":"mcp_set_servers"}"#; + assert_eq!(take_t3_server_from(prompt, &credentials, None), None); + } + + #[test] + fn maps_tool_and_server_names_both_ways() { + let from_claude = br#"{"tool":"mcp__plugin_t3-code_t3-code__t3_thread_wait","server":"plugin:t3-code:t3-code","other":"mcp__x__y"}"#; + assert_eq!( + to_t3_names(from_claude).as_ref(), + br#"{"tool":"mcp__t3-code__t3_thread_wait","server":"t3-code","other":"mcp__x__y"}"# + ); + assert_eq!( + to_plugin_names_arg(&OsString::from("Read,mcp__t3-code__*")), + OsString::from("Read,mcp__plugin_t3-code_t3-code__*") + ); + } +} diff --git a/native/t3-mcp-plugin/src/codex.rs b/native/t3-mcp-plugin/src/codex.rs new file mode 100644 index 0000000000000..df423a1557f9f --- /dev/null +++ b/native/t3-mcp-plugin/src/codex.rs @@ -0,0 +1,253 @@ +use crate::child::{self, Stdin}; +use crate::credentials::{Binding, Credential, Credentials, ParentLookup, RequestId, ThreadId}; +use crate::lock; +use crate::proxy::Identity; +use serde::Deserialize; +use serde::de::IgnoredAny; +use serde_json::value::RawValue; +use serde_json::{Map, Value, json}; +use std::borrow::Cow; +use std::collections::HashMap; +use std::ffi::OsString; +use std::path::Path; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::sync::oneshot; + +const REAL_CODEX_ENV: &str = "T3_CODEX_REAL"; +const T3_SERVERS: [&str; 2] = ["t3-code", "t3-code-direct"]; +const THREAD_METHODS: [&str; 3] = ["thread/start", "thread/resume", "thread/fork"]; +/// Keeps the wrapper's endpoint out of agent shell commands; bridges get it through `env_vars`. +const SHELL_FILTER_KEY: &str = "shell_environment_policy.filters.T3_MCP_*"; +const READ_TIMEOUT: Duration = Duration::from_secs(10); + +pub fn run(args: Vec) -> crate::Result { + let real = child::real_binary("codex", REAL_CODEX_ENV)?; + if args.first().is_none_or(|arg| arg != "app-server") { + return child::passthrough(&real, &args); + } + Ok(crate::runtime()?.block_on(app_server(&real, &args))?) +} + +async fn app_server(real: &Path, args: &[OsString]) -> std::io::Result { + let (mut process, listener) = child::spawn(real, args)?; + let codex = Arc::new(Codex { + stdin: Arc::new(Stdin::start(process.stdin.take().expect("piped stdin"))), + replies: Replies::default(), + next_id: AtomicU64::new(1), + }); + let lookup = Arc::clone(&codex); + let parent_of: ParentLookup = Box::new(move |thread| { + let codex = Arc::clone(&lookup); + Box::pin(async move { codex.parent_of(thread).await }) + }); + let credentials = Arc::new(Credentials::new(parent_of)); + + let t3_socket = child::t3_socket(); + let from_t3 = Arc::clone(&credentials); + child::read_from_t3(Arc::clone(&codex.stdin), move |line| { + take_t3_servers_from(&line, &from_t3, t3_socket.as_deref()).unwrap_or(line) + }); + let (to_t3, replies) = (Arc::clone(&credentials), Arc::clone(&codex)); + let stdout = process.stdout.take().expect("piped stdout"); + let to_t3 = child::relay_to_t3(stdout, move |line| { + (!note_reply(line, &replies.replies, &to_t3)).then_some(Cow::Borrowed(line)) + }); + child::serve(process, listener, credentials, Identity::Thread, to_t3).await +} + +struct Codex { + stdin: Arc, + replies: Replies, + next_id: AtomicU64, +} + +/// Replies the wrapper's own requests are waiting for. +#[derive(Default)] +struct Replies(Mutex>>); + +impl Replies { + fn is_waiting(&self) -> bool { + !lock(&self.0).is_empty() + } + + /// False if the reply is T3 Code's. + fn deliver(&self, id: &RequestId, line: &[u8]) -> bool { + let Some(waiter) = lock(&self.0).remove(id) else { + return false; + }; + if let Ok(message) = serde_json::from_slice(line) { + let _ = waiter.send(message); + } + true + } +} + +impl Codex { + async fn request(&self, method: &str, params: Value) -> Result { + let id = RequestId::String(format!( + "t3-codex:{}", + self.next_id.fetch_add(1, Ordering::SeqCst) + )); + let (waiter, reply) = oneshot::channel(); + lock(&self.replies.0).insert(id.clone(), waiter); + let mut line = json!({ "id": id, "method": method, "params": params }) + .to_string() + .into_bytes(); + line.push(b'\n'); + self.stdin.send(line); + let reply = tokio::time::timeout(READ_TIMEOUT, reply).await; + lock(&self.replies.0).remove(&id); + let reply = reply + .map_err(|_| format!("Codex did not answer {method}."))? + .map_err(|_| format!("Codex closed before answering {method}."))?; + match reply.get("error") { + Some(error) => Err(format!("Codex refused {method}: {error}")), + None => Ok(reply.get("result").cloned().unwrap_or(Value::Null)), + } + } + + async fn parent_of(&self, thread: ThreadId) -> Result, String> { + let read = self + .request("thread/read", json!({ "threadId": thread.0 })) + .await?; + Ok(read + .pointer("/thread/source/subAgent/thread_spawn/parent_thread_id") + .and_then(Value::as_str) + .map(|parent| ThreadId(parent.to_owned()))) + } +} + +/// T3 Code puts its MCP servers, with each thread's own credential, in every thread request's +/// config. The admin policy refuses MCP servers from config, so they move into the plugin. +/// `None` leaves the line as it is. +fn take_t3_servers_from( + line: &[u8], + credentials: &Credentials, + t3_socket: Option<&Path>, +) -> Option> { + #[derive(Deserialize)] + struct Head<'a> { + #[serde(borrow)] + method: Option>, + id: Option, + } + let head: Head = serde_json::from_slice(line).ok()?; + let is_thread_request = head + .method + .as_deref() + .is_some_and(|method| THREAD_METHODS.contains(&method)); + let id = head.id.filter(|_| is_thread_request)?; + let mut message: Value = serde_json::from_slice(line).ok()?; + let config = message.pointer_mut("/params/config")?.as_object_mut()?; + let binding = take_t3_servers(config, t3_socket)?; + config.insert(SHELL_FILTER_KEY.to_owned(), json!("exclude")); + if let Err(error) = &binding { + eprintln!("t3-codex: {error}"); + } + credentials.begin(id, binding); + let mut rewritten = message.to_string().into_bytes(); + rewritten.push(b'\n'); + Some(rewritten) +} + +/// Removes T3 Code's servers (`mcp_servers.t3-code`, ...). `None` if there were none. +fn take_t3_servers(config: &mut Map, t3_socket: Option<&Path>) -> Option { + let taken: Vec = T3_SERVERS + .iter() + .filter_map(|name| config.remove(&format!("mcp_servers.{name}"))) + .collect(); + if taken.is_empty() { + return None; + } + let credential = taken.iter().find_map(|server| { + let headers = server.get("http_headers")?.as_object()?; + let (_, authorization) = headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case("authorization"))?; + Some(Credential { + url: server.get("url")?.as_str()?.to_owned(), + authorization: authorization.as_str()?.to_owned(), + socket: t3_socket.map(Path::to_path_buf), + }) + }); + Some(credential.ok_or_else(|| { + "T3 Code sent its MCP servers in a form t3-codex does not understand (expected `url` and an \ + `Authorization` header in `http_headers`). Update t3-codex to match this T3 Code version." + .to_owned() + })) +} + +/// Parses only while something is waiting; the app-server's stream otherwise passes unread. +/// True for a reply to the wrapper's own request, which must not reach T3 Code. +fn note_reply(line: &[u8], replies: &Replies, credentials: &Credentials) -> bool { + if !credentials.is_waiting() && !replies.is_waiting() { + return false; + } + #[derive(Deserialize)] + struct Reply<'a> { + id: Option, + method: Option, + #[serde(borrow)] + result: Option<&'a RawValue>, + } + #[derive(Deserialize)] + struct ThreadResult { + thread: ThreadFields, + } + #[derive(Deserialize)] + struct ThreadFields { + id: String, + } + let Ok(Reply { + id: Some(id), + method: None, + result, + }) = serde_json::from_slice::(line) + else { + return false; + }; + if replies.deliver(&id, line) { + return true; + } + let thread = result + .and_then(|result| serde_json::from_str::(result.get()).ok()) + .map(|result| ThreadId(result.thread.id)); + credentials.finish(&id, thread); + false +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn moves_t3_servers_out_of_thread_requests() { + let credentials = Credentials::without_parents(); + let mut config: Map = + serde_json::from_str(include_str!("../fixtures/codex-mcp-servers.json")).unwrap(); + config.insert("model".into(), json!("m")); + let start = json!({ "id": 7, "method": "thread/start", "params": { "config": config } }); + let rewritten = take_t3_servers_from(start.to_string().as_bytes(), &credentials, None); + let rewritten: Value = serde_json::from_slice(&rewritten.unwrap()).unwrap(); + assert_eq!( + rewritten["params"]["config"], + json!({ "model": "m", SHELL_FILTER_KEY: "exclude" }) + ); + + assert!(!note_reply( + br#"{"id":7,"result":{"thread":{"id":"th"}}}"#, + &Replies::default(), + &credentials + )); + assert_eq!( + credentials.for_thread(&ThreadId("th".into())).await, + Ok(Credential { + url: "http://127.0.0.1:3773/mcp".into(), + authorization: "Bearer token".into(), + socket: None, + }) + ); + } +} diff --git a/native/t3-mcp-plugin/src/credentials.rs b/native/t3-mcp-plugin/src/credentials.rs new file mode 100644 index 0000000000000..ac9cb5c2c1312 --- /dev/null +++ b/native/t3-mcp-plugin/src/credentials.rs @@ -0,0 +1,238 @@ +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::future::Future; +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::Mutex; +use std::time::Duration; +use tokio::sync::watch; + +/// Subagents rarely nest; this only stops a cycle from looping forever. +const MAX_PARENT_DEPTH: usize = 8; + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct Credential { + pub url: String, + pub authorization: String, + /// Reach `url` through this socket instead of TCP (T3 Code's `T3CODE_MCP_SOCKET`). + pub socket: Option, +} + +/// A thread's credential, or why T3 Code's config for it could not be read. +pub type Binding = Result; + +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(untagged)] +pub enum RequestId { + Number(i64), + String(String), +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct ThreadId(pub String); + +pub type ParentLookup = Box< + dyn Fn(ThreadId) -> Pin, String>> + Send>> + + Send + + Sync, +>; + +#[derive(Default)] +struct State { + threads: HashMap, + /// Subagent -> the T3 Code thread it acts as. Its credential is read from there, so it follows + /// the parent's rotation. + parents: HashMap, + /// Thread requests awaiting Codex's reply, which names the thread. + pending: HashMap, + /// Credentials for opening a session before any thread is known, newest last. + recent: Vec, +} + +pub struct Credentials { + state: Mutex, + parent_of: ParentLookup, + /// Bumped whenever a credential arrives, for callers waiting on the first one. + arrived: watch::Sender, +} + +impl Credentials { + pub fn new(parent_of: ParentLookup) -> Self { + Self { + state: Mutex::default(), + parent_of, + arrived: watch::Sender::new(0), + } + } + + /// For a CLI whose subagents share its process, so no thread lineage is needed. + pub fn without_parents() -> Self { + Self::new(Box::new(|_| Box::pin(async { Ok(None) }))) + } + + fn state(&self) -> std::sync::MutexGuard<'_, State> { + crate::lock(&self.state) + } + + /// Called before the thread request reaches Codex, so a bridge Codex starts for it finds a + /// credential. + pub fn begin(&self, request: RequestId, binding: Binding) { + if let Ok(credential) = &binding { + self.set_newest(credential.clone()); + } + self.state().pending.insert(request, binding); + } + + /// A CLI process acts as one thread; T3 Code replaces its credential when it rotates it. + pub fn set_newest(&self, credential: Credential) { + let mut state = self.state(); + state.recent.retain(|known| known != &credential); + state.recent.push(credential); + drop(state); + self.arrived.send_modify(|count| *count += 1); + } + + /// Records Codex's reply to a thread request. `None` means the request failed; T3 Code may + /// revoke its credential, so it is no longer offered. + pub fn finish(&self, request: &RequestId, thread: Option) { + let mut state = self.state(); + let Some(binding) = state.pending.remove(request) else { + return; + }; + match thread { + Some(thread) => { + state.threads.insert(thread, binding); + } + None => { + if let Ok(credential) = &binding { + state.recent.retain(|known| known != credential); + } + } + } + } + + pub fn is_waiting(&self) -> bool { + !self.state().pending.is_empty() + } + + /// Codex: for initialize and discovery, before any thread is known. Claude: the process's own. + /// `None`: T3 Code gave none, so it did not start this session. + pub fn newest(&self) -> Option { + self.state().recent.last().cloned() + } + + /// Claude starts plugin servers before T3 Code sends the credential, so they wait for it. + pub async fn wait_for_newest(&self, timeout: Duration) -> Option { + let mut arrived = self.arrived.subscribe(); + let wait = async { + loop { + if let Some(credential) = self.newest() { + return credential; + } + if arrived.changed().await.is_err() { + std::future::pending::<()>().await; + } + } + }; + tokio::time::timeout(timeout, wait).await.ok() + } + + /// The credential a Codex thread acts with. A subagent T3 Code never saw acts as the T3 Code + /// thread it descends from. + pub async fn for_thread(&self, thread: &ThreadId) -> Binding { + let mut current = self + .state() + .parents + .get(thread) + .cloned() + .unwrap_or_else(|| thread.clone()); + for _ in 0..MAX_PARENT_DEPTH { + let found = self.state().threads.get(¤t).cloned(); + if let Some(binding) = found { + if ¤t != thread { + self.state().parents.insert(thread.clone(), current); + } + return binding; + } + match (self.parent_of)(current.clone()).await? { + Some(parent) => current = parent, + None => break, + } + } + Err(format!( + "Codex thread {} has no T3 Code credential.", + thread.0 + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn credential(token: &str) -> Credential { + Credential { + url: "http://h/mcp".into(), + authorization: format!("Bearer {token}"), + socket: None, + } + } + + fn thread(id: &str) -> ThreadId { + ThreadId(id.into()) + } + + fn lineage() -> Credentials { + Credentials::new(Box::new(|id: ThreadId| { + let parent = match id.0.as_str() { + "grandchild" => Some(thread("child")), + "child" => Some(thread("parent")), + _ => None, + }; + Box::pin(async move { Ok(parent) }) + })) + } + + #[tokio::test] + async fn subagents_follow_their_t3_thread_through_rotation() { + let credentials = lineage(); + credentials.begin(RequestId::Number(1), Ok(credential("p1"))); + credentials.finish(&RequestId::Number(1), Some(thread("parent"))); + assert_eq!( + credentials.for_thread(&thread("grandchild")).await, + Ok(credential("p1")) + ); + + credentials.begin(RequestId::Number(2), Ok(credential("p2"))); + credentials.finish(&RequestId::Number(2), Some(thread("parent"))); + assert_eq!( + credentials.for_thread(&thread("grandchild")).await, + Ok(credential("p2")) + ); + assert!(credentials.for_thread(&thread("stranger")).await.is_err()); + } + + #[tokio::test] + async fn failed_thread_requests_leave_bootstrap() { + let credentials = lineage(); + credentials.begin(RequestId::Number(1), Ok(credential("a"))); + credentials.begin(RequestId::String("x".into()), Ok(credential("b"))); + assert_eq!(credentials.newest(), Some(credential("b"))); + credentials.finish(&RequestId::Number(1), Some(thread("a"))); + credentials.finish(&RequestId::String("x".into()), None); + assert_eq!(credentials.newest(), Some(credential("a"))); + assert!(!credentials.is_waiting()); + } + + #[tokio::test] + async fn waits_for_a_credential_that_arrives_late() { + let credentials = std::sync::Arc::new(Credentials::without_parents()); + let late = std::sync::Arc::clone(&credentials); + tokio::spawn(async move { + tokio::task::yield_now().await; + late.set_newest(credential("late")); + }); + let waited = credentials.wait_for_newest(Duration::from_secs(5)).await; + assert_eq!(waited, Some(credential("late"))); + } +} diff --git a/native/t3-mcp-plugin/src/http.rs b/native/t3-mcp-plugin/src/http.rs new file mode 100644 index 0000000000000..a247513a50475 --- /dev/null +++ b/native/t3-mcp-plugin/src/http.rs @@ -0,0 +1,168 @@ +use crate::ipc::{self, Stream}; +use bytes::Bytes; +use http_body_util::{BodyExt, Full}; +use hyper::body::Incoming; +use hyper::{Method, Request, Response, Uri}; +use hyper_util::rt::TokioIo; +use std::io; +use std::path::PathBuf; + +#[derive(Clone, Debug)] +pub struct Endpoint { + authority: String, + /// `authority` with the default port filled in, for connecting over TCP. + address: String, + path: String, + socket: Option, +} + +impl Endpoint { + pub fn new(url: &str, socket: Option) -> Result { + let invalid = || format!("T3 Code gave an MCP URL the plugin cannot use: {url}"); + let uri: Uri = url.parse().map_err(|_| invalid())?; + let authority = uri + .authority() + .filter(|_| uri.scheme_str() == Some("http")) + .ok_or_else(invalid)?; + Ok(Self { + address: format!( + "{}:{}", + authority.host(), + authority.port_u16().unwrap_or(80) + ), + authority: authority.to_string(), + path: uri + .path_and_query() + .map_or_else(|| "/".to_owned(), ToString::to_string), + socket, + }) + } +} + +pub async fn send( + endpoint: &Endpoint, + method: Method, + headers: &[(&str, &str)], + body: Vec, +) -> Result, String> { + let request = headers + .iter() + .fold( + Request::builder() + .method(method) + .uri(&endpoint.path) + .header(hyper::header::HOST, &endpoint.authority), + |request, (name, value)| request.header(*name, *value), + ) + .body(Full::new(Bytes::from(body))) + .map_err(|error| format!("T3 Code MCP request is not valid: {error}"))?; + match &endpoint.socket { + Some(socket) => exchange(ipc::connect(socket.as_os_str()).await, request).await, + None => { + exchange( + tokio::net::TcpStream::connect(&endpoint.address).await, + request, + ) + .await + } + } +} + +async fn exchange( + connection: io::Result, + request: Request>, +) -> Result, String> { + let connection = connection.map_err(|error| format!("cannot reach T3 Code: {error}"))?; + let (mut sender, driver) = hyper::client::conn::http1::handshake(TokioIo::new(connection)) + .await + .map_err(|error| format!("cannot reach T3 Code: {error}"))?; + tokio::spawn(driver); + sender + .send_request(request) + .await + .map_err(|error| format!("T3 Code MCP request failed: {error}")) +} + +/// `None` at the end of the body. A body cut short is an error. +pub async fn next_chunk(body: &mut Incoming) -> Result, String> { + while let Some(frame) = body.frame().await { + let frame = frame.map_err(|error| format!("T3 Code MCP response failed: {error}"))?; + if let Ok(data) = frame.into_data() { + return Ok(Some(data)); + } + } + Ok(None) +} + +#[derive(Default)] +pub struct EventStream { + line: Vec, + data: Option>, + /// The last byte ended a line with CR, so an LF right after it (even in the next chunk) belongs + /// to the same line ending. + after_cr: bool, +} + +impl EventStream { + /// An event cut off by the end of the stream is never returned. + pub fn push(&mut self, chunk: &[u8]) -> Vec> { + let mut events = Vec::new(); + for &byte in chunk { + let continues_crlf = self.after_cr && byte == b'\n'; + self.after_cr = byte == b'\r'; + match byte { + _ if continues_crlf => {} + b'\r' | b'\n' => self.end_line(&mut events), + _ => self.line.push(byte), + } + } + events + } + + fn end_line(&mut self, events: &mut Vec>) { + let line = std::mem::take(&mut self.line); + if line.is_empty() { + events.extend(self.data.take()); + return; + } + let (field, value) = match line.iter().position(|&byte| byte == b':') { + Some(colon) => { + let value = &line[colon + 1..]; + (&line[..colon], value.strip_prefix(b" ").unwrap_or(value)) + } + None => (&line[..], &[][..]), + }; + if field == b"data" { + match &mut self.data { + Some(data) => { + data.push(b'\n'); + data.extend_from_slice(value); + } + None => self.data = Some(value.to_vec()), + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decodes_events_as_soon_as_they_end() { + let mut stream = EventStream::default(); + let mut push = |chunk: &str| -> Vec { + let events = stream.push(chunk.as_bytes()); + events + .into_iter() + .map(|data| String::from_utf8(data).unwrap()) + .collect() + }; + assert!(push("event: message\r").is_empty()); + assert!(push("\ndata: {\"a\":").is_empty()); + assert_eq!(push("1}\r\n\r\n: comment\n"), [r#"{"a":1}"#]); + // A lone CR ends a line at once: a stream held open must not wait for another byte. + assert_eq!(push("data: x\rdata: y\r\r"), ["x\ny"]); + assert!(push("data: cut off").is_empty()); + } +} diff --git a/native/t3-mcp-plugin/src/ipc.rs b/native/t3-mcp-plugin/src/ipc.rs new file mode 100644 index 0000000000000..c844df703e91b --- /dev/null +++ b/native/t3-mcp-plugin/src/ipc.rs @@ -0,0 +1,147 @@ +use std::collections::hash_map::RandomState; +use std::ffi::{OsStr, OsString}; +use std::hash::{BuildHasher, Hasher}; +use std::io; +use tokio::io::{AsyncRead, AsyncWrite}; + +/// Names the wrapper's endpoint for the bridges Codex starts. Plugin `env_vars` pass it through. +pub const ENV: &str = "T3_MCP_BRIDGE"; + +pub trait Stream: AsyncRead + AsyncWrite + Unpin + Send + 'static {} +impl Stream for T {} + +/// Unguessable without extra dependencies: `RandomState` is seeded from the OS. +fn random_suffix() -> String { + format!("{:016x}", RandomState::new().build_hasher().finish()) +} + +#[cfg(unix)] +pub use unix::{Listener, connect}; +#[cfg(windows)] +pub use windows::{Listener, connect}; + +#[cfg(unix)] +mod unix { + use super::*; + use std::fs::{self, DirBuilder}; + use std::os::unix::fs::DirBuilderExt; + use std::path::{Path, PathBuf}; + use tokio::net::{UnixListener, UnixStream}; + + pub struct Listener { + listener: UnixListener, + dir: PathBuf, + pub name: OsString, + } + + impl Listener { + /// Binds in a fresh owner-only directory, so no other user can connect. + pub fn bind() -> io::Result { + let base = std::env::var_os("XDG_RUNTIME_DIR") + .filter(|dir| !dir.is_empty()) + .map_or_else(std::env::temp_dir, PathBuf::from); + remove_stale_dirs(&base); + let dir = base.join(format!( + "t3-codex-{}-{}", + std::process::id(), + random_suffix() + )); + DirBuilder::new().mode(0o700).create(&dir)?; + let path = dir.join("bridge.sock"); + let listener = UnixListener::bind(&path).inspect_err(|_| { + let _ = fs::remove_dir_all(&dir); + })?; + Ok(Self { + listener, + dir, + name: path.into_os_string(), + }) + } + + pub async fn accept(&mut self) -> io::Result { + Ok(self.listener.accept().await?.0) + } + } + + impl Drop for Listener { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.dir); + } + } + + pub async fn connect(name: &OsStr) -> io::Result { + UnixStream::connect(name).await + } + + /// Removes directories whose wrapper is gone, as after SIGKILL. + fn remove_stale_dirs(base: &Path) { + let Ok(entries) = fs::read_dir(base) else { + return; + }; + for entry in entries.flatten() { + let name = entry.file_name(); + let Some(pid) = name + .to_str() + .and_then(|name| name.strip_prefix("t3-codex-")) + .and_then(|rest| rest.split('-').next()) + .and_then(|pid| pid.parse().ok()) + .and_then(rustix::process::Pid::from_raw) + else { + continue; + }; + let gone = rustix::process::test_kill_process(pid) == Err(rustix::io::Errno::SRCH); + if gone { + let _ = fs::remove_dir_all(entry.path()); + } + } + } +} + +#[cfg(windows)] +mod windows { + use super::*; + use std::time::Duration; + use tokio::net::windows::named_pipe::{ClientOptions, NamedPipeServer, ServerOptions}; + + const ERROR_PIPE_BUSY: i32 = 231; + + /// Named pipes need no cleanup: they vanish with their last handle. + pub struct Listener { + next: NamedPipeServer, + pub name: OsString, + } + + impl Listener { + pub fn bind() -> io::Result { + let name = OsString::from(format!( + r"\\.\pipe\t3-codex-{}-{}", + std::process::id(), + random_suffix() + )); + let next = ServerOptions::new() + .first_pipe_instance(true) + .reject_remote_clients(true) + .create(&name)?; + Ok(Self { next, name }) + } + + pub async fn accept(&mut self) -> io::Result { + self.next.connect().await?; + let following = ServerOptions::new() + .reject_remote_clients(true) + .create(&self.name)?; + Ok(std::mem::replace(&mut self.next, following)) + } + } + + pub async fn connect(name: &OsStr) -> io::Result { + loop { + match ClientOptions::new().open(name) { + Err(error) if error.raw_os_error() == Some(ERROR_PIPE_BUSY) => { + tokio::time::sleep(Duration::from_millis(20)).await; + } + result => return result, + } + } + } +} diff --git a/native/t3-mcp-plugin/src/lib.rs b/native/t3-mcp-plugin/src/lib.rs new file mode 100644 index 0000000000000..b89fa781df16b --- /dev/null +++ b/native/t3-mcp-plugin/src/lib.rs @@ -0,0 +1,45 @@ +//! Lets T3 Code's MCP tools reach Codex and Claude Code through T3 Code plugins where an admin +//! policy allows MCP servers only from plugins. `t3-codex` and `t3-claude` stand in for each CLI; +//! ` mcp-bridge` is the plugin's server. + +mod bridge; +mod child; +pub mod claude; +pub mod codex; +mod credentials; +mod http; +mod ipc; +mod proxy; + +use std::ffi::OsString; +use std::sync::{Mutex, MutexGuard}; + +pub type Result = std::result::Result>; + +/// A panic elsewhere never leaves this crate's shared state half-updated, so it stays usable. +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn runtime() -> std::io::Result { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() +} + +pub fn main(wrapper: fn(Vec) -> Result) -> ! { + let args: Vec = std::env::args_os().skip(1).collect(); + let outcome = match args.first() { + Some(mode) if mode == "mcp-bridge" => bridge::run(), + _ => wrapper(args), + }; + match outcome { + Ok(code) => std::process::exit(code), + Err(error) => { + eprintln!("{error}"); + std::process::exit(1); + } + } +} diff --git a/native/t3-mcp-plugin/src/proxy.rs b/native/t3-mcp-plugin/src/proxy.rs new file mode 100644 index 0000000000000..8d1d590cead03 --- /dev/null +++ b/native/t3-mcp-plugin/src/proxy.rs @@ -0,0 +1,542 @@ +use crate::bridge::offline_reply; +use crate::credentials::{Credential, Credentials, RequestId, ThreadId}; +use crate::http::{self, Endpoint, EventStream}; +use crate::ipc::Stream; +use crate::lock; +use hyper::Method; +use serde::Deserialize; +use serde::de::IgnoredAny; +use serde_json::value::RawValue; +use serde_json::{Value, json}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::sync::{OnceCell, mpsc, watch}; +use tokio::task::JoinSet; + +/// Opening a session is quick. Tool calls get no limit: `t3_thread_wait` can take an hour. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30); +const END_SESSION_TIMEOUT: Duration = Duration::from_secs(2); +/// Covers T3 Code sending Claude its credential just after the plugin servers start. +const CREDENTIAL_WAIT: Duration = Duration::from_secs(15); +const INITIALIZE_ID: &str = "t3-mcp-plugin:initialize"; +const NO_REPLY: &str = "T3 Code MCP endpoint ended the request without a reply."; +const NO_THREAD: &str = "Codex sent a tool call without _meta.threadId, so t3-codex cannot tell which \ + T3 Code thread is calling. This Codex version is not supported."; + +/// Which T3 Code thread a tool call acts as. +#[derive(Clone, Copy, Debug)] +pub enum Identity { + /// Codex: one app-server hosts many threads, and `_meta.threadId` names the caller. + Thread, + /// Claude: one process per thread; subagents share it, so every call is that thread's. + Process, +} + +#[derive(Debug)] +enum Error { + /// T3 Code no longer accepts this credential or session (HTTP 401 or 404). Nothing was + /// relayed, so the message can go again on a fresh session. + Stale(u16), + /// T3 Code gave no credential: it did not start this session. + Offline, + Failed(String), +} + +impl Error { + fn message(&self) -> String { + match self { + Self::Stale(status) => format!("T3 Code MCP endpoint responded with HTTP {status}."), + Self::Offline => "T3 Code did not start this session.".to_owned(), + Self::Failed(message) => message.clone(), + } + } +} + +impl From for Error { + fn from(message: String) -> Self { + Self::Failed(message) + } +} + +struct Session { + credential: Credential, + endpoint: Endpoint, + id: Mutex>, + protocol_version: OnceLock, + /// T3 Code's initialize result, which Codex receives as this server's own. + init_result: OnceLock>, +} + +struct Connection { + credentials: Arc, + identity: Identity, + output: mpsc::UnboundedSender, + sessions: Mutex>>>>, + /// Serves everything not tied to a thread: discovery, pings, notifications. + primary: Mutex>>, + /// Codex's initialize params, replayed to open each session. + init_params: Mutex, + initialized: watch::Sender, + offline: AtomicBool, + /// Server request -> its session, so Codex's reply goes back there. + server_requests: Mutex>>, + /// Codex request -> its session, so a cancellation reaches the right one. + in_flight: Mutex>>, +} + +pub async fn serve(stream: impl Stream, credentials: Arc, identity: Identity) { + let (reader, mut writer) = tokio::io::split(stream); + let (output, mut outgoing) = mpsc::unbounded_channel::(); + tokio::spawn(async move { + // Every message is one line by construction (see `one_line`). + while let Some(mut line) = outgoing.recv().await { + line.push('\n'); + if writer.write_all(line.as_bytes()).await.is_err() { + break; + } + } + }); + let connection = Arc::new(Connection { + credentials, + identity, + output, + sessions: Mutex::default(), + primary: Mutex::default(), + init_params: Mutex::new(json!({})), + initialized: watch::Sender::new(false), + offline: AtomicBool::new(false), + server_requests: Mutex::default(), + in_flight: Mutex::default(), + }); + let mut handlers = JoinSet::new(); + let mut lines = BufReader::new(reader); + let mut line = Vec::new(); + loop { + line.clear(); + match lines.read_until(b'\n', &mut line).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + let message = line.trim_ascii(); + if !message.is_empty() { + handlers.spawn(Arc::clone(&connection).handle(message.to_vec())); + } + while handlers.try_join_next().is_some() {} + } + // The bridge is gone, so no one is left to read replies. + handlers.abort_all(); + connection.end_sessions().await; +} + +#[derive(Deserialize)] +struct Incoming { + id: Option, + method: Option, + #[serde(default)] + params: Value, +} + +impl Connection { + fn send(&self, line: String) { + let _ = self.output.send(line); + } + + fn send_error(&self, id: Option<&RequestId>, code: i64, message: &str) { + self.send( + json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } }) + .to_string(), + ); + } + + async fn handle(self: Arc, line: Vec) { + let Ok(message) = serde_json::from_slice::(&line) else { + return self.send_error(None, -32700, "Parse error"); + }; + let outcome = match (message.method.as_deref(), message.id.as_ref()) { + (Some("initialize"), Some(id)) => self.initialize(id, &message.params).await, + // Each session sends its own while opening. + (Some("notifications/initialized"), _) => Ok(()), + (Some(method), id) => { + let _ = self.initialized.subscribe().wait_for(|done| *done).await; + if self.offline.load(Ordering::SeqCst) { + if let Some(id) = id { + self.send(offline_reply(id, method, &message.params)); + } + Ok(()) + } else { + self.forward(&line, method, id, &message.params).await + } + } + (None, Some(id)) => self.forward_reply(&line, id).await, + (None, None) => Ok(()), + }; + if let (Err(error), Some(id), Some(_)) = (outcome, &message.id, &message.method) { + self.send_error(Some(id), -32603, &error.message()); + } + } + + async fn initialize(&self, id: &RequestId, params: &Value) -> Result<(), Error> { + *lock(&self.init_params) = params.clone(); + let opened = self.primary().await; + self.initialized.send_replace(true); + match opened { + Ok(session) => { + let result = session + .init_result + .get() + .map_or("{}", |result| result.get()); + let id = serde_json::to_string(id).unwrap_or_default(); + self.send(format!( + r#"{{"jsonrpc":"2.0","id":{id},"result":{result}}}"# + )); + Ok(()) + } + Err(Error::Offline) => { + self.offline.store(true, Ordering::SeqCst); + self.send(offline_reply(id, "initialize", params)); + Ok(()) + } + Err(error) => Err(error), + } + } + + async fn forward( + &self, + line: &[u8], + method: &str, + id: Option<&RequestId>, + params: &Value, + ) -> Result<(), Error> { + let thread = match (method, self.identity) { + // Without it, the call would run as whichever thread a session was opened for. + ("tools/call", Identity::Thread) => Some( + params + .pointer("/_meta/threadId") + .and_then(Value::as_str) + .filter(|thread| !thread.is_empty()) + .map(|thread| ThreadId(thread.to_owned())) + .ok_or_else(|| Error::Failed(NO_THREAD.to_owned()))?, + ), + _ => None, + }; + for attempt in 0..2 { + let session = match (&thread, method) { + (Some(thread), _) => { + self.session(self.credentials.for_thread(thread).await?) + .await? + } + // A rotated credential takes effect on the next call. + (None, "tools/call") => { + let credential = self.credentials.newest().ok_or(Error::Offline)?; + self.session(credential).await? + } + _ => match self.cancelled_request_session(method, params) { + Some(session) => session, + None => self.primary().await?, + }, + }; + if let Some(id) = id { + lock(&self.in_flight).insert(id.clone(), Arc::clone(&session)); + } + let outcome = self.exchange(&session, line.to_vec(), id, false).await; + if let Some(id) = id { + lock(&self.in_flight).remove(id); + } + match outcome { + Ok(_) => return Ok(()), + Err(Error::Stale(_)) if attempt == 0 => self.forget(&session), + Err(error) => return Err(error), + } + } + unreachable!("the loop returns on its second attempt") + } + + fn cancelled_request_session(&self, method: &str, params: &Value) -> Option> { + let request = params + .get("requestId") + .filter(|_| method == "notifications/cancelled")?; + let request = RequestId::deserialize(request).ok()?; + lock(&self.in_flight).get(&request).cloned() + } + + async fn forward_reply(&self, line: &[u8], id: &RequestId) -> Result<(), Error> { + let session = lock(&self.server_requests).remove(id); + let session = match session { + Some(session) => session, + None => self.primary().await?, + }; + self.exchange(&session, line.to_vec(), None, false) + .await + .map(|_| ()) + } + + /// The session opened with any current credential. One revoked since gets a fresh lookup. + async fn primary(&self) -> Result, Error> { + if let Some(session) = lock(&self.primary).clone() { + return Ok(session); + } + for attempt in 0..2 { + let credential = match self.identity { + Identity::Thread => self.credentials.newest(), + Identity::Process => self.credentials.wait_for_newest(CREDENTIAL_WAIT).await, + }; + let credential = credential.ok_or(Error::Offline)?; + match self.session(credential).await { + Ok(session) => { + *lock(&self.primary) = Some(Arc::clone(&session)); + return Ok(session); + } + Err(Error::Stale(_)) if attempt == 0 => {} + Err(error) => return Err(error), + } + } + unreachable!("the loop returns on its second attempt") + } + + /// Opens at most one session per credential; the map lock is never held while opening. + async fn session(&self, credential: Credential) -> Result, Error> { + let cell = Arc::clone(lock(&self.sessions).entry(credential.clone()).or_default()); + cell.get_or_try_init(|| self.open(credential)) + .await + .cloned() + } + + fn forget(&self, session: &Arc) { + let mut sessions = lock(&self.sessions); + let current = sessions + .get(&session.credential) + .and_then(|cell| cell.get()); + if current.is_some_and(|current| Arc::ptr_eq(current, session)) { + sessions.remove(&session.credential); + } + let mut primary = lock(&self.primary); + if primary + .as_ref() + .is_some_and(|current| Arc::ptr_eq(current, session)) + { + *primary = None; + } + } + + async fn open(&self, credential: Credential) -> Result, Error> { + let session = Arc::new(Session { + endpoint: Endpoint::new(&credential.url, credential.socket.clone())?, + credential, + id: Mutex::default(), + protocol_version: OnceLock::new(), + init_result: OnceLock::new(), + }); + let params = lock(&self.init_params).clone(); + let initialize = json!({ "jsonrpc": "2.0", "id": INITIALIZE_ID, "method": "initialize", "params": params }); + let initialized = json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }); + let handshake = async { + let init_id = RequestId::String(INITIALIZE_ID.to_owned()); + let result = self + .exchange( + &session, + initialize.to_string().into_bytes(), + Some(&init_id), + true, + ) + .await? + .ok_or_else(|| Error::Failed(NO_REPLY.to_owned()))?; + if let Some(version) = protocol_version(&result) { + let _ = session.protocol_version.set(version); + } + let _ = session.init_result.set(result); + self.exchange(&session, initialized.to_string().into_bytes(), None, false) + .await + }; + tokio::time::timeout(HANDSHAKE_TIMEOUT, handshake) + .await + .map_err(|_| { + Error::Failed("T3 Code did not finish opening an MCP session.".to_owned()) + })??; + Ok(session) + } + + /// With `capture`, the result for `expect` is returned instead of relayed to Codex. + async fn exchange( + &self, + session: &Arc, + body: Vec, + expect: Option<&RequestId>, + capture: bool, + ) -> Result>, Error> { + let session_id = lock(&session.id).clone(); + let mut headers = vec![ + ("content-type", "application/json"), + ("accept", "application/json, text/event-stream"), + ("authorization", session.credential.authorization.as_str()), + ]; + if let Some(session_id) = &session_id { + headers.push(("mcp-session-id", session_id)); + } + if let Some(version) = session.protocol_version.get() { + headers.push(("mcp-protocol-version", version)); + } + let mut response = http::send(&session.endpoint, Method::POST, &headers, body).await?; + if let Some(id) = response + .headers() + .get("mcp-session-id") + .and_then(|value| value.to_str().ok()) + { + *lock(&session.id) = Some(id.to_owned()); + } + let status = response.status().as_u16(); + if matches!(status, 401 | 404) { + return Err(Error::Stale(status)); + } + if !response.status().is_success() { + return Err(Error::Failed(format!( + "T3 Code MCP endpoint responded with HTTP {status}." + ))); + } + let mut relay = Relay { + connection: self, + session, + expect, + capture, + answered: false, + captured: None, + }; + if !matches!(status, 202 | 204) { + let is_event_stream = response + .headers() + .get(hyper::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.contains("text/event-stream")); + let body = response.body_mut(); + if is_event_stream { + let mut events = EventStream::default(); + while let Some(chunk) = http::next_chunk(body).await? { + for data in events.push(&chunk) { + relay.payload(&data)?; + } + } + } else { + let mut whole = Vec::new(); + while let Some(chunk) = http::next_chunk(body).await? { + whole.extend_from_slice(&chunk); + } + if !whole.trim_ascii().is_empty() { + relay.payload(&whole)?; + } + } + } + if expect.is_some() && !relay.answered { + return Err(Error::Failed(NO_REPLY.to_owned())); + } + match relay.captured { + Some(Err(error)) => Err(Error::Failed(format!( + "T3 Code refused the MCP session: {error}" + ))), + Some(Ok(result)) => Ok(Some(result)), + None => Ok(None), + } + } + + /// Best effort: T3 Code keeps a session until it is deleted or T3 Code restarts. + async fn end_sessions(&self) { + let sessions: Vec<_> = lock(&self.sessions) + .values() + .filter_map(|cell| cell.get().cloned()) + .collect(); + let mut ending = JoinSet::new(); + for session in sessions { + let Some(session_id) = lock(&session.id).clone() else { + continue; + }; + ending.spawn(async move { + let mut headers = vec![ + ("authorization", session.credential.authorization.as_str()), + ("mcp-session-id", session_id.as_str()), + ]; + if let Some(version) = session.protocol_version.get() { + headers.push(("mcp-protocol-version", version)); + } + let delete = http::send(&session.endpoint, Method::DELETE, &headers, Vec::new()); + let _ = tokio::time::timeout(END_SESSION_TIMEOUT, delete).await; + }); + } + ending.join_all().await; + } +} + +/// Everything else in a message from T3 Code passes through verbatim. +#[derive(Deserialize)] +struct ServerMessage<'a> { + id: Option, + method: Option, + #[serde(borrow)] + result: Option<&'a RawValue>, + #[serde(borrow)] + error: Option<&'a RawValue>, +} + +struct Relay<'a> { + connection: &'a Connection, + session: &'a Arc, + expect: Option<&'a RequestId>, + capture: bool, + answered: bool, + captured: Option, String>>, +} + +impl Relay<'_> { + fn payload(&mut self, payload: &[u8]) -> Result<(), Error> { + let unreadable = |_| Error::Failed("T3 Code sent an unreadable MCP message.".to_owned()); + let entries: Vec<&RawValue> = if payload.trim_ascii_start().starts_with(b"[") { + serde_json::from_slice(payload).map_err(unreadable)? + } else { + vec![serde_json::from_slice(payload).map_err(unreadable)?] + }; + for entry in entries { + let message: ServerMessage = serde_json::from_str(entry.get()).map_err(unreadable)?; + match (message.id, message.method) { + (Some(id), None) if Some(&id) == self.expect => { + self.answered = true; + if self.capture { + self.captured = Some(match (message.result, message.error) { + (Some(result), _) => Ok(result.to_owned()), + (None, error) => { + Err(error.map_or_else(String::new, |error| error.get().to_owned())) + } + }); + continue; + } + } + (Some(id), Some(_)) => { + lock(&self.connection.server_requests).insert(id, Arc::clone(self.session)); + } + _ => {} + } + self.connection.send(one_line(entry.get())); + } + Ok(()) + } +} + +/// Codex reads one message per line. Raw CR and LF can appear in JSON only as whitespace between +/// tokens (inside strings they must be escaped), so removing them never changes a message. +fn one_line(json: &str) -> String { + json.chars().filter(|c| !matches!(c, '\r' | '\n')).collect() +} + +fn protocol_version(init_result: &RawValue) -> Option { + #[derive(Deserialize)] + struct Probe { + #[serde(rename = "protocolVersion")] + protocol_version: Option, + } + match serde_json::from_str::(init_result.get()) + .ok()? + .protocol_version? + { + Value::String(version) if !version.is_empty() => Some(version), + _ => None, + } +}