1import {
2 type CommandId,
3 type RuntimeRequestId,
4 ThreadId,
5 type OrchestrationV2ThreadProjection,
6 type RunId,
7 OrchestratorMcpFailure,
8 type OrchestrationV2Command,
9} from "@t3tools/contracts";
10import * as Effect from "effect/Effect";
11import { modelSelectionCommandType } from "@t3tools/shared/model";
12
13import * as McpToolAccess from "../../McpToolAccess.ts";
14import {
15 dispatchFailure,
16 newCommandId,
17 readCaller,
18 readThread,
19 unavailable,
20} from "../../threadAccess.ts";
21import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts";
22import * as ScheduledTasks from "../../../scheduledTasks/ScheduledTaskService.ts";
23import { queuedRunsInDeliveryOrder } from "../../../orchestration-v2/QueuedRunOrder.ts";
24import { ThreadToolkit } from "./tools.ts";
25
26function queueEntry(
27 projection: Pick<OrchestrationV2ThreadProjection, "runs" | "messages">,
28 runId: RunId,
29 limit: number,
30) {
31 const run = projection.runs.find((run) => run.id === runId && run.status === "queued");
32 const message = projection.messages.find((message) => message.id === run?.userMessageId);
33 if (run === undefined || message === undefined) return undefined;
34 const characters = Array.from(message.text);
35 return {
36 queuedRunId: run.id,
37 text: characters.slice(0, limit).join(""),
38 truncated: characters.length > limit,
39 };
40}
41const dispatch = Effect.fn("mcp.dispatchThreadCommand")(function* (
42 threadId: ThreadId | undefined,
43 command: (common: { commandId: CommandId; threadId: ThreadId }) => OrchestrationV2Command,
44) {
45 const { threads, projection } = yield* readThread(threadId);
46 const result = yield* threads
47 .dispatch(command({ commandId: yield* newCommandId(), threadId: projection.thread.id }))
48 .pipe(Effect.mapError(dispatchFailure));
49 return { sequence: result.sequence };
50});
51
52const readQuestion = Effect.fn("mcp.readQuestion")(function* (input: {
53 threadId?: ThreadId | undefined;
54 requestId: RuntimeRequestId;
55}) {
56 const context = yield* readThread(input.threadId, ["runtimeRequests", "turnItems"]);
57 const request = context.projection.runtimeRequests.find(
58 (request) =>
59 request.id === input.requestId &&
60 request.kind === "user_input" &&
61 request.status === "pending",
62 );
63 const item = context.projection.turnItems.find(
64 (item) => item.type === "user_input_request" && item.requestId === input.requestId,
65 );
66 if (request === undefined || item?.type !== "user_input_request")
67 return yield* new OrchestratorMcpFailure({
68 code: "invalid_request",
69 message: "The pending user-input request was not found.",
70 });
71 return { ...context, request, item };
72});
73/** A tool that changes `threadId`, or the caller's own thread when it is omitted. */
74const writesThread = <P extends { readonly threadId?: ThreadId | undefined }, A, E, R>(
75 handle: (params: P) => Effect.Effect<A, E, R>,
76) => McpToolAccess.writesThreads((params: P) => [params.threadId], handle);
77
78export const layer = McpToolAccess.toLayer(ThreadToolkit, {
79 run_scheduled_task_now: McpToolAccess.writesEnvironment((input) =>
80 Effect.gen(function* () {
81 const scheduler = yield* ScheduledTasks.ScheduledTaskService;
82 const { tasks } = yield* scheduler.list().pipe(Effect.mapError(unavailable));
83 if (!tasks.some((task) => task.id === input.taskId))
84 return yield* new OrchestratorMcpFailure({
85 code: "invalid_request",
86 message: "The scheduled task was not found.",
87 });
88 const { task } = yield* scheduler
89 .runNow({ id: input.taskId })
90 .pipe(Effect.mapError(unavailable));
91 return {
92 taskId: task.id,
93 threadId: task.threadId,
94 lastRunStatus: task.lastRunStatus,
95 runCount: task.runCount,
96 nextRunAt: task.nextRunAt,
97 };
98 }),
99 ),
100 t3_thread_search: McpToolAccess.reads((input) =>
101 Effect.gen(function* () {
102 const { caller } = yield* readCaller();
103 const { projectId: requested, ...query } = input;
104 // Like the other project tools, an omitted project means the caller's own; a client
105 // outside a thread searches every project.
106 const projectId = requested ?? caller?.projectId;
107 const threadSearch = yield* ThreadSearch.ThreadSearch;
108 const result = yield* threadSearch.search(query).pipe(Effect.mapError(unavailable));
109 return {
110 matches:
111 projectId === undefined
112 ? result.matches
113 : result.matches.filter((match) => match.projectId === projectId),
114 };
115 }),
116 ),
117 t3_thread_fork: writesThread((input) =>
118 Effect.gen(function* () {
119 const { threads, projection } = yield* readThread(input.threadId);
120 const commandId = yield* newCommandId();
121 const targetThreadId = ThreadId.make(`${commandId}:fork`);
122 const result = yield* threads
123 .dispatch({
124 type: "thread.fork",
125 commandId,
126 sourceThreadId: projection.thread.id,
127 targetThreadId,
128 sourcePoint: input.sourcePoint,
129 ...(input.title === undefined ? {} : { title: input.title }),
130 createdBy: "agent",
131 creationSource: "mcp",
132 })
133 .pipe(Effect.mapError(dispatchFailure));
134 return { sequence: result.sequence, targetThreadId };
135 }),
136 ),
137 t3_thread_merge_back: McpToolAccess.writesThreads(
138 (input) => [input.targetThreadId, input.sourceThreadId],
139 (input) =>
140 Effect.gen(function* () {
141 const context = yield* readThread(input.targetThreadId);
142 const source = yield* readThread(input.sourceThreadId);
143 const result = yield* context.threads
144 .dispatch({
145 type: "thread.merge_back",
146 commandId: yield* newCommandId(),
147 sourceThreadId: source.projection.thread.id,
148 targetThreadId: input.targetThreadId,
149 sourcePoint: input.sourcePoint,
150 createdBy: "agent",
151 creationSource: "mcp",
152 })
153 .pipe(Effect.mapError(dispatchFailure));
154 return { sequence: result.sequence, targetThreadId: input.targetThreadId };
155 }),
156 ),
157 t3_thread_transfers: McpToolAccess.reads((input) =>
158 Effect.gen(function* () {
159 const { projection } = yield* readThread(input.threadId, ["contextTransfers"]);
160 return {
161 transfers: projection.contextTransfers.map(
162 ({ id, sourceThreadId, targetThreadId, status }) => ({
163 id,
164 sourceThreadId,
165 targetThreadId,
166 status,
167 }),
168 ),
169 };
170 }),
171 ),
172 t3_thread_configuration: McpToolAccess.reads((input) =>
173 Effect.gen(function* () {
174 const {
175 projection: { thread },
176 } = yield* readThread(input.threadId);
177 return {
178 threadId: thread.id,
179 modelSelection: thread.modelSelection,
180 runtimeMode: thread.runtimeMode,
181 interactionMode: thread.interactionMode,
182 };
183 }),
184 ),
185 t3_thread_configure: writesThread((input) =>
186 Effect.gen(function* () {
187 const {
188 threads,
189 projection: { thread },
190 } = yield* readThread(input.threadId);
191 const type = modelSelectionCommandType(thread.providerInstanceId, input.modelSelection);
192 const result = yield* threads
193 .dispatch({
194 type,
195 threadId: thread.id,
196 commandId: yield* newCommandId(),
197 modelSelection: input.modelSelection,
198 })
199 .pipe(Effect.mapError(dispatchFailure));
200 return { sequence: result.sequence };
201 }),
202 ),
203 t3_pending_request_list: McpToolAccess.reads((input) =>
204 Effect.gen(function* () {
205 const { projection } = yield* readThread(input.threadId, ["runtimeRequests"]);
206 return {
207 requestIds: projection.runtimeRequests
208 .filter((request) => request.kind === "user_input" && request.status === "pending")
209 .map((request) => request.id),
210 };
211 }),
212 ),
213 t3_pending_request_read: McpToolAccess.reads((input) =>
214 Effect.gen(function* () {
215 const { item } = yield* readQuestion(input);
216 return { requestId: input.requestId, questions: item.questions };
217 }),
218 ),
219 t3_pending_request_respond: writesThread((input) =>
220 Effect.gen(function* () {
221 const { threads, projection } = yield* readQuestion(input);
222 const result = yield* threads
223 .dispatch({
224 type: "runtime-request.respond",
225 threadId: projection.thread.id,
226 commandId: yield* newCommandId(),
227 requestId: input.requestId,
228 answers: input.answers,
229 })
230 .pipe(Effect.mapError(dispatchFailure));
231 return { sequence: result.sequence };
232 }),
233 ),
234 t3_queue_list: McpToolAccess.reads((input) =>
235 Effect.gen(function* () {
236 const { projection } = yield* readThread(input.threadId, ["runs", "messages"]);
237 const runs = queuedRunsInDeliveryOrder(projection);
238 const cursor = input.cursor ?? 0;
239 const end = cursor + (input.limit ?? 20);
240 return {
241 items: runs.slice(cursor, end).flatMap((run) => {
242 const entry = queueEntry(projection, run.id, 1000);
243 return entry === undefined ? [] : [entry];
244 }),
245 nextCursor: end < runs.length ? end : null,
246 };
247 }),
248 ),
249 t3_queue_read: McpToolAccess.reads((input) =>
250 Effect.gen(function* () {
251 const { projection } = yield* readThread(input.threadId, ["runs", "messages"]);
252 const entry = queueEntry(projection, input.queuedRunId, 16000);
253 return (
254 entry ??
255 (yield* new OrchestratorMcpFailure({
256 code: "invalid_request",
257 message: "The queued message was not found.",
258 }))
259 );
260 }),
261 ),
262 t3_queue_edit: writesThread((input) =>
263 dispatch(input.threadId, (common) => ({
264 ...common,
265 type: "queued-run.edit",
266 runId: input.queuedRunId,
267 text: input.text,
268 })),
269 ),
270 t3_queue_cancel: writesThread((input) =>
271 dispatch(input.threadId, (common) => ({
272 ...common,
273 type: "queued-run.cancel",
274 runId: input.queuedRunId,
275 })),
276 ),
277 t3_queue_reorder: writesThread((input) =>
278 dispatch(input.threadId, (common) => ({
279 ...common,
280 type: "queued-run.reorder",
281 runId: input.queuedRunId,
282 beforeRunId: input.beforeRunId,
283 })),
284 ),
285 t3_queue_promote_to_steer: writesThread((input) =>
286 dispatch(input.threadId, (common) => ({
287 ...common,
288 type: "queued-message.promote-to-steer",
289 queuedRunId: input.queuedRunId,
290 targetRunId: input.targetRunId,
291 })),
292 ),
293 t3_thread_organize: writesThread((input) =>
294 Effect.gen(function* () {
295 const { threads, projection, caller } = yield* readThread(input.threadId);
296 const common = { commandId: yield* newCommandId(), threadId: projection.thread.id };
297 if (input.action === "settle") {
298 return yield* threads
299 .settleThread({ ...common, byOwnAgent: caller?.id === projection.thread.id })
300 .pipe(Effect.mapError(dispatchFailure));
301 }
302 let command: OrchestrationV2Command;
303 switch (input.action) {
304 case "snooze":
305 if (input.snoozedUntil === undefined) {
306 return yield* new OrchestratorMcpFailure({
307 code: "invalid_request",
308 message: "snooze requires snoozedUntil.",
309 });
310 }
311 command = { ...common, type: "thread.snooze", snoozedUntil: input.snoozedUntil };
312 break;
313 case "unsnooze":
314 case "unsettle":
315 command = { ...common, type: `thread.${input.action}`, reason: "user" };
316 break;
317 case "mark_unread":
318 command = { ...common, type: "thread.mark-unread" };
319 break;
320 default:
321 command = { ...common, type: `thread.${input.action}` };
322 }
323 const result = yield* threads.dispatch(command).pipe(Effect.mapError(dispatchFailure));
324 return { sequence: result.sequence };
325 }),
326 ),
327});