1import { expect, it } from "@effect/vitest";
2import {
3 DEFAULT_SERVER_SETTINGS,
4 EnvironmentId,
5 ProviderInstanceId,
6 ThreadId,
7 type OrchestrationV2ThreadShell,
8} from "@t3tools/contracts";
9import * as Deferred from "effect/Deferred";
10import * as Effect from "effect/Effect";
11import * as Fiber from "effect/Fiber";
12import * as Layer from "effect/Layer";
13import * as Ref from "effect/Ref";
14import * as Stream from "effect/Stream";
15import { McpSchema, McpServer } from "effect/ai";
16
17import * as Environment from "../../../environment/ServerEnvironment.ts";
18import * as ThreadCommandExecutor from "../../../orchestration-v2/ThreadCommandExecutor.ts";
19import * as ThreadManagement from "../../../orchestration-v2/ThreadManagementService.ts";
20import * as Settings from "../../../serverSettings.ts";
21import * as McpHttpServer from "../../McpHttpServer.ts";
22import * as McpInvocationContext from "../../McpInvocationContext.ts";
23import * as McpToolAccess from "../../McpToolAccess.ts";
24import { liveThreadShell } from "../../McpToolAccess.testkit.ts";
25import * as EnvironmentHandlers from "./handlers.ts";
26import { EnvironmentToolkit } from "./tools.ts";
27
28const environmentId = EnvironmentId.make("environment:preferences");
29const threadId = ThreadId.make("thread:preferences");
30const caller: McpInvocationContext.McpInvocationScope = {
31 environmentId,
32 requestNamespace: "provider:preferences",
33 thread: {
34 threadId,
35 providerSessionId: "provider:preferences",
36 providerInstanceId: ProviderInstanceId.make("codex"),
37 },
38 client: undefined,
39 capabilities: new Set(["orchestration" as const]),
40 issuedAt: 0,
41};
42const layerEnvironment = Layer.mock(Environment.ServerEnvironment)({
43 getDescriptor: Effect.succeed({
44 environmentId,
45 label: "Test",
46 platform: { os: "linux", arch: "x64" },
47 serverVersion: "0.0.0",
48 capabilities: { repositoryIdentity: false },
49 }),
50});
51
52it.effect("updates preferences for a thread caller through the /mcp registration", () =>
53 Effect.gen(function* () {
54 const server = yield* McpServer.McpServer;
55 const result = yield* server
56 .callTool({
57 name: "t3_environment_preferences_update",
58 arguments: { newWorktreesStartFromOrigin: true },
59 })
60 .pipe(
61 Effect.provideService(McpInvocationContext.McpInvocationContext, caller),
62 Effect.provideService(
63 McpSchema.McpServerClient,
64 McpSchema.McpServerClient.of({
65 clientId: 1,
66 protocolVersion: "2025-06-18",
67 clientCapabilities: {},
68 clientInfo: { name: "preferences", version: "1" },
69 initializePayload: {
70 protocolVersion: "2025-06-18",
71 capabilities: {},
72 clientInfo: { name: "preferences", version: "1" },
73 },
74 getClient: Effect.die("unused"),
75 }),
76 ),
77 );
78 expect(result.isError).toBe(false);
79 expect(result.structuredContent).toMatchObject({ newWorktreesStartFromOrigin: true });
80 }).pipe(
81 Effect.provide(
82 McpHttpServer.layerEnvironmentToolkit.pipe(
83 Layer.provideMerge(McpServer.McpServer.layer),
84 Layer.provide(
85 Layer.mock(ThreadManagement.ThreadManagementService)({
86 getThreadShell: (id) => Effect.succeed(liveThreadShell(id)),
87 }),
88 ),
89 Layer.provide(layerEnvironment),
90 Layer.provide(Settings.layerTest()),
91 ),
92 ),
93 ),
94);
95
96it.effect("refuses a preferences update when the caller's turn ends while it waits", () =>
97 Effect.gen(function* () {
98 const caller = yield* Ref.make<OrchestrationV2ThreadShell>(liveThreadShell(threadId));
99 const updates = yield* Ref.make(0);
100 // Completes once the declaration's own check has read the caller.
101 const checked = yield* Deferred.make<void>();
102 const layerDependencies = Layer.mergeAll(
103 ThreadCommandExecutor.layer,
104 Layer.succeed(McpInvocationContext.McpInvocationContext, {
105 environmentId,
106 requestNamespace: "provider:preferences",
107 thread: {
108 threadId,
109 providerSessionId: "provider:preferences",
110 providerInstanceId: ProviderInstanceId.make("codex"),
111 },
112 client: undefined,
113 capabilities: new Set(["orchestration" as const]),
114 issuedAt: 0,
115 }),
116 Layer.mock(ThreadManagement.ThreadManagementService)({
117 getThreadShell: () =>
118 Ref.get(caller).pipe(Effect.tap(() => Deferred.succeed(checked, undefined))),
119 }),
120 Layer.mock(Environment.ServerEnvironment)({
121 getDescriptor: Effect.succeed({
122 environmentId,
123 label: "Test",
124 platform: { os: "linux", arch: "x64" },
125 serverVersion: "0.0.0",
126 capabilities: { repositoryIdentity: false },
127 }),
128 }),
129 Layer.mock(Settings.ServerSettingsService)({
130 getSettings: Effect.succeed(DEFAULT_SERVER_SETTINGS),
131 updateSettings: () =>
132 Ref.update(updates, (count) => count + 1).pipe(Effect.as(DEFAULT_SERVER_SETTINGS)),
133 }),
134 );
135 yield* Effect.gen(function* () {
136 const toolkit = yield* EnvironmentToolkit;
137 const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor;
138 const held = yield* Deferred.make<void>();
139 const release = yield* Deferred.make<void>();
140 // A turn-completion command holds the thread's lock.
141 const holder = yield* executor
142 .withLock(
143 threadId,
144 Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))),
145 )
146 .pipe(Effect.forkChild);
147 yield* Deferred.await(held);
148 const update = yield* toolkit
149 .handle("t3_environment_preferences_update", { newWorktreesStartFromOrigin: true })
150 .pipe(Stream.unwrap, Stream.runCollect, Effect.forkChild);
151 // The update passed its first check and waits for the lock; the turn then ends.
152 yield* Deferred.await(checked);
153 yield* Ref.update(caller, (shell) => ({ ...shell, activeRunId: null }));
154 yield* Deferred.succeed(release, undefined);
155 yield* Fiber.join(holder);
156 const result = yield* Fiber.join(update);
157 expect(result.at(-1)?.result).toMatchObject({ code: "parent_not_active" });
158 expect(yield* Ref.get(updates)).toBe(0);
159 }).pipe(
160 Effect.provide(
161 McpToolAccess.HandlersLayer.layer(EnvironmentHandlers.layer).pipe(
162 Layer.provideMerge(layerDependencies),
163 ),
164 ),
165 );
166 }),
167);