1import type { ProjectionRecordField } from "../orchestration-v2/ProjectionStore.ts";
2import {
3 CommandId,
4 OrchestratorMcpFailure,
5 type ProjectId,
6 type ProviderInstanceId,
7 type ProviderInteractionMode,
8 type RuntimeMode,
9 ThreadId,
10 type OrchestrationV2ThreadShell,
11} from "@t3tools/contracts";
12import * as SecureRandom from "effect/crypto/SecureRandom";
13import * as Effect from "effect/Effect";
14
15import type { OrchestratorV2Error } from "../orchestration-v2/Orchestrator.ts";
16import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
17import { resolveInteractionMode, resolveRuntimeMode } from "./delegatedTaskModes.ts";
18import * as McpInvocationContext from "./McpInvocationContext.ts";
19
20/**
21 * Threads an agent starts over MCP are `mcp:<uuid>`, and agents often paste the
22 * bare uuid. The id is never rewritten, so permission checks keyed on it still
23 * hold; the miss only names a live thread the caller could read by its full id.
24 */
25export const threadNotFound = Effect.fn("OrchestratorMcpService.threadNotFound")(function* (
26 threads: Pick<ThreadManagement.ThreadManagementService["Service"], "getThreadShell">,
27 threadId: ThreadId,
28 message: string,
29) {
30 const prefixed = ThreadId.make(`mcp:${threadId}`);
31 const shell = yield* threads.getThreadShell(prefixed).pipe(Effect.orElseSucceed(() => null));
32 return yield* new OrchestratorMcpFailure({
33 code: "thread_not_found",
34 message:
35 shell === null || shell.deletedAt !== null
36 ? message
37 : `${message} Did you mean \`${prefixed}\`?`,
38 });
39});
40
41export const unavailable = () =>
42 new OrchestratorMcpFailure({
43 code: "orchestration_error",
44 message: "The operation could not be completed.",
45 });
46
47export const dispatchFailure = (
48 error: OrchestratorV2Error | ThreadManagement.ThreadManagementProviderInstanceDisabledError,
49) =>
50 error._tag === "ThreadManagementProviderInstanceDisabledError" ||
51 error._tag === "OrchestratorRuntimeModeAboveCeilingError" ||
52 error._tag === "OrchestratorProviderSelectionBusyError"
53 ? new OrchestratorMcpFailure({ code: "invalid_request", message: error.message })
54 : (error._tag === "OrchestratorDispatchError" ||
55 error._tag === "OrchestratorCommandRejectedError") &&
56 typeof error.cause === "string" &&
57 error.cause.length > 0
58 ? new OrchestratorMcpFailure({
59 code: "orchestration_error",
60 message: Array.from(error.cause).slice(0, 1000).join(""),
61 })
62 : unavailable();
63
64export interface CallerLimits {
65 readonly runtimeMode: RuntimeMode;
66 readonly interactionMode: ProviderInteractionMode;
67}
68
69export interface Caller {
70 readonly scope: McpInvocationContext.McpInvocationScope;
71 readonly threads: ThreadManagement.ThreadManagementService["Service"];
72 readonly caller: OrchestrationV2ThreadShell | undefined;
73 readonly limits: CallerLimits;
74}
75
76export const readCaller = Effect.fn("mcp.readCaller")(function* () {
77 const scope = yield* McpInvocationContext.McpInvocationContext;
78 if (!scope.capabilities.has("orchestration")) {
79 return yield* new OrchestratorMcpFailure({
80 code: "capability_denied",
81 message: "This credential cannot control threads.",
82 });
83 }
84 return yield* loadCaller();
85});
86
87export const loadCaller = Effect.fn("mcp.loadCaller")(function* () {
88 const scope = yield* McpInvocationContext.McpInvocationContext;
89 const threads = yield* ThreadManagement.ThreadManagementService;
90 if (scope.thread === undefined) {
91 return {
92 scope,
93 threads,
94 caller: undefined,
95 limits: McpInvocationContext.clientModeCeiling(scope.client),
96 } satisfies Caller;
97 }
98 const caller = yield* threads
99 .getThreadShell(scope.thread.threadId)
100 .pipe(Effect.mapError(unavailable));
101 if (caller === null || caller.deletedAt !== null) {
102 return yield* new OrchestratorMcpFailure({
103 code: "thread_not_found",
104 message: "The calling thread was not found.",
105 });
106 }
107 return {
108 scope,
109 threads,
110 caller,
111 limits: { runtimeMode: caller.runtimeMode, interactionMode: caller.interactionMode },
112 } satisfies Caller;
113});
114
115export const assertTargetWithinLimits = (
116 limits: CallerLimits,
117 target: { readonly runtimeMode: RuntimeMode; readonly interactionMode: ProviderInteractionMode },
118) =>
119 resolveRuntimeMode(limits.runtimeMode, target.runtimeMode).pipe(
120 Effect.andThen(resolveInteractionMode(limits.interactionMode, target.interactionMode)),
121 Effect.asVoid,
122 );
123
124/** Background subagents retain credentials until provider exit; archived threads and other instances cannot act. */
125export const isLiveCallerShell = (
126 shell: Pick<
127 OrchestrationV2ThreadShell,
128 "archivedAt" | "activeRunId" | "providerInstanceId" | "pendingBackgroundTasks"
129 > | null,
130 providerInstanceId: ProviderInstanceId | undefined,
131) =>
132 shell !== null &&
133 shell.archivedAt === null &&
134 shell.providerInstanceId === providerInstanceId &&
135 (shell.activeRunId !== null ||
136 (shell.pendingBackgroundTasks?.some(
137 (task) => task.kind === "subagent" && task.providerInstanceId === providerInstanceId,
138 ) ??
139 false));
140
141/** Thread callers need a live run or its background subagent in an unarchived thread. OAuth clients rely on their session and ceiling. */
142export function assertLiveCaller({ caller, scope }: Caller) {
143 if (caller === undefined) return Effect.void;
144 return isLiveCallerShell(caller, scope.thread?.providerInstanceId)
145 ? Effect.void
146 : Effect.fail(
147 new OrchestratorMcpFailure({
148 code: "parent_not_active",
149 message: "The calling provider no longer owns an active thread run.",
150 }),
151 );
152}
153
154export const assertFullAccess = (context: Caller, message: string) =>
155 context.limits.runtimeMode === "full-access" && context.limits.interactionMode === "default"
156 ? Effect.void
157 : Effect.fail(new OrchestratorMcpFailure({ code: "capability_denied", message }));
158
159export const resolveProjectId = (context: Caller, projectId: ProjectId | undefined) =>
160 projectId !== undefined
161 ? Effect.succeed(projectId)
162 : context.caller !== undefined
163 ? Effect.succeed(context.caller.projectId)
164 : Effect.fail(
165 new OrchestratorMcpFailure({
166 code: "target_required",
167 message: "Pass projectId: this MCP client is not running inside a T3 thread.",
168 }),
169 );
170
171const resolveThreadId = (context: Caller, threadId: ThreadId | undefined) =>
172 threadId !== undefined
173 ? Effect.succeed(threadId)
174 : context.caller !== undefined
175 ? Effect.succeed(context.caller.id)
176 : Effect.fail(
177 new OrchestratorMcpFailure({
178 code: "target_required",
179 message: "Pass threadId: this MCP client is not running inside a T3 thread.",
180 }),
181 );
182
183export const readThread = Effect.fn("mcp.readThread")(function* <
184 K extends ProjectionRecordField = never,
185>(threadId?: ThreadId, fields: ReadonlyArray<K> = []) {
186 const context = yield* readCaller();
187 const targetId = yield* resolveThreadId(context, threadId);
188 const shell = yield* context.threads.getThreadShell(targetId).pipe(Effect.mapError(unavailable));
189 if (shell === null || shell.deletedAt !== null) {
190 return yield* threadNotFound(context.threads, targetId, "The thread was not found.");
191 }
192 const projection = yield* context.threads
193 .getProjectThreadRecords({ projectId: shell.projectId, threadId: targetId }, fields, {
194 turnItemTypes: ["user_input_request"],
195 })
196 .pipe(
197 Effect.mapError((error) =>
198 error._tag === "ThreadManagementThreadNotFoundError"
199 ? new OrchestratorMcpFailure({
200 code: "thread_not_found",
201 message: "The thread was not found.",
202 })
203 : unavailable(),
204 ),
205 );
206 return { ...context, projection };
207});
208
209export const newCommandId = Effect.fn("mcp.newCommandId")(function* () {
210 const random = yield* SecureRandom.SecureRandom;
211 return CommandId.make(`mcp:${yield* random.uuidV4()}`);
212});