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
11 changes: 11 additions & 0 deletions apps/server/src/mcp/McpHttpServer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,12 @@ import * as McpInvocationContext from "./McpInvocationContext.ts";
import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
import * as McpSessionRegistry from "./McpSessionRegistry.ts";
import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
import * as ProjectMcpService from "./ProjectMcpService.ts";
import * as T3ProjectFileLoader from "../project/T3ProjectFileLoader.ts";
import { OrchestratorToolkitHandlersLive } from "./toolkits/orchestrator/handlers.ts";
import { OrchestratorToolkit } from "./toolkits/orchestrator/tools.ts";
import { ProjectToolkitHandlersLive } from "./toolkits/project/handlers.ts";
import { ProjectToolkit } from "./toolkits/project/tools.ts";
import {
PreviewSnapshotToolkitHandlersLive,
PreviewStandardToolkitHandlersLive,
Expand Down Expand Up @@ -227,6 +231,12 @@ export const OrchestratorToolkitRegistrationLive = McpServer.toolkit(Orchestrato
Layer.provide(OrchestratorMcpService.layer),
);

export const ProjectToolkitRegistrationLive = McpServer.toolkit(ProjectToolkit).pipe(
Layer.provide(ProjectToolkitHandlersLive),
Layer.provide(ProjectMcpService.layer),
Layer.provide(T3ProjectFileLoader.layer),
);

export const WorktreeToolkitRegistrationLive = McpServer.toolkit(WorktreeToolkit).pipe(
Layer.provide(WorktreeToolkitHandlersLive),
Layer.provide(WorktreeMcpService.layer),
Expand All @@ -242,5 +252,6 @@ const McpTransportLive = McpServer.layerHttp({
export const layer = Layer.mergeAll(
PreviewToolkitRegistrationLive,
OrchestratorToolkitRegistrationLive,
ProjectToolkitRegistrationLive,
WorktreeToolkitRegistrationLive,
).pipe(Layer.provideMerge(McpTransportLive));
137 changes: 137 additions & 0 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,20 @@ import * as NodeServices from "@effect/platform-node/NodeServices";
import { assert, describe, it } from "@effect/vitest";
import {
EnvironmentId,
MessageId,
NodeId,
ProjectId,
ProviderDriverKind,
ProviderInstanceId,
RunId,
type ServerProvider,
ThreadId,
type OrchestrationV2ThreadProjection,
} from "@t3tools/contracts";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Ref from "effect/Ref";

import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts";
Expand All @@ -19,6 +25,137 @@ import type { McpInvocationScope } from "./McpInvocationContext.ts";
import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";

describe("OrchestratorMcpService", () => {
it.effect("preserves a structured parent admission failure", () =>
Effect.gen(function* () {
const parentThreadId = ThreadId.make("thread:mcp-create-admission-parent");
const parentRunId = RunId.make("run:mcp-create-admission-parent");
const parentNodeId = NodeId.make("node:mcp-create-admission-parent");
const projectId = ProjectId.make("project:mcp-create-admission");
const providerInstanceId = ProviderInstanceId.make("codex");
const now = DateTime.makeUnsafe("2026-08-30T12:00:00.000Z");
const activeParent: OrchestrationV2ThreadProjection = {
thread: {
createdBy: "agent",
creationSource: "mcp",
id: parentThreadId,
projectId,
title: "Admission parent",
providerInstanceId,
modelSelection: { instanceId: providerInstanceId, model: "gpt-test" },
runtimeMode: "full-access",
interactionMode: "default",
branch: "main",
worktreePath: null,
activeProviderThreadId: null,
lineage: {
parentThreadId: null,
relationshipToParent: null,
rootThreadId: parentThreadId,
},
forkedFrom: null,
createdAt: now,
updatedAt: now,
archivedAt: null,
settledOverride: null,
settledAt: null,
lastVisitedAt: null,
deletedAt: null,
},
runs: [
{
id: parentRunId,
threadId: parentThreadId,
ordinal: 1,
providerInstanceId,
modelSelection: { instanceId: providerInstanceId, model: "gpt-test" },
providerThreadId: null,
userMessageId: MessageId.make("message:mcp-create-admission-parent"),
rootNodeId: parentNodeId,
activeAttemptId: null,
status: "running",
requestedAt: now,
startedAt: now,
completedAt: null,
checkpointId: null,
contextHandoffId: null,
},
],
attempts: [],
nodes: [],
subagents: [],
providerSessions: [],
providerThreads: [],
providerTurns: [],
runtimeRequests: [],
messages: [],
plans: [],
turnItems: [],
checkpointScopes: [],
checkpoints: [],
contextHandoffs: [],
contextTransfers: [],
visibleTurnItems: [],
updatedAt: now,
};
const archivedParent = {
...activeParent,
thread: { ...activeParent.thread, archivedAt: now },
} as OrchestrationV2ThreadProjection;
const provider = {
instanceId: providerInstanceId,
driver: ProviderDriverKind.make("codex"),
enabled: true,
installed: true,
version: "test",
status: "ready",
auth: { status: "authenticated" },
checkedAt: "2026-08-30T12:00:00.000Z",
models: [{ slug: "gpt-test", name: "GPT Test", isCustom: false, capabilities: null }],
slashCommands: [],
skills: [],
} satisfies ServerProvider;
const projectionReads = yield* Ref.make(0);
const dependencies = Layer.mergeAll(
NodeServices.layer,
Layer.mock(ThreadManagementService)({
getThreadProjection: () =>
Ref.updateAndGet(projectionReads, (count) => count + 1).pipe(
Effect.map((count) => (count === 1 ? activeParent : archivedParent)),
),
withProjectCreationAdmission: (_input, effect) => effect(Option.none()),
dispatch: () => Effect.die("dispatch must not run after parent admission fails"),
}),
Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([provider]) }),
Layer.mock(ScheduledTaskService)({}),
);
const scope: McpInvocationScope = {
environmentId: EnvironmentId.make("environment:mcp-create-admission"),
threadId: parentThreadId,
providerSessionId: "provider-session:mcp-create-admission",
providerInstanceId,
capabilities: new Set(["orchestration"]),
issuedAt: 1,
};

const error = yield* Effect.gen(function* () {
const service = yield* OrchestratorMcpService.OrchestratorMcpService;
return yield* service
.createThreads(scope, {
clientRequestId: "parent-admission-failure",
threads: [{ title: "Must not be created" }],
})
.pipe(Effect.flip);
}).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(dependencies))));

assert.equal(error.code, "parent_not_active");
assert.equal(
error.message,
"Thread creation requires an active parent in the target project.",
);
assert.equal(yield* Ref.get(projectionReads), 2);
}),
);

it.effect("retries terminal acknowledgement with a fresh command id", () =>
Effect.gen(function* () {
const parentThreadId = ThreadId.make("thread:mcp-ack-parent");
Expand Down
74 changes: 51 additions & 23 deletions apps/server/src/mcp/OrchestratorMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,7 @@ export class OrchestratorMcpService extends Context.Service<
>()("t3/mcp/OrchestratorMcpService") {}

const isThreadManagementError = Schema.is(ThreadManagementError);
const isOrchestratorMcpFailure = Schema.is(OrchestratorMcpFailure);

function failure(code: OrchestratorMcpFailure["code"], message: string): OrchestratorMcpFailure {
return new OrchestratorMcpFailure({ code, message });
Expand Down Expand Up @@ -1370,32 +1371,59 @@ const make = Effect.gen(function* () {
title: request.title,
index,
});
const createCommandId = stableCommandId({
scope,
requestKey: key,
operation: "create-thread",
index,
});
const createCommand = {
type: "thread.create",
createdBy: "agent",
creationSource: "mcp",
commandId: createCommandId,
threadId,
projectId: parent.thread.projectId,
title,
modelSelection: target.modelSelection,
runtimeMode,
interactionMode,
branch: parent.thread.branch,
worktreePath: parent.thread.worktreePath,
} as const;
yield* threadManagement
.dispatch({
type: "thread.create",
createdBy: "agent",
creationSource: "mcp",
commandId: stableCommandId({
scope,
requestKey: key,
operation: "create-thread",
index,
}),
threadId,
projectId: parent.thread.projectId,
title,
modelSelection: target.modelSelection,
runtimeMode,
interactionMode,
branch: parent.thread.branch,
worktreePath: parent.thread.worktreePath,
})
.withProjectCreationAdmission(
{ projectId: parent.thread.projectId, commandId: createCommandId },
(receipt) =>
Effect.gen(function* () {
if (Option.isNone(receipt)) {
const freshParent = yield* loadProjection(scope.threadId);
const freshParentRun = latestActiveRun(freshParent);
if (
freshParent.thread.projectId !== parent.thread.projectId ||
freshParent.thread.deletedAt !== null ||
freshParent.thread.archivedAt !== null ||
freshParentRun === undefined ||
freshParentRun.rootNodeId === null ||
freshParentRun.providerInstanceId !== scope.providerInstanceId
) {
return yield* failure(
"parent_not_active",
"Thread creation requires an active parent in the target project.",
);
}
}
return yield* threadManagement.dispatch(createCommand);
}),
)
Comment thread
juliusmarminge marked this conversation as resolved.
.pipe(
Effect.mapError((error) =>
failure(
"orchestration_error",
`Unable to create thread ${index + 1}: ${errorMessage(error)}`,
),
isOrchestratorMcpFailure(error)
? error
: failure(
"orchestration_error",
`Unable to create thread ${index + 1}: ${errorMessage(error)}`,
),
),
);
if (request.prompt !== undefined) {
Expand Down
Loading
Loading