Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
47 changes: 47 additions & 0 deletions apps/server/src/mcp/mcpPlugin.test.ts
Original file line number Diff line number Diff line change
@@ -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<string>) =>
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"),
);
});
});
90 changes: 90 additions & 0 deletions apps/server/src/mcp/mcpUnixSocket.test.ts
Original file line number Diff line number Diff line change
@@ -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)),
);
115 changes: 115 additions & 0 deletions apps/server/src/mcp/mcpUnixSocket.ts
Original file line number Diff line number Diff line change
@@ -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>()("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 = <E, R>(app: Effect.Effect<HttpServerResponse.HttpServerResponse, E, R>) =>
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<boolean>((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 = <E, R>(
server: HttpServer.HttpServer["Service"],
app: Effect.Effect<HttpServerResponse.HttpServerResponse, E, R>,
middleware: HttpMiddleware.HttpMiddleware | undefined,
): Effect.Effect<void, never, Exclude<R, HttpServerRequest.HttpServerRequest> | 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<NodeHttp.Server>,
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: <E, R>(
app: Effect.Effect<HttpServerResponse.HttpServerResponse, E, R>,
middleware?: HttpMiddleware.HttpMiddleware,
) =>
Effect.andThen(
serveWith(main, app, middleware),
serveWith(socket, onlyMcp(app), middleware),
),
});
}),
),
NodeHttpServer.layerHttpServices,
);
Original file line number Diff line number Diff line change
Expand Up @@ -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 };
Expand Down
4 changes: 2 additions & 2 deletions apps/server/src/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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";
Expand Down Expand Up @@ -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()),
Expand Down
Loading
Loading