1import type { ProjectionRecordField } from "../orchestration-v2/ProjectionStore.ts";
2import {
3 CommandId,
4 OrchestratorMcpFailure,
5 type ProjectId,
6 type ProviderInteractionMode,
7 type RuntimeMode,
8 type ThreadId,
9 type OrchestrationV2ThreadShell,
10} from "@t3tools/contracts";
11import * as Crypto from "effect/Crypto";
12import * as Effect from "effect/Effect";
13
14import type { OrchestratorV2Error } from "../orchestration-v2/Orchestrator.ts";
15import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
16import * as OrchestrationMcp from "./OrchestratorMcpService.ts";
17import * as McpInvocationContext from "./McpInvocationContext.ts";
18
19export const unavailable = () =>
20 new OrchestratorMcpFailure({
21 code: "orchestration_error",
22 message: "The operation could not be completed.",
23 });
24
25/** Decider string rejections are public; wrapped storage and hydration causes are not. */
26export const dispatchFailure = (
27 error: OrchestratorV2Error | ThreadManagement.ThreadManagementProviderInstanceDisabledError,
28) =>
29 error._tag === "ThreadManagementProviderInstanceDisabledError"
30 ? new OrchestratorMcpFailure({ code: "invalid_request", message: error.message })
31 : (error._tag === "OrchestratorDispatchError" ||
32 error._tag === "OrchestratorCommandRejectedError") &&
33 typeof error.cause === "string" &&
34 error.cause.length > 0
35 ? new OrchestratorMcpFailure({
36 code: "orchestration_error",
37 message: Array.from(error.cause).slice(0, 1000).join(""),
38 })
39 : unavailable();
40
41/**
42 * The most a caller may hand to the threads it targets. A thread caller is
43 * capped by its own thread's modes; an OAuth client by the ceiling chosen
44 * when it was approved.
45 */
46export interface CallerLimits {
47 readonly runtimeMode: RuntimeMode;
48 readonly interactionMode: ProviderInteractionMode;
49}
50
51export interface Caller {
52 readonly scope: McpInvocationContext.McpInvocationScope;
53 readonly threads: ThreadManagement.ThreadManagementService["Service"];
54 /** The calling thread, absent for a client signed in from outside T3. */
55 readonly caller: OrchestrationV2ThreadShell | undefined;
56 readonly limits: CallerLimits;
57}
58
59export const readCaller = Effect.fn("mcp.readCaller")(function* () {
60 const scope = yield* McpInvocationContext.McpInvocationContext;
61 if (!scope.capabilities.has("orchestration")) {
62 return yield* new OrchestratorMcpFailure({
63 code: "capability_denied",
64 message: "This credential cannot control threads.",
65 });
66 }
67 return yield* loadCaller();
68});
69
70/**
71 * The caller and its limits, whichever tools its credential grants. Tools
72 * check their own capability; `McpToolAccess` uses this for every tool.
73 */
74export const loadCaller = Effect.fn("mcp.loadCaller")(function* () {
75 const scope = yield* McpInvocationContext.McpInvocationContext;
76 const threads = yield* ThreadManagement.ThreadManagementService;
77 if (scope.thread === undefined) {
78 return {
79 scope,
80 threads,
81 caller: undefined,
82 limits: {
83 runtimeMode: McpInvocationContext.clientRuntimeModeCeiling(scope.client),
84 interactionMode: "default",
85 },
86 } satisfies Caller;
87 }
88 const caller = yield* threads
89 .getThreadShell(scope.thread.threadId)
90 .pipe(Effect.mapError(unavailable));
91 if (caller === null || caller.deletedAt !== null) {
92 return yield* new OrchestratorMcpFailure({
93 code: "thread_not_found",
94 message: "The calling thread was not found.",
95 });
96 }
97 return {
98 scope,
99 threads,
100 caller,
101 limits: { runtimeMode: caller.runtimeMode, interactionMode: caller.interactionMode },
102 } satisfies Caller;
103});
104
105/**
106 * A caller may change another thread only if that thread runs within the
107 * caller's own modes. Its own thread is always within them.
108 */
109export const assertTargetWithinLimits = (
110 limits: CallerLimits,
111 target: { readonly runtimeMode: RuntimeMode; readonly interactionMode: ProviderInteractionMode },
112) =>
113 OrchestrationMcp.resolveRuntimeMode(limits.runtimeMode, target.runtimeMode).pipe(
114 Effect.andThen(
115 OrchestrationMcp.resolveInteractionMode(limits.interactionMode, target.interactionMode),
116 ),
117 Effect.asVoid,
118 );
119
120/**
121 * A thread caller acts only while it owns a live run of a thread that is not
122 * archived, so a provider token that outlived its session cannot act. An
123 * OAuth client has no run; its session and ceiling are its authority.
124 */
125export function assertLiveCaller({ caller, scope }: Caller) {
126 if (caller === undefined) return Effect.void;
127 return caller.archivedAt !== null ||
128 caller.activeRunId === null ||
129 caller.providerInstanceId !== scope.thread?.providerInstanceId
130 ? Effect.fail(
131 new OrchestratorMcpFailure({
132 code: "parent_not_active",
133 message: "The calling provider no longer owns an active thread run.",
134 }),
135 )
136 : Effect.void;
137}
138
139/**
140 * Actions that change the environment itself (projects, preferences) need full
141 * access: a thread caller in full-access/default mode, or a client approved
142 * with a full-access ceiling.
143 */
144export const assertFullAccess = (context: Caller, message: string) =>
145 context.limits.runtimeMode === "full-access" && context.limits.interactionMode === "default"
146 ? Effect.void
147 : Effect.fail(new OrchestratorMcpFailure({ code: "capability_denied", message }));
148
149/** A target project: the one passed, else the calling thread's. */
150export const resolveProjectId = (context: Caller, projectId: ProjectId | undefined) =>
151 projectId !== undefined
152 ? Effect.succeed(projectId)
153 : context.caller !== undefined
154 ? Effect.succeed(context.caller.projectId)
155 : Effect.fail(
156 new OrchestratorMcpFailure({
157 code: "target_required",
158 message: "Pass projectId: this MCP client is not running inside a T3 thread.",
159 }),
160 );
161
162/** A target thread: the one passed, else the calling thread. */
163const resolveThreadId = (context: Caller, threadId: ThreadId | undefined) =>
164 threadId !== undefined
165 ? Effect.succeed(threadId)
166 : context.caller !== undefined
167 ? Effect.succeed(context.caller.id)
168 : Effect.fail(
169 new OrchestratorMcpFailure({
170 code: "target_required",
171 message: "Pass threadId: this MCP client is not running inside a T3 thread.",
172 }),
173 );
174
175/** Load a target thread anywhere in the environment; an omitted id means the calling thread. */
176export const readThread = Effect.fn("mcp.readThread")(function* <
177 K extends ProjectionRecordField = never,
178>(threadId?: ThreadId, fields: ReadonlyArray<K> = []) {
179 const context = yield* readCaller();
180 const targetId = yield* resolveThreadId(context, threadId);
181 const shell = yield* context.threads.getThreadShell(targetId).pipe(Effect.mapError(unavailable));
182 if (shell === null || shell.deletedAt !== null) {
183 return yield* new OrchestratorMcpFailure({
184 code: "thread_not_found",
185 message: "The thread was not found.",
186 });
187 }
188 const projection = yield* context.threads
189 .getProjectThreadRecords({ projectId: shell.projectId, threadId: targetId }, fields, {
190 turnItemTypes: ["user_input_request"],
191 })
192 .pipe(
193 Effect.mapError((error) =>
194 error._tag === "ThreadManagementThreadNotFoundError"
195 ? new OrchestratorMcpFailure({
196 code: "thread_not_found",
197 message: "The thread was not found.",
198 })
199 : unavailable(),
200 ),
201 );
202 return { ...context, projection };
203});
204
205export const newCommandId = Effect.fn("mcp.newCommandId")(function* () {
206 const crypto = yield* Crypto.Crypto;
207 return CommandId.make(`mcp:${yield* crypto.randomUUIDv4().pipe(Effect.orDie)}`);
208});