History

Merge remote-tracking branch 'origin/codex/batch17-cloud-mcp-boundaries' into codex/effect-lint-integration

Rhys Sullivan committed 6 months ago

Commit 3a856292 parents cd10e67 fcae4133 files · +158 −90Browse files

Unified · Split

3 files
  1. apps/cloud/src/mcp-session.ts +27 −13
  2. apps/cloud/src/mcp.ts +28 −24
  3. apps/cloud/src/mcp/response-peek.ts +103 −53
apps/cloud/src/mcp-session.ts+27−13
@@ -145,6 +145,7 @@ const makeDbHandle = (options: { · 144 unmodified lines
145145 return {
146146 sql,
147147 db: drizzle(sql, { schema: combinedSchema }) as DrizzleDb,
148+ // oxlint-disable-next-line executor/no-promise-catch -- boundary: postgres.js close is best-effort during DO/runtime cleanup
148149 end: () => sql.end({ timeout: 0 }).catch(() => undefined),
149150 };
150151 };
@@ -268,9 +269,13 @@ export class McpSessionDO extends DurableObject { · 117 unmodified lines
268269 this.transportJsonResponseMode = null;
269270
270271 await Promise.all([
272+ // oxlint-disable-next-line executor/no-promise-catch -- boundary: Durable Object storage cleanup is best-effort after session invalidation
271273 this.ctx.storage.delete(TRANSPORT_STATE_KEY).catch(() => false),
274+ // oxlint-disable-next-line executor/no-promise-catch -- boundary: Durable Object storage cleanup is best-effort after session invalidation
272275 this.ctx.storage.delete(SESSION_META_KEY).catch(() => false),
276+ // oxlint-disable-next-line executor/no-promise-catch -- boundary: Durable Object storage cleanup is best-effort after session invalidation
273277 this.ctx.storage.delete(LAST_ACTIVITY_KEY).catch(() => false),
278+ // oxlint-disable-next-line executor/no-promise-catch -- boundary: Durable Object alarm cleanup is best-effort after session invalidation
274279 this.ctx.storage.deleteAlarm().catch(() => undefined),
275280 ]);
276281 }).pipe(Effect.withSpan("mcp.session.clear_state"));
@@ -322,6 +327,7 @@ export class McpSessionDO extends DurableObject { · 45 unmodified lines
322327 }
323328 if (self.mcpServer) {
324329 const mcpServer = self.mcpServer;
330+ // oxlint-disable-next-line executor/no-promise-catch -- boundary: MCP SDK close failure is ignored during best-effort runtime teardown
325331 yield* Effect.promise(() => mcpServer.close().catch(() => undefined));
326332 self.mcpServer = null;
327333 }
@@ -332,7 +338,10 @@ export class McpSessionDO extends DurableObject { · 4 unmodified lines
332338 }
333339 self.initialized = false;
334340 self.transportJsonResponseMode = null;
335− }).pipe(Effect.orDie);
341+ }).pipe(
342+ // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: DO cleanup has no typed failure surface
343+ Effect.orDie,
344+ );
336345 }
337346
338347 private installRuntime(
@@ -388,6 +397,7 @@ export class McpSessionDO extends DurableObject { · 49 unmodified lines
388397 "mcp.request.session_id_present": !!request.headers.get("mcp-session-id"),
389398 },
390399 }),
400+ // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: cold DO restore is re-entered from Promise-only Durable Object method
391401 Effect.orDie,
392402 );
393403 }
@@ -409,7 +419,11 @@ export class McpSessionDO extends DurableObject { · 15 unmodified lines
409419 yield* Effect.annotateCurrentSpan({
410420 "mcp.session.transport_upgraded_json_response": true,
411421 });
412− }).pipe(Effect.withSpan("McpSessionDO.ensureJsonResponseTransportForPost"), Effect.orDie);
422+ }).pipe(
423+ Effect.withSpan("McpSessionDO.ensureJsonResponseTransportForPost"),
424+ // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: transport rebuild is internal DO runtime state
425+ Effect.orDie,
426+ );
413427 }
414428
415429 private validateSessionOwner(request: Request): Effect.Effect<Response | null> {
@@ -435,17 +449,15 @@ export class McpSessionDO extends DurableObject { · 19 unmodified lines
435449 const self = this;
436450 return Effect.gen(function* () {
437451 const dbHandle = makeEphemeralDb();
438− try {
439− const sessionMeta = yield* resolveSessionMeta(token.organizationId, token.userId).pipe(
440− Effect.provide(makeResolveOrganizationServices(dbHandle)),
441− );
442− yield* Effect.promise(() => self.saveSessionMeta(sessionMeta)).pipe(
443− Effect.withSpan("mcp.session.save_meta"),
444− );
445− return sessionMeta;
446− } finally {
447− yield* Effect.promise(() => dbHandle.end());
448− }
452+ return yield* resolveSessionMeta(token.organizationId, token.userId).pipe(
453+ Effect.provide(makeResolveOrganizationServices(dbHandle)),
454+ Effect.tap((sessionMeta) =>
455+ Effect.promise(() => self.saveSessionMeta(sessionMeta)).pipe(
456+ Effect.withSpan("mcp.session.save_meta"),
457+ ),
458+ ),
459+ Effect.ensuring(Effect.promise(() => dbHandle.end())),
460+ );
449461 }).pipe(Effect.withSpan("mcp.session.resolve_and_store_meta"));
450462 }
451463
@@ -463,6 +475,7 @@ export class McpSessionDO extends DurableObject { · 11 unmodified lines
463475 }),
464476 (eff) => withIncomingParent(incoming, eff),
465477 Effect.provide(DoTelemetryLive),
478+ // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: Durable Object init method can only reject its Promise
466479 Effect.orDie,
467480 ),
468481 );
@@ -511,6 +524,7 @@ export class McpSessionDO extends DurableObject { · 42 unmodified lines
511524 return yield* Effect.failCause(cause);
512525 }),
513526 ),
527+ // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: doInit is called only from Promise-only Durable Object init
514528 Effect.orDie,
515529 );
516530 }
apps/cloud/src/mcp.ts+28−24
@@ -16,7 +16,7 @@ · 15 unmodified lines
1616
1717 import { env } from "cloudflare:workers";
1818 import { HttpEffect, HttpServerRequest, HttpServerResponse } from "effect/unstable/http";
19−import { Cause, Context, Effect, Layer, Option, Schema } from "effect";
19+import { Cause, Context, Effect, Layer, Option, Predicate, Result, Schema } from "effect";
2020
2121 import { createCachedRemoteJWKSet } from "./jwks-cache";
2222 import { captureCause } from "./observability";
@@ -171,7 +171,7 @@ export const McpAuthLive = Layer.succeed(McpAuth)({ · 148 unmodified lines
171171 }),
172172 );
173173 if (!verified) return mcpUnauthorized("invalid_token", "The access token is invalid");
174− if ("_tag" in verified) return verified;
174+ if (Predicate.isTagged(verified, "Unauthorized")) return verified;
175175 if (!verified.accountId) {
176176 yield* Effect.annotateCurrentSpan({ "mcp.auth.outcome": "missing_subject" });
177177 return mcpUnauthorized("invalid_token", "The access token is invalid");
@@ -211,8 +211,7 @@ type CfRequestMetadata = { · 33 unmodified lines
211211 const requestWithCf = (request: Request): Request & { cf?: CfRequestMetadata } =>
212212 request as Request & { cf?: CfRequestMetadata };
213213
214−const getCfMeta = (request: Request): CfRequestMetadata =>
215− requestWithCf(request).cf ?? {};
214+const getCfMeta = (request: Request): CfRequestMetadata => requestWithCf(request).cf ?? {};
216215
217216 const HEADERS_TO_DUMP = [
218217 "accept",
@@ -285,22 +284,28 @@ const InitializeParams = Schema.Struct({ · 66 unmodified lines
285284 const NamedParams = Schema.Struct({ name: Schema.optional(Schema.String) });
286285 const UriParams = Schema.Struct({ uri: Schema.optional(Schema.String) });
287286
288−const decodeJsonRpcEnvelope = Schema.decodeUnknownOption(JsonRpcEnvelope);
287+const decodeJsonRpcEnvelopeString = Schema.decodeUnknownOption(
288+ Schema.fromJsonString(JsonRpcEnvelope),
289+);
289290 const decodeInitializeParams = Schema.decodeUnknownOption(InitializeParams);
290291 const decodeNamedParams = Schema.decodeUnknownOption(NamedParams);
291292 const decodeUriParams = Schema.decodeUnknownOption(UriParams);
292293 const decodeElicitationReplyResult = Schema.decodeUnknownOption(ElicitationReplyResult);
293294
295+const isMcpAuthorized = (value: McpAuthResult): value is McpAuthorizedResult =>
296+ Predicate.isTagged(value, "Authorized");
297+const isMcpUnauthorized = (value: McpAuthResult): value is McpUnauthorizedResult =>
298+ Predicate.isTagged(value, "Unauthorized");
299+
294300 const readJsonRpcEnvelope = (request: Request): Effect.Effect<Option.Option<JsonRpcEnvelope>> =>
295− Effect.promise(async () => {
296− try {
297− const text = await request.clone().text();
298− if (!text) return Option.none();
299− return decodeJsonRpcEnvelope(JSON.parse(text));
300− } catch {
301− return Option.none();
302− }
303− }).pipe(Effect.withSpan("mcp.request.read_json_rpc"));
301+ Effect.tryPromise({
302+ try: () => request.clone().text(),
303+ catch: () => undefined,
304+ }).pipe(
305+ Effect.map((text) => (text ? decodeJsonRpcEnvelopeString(text) : Option.none())),
306+ Effect.catchCause(() => Effect.succeed(Option.none())),
307+ Effect.withSpan("mcp.request.read_json_rpc"),
308+ );
304309
305310 const methodAttrs = (envelope: JsonRpcEnvelope): Record<string, unknown> => {
306311 const params = envelope.params ?? {};
@@ -409,15 +414,14 @@ const protectedResourceMetadata = Effect.sync(() => · 102 unmodified lines
409414 }),
410415 );
411416
412−const authorizationServerMetadata = Effect.promise(async () => {
413− try {
417+const authorizationServerMetadata = Effect.tryPromise({
418+ try: async () => {
414419 const res = await fetch(`${AUTHKIT_DOMAIN}/.well-known/oauth-authorization-server`);
415420 if (!res.ok) return jsonResponse({ error: "upstream_error" }, 502);
416421 return jsonResponse(await res.json());
417− } catch {
418− return jsonResponse({ error: "upstream_error" }, 502);
419− }
420−});
422+ },
423+ catch: () => undefined,
424+}).pipe(Effect.catchCause(() => Effect.succeed(jsonResponse({ error: "upstream_error" }, 502))));
421425
422426 // ---------------------------------------------------------------------------
423427 // DO dispatch
@@ -545,7 +549,7 @@ const authorizeMcpOrganization = ( · 121 unmodified lines
545549 Effect.catchCause((error) =>
546550 Effect.gen(function* () {
547551 yield* Effect.annotateCurrentSpan({
548− "mcp.auth.organization_authorize_error": String(error),
552+ "mcp.auth.organization_authorize_error": Cause.pretty(error),
549553 });
550554 return false;
551555 }),
@@ -662,7 +666,7 @@ export const mcpApp: Effect.Effect< · 110 unmodified lines
662666 const auth = yield* McpAuth;
663667 const authResult = yield* auth.verifyBearer(request).pipe(Effect.result);
664668
665− if (authResult._tag === "Failure") {
669+ if (Result.isFailure(authResult)) {
666670 yield* annotateMcpRequest(request, {
667671 token: null,
668672 parseBody: request.method === "POST",
@@ -675,11 +679,11 @@ export const mcpApp: Effect.Effect< · 6 unmodified lines
675679 // POST bodies are JSON-RPC payloads worth parsing; GET (SSE) and DELETE
676680 // don't carry one.
677681 yield* annotateMcpRequest(request, {
678− token: authValue._tag === "Authorized" ? authValue.token : null,
682+ token: isMcpAuthorized(authValue) ? authValue.token : null,
679683 parseBody: request.method === "POST",
680684 });
681685
682− if (authValue._tag === "Unauthorized") {
686+ if (isMcpUnauthorized(authValue)) {
683687 return unauthorized(authValue, PROTECTED_RESOURCE_METADATA_URL);
684688 }
685689 const token = authValue.token;
apps/cloud/src/mcp/response-peek.ts+103−53
@@ -1,23 +1,57 @@
11 import * as Sentry from "@sentry/cloudflare";
2−import { Effect } from "effect";
2+import { Cause, Data, Effect, Exit, Option, Schema } from "effect";
33
44 import { jsonRpcWebResponse } from "./responses";
55
66 const SSE_PEEK_TIMEOUT_MS = 10_000;
77
8−type SandboxOutcome = {
9− readonly status?: string;
10− readonly error?: { readonly kind?: string; readonly message?: string };
11−};
12−
13−type JsonRpcResponseBody = {
14− readonly jsonrpc?: string;
15− readonly error?: { readonly code?: number; readonly message?: string };
16− readonly result?: {
17− readonly isError?: boolean;
18− readonly structuredContent?: SandboxOutcome;
19− };
20−};
8+class ResponseBodyTimeoutError extends Data.TaggedError("ResponseBodyTimeoutError")<{
9+ readonly timeoutMs: number;
10+}> {}
11+
12+class ResponseBodyReadError extends Data.TaggedError("ResponseBodyReadError") {}
13+
14+class McpInternalJsonRpcError extends Data.TaggedError("McpInternalJsonRpcError")<{
15+ readonly message: string;
16+}> {}
17+
18+const ResponseBodyTimeoutErrorData = Schema.Struct({
19+ _tag: Schema.Literal("ResponseBodyTimeoutError"),
20+ timeoutMs: Schema.Number,
21+});
22+const decodeResponseBodyTimeoutError = Schema.decodeUnknownOption(ResponseBodyTimeoutErrorData);
23+
24+const SandboxOutcomeSchema = Schema.Struct({
25+ status: Schema.optional(Schema.String),
26+ error: Schema.optional(
27+ Schema.Struct({
28+ kind: Schema.optional(Schema.String),
29+ message: Schema.optional(Schema.String),
30+ }),
31+ ),
32+});
33+
34+const JsonRpcResponseBodySchema = Schema.Struct({
35+ jsonrpc: Schema.optional(Schema.String),
36+ error: Schema.optional(
37+ Schema.Struct({
38+ code: Schema.optional(Schema.Number),
39+ message: Schema.optional(Schema.String),
40+ }),
41+ ),
42+ result: Schema.optional(
43+ Schema.Struct({
44+ isError: Schema.optional(Schema.Boolean),
45+ structuredContent: Schema.optional(SandboxOutcomeSchema),
46+ }),
47+ ),
48+});
49+
50+const decodeJsonRpcResponseBody = Schema.decodeUnknownOption(
51+ Schema.fromJsonString(JsonRpcResponseBodySchema),
52+);
53+
54+type JsonRpcResponseBody = typeof JsonRpcResponseBodySchema.Type;
2155
2256 const responseBodyShape = (body: string): string => {
2357 const trimmed = body.trimStart();
@@ -31,18 +65,18 @@ const responseBodyShape = (body: string): string => { · 7 unmodified lines
3165
3266 const parseFirstJsonRpc = (contentType: string, body: string): JsonRpcResponseBody | null => {
3367 if (!body) return null;
34− try {
35− if (contentType.includes("text/event-stream")) {
36− for (const line of body.split(/\r?\n/)) {
37− if (line.startsWith("data:")) return JSON.parse(line.slice(5).trimStart());
68+ if (contentType.includes("text/event-stream")) {
69+ for (const line of body.split(/\r?\n/)) {
70+ if (line.startsWith("data:")) {
71+ return Option.getOrNull(decodeJsonRpcResponseBody(line.slice(5).trimStart()));
3872 }
39− return null;
4073 }
41− if (contentType.includes("application/json")) return JSON.parse(body);
42− return null;
43− } catch {
4474 return null;
4575 }
76+ if (contentType.includes("application/json")) {
77+ return Option.getOrNull(decodeJsonRpcResponseBody(body));
78+ }
79+ return null;
4680 };
4781
4882 const jsonRpcResponseAttrs = (payload: JsonRpcResponseBody | null): Record<string, unknown> => {
@@ -52,8 +86,10 @@ const jsonRpcResponseAttrs = (payload: JsonRpcResponseBody | null): Record<strin · 3 unmodified lines
5286 if (err && typeof err === "object") {
5387 attrs["mcp.rpc.is_error"] = true;
5488 if (typeof err.code === "number") attrs["mcp.rpc.error.code"] = err.code;
55− if (typeof err.message === "string") {
56− attrs["mcp.rpc.error.message"] = err.message.slice(0, 500);
89+ // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: schema-decoded JSON-RPC error message is protocol telemetry
90+ const { message } = err;
91+ if (typeof message === "string") {
92+ attrs["mcp.rpc.error.message"] = message.slice(0, 500);
5793 }
5894 }
5995 if (payload.result?.isError === true) attrs["mcp.tool.result.is_error"] = true;
@@ -61,20 +97,14 @@ const jsonRpcResponseAttrs = (payload: JsonRpcResponseBody | null): Record<strin · 1 unmodified line
6197 if (structured && typeof structured.status === "string") {
6298 attrs["mcp.tool.sandbox.status"] = structured.status;
6399 if (structured.error?.kind) attrs["mcp.tool.sandbox.error.kind"] = structured.error.kind;
64− if (typeof structured.error?.message === "string") {
65− attrs["mcp.tool.sandbox.error.message"] = structured.error.message.slice(0, 500);
100+ const message = structured.error?.["message"];
101+ if (typeof message === "string") {
102+ attrs["mcp.tool.sandbox.error.message"] = message.slice(0, 500);
66103 }
67104 }
68105 return attrs;
69106 };
70107
71−class ResponseBodyTimeoutError extends Error {
72− constructor(readonly timeoutMs: number) {
73− super(`Timed out waiting for MCP response body after ${timeoutMs}ms`);
74− this.name = "ResponseBodyTimeoutError";
75− }
76−}
77−
78108 const readResponseText = async (response: Response, timeoutMs: number | null): Promise<string> => {
79109 if (timeoutMs === null) return await response.text();
80110
@@ -85,11 +115,14 @@ const readResponseText = async (response: Response, timeoutMs: number | null): P · 4 unmodified lines
85115 let timeout: ReturnType<typeof setTimeout> | undefined;
86116 const timeoutPromise = new Promise<never>((_, reject) => {
87117 timeout = setTimeout(() => {
118+ // oxlint-disable-next-line executor/no-promise-catch -- boundary: best-effort stream cancellation inside timeout callback
88119 void reader.cancel().catch(() => undefined);
89− reject(new ResponseBodyTimeoutError(timeoutMs));
120+ // oxlint-disable-next-line executor/no-promise-reject -- boundary: Promise.race timeout adapter for Web ReadableStream
121+ reject(new ResponseBodyTimeoutError({ timeoutMs }));
90122 }, timeoutMs);
91123 });
92124 const readPromise = (async () => {
125+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: Web stream reader cleanup must clear host timeout after success or failure
93126 try {
94127 let text = "";
95128 for (;;) {
@@ -125,9 +158,18 @@ const withoutBodyHeaders = (response: Response) => { · 29 unmodified lines
125158 });
126159 };
127160
161+const isResponseBodyTimeoutError = (error: unknown) =>
162+ Option.isSome(decodeResponseBodyTimeoutError(error));
163+
164+const responsePeekError = (error: unknown): ResponseBodyTimeoutError | ResponseBodyReadError =>
165+ Option.match(decodeResponseBodyTimeoutError(error), {
166+ onNone: () => new ResponseBodyReadError(),
167+ onSome: ({ timeoutMs }) => new ResponseBodyTimeoutError({ timeoutMs }),
168+ });
169+
128170 const responseReadFailure = (error: unknown) =>
129171 Effect.gen(function* () {
130− const timedOut = error instanceof ResponseBodyTimeoutError;
172+ const timedOut = isResponseBodyTimeoutError(error);
131173 yield* Effect.annotateCurrentSpan({
132174 "mcp.response.status_code": timedOut ? 504 : 500,
133175 "mcp.response.content_type": "application/json",
@@ -135,7 +177,7 @@ const responseReadFailure = (error: unknown) => · 1 unmodified line
135177 "mcp.response.body.length": 0,
136178 "mcp.response.jsonrpc.detected": true,
137179 "mcp.peek_response.timed_out": timedOut,
138− "mcp.peek_response.error": String(error),
180+ "mcp.peek_response.error": timedOut ? "ResponseBodyTimeoutError" : "ResponseBodyReadError",
139181 });
140182 return jsonRpcWebResponse(
141183 timedOut ? 504 : 500,
@@ -149,8 +191,8 @@ const responseReadFailure = (error: unknown) => · 7 unmodified lines
149191 const reportInternalJsonRpcError = (payload: JsonRpcResponseBody | null) =>
150192 Effect.sync(() => {
151193 if (payload?.error?.code !== -32603) return;
152− const msg = payload.error.message ?? "unknown";
153− Sentry.captureException(new Error(`MCP internal error (-32603): ${msg}`));
194+ const message = payload.error["message"] ?? "unknown";
195+ Sentry.captureException(new McpInternalJsonRpcError({ message }));
154196 });
155197
156198 export const peekAndAnnotate = (response: Response): Effect.Effect<Response> =>
@@ -167,21 +209,29 @@ export const peekAndAnnotate = (response: Response): Effect.Effect<Response> => · 10 unmodified lines
167209
168210 const isSseResponse = contentType.includes("text/event-stream");
169211 const timeoutMs = isSseResponse ? SSE_PEEK_TIMEOUT_MS : null;
170− const textResult = yield* Effect.result(Effect.tryPromise({
171− try: () => readResponseText(response, timeoutMs),
172− catch: (error) => error,
173− }).pipe(
174− Effect.withSpan("mcp.peek_response", {
175− attributes: {
176− "http.response.content_type": contentType,
177− "http.response.status_code": response.status,
178− "mcp.peek_response.timeout_ms": timeoutMs ?? 0,
179− },
180− }),
181− ));
182− if (textResult._tag === "Failure") return yield* responseReadFailure(textResult.failure);
183−
184− const text = textResult.success;
212+ const textExit = yield* Effect.exit(
213+ Effect.tryPromise({
214+ try: () => readResponseText(response, timeoutMs),
215+ catch: responsePeekError,
216+ }).pipe(
217+ Effect.withSpan("mcp.peek_response", {
218+ attributes: {
219+ "http.response.content_type": contentType,
220+ "http.response.status_code": response.status,
221+ "mcp.peek_response.timeout_ms": timeoutMs ?? 0,
222+ },
223+ }),
224+ ),
225+ );
226+ if (Exit.isFailure(textExit)) {
227+ const error = Option.getOrElse(
228+ Cause.findErrorOption(textExit.cause),
229+ () => new ResponseBodyReadError(),
230+ );
231+ return yield* responseReadFailure(error);
232+ }
233+
234+ const text = textExit.value;
185235 const payload = parseFirstJsonRpc(contentType, text);
186236 yield* Effect.annotateCurrentSpan({
187237 "mcp.response.status_code": response.status,