import type { ComponentApi } from "../component/_generated/component.js";
import {
  buildProtectedResourceMetadataUrl,
  isDeliberateConvexError,
  AUTHORIZER_INVALID_SHAPE_REASON,
  parseAuthorizerDecision,
  type McpAuthorizerArgs,
  type McpAuthorizerDecision,
  type McpAuthorizerHandler,
  type McpBeforeResourceReadHandler,
  type McpCompleteReadResult,
  type McpDeclineReadResult,
  type McpIcon,
  type McpInputRequiredFallback,
  type McpInputRequiredResult,
  type McpServerInfo,
  type McpToolRegistration,
  mcpTaskSupportLevel,
} from "../shared.js";

/**
 * Browser-based MCP clients (e.g. anything served from a webapp
 * origin) issue a CORS preflight before each `/mcp/` call. Set this
 * option to enable preflight handling and the matching response
 * headers; non-browser clients (CLIs, server-to-server) work without
 * it.
 *
 * - `true`, permissive: `Access-Control-Allow-Origin: *`,
 *   `Access-Control-Allow-Credentials: false` (the spec forbids
 *   credentials with the wildcard origin). Tokens are passed via
 *   `Authorization: Bearer ...` so this works for OAuth flows.
 * - `string` / `string[]`, exact-match allowlist of origins. The
 *   request's `Origin` header is echoed back if it matches, otherwise
 *   no CORS headers are emitted (the browser then blocks the call).
 * - `(origin: string) => boolean`, custom matcher for things like
 *   subdomain wildcards or per-tenant rules.
 *
 * `Mcp-Session-Id` is automatically exposed via
 * `Access-Control-Expose-Headers` so JS clients can read it after
 * `initialize`.
 *
 * **Production note**: `cors: true` makes response bodies readable
 * from any origin. The gateway carries auth via `Authorization:
 * Bearer ...` (never cookies), so wildcard CORS does not transmit
 * the user's credentials, but a webapp running in the user's
 * browser with the Bearer in its own state can read responses
 * cross-origin. Prefer an explicit allowlist
 * (`cors: ["https://app.example.com"]`) for any deployment with
 * non-trivial auth coupling.
 */
export type McpCorsOption =
  | true
  | string
  | string[]
  | ((origin: string) => boolean);

/**
 * Optional Bearer-token validator for `handleMcpRequest`. When set,
 * the gateway calls this BEFORE `ctx.auth.getUserIdentity()` and uses
 * its return value as the identity for the audit row and as a hint
 * for the authorize callback (via `args.identity`).
 *
 * Useful when the upstream IdP issues opaque access tokens that
 * Convex's local JWT validation can't verify, typical pattern is
 * to call the IdP's userinfo endpoint:
 *
 * ```ts
 * resolveIdentity: async (token) => {
 *   const r = await fetch("https://id.example.com/api/oidc/userinfo", {
 *     headers: { Authorization: `Bearer ${token}` },
 *   });
 *   if (!r.ok) return null;
 *   const u = await r.json();
 *   return { subject: u.sub, claims: u };
 * }
 * ```
 *
 * Returning `null` means "token rejected" (treated identically to
 * "no token at all"). Throwing is treated as null with a warning
 * logged, rejection is not an error condition.
 *
 * When this option is omitted, the gateway falls back to
 * `ctx.auth.getUserIdentity()` (which only handles JWTs validated
 * by your `auth.config.ts`).
 */
export type McpIdentityResolver = (
  token: string,
) => Promise<{ subject: string; claims?: Record<string, unknown> } | null>;

/**
 * MCP resource/content annotations. All fields optional:
 * - `audience`: who the resource is for (`"user"` and/or `"assistant"`).
 * - `priority`: importance from `0` (least) to `1` (most).
 * - `lastModified`: timestamp of the last change, conventionally ISO 8601.
 *   Validated only as a string; the date format is not enforced.
 */
export type McpResourceAnnotations = {
  audience?: ("user" | "assistant")[];
  priority?: number;
  lastModified?: string;
};

export type McpResource = {
  uri: string;
  name: string;
  /** Human-friendly display name; falls back to `name` in clients. */
  title?: string;
  description?: string;
  mimeType?: string;
  annotations?: McpResourceAnnotations;
  icons?: McpIcon[];
  /** Raw size in bytes, if known. */
  size?: number;
};

export type McpResourceContent = {
  uri: string;
  mimeType?: string;
  text?: string;
  blob?: string;
};

/**
 * The caller a resource provider or template read handler sees.
 *
 * `null` only on a mount that set `anonymousResources`. Without that
 * option every one of these calls carries a principal, exactly as before
 * the option existed, so a provider that does not serve anonymous callers
 * can narrow once and carry on.
 */
export type McpCallerIdentity = {
  subject: string;
  claims?: Record<string, unknown>;
};

export type McpResourceCaller = McpCallerIdentity | null;

/**
 * The three resource methods, in the vocabulary the audit log already
 * uses. Named because it appears in three places: the `operation` an
 * anonymous authorizer decision carries, the `resourceOperation` on an
 * audit row, and the helper that builds the authorizer input. (The
 * `auditResources` OPTION spells the middle one `templatesList`; that is
 * a host-facing config key rather than this vocabulary, and it stays.)
 */
export type McpResourceOperation = "list" | "templates_list" | "read";

export type McpResourceProvider = {
  name: string;
  list: (
    ctx: McpHandlerCtx,
    args: { identity: McpResourceCaller },
  ) => Promise<McpResource[]>;
  read: (
    ctx: McpHandlerCtx,
    args: {
      uri: string;
      identity: McpResourceCaller;
    },
  ) => Promise<McpResourceContent[] | null>;
};

/**
 * An RFC 6570 resource template advertised via `resources/templates/list`.
 * `uriTemplate` is a level-1 template (simple `{var}` placeholders, each
 * matching a single URI path segment); clients expand it to a concrete URI
 * and read it through `resources/read`.
 */
export type McpResourceTemplate = {
  uriTemplate: string;
  name: string;
  /** Human-friendly display name; falls back to `name` in clients. */
  title?: string;
  description?: string;
  mimeType?: string;
  annotations?: McpResourceAnnotations;
  icons?: McpIcon[];
};

/**
 * Server-side read handler for a resource template: invoked when a
 * `resources/read` URI matches the template, with the extracted template
 * variables in `params`. Returns `null` to decline the URI (a later
 * template or a not-found is then used).
 */
export type McpResourceTemplateReadHandler = (
  ctx: McpHandlerCtx,
  args: {
    uri: string;
    params: Record<string, string>;
    identity: McpResourceCaller;
  },
) => Promise<McpResourceContent[] | null>;

/**
 * Runtime form of a resource template, as produced by
 * `defineMcpResourceTemplate`. `match` returns the extracted template
 * variables when a concrete URI matches `template.uriTemplate`, or `null`
 * when it doesn't. `read` is optional: present means the gateway resolves
 * expanded-URI reads server-side; absent means the template is
 * listing-only (the client reads the expansion via another provider).
 */
export type McpResourceTemplateProvider = {
  template: McpResourceTemplate;
  match: (uri: string) => Record<string, string> | null;
  read?: McpResourceTemplateReadHandler;
};

/**
 * The authorizer input for a caller the gateway has authenticated. Every
 * resource method resolves to one of these unless the mount opted into
 * `anonymousResources`.
 */
export interface McpIdentifiedResourceAuthorizerArgs {
  /**
   * `"resource_list"` when filtering `resources/list`,
   * `"resource_read"` before a `resources/read` handler runs,
   * `"resource_templates_list"` when filtering `resources/templates/list`
   * (here `resourceUri` carries the template's `uriTemplate`).
   *
   * Note on templates: `resources/read` of a template-expanded URI is
   * authorized under `"resource_read"` with the **concrete expanded URI**
   * (e.g. `weather://london/current`), not the `uriTemplate`, and with
   * `resourceMetadata: null`. So a template hidden at list time
   * (`"resource_templates_list"` → denied) is NOT automatically unreadable:
   * `"resource_read"` is the read gate for both concrete and template URIs.
   * Enforce read access in the `resource_read` branch (match the URI shape)
   * and/or inside the template's own `read` handler.
   */
  mode: "resource_list" | "resource_read" | "resource_templates_list";
  resourceUri: string;
  /**
   * Free-form metadata attached to a registered resource. Runtime-only
   * provider resources that are not present in the registry pass `null`.
   */
  resourceMetadata: unknown;
  /**
   * The caller's identity resolved once at the gateway boundary. These
   * three modes run only for an authenticated caller, so this is non-null
   * when the callback runs, and the union enforces that: a null identity
   * on one of these modes does not typecheck.
   */
  identity: McpCallerIdentity;
}

/**
 * The authorizer input for an UNAUTHENTICATED caller. Only reachable on a
 * mount that set `anonymousResources: true`; without it the gateway
 * refuses anonymous resource requests before the authorizer runs and this
 * variant never occurs.
 *
 * It is a mode of its own rather than the three above with a null
 * `identity` so that an existing authorizer meets an unrecognised mode
 * rather than a familiar one with a surprising argument. What happens
 * next is that authorizer's default branch, and the two common shapes
 * differ:
 *
 *   - one that returns nothing for an unknown mode denies, because
 *     `parseAuthorizerDecision` reads a missing decision as a denial
 *   - one that ends in `return { allowed: true }` ALLOWS, and publishes
 *     whatever the anonymous caller asked for
 *
 * So this is a smaller guarantee than "cannot accept one by accident":
 * the lock is `anonymousResources` itself, which no existing mount has
 * set. Read your default branch before setting it.
 */
export interface McpAnonymousResourceAuthorizerArgs {
  mode: "resource_anonymous";
  /**
   * Which resource method the anonymous caller is attempting. One mode
   * keeps the "do I serve anonymous callers at all" decision in a single
   * branch; this field is there for a host that wants to allow anonymous
   * listing without allowing anonymous reads, or the reverse.
   *
   * `"list"` and `"read"` carry a concrete resource URI, `"templates_list"`
   * carries the template's `uriTemplate`.
   */
  operation: McpResourceOperation;
  resourceUri: string;
  /**
   * Free-form metadata attached to a registered resource, exactly as on an
   * authenticated call. `null` for a runtime-only provider resource, for
   * every template, and for a read whose URI matched a template rather
   * than a registered resource, so a public template must be recognised by
   * its URI shape rather than by metadata.
   */
  resourceMetadata: unknown;
  /** Always `null`. The discriminant is `mode`; this documents the fact. */
  identity: null;
}

/**
 * What `authorizeResource` receives. Narrow on `mode` before reading
 * `identity`: it is non-null for the three authenticated modes and null
 * for `"resource_anonymous"`.
 */
export type McpResourceAuthorizerArgs =
  | McpIdentifiedResourceAuthorizerArgs
  | McpAnonymousResourceAuthorizerArgs;

export type McpResourceAuthorizerHandler = (
  ctx: McpHandlerCtx,
  args: McpResourceAuthorizerArgs,
) => Promise<McpAuthorizerDecision> | McpAuthorizerDecision;

export type McpResourceAuditOption =
  | boolean
  | {
      list?: boolean;
      read?: boolean;
      templatesList?: boolean;
    };

export type McpMrtrOptions = {
  /** At least 32 bytes of private, stable key material for HMAC-SHA-256. */
  secret: string;
  /** Maximum continuation lifetime. Defaults to five minutes. */
  ttlMs?: number;
};

/**
 * The snapshot handed to a host task executor and to the update hooks.
 * `identity` is the caller resolved when the task was created; `args`
 * are the public tool arguments (identity/reserved keys already
 * stripped); `idempotencyKey` is issued once per task and must be
 * persisted by the tool around its side effect.
 */
export type McpTaskContext = {
  taskId: string;
  toolName: string;
  toolKind: "query" | "mutation" | "action";
  args: Record<string, unknown>;
  identity: { subject: string; claims?: Record<string, unknown> };
  idempotencyKey: string;
  expiresAt: number;
};

/**
 * Starts durable execution for a freshly created task, e.g.
 * `workflow.start(ctx, internal.tasks.runArchive, {...})` with
 * `@convex-dev/workflow`. It must only *start* the work and return;
 * the execution itself finalizes the task later via
 * `gateway.completeTask` / `gateway.failTask` (or pauses it via
 * `gateway.requireTaskInput`). A throw here fails the task immediately.
 */
export type McpTaskExecutor = (
  ctx: McpHandlerCtx,
  task: McpTaskContext,
) => Promise<void> | void;

export type McpTasksOptions = {
  /**
   * Host-owned durable execution. Omit it to use the built-in
   * scheduled executor, which runs the registered tool function once
   * and completes/fails the task (no retries, no input rounds).
   */
  execute?: McpTaskExecutor;
  /**
   * Whether THIS call should become a task, for a tool registered
   * `taskSupport: "optional"`. SEP-2663 makes the decision the server's:
   * a client opts in once through the extension capability and sends no
   * per-request flag, so something on this side has to choose, and the
   * knowledge of "is this one going to be slow" lives in the host rather
   * than in the gateway.
   *
   * Omit it and every eligible call becomes a task, which is what the
   * spec's own conformance scenario requires of a task-supporting tool.
   * Return `false` to answer inline instead; the spec allows that
   * explicitly for a fast operation.
   *
   * Only consulted when a task is possible at all: the tool says
   * `optional`, the client declared the extension, and the caller is
   * authenticated. A `required` tool never reaches it, and neither does
   * a call this mount's `authorize` denied.
   */
  shouldCreate?: (
    ctx: McpHandlerCtx,
    call: {
      toolName: string;
      toolKind: "query" | "mutation" | "action";
      args: Record<string, unknown>;
      identity: { subject: string; claims?: Record<string, unknown> };
    },
  ) => Promise<boolean> | boolean;
  /**
   * Called after a `tasks/update` accepted MRTR-shaped `inputResponses`
   * for an `input_required` task (now back in `working`). Hosts with a
   * custom `execute` resume their workflow here. Best-effort AND
   * at-least-once: a throw is logged and the update still succeeds (the
   * responses are already durably stored on the task row), and an
   * idempotent duplicate update re-fires the hook so a client that
   * re-sends the same responses retries a notification that previously
   * failed. The hook MUST therefore tolerate repeats.
   */
  onInputResponses?: (
    ctx: McpHandlerCtx,
    event: {
      taskId: string;
      toolName: string;
      inputResponses: Record<string, unknown>;
    },
  ) => Promise<void> | void;
  /**
   * Called after an owner cancellation, including idempotent repeats:
   * re-sending the cancel is the retry path for a notification that
   * previously threw, so the hook MUST tolerate being called for an
   * already-cancelled task. Hosts cancel their workflow run here.
   */
  onCancel?: (
    ctx: McpHandlerCtx,
    event: { taskId: string; toolName: string },
  ) => Promise<void> | void;
  /**
   * Default task retention. A task row (and with it the result) expires
   * `retentionMs` after creation; expired tasks answer like unknown ids
   * and are dropped by `gateway.pruneTasks`. Clamped to
   * [1 minute, 7 days]; defaults to 24 hours. A client may request a
   * shorter `ttlMs` per call, clamped the same way.
   */
  retentionMs?: number;
  /**
   * Identifies THIS mount for task ownership. Stored on every task the
   * mount creates, and required to match on every `tasks/get` /
   * `tasks/update`; a mismatch answers exactly like an unknown task id.
   *
   * Set it whenever the gateway is mounted more than once with different
   * `authorize` policies over the same identity namespace. The task table
   * is component-wide and `authorize` runs only at creation, so without a
   * scope a caller permitted on a broad mount can start a privileged task
   * there and collect its result through a narrower one: bypassing the
   * narrower mount's policy without any bug in it.
   *
   * A sealed MRTR `requestState` is not bound to the mount, so two mounts
   * sharing `mrtr.secret` both accept the same continuation and each ends
   * up owning its own task row for that chain (their rows are
   * scope-isolated). The tool still dedupes, because both runs receive the
   * same chain key in `mrtrArgs`. Give differently-scoped mounts different
   * secrets if you want continuations to be non-transferable.
   *
   * Unset (the default) keeps the pre-scope behaviour, so a single-mount
   * host needs no migration. Adopting it later only affects tasks created
   * from then on: rows already in flight have no scope and stay visible
   * only to unscoped mounts until they expire.
   */
  scope?: string;
  /** Advertised polling interval hint, defaults to 2000 ms. */
  pollIntervalMs?: number;
};

/**
 * Origin allowlist for `handleMcpRequest`. MCP Streamable HTTP requires
 * servers to validate the `Origin` header to prevent DNS-rebinding
 * attacks: a request whose `Origin` is present but not allowed is
 * rejected with HTTP 403 before identity resolution, authorization,
 * auditing, or dispatch. Requests without an `Origin` header (every
 * CLI and server-to-server client) are unaffected.
 *
 * - `string` / `string[]`, exact-match allowlist of origins.
 * - `(origin: string) => boolean`, custom matcher for subdomain
 *   wildcards or per-tenant rules.
 *
 * This is deliberately independent of `cors`. CORS is a browser
 * mechanism that decides what a browser is allowed to *read*;
 * `allowedOrigins` is an authorization gate that decides what the
 * gateway is willing to *serve*. Coupling the two makes the permissive
 * `cors: true` silently disable the origin gate, so they are separate
 * options.
 *
 * **Omitting this option disables origin validation entirely.** That is
 * the default because a Convex deployment is reachable at a fixed
 * public URL rather than on localhost, which is the DNS-rebinding
 * scenario the requirement targets. Set it for any deployment that
 * serves browser clients.
 */
export type McpAllowedOriginsOption =
  | string
  | string[]
  | ((origin: string) => boolean);

/**
 * Options for `gateway.handleMcpRequest`. The host supplies an
 * `authorize` callback that decides allowed vs denied per
 * `tools/call` and per tool in a filtered `tools/list`. The callback
 * runs in the host's HTTP-action context, so it has the host's
 * `ctx.auth` and can call `ctx.auth.getUserIdentity()` directly.
 */
export interface HandleMcpRequestOptions {
  authorize: McpAuthorizerHandler;
  /** See `McpCorsOption`. Omit for non-browser-only deployments. */
  cors?: McpCorsOption;
  /**
   * See `McpAllowedOriginsOption`. Omit to disable origin validation.
   */
  allowedOrigins?: McpAllowedOriginsOption;
  /**
   * See `McpIdentityResolver`. Omit to use Convex's built-in JWT
   * validation via `ctx.auth.getUserIdentity()`.
   */
  resolveIdentity?: McpIdentityResolver;
  /**
   * Override the `serverInfo` returned in the `initialize` response and
   * in the `_meta` of every stateless result. Defaults to this package's
   * own name and version; that constant is intentionally static, because
   * Convex doesn't expose `package.json` to the runtime.
   *
   * Supplying this replaces the whole block rather than merging into it,
   * so a host that only wants to add an icon still restates `name` and
   * `version`. Beyond those two, the spec's `Implementation` carries
   * `title`, `description`, `websiteUrl` and `icons` for hosts that
   * white-label or want telemetry-grade version reporting.
   *
   * Two things to know before adding `icons` here, both measured rather
   * than reasoned about.
   *
   * A SPEC-CORRECT icons block takes the connection down on SDK 1.18.0
   * through 1.18.2, and `describeServerInfoProblem` cannot stop it. Those
   * builds typed `Icon.sizes` as a bare string, so `sizes: ["48x48"]`,
   * which is what the spec mandates and what the validator rightly
   * accepts, fails their parse of the whole `InitializeResult`
   * ("Expected string, received array at serverInfo.icons.0.sizes").
   * Omitting `sizes` is accepted by every build we have. The validator
   * guards the MALFORMED shape, which is a different hazard: it stops the
   * gateway emitting a block a spec-conformant client would reject. The
   * old-client hazard is a client-side bug and the only lever here is not
   * sending `sizes`. Note the escalation over the same field on a tool:
   * there it costs one `tools/list`, here the client never connects.
   *
   * Keep it small, and prefer an `https:` src over a `data:` one. Unlike
   * a tool descriptor, this block is repeated on every stateless result,
   * `tools/call` and `resources/read` and every `tasks/get` poll
   * included, not just the handshake. Measured with the smallest real
   * inline icon there is, a 1x1 transparent PNG: `name` + `version`
   * alone is 47 bytes, the same block with two data-URI icons is 534, so
   * 11x per result. A realistic 48x48 icon is several KB.
   */
  serverInfo?: McpServerInfo;
  /**
   * Challenge anonymous requests with `401` instead of letting them
   * through to `initialize` / `tools/list`. Default `false`.
   *
   * Leave this off for **mixed** servers (some tools `public`,
   * some private): anonymous callers should still see the public
   * catalog, so the default 200-with-filtered-list is correct.
   *
   * Turn it on for **all-private** servers that browser MCP clients
   * (claude.ai) connect to. Such a client only does `initialize` +
   * `tools/list` when a connector is added; with the default both
   * return 200 (an empty, authorize-filtered list), so the client
   * concludes "connected, no tools" and never starts the OAuth flow,
   * its only trigger is a `401` + `WWW-Authenticate`. With
   * `requireAuth: true` an anonymous POST gets that 401, so the login
   * is prompted and discovery begins.
   *
   * Needs `setOAuthConfig` to have run so the `WWW-Authenticate`
   * header can carry the protected-resource metadata URL. If
   * `requireAuth` is set but no OAuth config exists, the gate still
   * returns 401, but without the header (and `console.warn`s once);
   * browser clients can't begin discovery until `setOAuthConfig` is
   * called.
   *
   * Applies to `POST` only. `GET` already 405s, `DELETE` is
   * identity-bound, and `OPTIONS` (CORS preflight) is left untouched.
   */
  requireAuth?: boolean;
  /**
   * Declarative tool catalog. When set, the registry is reconciled from
   * this list on `initialize` (change-detected, so an unchanged list is
   * a cheap no-op), and no separate registration mutation is needed.
   * Omit it to manage the registry yourself via `gateway.register(...)`.
   * Annotate an exported list with `McpToolRegistration[]` to avoid a
   * Convex codegen circular-type error (see that type's docs).
   */
  tools?: McpToolRegistration[];
  /**
   * Server-level guidance returned in the MCP `initialize` result's
   * `instructions` field (see the spec's `InitializeResult.instructions`).
   * Clients may hand this to the LLM to explain how to use the server as a
   * whole, e.g. "call `kira_load_skill` before answering", without bloating
   * individual tool descriptions. Omitted from the response entirely when
   * unset, so the default `initialize` shape is unchanged.
   *
   * Best-effort hint, not a guarantee: the spec says clients MAY add it to
   * the system prompt, and some ignore it entirely. Clients that honor it
   * tend to cap and front-truncate the text, so keep it short and put the
   * critical guidance first. Enforce hard constraints in each tool's
   * `authorize` / handler, never here.
   */
  initializeInstructions?: string;
  /**
   * Optional MCP resources exposed by this gateway. Resources are listed
   * in `initialize.capabilities.resources`, served via `resources/list`,
   * and read via `resources/read`. A provider receives the resolved
   * caller identity, or `null` on a mount that set `anonymousResources`,
   * which is also the only mount where an anonymous resource request is
   * served rather than refused.
   */
  resources?: McpResourceProvider[];
  /**
   * Optional MCP resource templates (RFC 6570) exposed by this gateway.
   * Advertised via `resources/templates/list` and, for templates declared
   * with a `read` handler, resolved server-side when `resources/read`
   * requests a URI that matches the template (concrete resources take
   * precedence). Build these with `defineMcpResourceTemplate`.
   */
  resourceTemplates?: McpResourceTemplateProvider[];
  /**
   * Optional central authorization hook for MCP resources. If omitted,
   * authenticated callers can list/read all resources exposed by providers.
   * If set, `resources/list` filters resources through `resource_list`, and
   * `resources/read` checks `resource_read` before invoking the provider.
   */
  authorizeResource?: McpResourceAuthorizerHandler;
  /**
   * Serve `resources/list`, `resources/templates/list` and
   * `resources/read` to unauthenticated callers, subject to
   * `authorizeResource`. Default `false`, and with it off the three
   * methods refuse an anonymous caller with `-32001` before the
   * authorizer runs, which is the behaviour every mount had before this
   * option existed.
   *
   * This is the resource counterpart of a public tool, but it is a
   * gateway option rather than the host-side `metadata.public`
   * convention tools use, because `authorizeResource` cannot be the only
   * lock: a mount that never configured one authorizes every resource by
   * default, so "let the authorizer decide" would silently publish the
   * whole catalog. Setting this without an `authorizeResource` therefore
   * throws on the first request through the mount; there is no
   * deploy-time hook to fail at.
   *
   * An anonymous caller reaches the authorizer under
   * `mode: "resource_anonymous"`, never under the three authenticated
   * modes. That keeps an existing policy from being applied to a caller
   * it was not written for, but it does not decide the outcome: an
   * authorizer whose default branch returns `{ allowed: true }` allows
   * the anonymous caller too. Audit that branch before opting in.
   *
   * Three things it deliberately does not do:
   *
   *   - `resources/subscribe` and `resources/unsubscribe` stay
   *     authenticated. A subscription is server-side state an anonymous
   *     caller could accumulate, and it buys a client nothing here: this
   *     transport does not push, so the host delivers
   *     `notifications/resources/updated` over its own channel.
   *   - It cannot be combined with `beforeResourceRead`, which throws on
   *     the first request. That hook's contract passes a non-null identity and an
   *     MRTR continuation must bind to a principal, the same reason a
   *     tool's `beforeCall` requires an authenticated caller.
   *   - It does not override `requireAuth`. That gate answers anonymous
   *     POSTs with `401` before the method switch, so a mount setting
   *     both serves no anonymous resource. Same as `requireAuth` with a
   *     public tool.
   *
   * Auditing a mount that serves anonymous callers wants a retention
   * cron. A failing anonymous outcome is never recorded, but a SUCCEEDING
   * one is, one row per request, so `gateway.pruneAuditEntries` is what
   * bounds the table. See `docs/audit-log.md`.
   */
  anonymousResources?: boolean;
  /**
   * Optional MRTR hook for `resources/read`: the read counterpart of a
   * tool's `beforeCall`. Runs after `authorizeResource` allowed the read
   * and before any provider or template is consulted, on the first read
   * AND on every verified continuation of it (where it additionally
   * receives the decoded `state`, the client's untrusted `inputResponses`,
   * and the `round`).
   *
   * Return `inputRequired()` to ask the client for input (per MCP
   * 2026-07-28, `resources/read` may answer with an `InputRequiredResult`),
   * optionally with `onUnsupported` for clients that cannot satisfy the
   * requested capabilities,
   * `completeRead(contents)` to serve content yourself, `declineRead(reason)`
   * to refuse after the answer, or `null` to fall through to the normal
   * read path.
   *
   * Mount-level rather than per-resource, mirroring `authorizeResource`: a
   * provider serves many URIs and the gateway cannot know which one owns a
   * URI without calling it, so the gate has to sit where the URI is known
   * and nothing has run yet. Branch on `uri` inside the hook.
   *
   * Requires the `mrtr` option (the continuation is sealed with its
   * secret) and the modern protocol; a read that demands input on a legacy
   * request fails closed rather than silently serving the resource.
   *
   * Incompatible with `anonymousResources`: the two together throw on the
   * first request through the mount, because this hook's contract passes
   * a non-null identity and the MRTR chain it can open must bind to a
   * principal.
   */
  beforeResourceRead?: McpBeforeResourceReadHandler;
  /**
   * Opt-in audit for MCP resource operations. Defaults to `false`.
   * `true` records `resources/list`, `resources/read`, and
   * `resources/templates/list`; the object form (`{ list, read,
   * templatesList }`) enables each operation independently. Resource
   * contents are never stored.
   */
  auditResources?: McpResourceAuditOption;
  /**
   * Opt-in MCP resource subscription support. **Off by default**, because
   * this gateway's HTTP transport is request-scoped and cannot push
   * server-initiated notifications (`notifications/resources/updated` /
   * `notifications/resources/list_changed`). With both flags off,
   * `initialize` advertises neither capability and `resources/subscribe` /
   * `resources/unsubscribe` return a clear `-32601`.
   *
   * Set these flags ONLY when the host fronts the gateway with a transport
   * that CAN deliver notifications (its own SSE/WebSocket layer). The
   * gateway then advertises the capability and tracks subscribe/unsubscribe
   * state per session; the host owns delivery, it reads
   * `gateway.listResourceSubscribers(uri)` and ships payloads built with
   * `gateway.buildResourceUpdatedNotification` /
   * `gateway.buildResourceListChangedNotification`.
   *
   * - `subscribe`: advertise `capabilities.resources.subscribe` and handle
   *   `resources/subscribe` / `resources/unsubscribe`.
   * - `listChanged`: advertise `capabilities.resources.listChanged` (the
   *   host emits `notifications/resources/list_changed` itself when its
   *   catalog changes).
   */
  resourceSubscriptions?: {
    subscribe?: boolean;
    listChanged?: boolean;
  };
  /**
   * Opt-in support for stateless-era multi-round-trip requests (MRTR). A
   * declarative tool's host-side `beforeCall` hook is the state machine:
   * on the first call it can return `inputRequired(inputRequests, state)`
   * or provide an `onUnsupported` fallback for clients that cannot satisfy
   * the requested capabilities
   * before the underlying Convex function can run; on every verified
   * continuation it runs again with the decoded state, the client's
   * untrusted `inputResponses`, and the chain's stable idempotency key,
   * and decides whether to ask for another round, finish without
   * dispatching (`completeCall()`), or continue to the Convex function
   * (which stays MCP-unaware; only the idempotency key is injectable via
   * `mrtrArgs`). Continuations are HMAC-sealed, TTL-bound, bound to the
   * caller/tool/arguments, and redeemed once server-side so a captured
   * state cannot be replayed with different responses.
   */
  mrtr?: McpMrtrOptions;
  /**
   * Opt-in MCP Tasks (`io.modelcontextprotocol/tasks`). **Off by
   * default**; the capability is advertised in `server/discover` only
   * when this option is set, and only tools registered with
   * `taskSupport: true` accept a task-augmented stateless `tools/call`.
   *
   * Without `execute`, the gateway runs the tool once via the component's
   * built-in scheduled executor (durable across restarts, no retries) and
   * completes or fails the task. Hosts that need retry policy, delays,
   * or `input_required` rounds supply `execute` to start their own
   * durable execution (typically a `@convex-dev/workflow` run), and
   * finalize via `gateway.completeTask` / `failTask` /
   * `requireTaskInput`. See docs/tasks.md.
   */
  tasks?: McpTasksOptions;
}

/**
 * Internal handler options: the public `HandleMcpRequestOptions` plus the
 * catalog synchronizer that `McpGateway.handleMcpRequest` derives from the
 * declarative catalog options. Not exported, hosts never set it directly.
 */
type InternalHandleMcpRequestOptions = HandleMcpRequestOptions & {
  ensureCatalogSynced?: () => Promise<void>;
  declarativeTools?: McpToolRegistration[];
};

export type McpHandlerCtx = {
  runQuery: (ref: any, args: any) => Promise<any>;
  runMutation: (ref: any, args: any) => Promise<any>;
  runAction: (ref: any, args: any) => Promise<any>;
  auth: { getUserIdentity: () => Promise<any> };
};

type HandlerCtx = McpHandlerCtx;

type JsonRpcMessage = {
  jsonrpc?: "2.0";
  id?: string | number | null;
  method?: string;
  params?: Record<string, any>;
};

type RegisteredTool = {
  name: string;
  description: string;
  kind: "query" | "mutation" | "action";
  functionHandle: string;
  inputSchema: unknown;
  outputSchema?: unknown;
  /** See `advertisedSchema`: the wire form, when the row carries it. */
  authoredInputSchemaJson?: string;
  authoredOutputSchemaJson?: string;
  identityArg?: string;
  mrtrArgs?: {
    idempotencyKey: string;
  };
  mrtrGated?: boolean;
  taskSupport?: boolean;
  protocolMetadata?: {
    title?: string;
    annotations?: unknown;
    _meta?: unknown;
    securitySchemes?: unknown;
  };
  metadata?: unknown;
};

// A row from the resource registry. Narrower than `McpResource`: the
// registry persists only these catalog fields (the richer title/annotations/
// size are runtime-only and never stored), so the row type reflects that.
type RegisteredResource = {
  uri: string;
  name: string;
  description?: string;
  mimeType?: string;
  metadata?: unknown;
};

type ResourceCandidate = {
  resource: McpResource;
  metadata: unknown;
};

// A row from the resource-template registry. Unlike concrete resources,
// templates persist their full descriptor (incl. title/annotations).
type RegisteredResourceTemplate = {
  uriTemplate: string;
  name: string;
  title?: string;
  description?: string;
  mimeType?: string;
  annotations?: McpResourceAnnotations;
  /**
   * Persisted, unlike a concrete resource's icons: a registry-only template
   * must still list its full descriptor, which is the whole reason the
   * column exists. Declared here so narrowing this row through an
   * explicitly-typed literal cannot silently drop it.
   */
  icons?: McpIcon[];
};

/**
 * The four MCP revisions this gateway speaks, split by TRANSPORT MODEL
 * rather than by age.
 *
 * That is the axis the code actually branches on: a session-based
 * request carries an `Mcp-Session-Id` minted by `initialize`, a
 * stateless one carries its protocol metadata per request and creates
 * no session at all. Naming them for the property rather than for
 * recency is deliberate. Whichever revision is newest changes with every
 * release of the spec, so a name like "modern" describes the reader's
 * calendar instead of the code, and stops being true the day it matters
 * most. When a further stateless revision lands, it joins
 * `STATELESS_PROTOCOL_VERSIONS` and nothing else has to be renamed.
 */
const SESSION_PROTOCOL_VERSIONS = [
  "2025-11-25",
  "2025-06-18",
  "2025-03-26",
] as const;

/**
 * The stateless era. One entry today; kept as a list so a second
 * stateless revision is an addition rather than a rename.
 *
 * `STATELESS_PROTOCOL_VERSION` is the one a client is told to speak,
 * i.e. the newest we serve.
 */
const STATELESS_PROTOCOL_VERSIONS = ["2026-07-28"] as const;
const STATELESS_PROTOCOL_VERSION = STATELESS_PROTOCOL_VERSIONS[0];

/**
 * What `initialize` may negotiate, which is the session-based set ONLY.
 *
 * The exclusion is the point, and it is why this is not called
 * "supported": a stateless request never runs `initialize`, it declares
 * its revision per request through the routing header plus `_meta`. A
 * client asking `initialize` for `2026-07-28` is therefore answered with
 * the newest SESSION revision, because that is the only kind of session
 * this call can open. Adding the stateless revision to this list would
 * hand out a session id for a transport that has no sessions.
 *
 * `2025-11-25`'s additions over `2025-06-18` are its optional SSE
 * resumability framing (which needs an event store + GET replay this
 * gateway does not have, so it is not emitted, see `sseResponseFrame`)
 * and additive capabilities (tasks, url-mode elicitation, not
 * advertised), so all three are served with one wire contract. Not
 * emitting optional features is conforming.
 */
const SESSION_NEGOTIABLE_VERSIONS = SESSION_PROTOCOL_VERSIONS;

/**
 * Answered when a client requests a revision we do not serve: per spec
 * the server replies with its own latest, which for a session handshake
 * is the newest session-based one.
 */
const DEFAULT_PROTOCOL_VERSION = SESSION_PROTOCOL_VERSIONS[0];

const MAX_MCP_HEADER_VALUE_LENGTH = 8 * 1024;

function hasMcpHeaderControlCharacter(value: string): boolean {
  for (let index = 0; index < value.length; index += 1) {
    const codePoint = value.charCodeAt(index);
    if (codePoint <= 0x1f || codePoint === 0x7f) return true;
  }
  return false;
}
const SERVER_NAME = "convex-mcp-gateway";
// Kept in step with package.json by release-please; the trailing
// annotation is what it looks for. A host overrides the whole block
// with `options.serverInfo`.
const SERVER_VERSION = "2.0.0"; // x-release-please-version

const UNAUTHORIZED = -32001;
const FORBIDDEN = -32003;
const INVALID_PARAMS = -32602;
const INTERNAL_ERROR = -32603;
const UNSUPPORTED_PROTOCOL_VERSION = -32022;
const MISSING_REQUIRED_CLIENT_CAPABILITY = -32021;
const HEADER_MISMATCH = -32020;

/** Advertised default `tasks/get` polling hint (milliseconds). */
const TASK_POLL_INTERVAL_MS = 2000;

/**
 * Default task retention ceiling applied when the host did not set
 * `tasks.retentionMs`. Mirrors the component's `TASK_DEFAULT_TTL_MS`
 * (not imported to keep component server code out of the host bundle).
 */
const TASK_DEFAULT_RETENTION_MS = 24 * 60 * 60 * 1000;

/** Capability / extension key for MCP Tasks. */
const TASKS_CAPABILITY_KEY = "io.modelcontextprotocol/tasks";

/**
 * Whether the caller declared MCP Tasks. SEP-2663 puts the declaration
 * under `extensions`, on both sides: the client nests it in the
 * `clientCapabilities` it sends per request, the server in the
 * capabilities `server/discover` returns.
 */
function declaresTasksExtension(
  clientCapabilities: Record<string, unknown> | null,
): boolean {
  const extensions = clientCapabilities?.extensions;
  return (
    isPlainObject(extensions) && isPlainObject(extensions[TASKS_CAPABILITY_KEY])
  );
}

/**
 * `tasks.shouldCreate`, with the host's throw contained.
 *
 * Every other host callback on this path is wrapped, and this one has no
 * envelope of its own to fall back on: thrown bare it escapes
 * `handlePost` as a raw 500 with no JSON-RPC body. A throw falls back to
 * the default, which is to create the task. That direction is the safe
 * one: a task is durable and pollable, while an inline dispatch of an
 * operation the host thought worth deferring can exceed the request
 * timeout and lose the result outright.
 */
async function safeShouldCreateTask(
  shouldCreate: McpTasksOptions["shouldCreate"],
  ctx: HandlerCtx,
  call: {
    toolName: string;
    toolKind: "query" | "mutation" | "action";
    args: Record<string, unknown>;
    identity: { subject: string; claims?: Record<string, unknown> };
  },
): Promise<boolean> {
  if (!shouldCreate) return true;
  try {
    return await shouldCreate(ctx, call);
  } catch (err) {
    console.error(
      "[mcp-gateway] tasks shouldCreate threw; creating the task anyway",
      call.toolName,
      err,
    );
    return true;
  }
}

/**
 * The `data.requiredCapabilities` of a `-32021`, which names only what
 * is MISSING and mirrors the shape the client would have had to send.
 */
const TASKS_REQUIRED_CAPABILITIES = {
  extensions: { [TASKS_CAPABILITY_KEY]: {} },
} as const;

/**
 * A component task descriptor as the wire wants it.
 *
 * SEP-2663 flattened the envelope: a `CreateTaskResult` is `Result &
 * Task` with no nested `task` key, timestamps are ISO-8601, and the
 * lifetime is a duration (`ttlMs`) rather than the absolute instant the
 * component stores. `pollIntervalMs` rides on the task rather than on
 * the capability, which is where a client is told how often to poll.
 *
 * `ttlMs` counts from now, not from creation, so a client that polls a
 * task twice is told how much longer it may keep polling rather than how
 * long it could have.
 */
function taskWireFields(
  descriptor: {
    createdAt: number;
    updatedAt: number;
    expiresAt: number;
    [key: string]: unknown;
  },
  pollIntervalMs: number,
): Record<string, unknown> {
  const { createdAt, updatedAt, expiresAt, ...rest } = descriptor;
  return {
    ...rest,
    createdAt: new Date(createdAt).toISOString(),
    lastUpdatedAt: new Date(updatedAt).toISOString(),
    ttlMs: Math.max(0, expiresAt - Date.now()),
    pollIntervalMs,
  };
}

/**
 * What the MCP client is told when host code threw instead of returning
 * a decision or declining cleanly.
 *
 * The gateway never puts an accidental exception message on the wire:
 * a thrown error can quote a signed URL, an `Authorization` header, a
 * provider response, or a connection string, and the caller is an LLM
 * (often relaying to a third party). `dispatch.runTool` has always done
 * this for tool execution; these constants extend the same rule to the
 * authorize callbacks and the resource paths.
 *
 * The full text is not lost: it goes to the audit row and to the Convex
 * deployment log, both server-side. Hosts that want a specific message
 * to reach the caller throw `ConvexError`, the deliberate channel.
 */
const GENERIC_AUTHORIZER_ERROR = "Authorization check failed";
const GENERIC_RESOURCE_READ_ERROR = "Resource read failed";
const GENERIC_RESOURCE_LIST_ERROR = "Resource listing failed";
const GENERIC_RESOURCE_TEMPLATES_LIST_ERROR =
  "Resource template listing failed";

function resolveCorsOrigin(
  cors: McpCorsOption | undefined,
  requestOrigin: string | null,
): string | null {
  if (cors === undefined) return null;
  if (cors === true) return "*";
  if (!requestOrigin) return null;
  if (typeof cors === "string") {
    return cors === requestOrigin ? requestOrigin : null;
  }
  if (Array.isArray(cors)) {
    return cors.includes(requestOrigin) ? requestOrigin : null;
  }
  return cors(requestOrigin) ? requestOrigin : null;
}

/**
 * MCP Streamable HTTP: "Servers MUST validate the `Origin` header on all
 * incoming connections to prevent DNS rebinding attacks. If the `Origin`
 * header is present and invalid, servers MUST respond with HTTP 403."
 *
 * Applies to both protocol eras and runs before identity resolution,
 * authorization, auditing, and dispatch. Returns `null` when the request
 * may proceed.
 */
function originRejection(
  request: Request,
  options: HandleMcpRequestOptions,
): Response | null {
  const allowed = options.allowedOrigins;
  if (allowed === undefined) return null;
  const origin = request.headers.get("origin");
  if (origin === null) return null;
  let ok: boolean;
  if (typeof allowed === "string") {
    ok = allowed === origin;
  } else if (Array.isArray(allowed)) {
    ok = allowed.includes(origin);
  } else {
    try {
      ok = allowed(origin);
    } catch (err) {
      // A host matcher written as `new URL(origin).hostname.endsWith(...)`
      // throws on the literal `Origin: null` that sandboxed iframes and
      // some redirects send. Fail closed rather than letting the throw
      // escape as an opaque 500 with no gateway-prefixed log line.
      console.error(
        `[mcp-gateway] allowedOrigins matcher threw for origin ${origin}; ` +
          `treating it as not allowed.`,
        err,
      );
      ok = false;
    }
  }
  if (ok) return null;
  // The spec allows a JSON-RPC error body with no id here. POST callers
  // speak JSON-RPC, so give them one; DELETE has no JSON-RPC envelope.
  const body =
    request.method === "POST"
      ? jsonErrorEnvelope(null, FORBIDDEN, "Forbidden: origin is not allowed")
      : "Forbidden: origin is not allowed";
  return new Response(body, {
    status: 403,
    headers:
      request.method === "POST" ? { "content-type": "application/json" } : {},
  });
}

function corsHeaders(
  cors: McpCorsOption | undefined,
  request: Request,
): Record<string, string> {
  const allowOrigin = resolveCorsOrigin(cors, request.headers.get("origin"));
  if (allowOrigin === null) return {};
  const headers: Record<string, string> = {
    "access-control-allow-origin": allowOrigin,
    "access-control-expose-headers": "mcp-session-id",
    vary: "Origin",
  };
  // The wildcard origin forbids credentials per spec; with an exact
  // origin we leave credentials off too because MCP carries auth via
  // Bearer tokens, not cookies.
  return headers;
}

function preflightResponse(
  cors: McpCorsOption | undefined,
  request: Request,
): Response {
  const baseHeaders = corsHeaders(cors, request);
  if (Object.keys(baseHeaders).length === 0) {
    // CORS not configured for this origin, let the browser block it.
    return new Response(null, { status: 204 });
  }
  const requestedHeaders =
    request.headers.get("access-control-request-headers") ??
    "content-type, authorization, mcp-session-id, accept";
  return new Response(null, {
    status: 204,
    headers: {
      ...baseHeaders,
      "access-control-allow-methods": "POST, GET, DELETE, OPTIONS",
      "access-control-allow-headers": requestedHeaders,
      "access-control-max-age": "86400",
    },
  });
}

function withCors(
  response: Response,
  cors: McpCorsOption | undefined,
  request: Request,
): Response {
  const extra = corsHeaders(cors, request);
  if (Object.keys(extra).length === 0) return response;
  const merged = new Headers(response.headers);
  for (const [key, value] of Object.entries(extra)) {
    merged.set(key, value);
  }
  return new Response(response.body, {
    status: response.status,
    statusText: response.statusText,
    headers: merged,
  });
}

function generateSessionId(): string {
  const bytes = new Uint8Array(16);
  crypto.getRandomValues(bytes);
  return Array.from(bytes, (b) => b.toString(16).padStart(2, "0")).join("");
}

function clientWantsSse(request: Request): boolean {
  const accept = (request.headers.get("accept") ?? "").toLowerCase();
  const sseIdx = accept.indexOf("text/event-stream");
  if (sseIdx === -1) return false;
  const jsonIdx = accept.indexOf("application/json");
  if (jsonIdx === -1) return true;
  // MCP 2025-06-18 requires clients to list BOTH content types. When
  // both are listed, the client signals preference by order: SSE is
  // picked when it appears before application/json. This is a
  // simpler heuristic than full RFC 9110 q-value parsing and lines
  // up with what every real MCP client emits.
  return sseIdx < jsonIdx;
}

function isJsonRpcRequest(message: JsonRpcMessage): boolean {
  return (
    message.method !== undefined &&
    message.id !== undefined &&
    message.id !== null
  );
}

function isJsonRpcNotificationOrResponse(message: JsonRpcMessage): boolean {
  if (
    message.method !== undefined &&
    (message.id === undefined || message.id === null)
  ) {
    return true;
  }
  if (
    message.method === undefined &&
    message.id !== undefined &&
    message.id !== null
  ) {
    return true;
  }
  return false;
}

/**
 * A violation of the gateway's own provider contract: a descriptor or a
 * read result that doesn't match the MCP shape. The message is written
 * here and names only the offending field, never host data, so it is
 * safe to return to the caller (and telling a developer *which* field is
 * wrong is the whole point). It gets its own class so the wire/audit
 * split can tell it apart from an arbitrary host exception.
 */
class ResourceContractError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "ResourceContractError";
  }
}

/**
 * Split a thrown error into the text the audit row keeps and the text
 * the MCP client is allowed to see. Mirrors what `dispatch.runTool` does
 * for tool execution: a deliberate `ConvexError` passes through,
 * everything else collapses to `generic`.
 */
function splitErrorText(
  err: unknown,
  generic: string,
): { full: string; wire: string } {
  const full = err instanceof Error ? err.message : String(err);
  const deliberate =
    isDeliberateConvexError(err) || err instanceof ResourceContractError;
  return { full, wire: deliberate ? full : generic };
}

/**
 * Build the single-frame SSE body of a POST response.
 *
 * - `2026-07-28` (`isStateless`): a bare message event. The stateless revision
 *   removed `Last-Event-ID` resumability, so no event id at all.
 * - Every session-based revision, INCLUDING `2025-11-25`: one message
 *   event with id 1, byte-identical to what legacy sessions always
 *   received.
 *
 * `2025-11-25` adds an optional priming event + `retry` hint to SSE
 * framing, but the reference server emits those ONLY when it has an
 * event store backing GET + `Last-Event-ID` replay (its
 * `writePrimingEvent` returns early when `!this._eventStore`). This
 * gateway has neither an event store nor a GET channel (GET is a hard
 * 405), so priming here would advertise resumability it cannot honor:
 * a client whose connection dies mid-frame would schedule a GET
 * reconnect with `Last-Event-ID: 0`, get 405, and silently abandon the
 * request instead of failing cleanly. Not emitting the optional
 * additions is the conforming behavior for a server without replay,
 * and it keeps every legacy revision's frame identical.
 */
function sseResponseFrame(body: string, isStateless: boolean): string {
  const idLine = isStateless ? "" : "id: 1\n";
  return `${idLine}event: message\ndata: ${body}\n\n`;
}

function jsonResultEnvelope(id: JsonRpcMessage["id"], value: unknown): string {
  return JSON.stringify({ jsonrpc: "2.0", id: id ?? null, result: value });
}

/**
 * Whether an `outputSchema` may be advertised, and its value shipped as
 * `structuredContent`, to a client speaking `revision`.
 *
 * The two eras disagree, and the disagreement is not cosmetic. Through
 * `2025-11-25`, `Tool.outputSchema` must be rooted at `type: "object"`
 * and `CallToolResult.structuredContent` is typed `{ [key: string]:
 * unknown }`. A client validating to that revision rejects the WHOLE
 * `tools/list` response over one scalar-rooted schema, so a single
 * `returns: v.string()` tool hides every other tool from it. `2026-07-28`
 * widened both: any JSON Schema 2020-12, and `structuredContent` as
 * `unknown` ("object, array, string, number, boolean, or null").
 *
 * So a non-object schema is withheld from legacy clients and advertised
 * to stateless-era ones. The tool still works either way; a session-based client just
 * sees it untyped, with the value in the text block as always.
 */
function mayAdvertiseOutputSchema(
  outputSchema: unknown,
  isStateless: boolean,
): boolean {
  if (outputSchema === undefined) return false;
  if (isStateless) return true;
  return (
    isPlainObject(outputSchema) &&
    (outputSchema as { type?: unknown }).type === "object"
  );
}

/**
 * The schema a client is shown for a registry row.
 *
 * The stored object is the RESOLVED one: `$ref` inlined, `$defs` gone,
 * `$`-prefixed keywords stripped so Convex can store it at all. That is
 * the right form for the gateway's own `Mcp-Param-*` walk and the wrong
 * one for the wire, where SEP-1613 asks that those keywords survive. So
 * the authored JSON wins whenever the row carries it.
 *
 * Rows written before that field fall back to the resolved object, which
 * is exactly what they advertised before. A row whose JSON does not parse
 * can only come from a caller writing to the component mutation directly;
 * it falls back too rather than dropping the tool from the catalog.
 */
function advertisedSchema(
  authoredJson: unknown,
  stored: unknown,
  toolName: string,
): unknown {
  if (typeof authoredJson !== "string") return stored;
  try {
    return JSON.parse(authoredJson);
  } catch {
    console.error(
      "[mcp-gateway] tool has an unparsable authored schema, advertising the resolved one",
      toolName,
    );
    return stored;
  }
}

/**
 * `jsonResultEnvelope` for a host-authored result, where a value JSON
 * cannot represent (a `v.int64()` field read straight off a document)
 * would otherwise escape the switch as a raw 500: no JSON-RPC envelope,
 * no CORS headers, nothing the client can act on. Reported as an error
 * RESULT because the call itself already happened.
 */
function hostResultEnvelope(
  id: JsonRpcMessage["id"],
  value: unknown,
  toolName: string,
): string {
  try {
    return jsonResultEnvelope(id, value);
  } catch (err) {
    console.error(
      "[mcp-gateway] hook result cannot be serialized for the wire",
      toolName,
      err,
    );
    return jsonResultEnvelope(id, {
      content: [
        {
          type: "text",
          text: `Tool "${toolName}" returned a value that cannot be represented on the wire`,
        },
      ],
      isError: true,
    });
  }
}

function jsonErrorEnvelope(
  id: JsonRpcMessage["id"],
  code: number,
  message: string,
): string {
  return JSON.stringify({
    jsonrpc: "2.0",
    id: id ?? null,
    error: { code, message },
  });
}

function jsonErrorEnvelopeWithData(
  id: JsonRpcMessage["id"],
  code: number,
  message: string,
  data: Record<string, unknown>,
): string {
  return JSON.stringify({
    jsonrpc: "2.0",
    id: id ?? null,
    error: { code, message, data },
  });
}

function statelessErrorResponse(
  id: JsonRpcMessage["id"],
  code: number,
  message: string,
  data: Record<string, unknown> = {},
  status = 400,
): Response {
  return new Response(jsonErrorEnvelopeWithData(id, code, message, data), {
    status,
    headers: { "content-type": "application/json" },
  });
}

function statelessProtocolVersion(message: JsonRpcMessage): string | null {
  const meta = message.params?._meta;
  if (!isPlainObject(meta)) return null;
  const version = meta["io.modelcontextprotocol/protocolVersion"];
  return typeof version === "string" ? version : null;
}

/**
 * Canonical base64: the standard alphabet, padded to a multiple of four.
 * `atob` accepts an unpadded tail, so `SGVsbG8` decodes there exactly as
 * `SGVsbG8=` does, and SEP-2243 requires the former to be REFUSED. The
 * check has to happen before the decode for that reason, not after.
 */
const CANONICAL_BASE64 = /^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/;

function decodeMcpHeaderValue(value: string | null): string | null {
  if (value === null) return null;
  if (
    value.length > MAX_MCP_HEADER_VALUE_LENGTH ||
    hasMcpHeaderControlCharacter(value)
  ) {
    return null;
  }
  const prefix = "=?base64?";
  const suffix = "?=";
  if (!value.startsWith(prefix)) return value;
  // A wrapper the sender never closed is not an encoded value, so it is
  // the literal it looks like. SEP-2243 is explicit that only the full
  // `=?base64?...?=` form is decoded, and rejecting this instead refuses
  // a request whose header genuinely matches the argument.
  if (!value.endsWith(suffix)) return value;
  const encoded = value.slice(prefix.length, -suffix.length);
  if (encoded.length === 0 || !CANONICAL_BASE64.test(encoded)) return null;
  try {
    const binary = atob(encoded);
    const bytes = Uint8Array.from(binary, (character) =>
      character.charCodeAt(0),
    );
    const decoded = new TextDecoder("utf-8", { fatal: true }).decode(bytes);
    return decoded.length <= MAX_MCP_HEADER_VALUE_LENGTH &&
      !hasMcpHeaderControlCharacter(decoded)
      ? decoded
      : null;
  } catch {
    return null;
  }
}

function statelessNameMatches(message: JsonRpcMessage, request: Request): boolean {
  const headerName = decodeMcpHeaderValue(request.headers.get("mcp-name"));
  switch (message.method) {
    case "tools/call":
      return headerName === message.params?.name;
    case "resources/read":
      return headerName === message.params?.uri;
    case "prompts/get":
      return headerName === message.params?.name;
    case "tasks/get":
    case "tasks/update":
    case "tasks/cancel":
      return headerName === message.params?.taskId;
    default:
      return true;
  }
}

type McpHeaderParameter = {
  headerName: string;
  path: string[];
  type: "string" | "integer" | "boolean";
};

type McpHeaderParameterResult =
  | { parameters: McpHeaderParameter[]; problem?: never }
  | { parameters?: never; problem: string };

const HTTP_FIELD_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/;

function collectMcpHeaderParameters(schema: unknown): McpHeaderParameterResult {
  const parameters: McpHeaderParameter[] = [];
  const names = new Set<string>();

  function visit(
    node: unknown,
    path: string[],
    reachable: boolean,
  ): string | null {
    if (Array.isArray(node)) {
      for (const item of node) {
        const problem = visit(item, path, false);
        if (problem) return problem;
      }
      return null;
    }
    if (!isPlainObject(node)) return null;

    if ("x-mcp-header" in node) {
      const headerName = node["x-mcp-header"];
      const type = node.type;
      if (!reachable || path.length === 0) {
        return "x-mcp-header must be reachable through schema properties";
      }
      if (
        typeof headerName !== "string" ||
        !HTTP_FIELD_NAME.test(headerName) ||
        (type !== "string" && type !== "integer" && type !== "boolean")
      ) {
        return "x-mcp-header must name a string, integer, or boolean property";
      }
      const normalizedName = headerName.toLowerCase();
      if (names.has(normalizedName)) {
        return "x-mcp-header names must be case-insensitively unique";
      }
      names.add(normalizedName);
      parameters.push({ headerName, path, type });
    }

    for (const [key, value] of Object.entries(node)) {
      if (key === "x-mcp-header") continue;
      if (key === "properties" && isPlainObject(value)) {
        for (const [propertyName, propertySchema] of Object.entries(value)) {
          const problem = visit(
            propertySchema,
            [...path, propertyName],
            reachable,
          );
          if (problem) return problem;
        }
        continue;
      }
      if (typeof value === "object" && value !== null) {
        const problem = visit(value, path, false);
        if (problem) return problem;
      }
    }
    return null;
  }

  const problem = visit(schema, [], true);
  return problem ? { problem } : { parameters };
}

/**
 * Validate the `x-mcp-header` annotations in a tool's `inputSchema`.
 * Returns a human-readable problem string, or `null` when the schema is
 * valid. Called when a catalog is registered or synced, so a
 * schema-authoring mistake surfaces with the tool name attached instead
 * of failing every stateless `tools/call` for that tool at runtime.
 */
export function describeToolHeaderSchemaProblem(
  inputSchema: unknown,
): string | null {
  return collectMcpHeaderParameters(inputSchema).problem ?? null;
}

function headerValueForArgument(
  argumentsValue: unknown,
  parameter: McpHeaderParameter,
): string | null {
  let value = argumentsValue;
  for (const segment of parameter.path) {
    if (!isPlainObject(value) || !(segment in value)) return null;
    value = value[segment];
  }
  if (value === null || value === undefined) return null;
  if (parameter.type === "string" && typeof value === "string") return value;
  if (parameter.type === "boolean" && typeof value === "boolean") {
    return value ? "true" : "false";
  }
  if (parameter.type === "integer" && typeof value === "number") {
    return Number.isSafeInteger(value) ? String(value) : null;
  }
  return null;
}

/** Strict decimal literal, so `Number(" 42 ")` and `Number("")` don't slip through. */
const NUMERIC_HEADER_VALUE = /^[+-]?\d+(\.\d+)?$/;

type StatelessHeaderProblem =
  /** The tool's own schema is invalid. A server-side configuration error. */
  | { kind: "schema"; problem: string }
  /** The client's headers disagree with the body. `-32020` territory. */
  | { kind: "mismatch"; problem: string };

function validateStatelessToolParameterHeaders(
  request: Request,
  inputSchema: unknown,
  argumentsValue: unknown,
): StatelessHeaderProblem | null {
  const result = collectMcpHeaderParameters(inputSchema);
  if (result.parameters === undefined) {
    return { kind: "schema", problem: result.problem };
  }

  for (const parameter of result.parameters) {
    const header = request.headers.get(`mcp-param-${parameter.headerName}`);
    const expected = headerValueForArgument(argumentsValue, parameter);
    if (expected === null) {
      if (header !== null) {
        return {
          kind: "mismatch",
          problem: `Mcp-Param-${parameter.headerName} must be omitted`,
        };
      }
      continue;
    }
    const mismatch: StatelessHeaderProblem = {
      kind: "mismatch",
      problem: `Mcp-Param-${parameter.headerName} must match the request arguments`,
    };
    if (header === null) return mismatch;
    const decoded = decodeMcpHeaderValue(header);
    if (decoded === null) return mismatch;
    // Streamable HTTP: integer parameters SHOULD be compared numerically
    // rather than as strings, so `42.0` and `42` are considered equal.
    if (parameter.type === "integer") {
      if (
        !NUMERIC_HEADER_VALUE.test(decoded) ||
        Number(decoded) !== Number(expected)
      ) {
        return mismatch;
      }
      continue;
    }
    if (decoded !== expected) return mismatch;
  }
  return null;
}

function finalizeStatelessResult(
  body: string,
  method: string,
  options: HandleMcpRequestOptions,
): string {
  const envelope = JSON.parse(body) as { result?: Record<string, unknown> };
  if (!envelope.result) return body;
  envelope.result.resultType ??= "complete";
  const meta = isPlainObject(envelope.result._meta)
    ? envelope.result._meta
    : {};
  meta["io.modelcontextprotocol/serverInfo"] = options.serverInfo ?? {
    name: SERVER_NAME,
    version: SERVER_VERSION,
  };
  envelope.result._meta = meta;
  if (
    method === "server/discover" ||
    method === "tools/list" ||
    method === "resources/list" ||
    method === "resources/templates/list" ||
    method === "resources/read"
  ) {
    envelope.result.ttlMs ??= 0;
    envelope.result.cacheScope ??= "private";
  }
  // `tasks/get` is deliberately absent. SEP-2549 reads `ttlMs` on a
  // result as "how long you may cache this"; SEP-2663 puts a `ttlMs` on
  // the Task itself, meaning "how much lifetime is left", and a task
  // result carries the task's fields inline, so the two SEPs share one
  // key. The task meaning wins: a client that cannot read the remaining
  // lifetime cannot know when to stop polling. Injecting the cache hint
  // here would be swallowed by the task's own value and would then tell
  // a caching client to keep one poll for the task's whole life, which
  // is the single response that must never be cached. A poll is
  // uncacheable by nature, so there is no hint worth fighting over.
  return JSON.stringify(envelope);
}

type VerifiedMrtrState = {
  state: unknown;
  idempotencyKey: string;
  jti: string;
  round: number;
  exp: number;
};

const MRTR_DEFAULT_TTL_MS = 5 * 60 * 1000;
const MRTR_MAX_TTL_MS = 60 * 60 * 1000;
/**
 * Slack added to a chain claim's lifetime on top of the TTL ceiling.
 * The claim must outlive every continuation of its chain, and a branch
 * that read the chain as open can still be sealing one while the
 * winner's claim is being written. With a host `ttlMs` at the ceiling
 * that sibling's expiry exceeds the claim's by exactly that scheduling
 * delta, so cover it rather than reason about how small it is.
 */
const MRTR_CLAIM_SLACK_MS = 60 * 1000;
const MRTR_MAX_STATE_BYTES = 8 * 1024;
const MRTR_MIN_SECRET_BYTES = 32;
/**
 * Hard ceiling on continuation rounds per chain. MRTR server
 * requirement 8 permits repeated `InputRequiredResult`s, but an
 * unbounded chain lets a buggy hook ping-pong with a client forever.
 */
const MRTR_MAX_ROUNDS = 16;

function isMcpInputRequiredResult(
  value: unknown,
): value is McpInputRequiredResult<McpInputRequiredFallback> {
  return (
    isPlainObject(value) &&
    value.__mcpInputRequired === true &&
    (value.inputRequests === undefined || isPlainObject(value.inputRequests))
  );
}

function isMcpCompleteReadResult(
  value: unknown,
): value is McpCompleteReadResult {
  return (
    isPlainObject(value) &&
    value.__mcpCompleteRead === true &&
    Array.isArray(value.contents)
  );
}

function isMcpDeclineReadResult(value: unknown): value is McpDeclineReadResult {
  return (
    isPlainObject(value) &&
    value.__mcpDeclineRead === true &&
    typeof value.reason === "string"
  );
}

function isMcpCompleteCallResult(
  value: unknown,
): value is { __mcpCompleteCall: true; result: Record<string, unknown> } {
  return (
    isPlainObject(value) &&
    value.__mcpCompleteCall === true &&
    isPlainObject(value.result)
  );
}

/**
 * A well-formed MCP `tools/call` result carries a `content` array (the
 * spec-required field); `structuredContent` and `isError` are optional.
 * A `completeCall()` result is forwarded verbatim to the client, so the
 * gateway validates it here rather than shipping a shape a spec-
 * compliant client would reject. Returns a problem string, or `null`
 * when valid.
 */
function describeCompleteCallResultProblem(
  result: Record<string, unknown>,
  isStateless: boolean,
): string | null {
  if (!Array.isArray(result.content)) {
    return "result.content must be an array";
  }
  // Same era split the emit paths apply: `2026-07-28` types
  // `structuredContent` as `unknown`, everything before it as an object.
  // Judging a hook's result by a different rule than the gateway's own
  // output is what let the two drift apart before.
  if (
    result.structuredContent !== undefined &&
    !isStateless &&
    !isPlainObject(result.structuredContent)
  ) {
    return "result.structuredContent must be an object on this protocol revision";
  }
  if (result.isError !== undefined && typeof result.isError !== "boolean") {
    return "result.isError must be a boolean";
  }
  return null;
}

/**
 * Accumulate the client capabilities a set of `inputRequests` demands.
 * Elicitation is tracked per mode (`form` / `url`); a request without a
 * `mode` is a form request per the spec default. Returns `null` for a
 * request shape or method the gateway cannot vouch for.
 */
function mrtrRequiredCapabilities(
  inputRequests: Record<string, unknown>,
): Record<string, unknown> | null {
  const elicitationModes: { form?: object; url?: object } = {};
  const required: Record<string, unknown> = {};
  for (const request of Object.values(inputRequests)) {
    if (!isPlainObject(request) || typeof request.method !== "string") {
      return null;
    }
    switch (request.method) {
      case "elicitation/create": {
        const mode = isPlainObject(request.params)
          ? request.params.mode
          : undefined;
        if (mode !== undefined && mode !== "form" && mode !== "url")
          return null;
        // Merge instead of overwrite: two requests with different modes
        // require BOTH modes, and a later form request must not erase an
        // earlier url requirement.
        elicitationModes[(mode ?? "form") as "form" | "url"] = {};
        required.elicitation = elicitationModes;
        break;
      }
      case "sampling/createMessage":
        required.sampling = {};
        break;
      case "roots/list":
        required.roots = {};
        break;
      default:
        return null;
    }
  }
  return required;
}

/**
 * The subset of `required` the client did NOT declare, in the exact
 * shape `-32021`'s `data.requiredCapabilities` must carry (only the
 * missing entries, per `basic/index`). Empty object means fully
 * supported. Mode-aware in both directions: `elicitation: {}` is
 * equivalent to declaring form-only (spec backwards-compat rule), and a
 * url-only client does not support form requests.
 */
function missingMrtrCapabilities(
  clientCapabilities: Record<string, unknown>,
  required: Record<string, unknown>,
): Record<string, unknown> {
  const missing: Record<string, unknown> = {};
  for (const [name, requirement] of Object.entries(required)) {
    const declared = clientCapabilities[name];
    if (!isPlainObject(declared)) {
      missing[name] = requirement;
      continue;
    }
    if (name === "elicitation" && isPlainObject(requirement)) {
      const declaresNoModes =
        !isPlainObject(declared.form) && !isPlainObject(declared.url);
      const supportsForm = isPlainObject(declared.form) || declaresNoModes;
      const supportsUrl = isPlainObject(declared.url);
      const missingModes: Record<string, unknown> = {};
      if ("form" in requirement && !supportsForm) missingModes.form = {};
      if ("url" in requirement && !supportsUrl) missingModes.url = {};
      if (Object.keys(missingModes).length > 0) {
        missing.elicitation = missingModes;
      }
    }
  }
  return missing;
}

type MrtrCapabilityCheck = {
  missingCapabilities: Record<string, unknown>;
  supported: boolean;
};

function checkMrtrCapabilities(
  inputRequests: Record<string, unknown>,
  clientCapabilities: Record<string, unknown> | null,
): MrtrCapabilityCheck | null {
  const requiredCapabilities = mrtrRequiredCapabilities(inputRequests);
  if (!requiredCapabilities) return null;
  const missingCapabilities =
    clientCapabilities === null
      ? requiredCapabilities
      : missingMrtrCapabilities(clientCapabilities, requiredCapabilities);
  return {
    missingCapabilities,
    supported:
      clientCapabilities !== null &&
      Object.keys(missingCapabilities).length === 0,
  };
}

function base64UrlEncode(bytes: Uint8Array): string {
  let binary = "";
  for (const byte of bytes) binary += String.fromCharCode(byte);
  return btoa(binary)
    .replaceAll("+", "-")
    .replaceAll("/", "_")
    .replace(/=+$/, "");
}

function base64UrlDecode(value: string): Uint8Array | null {
  if (!/^[A-Za-z0-9_-]+$/.test(value)) return null;
  try {
    const padded = `${value.replaceAll("-", "+").replaceAll("_", "/")}${"=".repeat((4 - (value.length % 4)) % 4)}`;
    return Uint8Array.from(atob(padded), (char) => char.charCodeAt(0));
  } catch {
    return null;
  }
}

function cryptoBuffer(bytes: Uint8Array): ArrayBuffer {
  return bytes.buffer.slice(
    bytes.byteOffset,
    bytes.byteOffset + bytes.byteLength,
  ) as ArrayBuffer;
}

function stableJson(value: unknown): string {
  // Mirror JSON.stringify's undefined semantics so every output is
  // valid JSON: top-level and array-item undefined become null, and
  // undefined-valued properties are omitted. Without this, a call site
  // that forgot a `?? null` guard would embed a literal `undefined`
  // token (or return the value undefined) into digest input.
  if (value === undefined) return "null";
  if (value === null || typeof value !== "object") return JSON.stringify(value);
  if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;
  const record = value as Record<string, unknown>;
  return `{${Object.keys(record)
    .filter((key) => record[key] !== undefined)
    .sort()
    .map((key) => `${JSON.stringify(key)}:${stableJson(record[key])}`)
    .join(",")}}`;
}

async function sha256Base64Url(value: string): Promise<string> {
  const digest = await crypto.subtle.digest(
    "SHA-256",
    new TextEncoder().encode(value),
  );
  return base64UrlEncode(new Uint8Array(digest));
}

/**
 * True when `mrtr.secret` cannot key HMAC-SHA-256 safely. Checked up
 * front by the handler so a short secret surfaces as a `-32603` server
 * misconfiguration on BOTH the seal and verify paths, never as a
 * `-32602` that blames the client for a state that may be perfectly
 * valid.
 */
function mrtrSecretTooShort(options: McpMrtrOptions): boolean {
  return (
    new TextEncoder().encode(options.secret).byteLength < MRTR_MIN_SECRET_BYTES
  );
}

async function mrtrKey(secret: string): Promise<CryptoKey> {
  if (new TextEncoder().encode(secret).byteLength < MRTR_MIN_SECRET_BYTES) {
    throw new Error("MRTR secret must contain at least 32 bytes");
  }
  return await crypto.subtle.importKey(
    "raw",
    new TextEncoder().encode(secret),
    { name: "HMAC", hash: "SHA-256" },
    false,
    ["sign", "verify"],
  );
}

/**
 * Seal one continuation round. `argsDigest` is computed by the caller
 * BEFORE the hook runs, so a hook that mutates its (copied) argument
 * object can never poison the digest the retry is checked against.
 * `idempotencyKey` is minted once on round 1 and carried verbatim
 * through every later round of the chain; `jti` is fresh per round and
 * anchors the one-time redemption record.
 */
async function sealMrtrState(
  options: McpMrtrOptions,
  payloadFields: {
    toolName: string;
    identitySubject: string | null;
    argsDigest: string;
    state: unknown;
    idempotencyKey: string;
    round: number;
  },
): Promise<string> {
  const encodedState = stableJson(payloadFields.state);
  if (
    new TextEncoder().encode(encodedState).byteLength > MRTR_MAX_STATE_BYTES
  ) {
    throw new Error("MRTR state exceeds 8 KiB");
  }
  // `ttlMs` is a host option, not client input, so a nonsensical value
  // (0, negative, NaN, Infinity) is a server misconfiguration. Throw so
  // the seal path reports it as -32603, never as a -32602 that would
  // mint a dead-on-arrival continuation and then blame the client's
  // retry for the "expired" state it was handed.
  if (
    options.ttlMs !== undefined &&
    (!Number.isFinite(options.ttlMs) || options.ttlMs <= 0)
  ) {
    throw new Error("MRTR ttlMs must be a positive finite number");
  }
  const ttlMs = Math.min(options.ttlMs ?? MRTR_DEFAULT_TTL_MS, MRTR_MAX_TTL_MS);
  const now = Date.now();
  const payload = {
    v: 2,
    toolName: payloadFields.toolName,
    identitySubject: payloadFields.identitySubject,
    argsDigest: payloadFields.argsDigest,
    exp: now + ttlMs,
    jti: crypto.randomUUID(),
    round: payloadFields.round,
    idempotencyKey: payloadFields.idempotencyKey,
    state: payloadFields.state,
  };
  const encoded = base64UrlEncode(
    new TextEncoder().encode(JSON.stringify(payload)),
  );
  const signature = await crypto.subtle.sign(
    "HMAC",
    await mrtrKey(options.secret),
    new TextEncoder().encode(encoded),
  );
  return `${encoded}.${base64UrlEncode(new Uint8Array(signature))}`;
}

/**
 * Verify one continuation. Returns `null` when verification RAN and
 * said no (tampered, expired, wrong principal/tool/arguments): `-32602`
 * territory. Throws when verification could not run at all (e.g. a
 * misconfigured secret): the caller maps that to `-32603` instead of
 * blaming the client.
 */
async function verifyMrtrState(
  options: McpMrtrOptions,
  requestState: unknown,
  expected: {
    toolName: string;
    identitySubject: string | null;
    argsDigest: string;
  },
): Promise<VerifiedMrtrState | null> {
  if (typeof requestState !== "string") return null;
  const [encoded, signature, extra] = requestState.split(".");
  if (!encoded || !signature || extra !== undefined) return null;
  const signatureBytes = base64UrlDecode(signature);
  if (
    !signatureBytes ||
    !(await crypto.subtle.verify(
      "HMAC",
      await mrtrKey(options.secret),
      cryptoBuffer(signatureBytes),
      new TextEncoder().encode(encoded),
    ))
  ) {
    return null;
  }
  const payloadBytes = base64UrlDecode(encoded);
  if (!payloadBytes) return null;
  let payload: Record<string, unknown>;
  try {
    payload = JSON.parse(new TextDecoder().decode(payloadBytes)) as Record<
      string,
      unknown
    >;
  } catch {
    return null;
  }
  if (
    payload.v !== 2 ||
    payload.toolName !== expected.toolName ||
    payload.identitySubject !== expected.identitySubject ||
    typeof payload.exp !== "number" ||
    payload.exp < Date.now() ||
    typeof payload.jti !== "string" ||
    typeof payload.round !== "number" ||
    !Number.isSafeInteger(payload.round) ||
    payload.round < 1 ||
    typeof payload.idempotencyKey !== "string" ||
    payload.argsDigest !== expected.argsDigest
  ) {
    return null;
  }
  return {
    state: payload.state,
    idempotencyKey: payload.idempotencyKey,
    jti: payload.jti,
    round: payload.round,
    exp: payload.exp,
  };
}

async function ensureCatalogSynced(
  options: InternalHandleMcpRequestOptions,
): Promise<void> {
  await options.ensureCatalogSynced?.();
}

/**
 * Verify, redeem, and read the chain state for one MRTR continuation.
 *
 * Shared by `tools/call` and `resources/read` so the sequence exists
 * once: verify the seal, pin the answers with a one-time `jti`
 * redemption, then read how (or whether) the chain already resolved. A
 * second copy of this is how a future MRTR fix lands on one call path and
 * silently misses the other.
 *
 * `chainName` is both the value the seal binds and the label in logs, so
 * a continuation minted for one operation can never be presented as
 * another: tools pass the tool name, reads pass `resources/read:<uri>`.
 *
 * The three outcomes are kept apart on purpose. `"reject"` is the
 * client's fault and carries the wire code/message; `"error"` means a
 * step could not RUN (verification threw, the component predates a
 * table) and is always a logged `-32603`, never a raw 500 that would
 * skip the CORS wrapper.
 */
async function verifyMrtrContinuation(
  ctx: HandlerCtx,
  component: ComponentApi,
  options: McpMrtrOptions,
  input: {
    chainName: string;
    requestState: unknown;
    inputResponses: unknown;
    identitySubject: string | null;
    argsDigest: string;
  },
): Promise<
  | {
      status: "ok";
      continuation: VerifiedMrtrState;
      requestDigest: string | undefined;
      chain: MrtrChainState;
    }
  | { status: "reject"; code: number; message: string }
  /**
   * A step could not RUN. `message` overrides the caller's generic wire
   * text where a more precise one exists (a misconfigured secret is not a
   * verification failure), and is always paired with a console line.
   */
  | { status: "error"; message?: string }
> {
  if (mrtrSecretTooShort(options)) {
    // Server misconfiguration, never the client's fault: the state being
    // verified may be perfectly valid. Checked FIRST, so a misconfigured
    // gateway never blames the client for a malformed payload it was never
    // going to read.
    console.error(
      "[mcp-gateway] mrtr.secret is shorter than 32 bytes; " +
        "refusing to verify continuations",
    );
    return {
      status: "error",
      message: "MRTR is misconfigured on this gateway",
    };
  }
  if (
    input.inputResponses !== undefined &&
    !isPlainObject(input.inputResponses)
  ) {
    return {
      status: "reject",
      code: INVALID_PARAMS,
      message: "MRTR inputResponses must be an object",
    };
  }
  let continuation: VerifiedMrtrState | null;
  try {
    continuation = await verifyMrtrState(options, input.requestState, {
      toolName: input.chainName,
      identitySubject: input.identitySubject,
      argsDigest: input.argsDigest,
    });
  } catch (err) {
    // Verification could not run (as opposed to running and saying no).
    console.error("[mcp-gateway] MRTR state verification failed to run", err);
    return { status: "error" };
  }
  if (!continuation) {
    return {
      status: "reject",
      code: INVALID_PARAMS,
      message: "Invalid, expired, or mismatched MRTR requestState",
    };
  }
  // One-time redemption: a continuation stays cryptographically valid
  // until its TTL, so first use pins the responses it was answered with.
  // A byte-identical re-send is an idempotent replay (safe to
  // re-process); different responses for the same continuation would let
  // a captured state flip an already-resolved decision (decline ->
  // accept) and are rejected.
  const requestDigest =
    input.inputResponses !== undefined
      ? await sha256Base64Url(stableJson(input.inputResponses))
      : undefined;
  let redemption: "fresh" | "replay" | "conflict";
  try {
    redemption = await ctx.runMutation(component.mrtr.redeemContinuation, {
      jti: continuation.jti,
      ...(requestDigest !== undefined
        ? { responsesDigest: requestDigest }
        : {}),
      expiresAt: continuation.exp,
    });
  } catch (err) {
    console.error(
      "[mcp-gateway] MRTR continuation redemption failed to run " +
        "(is the component deployment up to date?)",
      input.chainName,
      err,
    );
    return { status: "error" };
  }
  if (redemption === "conflict") {
    // Either the replay-flip attack this mechanism exists for, or a
    // client re-collecting semantically identical answers into
    // byte-different responses. Make the event observable so an operator
    // can tell the two apart.
    console.warn(
      "[mcp-gateway] MRTR continuation redeemed with different " +
        "responses; rejecting",
      input.chainName,
      continuation.jti,
    );
    return {
      status: "reject",
      code: INVALID_PARAMS,
      message:
        "This MRTR continuation was already used with different input " +
        "responses. Re-send the exact previous responses, or restart the " +
        "call without requestState",
    };
  }
  const chain = await readMrtrChain(
    ctx,
    component,
    input.chainName,
    continuation.idempotencyKey,
  );
  if (chain.status === "error") return { status: "error" };
  return { status: "ok", continuation, requestDigest, chain };
}

/**
 * Resolve a chain exactly once, immediately before the gateway does the
 * thing that resolves it (dispatch a tool, finish a call itself, serve or
 * refuse a read). Shared by every such site so the repeat rule is stated
 * once.
 *
 * `"claimed"` means this branch owns the resolution. `"lost"` means
 * another branch settled the chain first and this one must be refused.
 * The resolving continuation re-sent with the same answer is NOT lost:
 * that is the idempotent retry of a response the client never received.
 *
 * `chain` is passed in rather than read here, and deliberately so: each
 * caller decides WHICH snapshot the repeat rule is judged against (the
 * pre-hook read on the terminal paths, a fresh post-hook re-read before a
 * dispatch). Reading it here would silently pick one for everybody.
 */
async function settleMrtrChain(
  ctx: HandlerCtx,
  component: ComponentApi,
  input: {
    chainName: string;
    continuation: VerifiedMrtrState;
    requestDigest: string | undefined;
    chain: MrtrChainState;
    resolution: "dispatched" | "completed";
  },
): Promise<"claimed" | "lost" | "error"> {
  const { chainName, continuation, requestDigest, chain, resolution } = input;
  if (
    chain.status === "resolved" &&
    isChainRepeat(chain, continuation.jti, requestDigest, resolution)
  ) {
    return "claimed";
  }
  const claim =
    chain.status === "resolved"
      ? { ...chain, status: "lost" as const }
      : await claimMrtrChain(
          ctx,
          component,
          chainName,
          continuation.idempotencyKey,
          continuation.jti,
          requestDigest,
          resolution,
          Date.now() + MRTR_MAX_TTL_MS + MRTR_CLAIM_SLACK_MS,
        );
  if (claim.status === "error") return "error";
  // Losing to a concurrent send of this very continuation with this very
  // answer is the same lost-response retry, just interleaved. Refusing it
  // would push the client to start a new chain, i.e. a new idempotency
  // key, i.e. the duplicate the key exists to prevent.
  if (
    claim.status === "lost" &&
    !isChainRepeat(claim, continuation.jti, requestDigest, resolution)
  ) {
    return "lost";
  }
  return "claimed";
}

/**
 * Ask the client for input: the shared tail of both MRTR gates, from the
 * last fail-closed checks through the sealed `input_required` envelope.
 *
 * Everything here is mechanical and must not drift between `tools/call`
 * and `resources/read`: the round ceiling, the capability gate, and above
 * all the sealed envelope, whose `toolName`/`argsDigest`/`idempotencyKey`
 * are what bind a continuation to exactly one chain. The checks that DO
 * differ (whether this request can carry a continuation at all) stay at
 * the call sites, where their wording belongs.
 *
 * `hookLabel` names the asking hook in the logs and in the one wire
 * message that mentions it. `{ response }` is the capability refusal,
 * which per spec is a `-32021` on the modern envelope; `{ body }` is every
 * other outcome, including success.
 */
async function askMrtrInput(
  mrtr: McpMrtrOptions,
  input: {
    id: JsonRpcMessage["id"];
    hookLabel: string;
    chainName: string;
    argsDigest: string;
    identitySubject: string | null;
    continuation: VerifiedMrtrState | null;
    clientCapabilities: Record<string, unknown>;
    inputRequests: Record<string, unknown>;
    state: unknown;
    capabilityCheck?: MrtrCapabilityCheck;
  },
): Promise<{ body: string } | { response: Response }> {
  if (mrtrSecretTooShort(mrtr)) {
    console.error(
      "[mcp-gateway] mrtr.secret is shorter than 32 bytes; " +
        "refusing to seal a continuation",
    );
    return {
      body: jsonErrorEnvelope(
        input.id,
        INTERNAL_ERROR,
        "MRTR is misconfigured on this gateway",
      ),
    };
  }
  const round = (input.continuation?.round ?? 0) + 1;
  if (round > MRTR_MAX_ROUNDS) {
    console.error(
      "[mcp-gateway] MRTR chain exceeded the round ceiling",
      input.chainName,
    );
    return {
      body: jsonErrorEnvelope(
        input.id,
        INTERNAL_ERROR,
        "MRTR continuation exceeded the round limit",
      ),
    };
  }
  const capabilityCheck =
    input.capabilityCheck ??
    checkMrtrCapabilities(input.inputRequests, input.clientCapabilities);
  if (!capabilityCheck) {
    // A host-side hook bug: a request method or shape the gateway cannot
    // vouch for (typo'd method, unknown elicitation mode). Name the
    // offending methods so the hook author can find it.
    console.error(
      `[mcp-gateway] ${input.hookLabel} returned unsupported input requests`,
      input.chainName,
      Object.values(input.inputRequests)
        .map((request) =>
          isPlainObject(request) ? String(request.method) : typeof request,
        )
        .join(", "),
    );
    return {
      body: jsonErrorEnvelope(
        input.id,
        INTERNAL_ERROR,
        `${input.hookLabel} returned unsupported input requests`,
      ),
    };
  }
  const missingCapabilities = capabilityCheck.missingCapabilities;
  if (Object.keys(missingCapabilities).length > 0) {
    return {
      response: statelessErrorResponse(
        input.id,
        MISSING_REQUIRED_CLIENT_CAPABILITY,
        "Client lacks a capability required for MRTR input requests",
        // Per spec, data.requiredCapabilities lists only what is MISSING,
        // not the full required set.
        { requiredCapabilities: missingCapabilities },
      ),
    };
  }
  try {
    return {
      body: jsonResultEnvelope(input.id, {
        resultType: "input_required",
        ...(Object.keys(input.inputRequests).length > 0
          ? { inputRequests: input.inputRequests }
          : {}),
        requestState: await sealMrtrState(mrtr, {
          toolName: input.chainName,
          identitySubject: input.identitySubject,
          argsDigest: input.argsDigest,
          state: input.state ?? null,
          // Round 1 mints the chain's key; later rounds carry it.
          idempotencyKey:
            input.continuation?.idempotencyKey ?? crypto.randomUUID(),
          round,
        }),
      }),
    };
  } catch (err) {
    console.error(
      "[mcp-gateway] failed to seal MRTR requestState",
      input.chainName,
      err,
    );
    return {
      body: jsonErrorEnvelope(
        input.id,
        INTERNAL_ERROR,
        "Failed to create MRTR request state",
      ),
    };
  }
}

/**
 * True when this request is the resolving continuation of an
 * already-resolved chain, re-sent with the same answer: the lost
 * response retry. Same continuation with a DIFFERENT answer is not a
 * repeat, which matters for a chain resolved on state alone, whose
 * continuation the redemption table never pinned.
 */
function isChainRepeat(
  chain: ResolvedChain,
  jti: string,
  responsesDigest: string | undefined,
  resolution: "dispatched" | "completed",
): boolean {
  return (
    chain.resolution === resolution &&
    chain.resolvedByJti === jti &&
    chain.resolvedByDigest === responsesDigest
  );
}

/**
 * Response for a continuation whose chain another branch already
 * resolved. Deliberately says nothing about how it resolved: whether
 * the call was accepted or declined is not the caller's business when
 * the caller is presenting a superseded continuation.
 */
function alreadyResolvedEnvelope(
  id: JsonRpcMessage["id"],
  toolName: string,
): string {
  console.warn(
    "[mcp-gateway] MRTR continuation presented for an already-resolved " +
      "chain; rejecting",
    toolName,
  );
  return jsonErrorEnvelope(
    id,
    INVALID_PARAMS,
    "This MRTR request has already been resolved. Start a new call " +
      "without requestState",
  );
}

/**
 * Claim an MRTR chain's single resolution, immediately before the
 * gateway dispatches or finishes the call itself. The claim IS the
 * decision: a continuation forked by an idempotent replay finds the
 * chain already resolved and is refused, which is what per-continuation
 * redemption cannot do (`jti` is fresh per round, so every
 * `inputRequired()` mints an unpinned sibling).
 *
 * Guarded like every other component call on this path: a claim that
 * cannot RUN is a logged `"error"` the caller turns into `-32603`,
 * never a raw 500 that skips the CORS wrapper.
 */
async function claimMrtrChain(
  ctx: HandlerCtx,
  component: ComponentApi,
  toolName: string,
  chainKey: string,
  jti: string,
  responsesDigest: string | undefined,
  resolution: "dispatched" | "completed",
  expiresAt: number,
): Promise<
  | { status: "claimed" }
  | ({ status: "lost" } & ResolvedChain)
  | { status: "error" }
> {
  try {
    const result = await ctx.runMutation(component.mrtr.claimChain, {
      chainKey,
      jti,
      ...(responsesDigest !== undefined ? { responsesDigest } : {}),
      resolution,
      expiresAt,
    });
    return result === "claimed"
      ? { status: "claimed" }
      : { status: "lost", ...result };
  } catch (err) {
    console.error(
      "[mcp-gateway] MRTR chain claim failed to run (is the component " +
        "deployment up to date?)",
      toolName,
      err,
    );
    return { status: "error" };
  }
}

/**
 * Read-only pre-check of a chain's resolution, run once per verified
 * continuation before the hook's decision is acted on. It exists so an
 * already-resolved chain refuses a continuation up front instead of
 * handing back one that could never resolve anything, and so an
 * idempotent re-send of a completed call can still reproduce its
 * result. The binding decision remains `claimMrtrChain`.
 */
type ResolvedChain = {
  resolution: "dispatched" | "completed";
  resolvedByJti: string;
  resolvedByDigest?: string;
};

type MrtrChainState =
  | { status: "open" }
  | ({ status: "resolved" } & ResolvedChain)
  | { status: "error" };

async function readMrtrChain(
  ctx: HandlerCtx,
  component: ComponentApi,
  toolName: string,
  chainKey: string,
): Promise<MrtrChainState> {
  try {
    const row = await ctx.runQuery(component.mrtr.getChainResolution, {
      chainKey,
    });
    return row === null ? { status: "open" } : { status: "resolved", ...row };
  } catch (err) {
    console.error(
      "[mcp-gateway] MRTR chain lookup failed to run (is the component " +
        "deployment up to date?)",
      toolName,
      err,
    );
    return { status: "error" };
  }
}

let warnedRequireAuthWithoutOAuth = false;

/**
 * Build a 401 challenge for an anonymous POST: `401` +
 * `WWW-Authenticate` when an OAuth server is configured (so the client
 * begins RFC 9728 discovery), or a bare 401 (plus a one-time warning)
 * when it isn't. Mirrors the `tools/call` UNAUTHORIZED branch.
 *
 * Called from the `requireAuth` gate, from the anonymous task-augmented
 * `tools/call` and `tasks/*` paths, and from the three resource methods
 * when the host's authorizer denies an anonymous caller with an
 * `unauth`-shaped reason. Only the read path passes a reason through; a
 * list discards its per-candidate reasons, so it takes the generic one.
 */
async function requireAuthChallenge(
  ctx: HandlerCtx,
  request: Request,
  component: ComponentApi,
  id: JsonRpcMessage["id"],
  reason = "Unauthorized: authentication required",
): Promise<Response> {
  const oauthConfig = await ctx.runQuery(component.registry.getOAuthConfig, {});
  const headers: Record<string, string> = {
    "content-type": "application/json",
  };
  if (oauthConfig) {
    const requestUrl = new URL(request.url);
    const mcpPath = requestUrl.pathname.replace(/\/+$/, "") || "/";
    const metadataUrl = buildProtectedResourceMetadataUrl(
      requestUrl.origin,
      mcpPath,
    );
    headers["www-authenticate"] = `Bearer resource_metadata="${metadataUrl}"`;
  } else if (!warnedRequireAuthWithoutOAuth) {
    warnedRequireAuthWithoutOAuth = true;
    console.warn(
      "[mcp-gateway] a 401 challenge was issued but no OAuth config " +
        "exists; returning 401 without WWW-Authenticate. Browser clients " +
        "can't begin OAuth discovery until setOAuthConfig is called.",
    );
  }
  return new Response(jsonErrorEnvelope(id, UNAUTHORIZED, reason), {
    status: 401,
    headers,
  });
}

async function safeAuthorize(
  authorize: McpAuthorizerHandler,
  ctx: HandlerCtx,
  args: McpAuthorizerArgs,
): Promise<{ decision: McpAuthorizerDecision; threw: boolean }> {
  try {
    const result = await authorize(ctx, args);
    return { decision: parseAuthorizerDecision(result), threw: false };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    return {
      decision: { allowed: false, reason: `Authorizer threw: ${message}` },
      threw: true,
    };
  }
}

/**
 * Build the `authorizeResource` input for one candidate.
 *
 * An anonymous caller only reaches here on a mount that set
 * `anonymousResources`, and is presented under `"resource_anonymous"`
 * rather than the mode its operation would otherwise use.
 *
 * The union is what makes an authenticated mode with a null identity
 * unrepresentable; this function is not load-bearing for that. What it
 * adds is the `undefined` case, which the types say cannot happen and an
 * untyped host's `resolveIdentity` can still produce.
 */
function resourceAuthorizerArgs(
  operation: McpResourceOperation,
  resourceUri: string,
  resourceMetadata: unknown,
  identity: McpResourceCaller,
): McpResourceAuthorizerArgs {
  // `== null`. `resolveCallerIdentity` normalizes an untyped host's
  // `undefined` away, so this is defence rather than a live case, but the
  // three method gates classify with `!identity`: were `undefined` ever to
  // reach here, `=== null` would build an AUTHENTICATED mode for a caller
  // the gates called anonymous.
  if (identity == null) {
    return {
      mode: "resource_anonymous",
      operation,
      resourceUri,
      resourceMetadata,
      identity: null,
    };
  }
  return {
    mode:
      operation === "read"
        ? "resource_read"
        : operation === "list"
          ? "resource_list"
          : "resource_templates_list",
    resourceUri,
    resourceMetadata,
    identity,
  };
}

async function safeAuthorizeResource(
  authorizeResource: McpResourceAuthorizerHandler | undefined,
  ctx: HandlerCtx,
  args: McpResourceAuthorizerArgs,
): Promise<{ decision: McpAuthorizerDecision; threw: boolean }> {
  if (!authorizeResource) {
    return { decision: { allowed: true }, threw: false };
  }
  try {
    const result = await authorizeResource(ctx, args);
    return { decision: parseAuthorizerDecision(result), threw: false };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    return {
      decision: {
        allowed: false,
        reason: `Resource authorizer threw: ${message}`,
      },
      threw: true,
    };
  }
}

function dedupeResourceCandidates(
  candidates: ResourceCandidate[],
): ResourceCandidate[] {
  const byUri = new Map<string, ResourceCandidate>();
  for (const candidate of candidates) {
    const existing = byUri.get(candidate.resource.uri);
    byUri.set(candidate.resource.uri, {
      resource: candidate.resource,
      metadata: candidate.metadata ?? existing?.metadata ?? null,
    });
  }
  return Array.from(byUri.values());
}

function isPlainObject(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

/** Max nesting the task path accepts in client-controlled JSON values. */
const MAX_TASK_VALUE_DEPTH = 64;

/**
 * True when `value` nests deeper than `MAX_TASK_VALUE_DEPTH`. Walked
 * with an EXPLICIT stack, never recursion, so checking a hostile value
 * cannot itself overflow. Task args and input responses are checked at
 * the HTTP boundary before they reach `ctx.runMutation`, whose own arg
 * serialization would otherwise overflow on a deeply nested body and
 * escape as a raw 500 instead of a clean JSON-RPC error.
 */
function nestsTooDeep(value: unknown): boolean {
  const stack: Array<{ node: unknown; depth: number }> = [
    { node: value, depth: 0 },
  ];
  while (stack.length > 0) {
    const { node, depth } = stack.pop()!;
    if (node === null || typeof node !== "object") continue;
    if (depth > MAX_TASK_VALUE_DEPTH) return true;
    for (const child of Array.isArray(node)
      ? node
      : Object.values(node as Record<string, unknown>)) {
      stack.push({ node: child, depth: depth + 1 });
    }
  }
  return false;
}

/**
 * Validate an MCP `icons` array (tools, resources, and resource templates all
 * carry the same shape). Per the spec's `Icon`: `src` is required, everything
 * else optional, `theme` constrained to `"light"` / `"dark"`.
 *
 * `src` is checked for being a non-empty string and nothing more. The gateway
 * is the producer here and never dereferences the URI; the spec puts the
 * fetch-side burden on consumers ("SHOULD take steps to ensure URLs serving
 * icons are from the same domain", "SHOULD take appropriate precautions when
 * consuming SVGs"). So an icon `src` is host-authored content that reaches
 * the client verbatim, exactly like a description.
 *
 * `label` names the enclosing descriptor in the message ("resource", "tool"),
 * so a rejection points at what to fix.
 */
export function describeIconsProblem(
  icons: unknown,
  label: string,
): string | null {
  if (icons === undefined) return null;
  if (!Array.isArray(icons)) return `${label}.icons must be an array`;
  for (const icon of icons) {
    if (!isPlainObject(icon)) return `${label}.icons entries must be objects`;
    if (typeof icon.src !== "string" || icon.src.length === 0) {
      return `${label}.icons[].src must be a non-empty string`;
    }
    if (icon.mimeType !== undefined && typeof icon.mimeType !== "string") {
      return `${label}.icons[].mimeType must be a string`;
    }
    if (
      icon.sizes !== undefined &&
      (!Array.isArray(icon.sizes) ||
        !icon.sizes.every((size) => typeof size === "string"))
    ) {
      return `${label}.icons[].sizes must be an array of strings`;
    }
    if (
      icon.theme !== undefined &&
      icon.theme !== "light" &&
      icon.theme !== "dark"
    ) {
      return `${label}.icons[].theme must be "light" or "dark"`;
    }
  }
  return null;
}

/**
 * Validate the `serverInfo` option against the spec's `Implementation`.
 * Returns a human-readable problem string, or `null` when valid
 * (including when `undefined`).
 *
 * The gateway's own default block is what this guards the wire against
 * replacing badly. A host builds this object once at mount time from
 * constants, so a problem here is wrong for every request rather than for
 * one caller's arguments, which is why `handleMcpRequest` throws on it
 * instead of answering the request with an error.
 */
export function describeServerInfoProblem(
  serverInfo: unknown,
): string | null {
  if (serverInfo === undefined) return null;
  if (!isPlainObject(serverInfo)) return "serverInfo must be an object";
  // `name` and `version` are the two the spec requires, and the default
  // block always carries both; an override that drops one would emit an
  // `Implementation` a validating client rejects outright.
  for (const key of ["name", "version"]) {
    const value = serverInfo[key];
    if (typeof value !== "string" || value.length === 0) {
      return `serverInfo.${key} must be a non-empty string`;
    }
  }
  for (const key of ["title", "description", "websiteUrl"]) {
    if (serverInfo[key] !== undefined && typeof serverInfo[key] !== "string") {
      return `serverInfo.${key} must be a string`;
    }
  }
  return describeIconsProblem(serverInfo.icons, "serverInfo");
}

/**
 * Validate MCP resource/content annotations. Returns a human-readable
 * problem string, or `null` when valid (including when `undefined`).
 * Exported so the `defineMcp*` helpers can fail loud at declaration time
 * with the same rules the request handler enforces on provider output.
 */
export function describeAnnotationsProblem(
  annotations: unknown,
): string | null {
  if (annotations === undefined) return null;
  if (!isPlainObject(annotations)) return "annotations must be an object";
  if (annotations.audience !== undefined) {
    if (
      !Array.isArray(annotations.audience) ||
      !annotations.audience.every((a) => a === "user" || a === "assistant")
    ) {
      return 'annotations.audience must be an array of "user" | "assistant"';
    }
  }
  if (annotations.priority !== undefined) {
    if (
      typeof annotations.priority !== "number" ||
      annotations.priority < 0 ||
      annotations.priority > 1
    ) {
      return "annotations.priority must be a number between 0 and 1";
    }
  }
  if (
    annotations.lastModified !== undefined &&
    typeof annotations.lastModified !== "string"
  ) {
    return "annotations.lastModified must be a string";
  }
  return null;
}

/**
 * Validate an MCP resource descriptor (a `resources/list` entry). Returns a
 * problem string or `null`. `uri` and `name` are required non-empty strings;
 * `title`/`description`/`mimeType` are optional strings; `size` is an
 * optional non-negative number; `annotations` is validated as above.
 */
export function describeResourceProblem(resource: unknown): string | null {
  if (!isPlainObject(resource)) return "resource must be an object";
  if (typeof resource.uri !== "string" || resource.uri.length === 0) {
    return "resource.uri must be a non-empty string";
  }
  if (typeof resource.name !== "string" || resource.name.length === 0) {
    return "resource.name must be a non-empty string";
  }
  for (const field of ["title", "description", "mimeType"] as const) {
    if (resource[field] !== undefined && typeof resource[field] !== "string") {
      return `resource.${field} must be a string`;
    }
  }
  if (
    resource.size !== undefined &&
    (typeof resource.size !== "number" ||
      !Number.isFinite(resource.size) ||
      resource.size < 0)
  ) {
    return "resource.size must be a non-negative number";
  }
  const iconsProblem = describeIconsProblem(resource.icons, "resource");
  if (iconsProblem) return iconsProblem;
  return describeAnnotationsProblem(resource.annotations);
}

/**
 * Validate an MCP resource template descriptor (a `resources/templates/list`
 * entry). Like `describeResourceProblem` but keyed on `uriTemplate` and
 * without `size`.
 */
export function describeResourceTemplateProblem(
  template: unknown,
): string | null {
  if (!isPlainObject(template)) return "resource template must be an object";
  if (
    typeof template.uriTemplate !== "string" ||
    template.uriTemplate.length === 0
  ) {
    return "template.uriTemplate must be a non-empty string";
  }
  if (typeof template.name !== "string" || template.name.length === 0) {
    return "template.name must be a non-empty string";
  }
  for (const field of ["title", "description", "mimeType"] as const) {
    if (template[field] !== undefined && typeof template[field] !== "string") {
      return `template.${field} must be a string`;
    }
  }
  const iconsProblem = describeIconsProblem(template.icons, "template");
  if (iconsProblem) return iconsProblem;
  return describeAnnotationsProblem(template.annotations);
}

/**
 * Validate the array a resource read handler returns. Must be an array; each
 * item needs a non-empty string `uri`, optional string `mimeType`, and at
 * least one of `text`/`blob` (each a string when present). Returns a problem
 * string or `null`.
 */
export function describeResourceContentsProblem(
  contents: unknown,
): string | null {
  if (!Array.isArray(contents)) {
    return "resource read result must be an array of content items";
  }
  for (const item of contents) {
    if (!isPlainObject(item)) return "each content item must be an object";
    if (typeof item.uri !== "string" || item.uri.length === 0) {
      return "content.uri must be a non-empty string";
    }
    if (item.mimeType !== undefined && typeof item.mimeType !== "string") {
      return "content.mimeType must be a string";
    }
    if (item.text !== undefined && typeof item.text !== "string") {
      return "content.text must be a string";
    }
    if (item.blob !== undefined && typeof item.blob !== "string") {
      return "content.blob must be a string";
    }
    if (item.text === undefined && item.blob === undefined) {
      return "content item must include text or blob";
    }
  }
  return null;
}

function publicResource(resource: RegisteredResource): McpResource {
  return {
    uri: resource.uri,
    name: resource.name,
    ...(resource.description !== undefined
      ? { description: resource.description }
      : {}),
    ...(resource.mimeType !== undefined ? { mimeType: resource.mimeType } : {}),
  };
}

/**
 * Project an arbitrary resource-shaped object down to exactly the known
 * `McpResource` fields. Applied to every `resources/list` entry before it
 * ships, so a provider's stray/internal keys never leak to the client (the
 * template path does the same via `pickTemplateFields`).
 */
function pickResourceFields(resource: McpResource): McpResource {
  return {
    uri: resource.uri,
    name: resource.name,
    ...(resource.title !== undefined ? { title: resource.title } : {}),
    ...(resource.description !== undefined
      ? { description: resource.description }
      : {}),
    ...(resource.mimeType !== undefined ? { mimeType: resource.mimeType } : {}),
    ...(resource.annotations !== undefined
      ? { annotations: resource.annotations }
      : {}),
    ...(resource.icons !== undefined ? { icons: resource.icons } : {}),
    ...(resource.size !== undefined ? { size: resource.size } : {}),
  };
}

/**
 * Project an arbitrary template-shaped object down to exactly the known
 * `McpResourceTemplate` fields. Shared by the request handler (response
 * shaping), `defineMcpResourceTemplate`, and the registry-sync projection so
 * the three never drift, and so a hand-built provider's extra keys never
 * reach the response or the registry's strict validator.
 */
export function pickTemplateFields(
  template: McpResourceTemplate,
): McpResourceTemplate {
  return {
    uriTemplate: template.uriTemplate,
    name: template.name,
    ...(template.title !== undefined ? { title: template.title } : {}),
    ...(template.description !== undefined
      ? { description: template.description }
      : {}),
    ...(template.mimeType !== undefined ? { mimeType: template.mimeType } : {}),
    ...(template.annotations !== undefined
      ? { annotations: template.annotations }
      : {}),
    ...(template.icons !== undefined ? { icons: template.icons } : {}),
  };
}

function registeredResourceCandidate(
  resource: RegisteredResource,
): ResourceCandidate {
  return {
    resource: publicResource(resource),
    metadata: resource.metadata ?? null,
  };
}

function shouldAuditResource(
  auditResources: McpResourceAuditOption | undefined,
  operation: "list" | "read" | "templatesList",
): boolean {
  if (auditResources === true) return true;
  if (!auditResources) return false;
  return auditResources[operation] === true;
}

/**
 * The caller half of a resource audit row. A union rather than two loose
 * fields, because `anonymous: true` always means a null subject: the two
 * are built together at the one place identity is resolved, so making
 * them disagree is not expressible.
 *
 * They are still two facts. `anonymous` routes the write, and is stripped
 * before the row crosses into the component; `identitySubject` is data on
 * the row. An authenticated request CAN carry a null subject, when a host
 * `resolveIdentity` returns an object without one, and its rows must be
 * kept: they are the signal that the resolver is misconfigured.
 */
type ResourceAuditCaller =
  | { anonymous: true; identitySubject: null }
  | { anonymous: false; identitySubject: string | null };

async function safeRecordResourceAudit(
  ctx: HandlerCtx,
  component: ComponentApi,
  entry: ResourceAuditCaller & {
    resourceUri?: string;
    resourceOperation: McpResourceOperation;
    args: unknown;
    outcome: "allowed" | "denied" | "error";
    durationMs: number;
    errorCode?: number;
    errorMessage?: string;
  },
): Promise<void> {
  // An anonymous outcome is recorded only when it is `allowed`.
  //
  // The reason resource denials went unaudited was that an unauthenticated
  // caller could grow the table without bound, and `resources/read` carries
  // a caller-controlled `uri`. `anonymousResources` would reopen exactly
  // that through the allow door: a public template accepts any expansion,
  // and every miss lands on the not-found branch, which does write a row.
  //
  // What this bounds is the number of rows an anonymous caller can cause
  // through a FAILING request, which is the unbounded part. It does not
  // bound successful ones: an allowed anonymous `resources/list` writes
  // one row per request like any other allowed outcome, which is what the
  // prune cron is for. The SIZE of a row is bounded separately, and lower
  // down, by `recordResourceEntry` in the component, so that it holds for
  // every writer of that public mutation rather than only for this one.
  if (entry.anonymous && entry.outcome !== "allowed") return;
  try {
    const { anonymous: _anonymous, ...row } = entry;
    await ctx.runMutation(component.audit.recordResourceEntry, row);
  } catch (err) {
    console.error(
      "[mcp-gateway] failed to record resource audit entry",
      entry.resourceOperation,
      entry.resourceUri ?? "(none)",
      entry.outcome,
      err,
    );
  }
}

export async function handleMcpRequest(
  ctx: HandlerCtx,
  request: Request,
  component: ComponentApi,
  options: InternalHandleMcpRequestOptions,
): Promise<Response> {
  // An empty task scope is a configuration error, not "unscoped": it
  // would be stored as its own third namespace, so a host writing
  // `scope: process.env.MOUNT_ID ?? ""` would silently get a namespace
  // nobody intended, unreachable from both scoped and unscoped mounts.
  // Fail at the mount rather than at the first poll.
  if (options.tasks?.scope === "") {
    throw new Error(
      "tasks.scope must be a non-empty string; omit it for an unscoped mount.",
    );
  }
  // Same class of error, same place: `serverInfo` is built once from
  // constants, so a malformed one is wrong for every request. It also
  // rides on EVERY result, `initialize` included, so emitting it and
  // letting the client decide costs a connection that fails with a zod
  // error on the other side of the wire instead of a named field here.
  const serverInfoProblem = describeServerInfoProblem(options.serverInfo);
  if (serverInfoProblem) throw new Error(serverInfoProblem);
  // Before the truthiness check below: `anonymousResources: process.env.X`
  // is the shape that turns "off" into on, and this option decides whether
  // unauthenticated callers are served at all. Same reasoning as the
  // `tasks.scope` check above, higher stakes.
  if (
    options.anonymousResources !== undefined &&
    typeof options.anonymousResources !== "boolean"
  ) {
    throw new Error("anonymousResources must be a boolean.");
  }
  if (options.anonymousResources) {
    // `safeAuthorizeResource` allows everything when no authorizer is
    // configured, so opting in without one would publish the entire
    // catalog rather than delegating the decision. Throw on the first
    // request rather than serve that; Convex offers no deploy-time hook,
    // so this is as early as the gateway can object. `!`, not
    // `=== undefined`, so the guard tests exactly what
    // `safeAuthorizeResource` tests and a config-shaped `null` cannot
    // slip through it.
    if (!options.authorizeResource) {
      throw new Error(
        "anonymousResources requires an authorizeResource callback; " +
          "without one every resource is allowed, so anonymous callers " +
          "would read the whole catalog.",
      );
    }
    // The hook's contract passes a non-null identity, and the read-side
    // MRTR chain it can open must bind to a principal. Rather than run it
    // with a null identity, or skip a hook the host installed as a gate,
    // refuse the combination on the first request through the mount.
    // `!= null`, so `null` is "no hook" here and at every other site that
    // reads this option: the two capability advertisements, the two
    // `-32601` branches, and the read case's own local. They used to test
    // `undefined` strictly, which meant a config-shaped `null` was absent
    // for this guard and installed for them, so such a mount advertised
    // and served a resources capability with no hook behind it. Treating
    // it as installed here would instead fail every request, preflight
    // included, over a hook the host never wrote.
    if (options.beforeResourceRead != null) {
      throw new Error(
        "anonymousResources cannot be combined with beforeResourceRead; " +
          "the hook requires an authenticated caller.",
      );
    }
  }
  // Before the preflight branch: telling a browser via CORS that a
  // cross-origin POST is permitted, only to 403 the POST itself, defeats
  // the point of preflight. A disallowed origin gets a bare 403 with no
  // CORS headers at all.
  const rejected = originRejection(request, options);
  if (rejected) return rejected;
  if (request.method === "OPTIONS") {
    return preflightResponse(options.cors, request);
  }
  let response: Response;
  switch (request.method) {
    case "POST":
      response = await handlePost(ctx, request, component, options);
      break;
    case "GET":
      response = new Response("Method Not Allowed", {
        status: 405,
        headers: { allow: "POST, DELETE, OPTIONS" },
      });
      break;
    case "DELETE":
      response = await handleDelete(ctx, request, component, options);
      break;
    default:
      response = new Response("Method Not Allowed", {
        status: 405,
        headers: { allow: "POST, DELETE, OPTIONS" },
      });
  }
  return withCors(response, options.cors, request);
}

/**
 * Resolve the caller's identity for a request. Used at three points:
 *   - `tools/list` and `tools/call`: identity drives audit + authorize
 *   - `initialize`: identity binds to the session row so DELETE later
 *     verifies teardown is authorised
 *   - `DELETE`: identity matches what was bound at create time
 *
 * Resolution order:
 *   1. `options.resolveIdentity` if configured AND a Bearer is present
 *   2. Convex's `ctx.auth.getUserIdentity()` (validates against
 *      `auth.config.ts`); `iss/aud` mismatches downgrade to null
 *      rather than 500 the request.
 */
async function resolveCallerIdentity(
  ctx: HandlerCtx,
  request: Request,
  options: HandleMcpRequestOptions,
): Promise<{ subject: string; claims?: Record<string, unknown> } | null> {
  if (options.resolveIdentity) {
    const authHeader = request.headers.get("authorization");
    const token = authHeader?.toLowerCase().startsWith("bearer ")
      ? authHeader.slice(7)
      : null;
    if (token) {
      try {
        // `?? null` normalizes here so every downstream check can be
        // strict. The signature says the resolver returns a caller or
        // null, but an untyped host can return `undefined`, and that
        // value used to reach two consumers that disagreed about it: the
        // authorizer treated it as anonymous, the audit rule as
        // authenticated. One shape at the boundary instead.
        return (await options.resolveIdentity(token)) ?? null;
      } catch (err) {
        console.warn(
          `[mcp-gateway] resolveIdentity threw; treating as anonymous. ` +
            `(${err instanceof Error ? err.message : String(err)})`,
        );
        return null;
      }
    }
    return null;
  }
  try {
    const raw = (await ctx.auth.getUserIdentity()) as
      | { subject?: string; [k: string]: unknown }
      | null
      | undefined;
    if (raw && typeof raw.subject === "string") {
      return { subject: raw.subject, claims: raw };
    }
  } catch (err) {
    console.warn(
      `[mcp-gateway] ctx.auth.getUserIdentity() threw; treating as anonymous. ` +
        `Likely a Bearer token whose iss/aud doesn't match auth.config.ts. ` +
        `(${err instanceof Error ? err.message : String(err)})`,
    );
  }
  return null;
}

async function handleDelete(
  ctx: HandlerCtx,
  request: Request,
  component: ComponentApi,
  options: HandleMcpRequestOptions,
): Promise<Response> {
  const sessionId = request.headers.get("mcp-session-id");
  if (!sessionId) {
    return new Response("Missing Mcp-Session-Id header", { status: 400 });
  }
  // Identity-bound DELETE: the session row remembers the subject that
  // initialised it. Teardown must come from the same subject (or both
  // sides anonymous), otherwise a leaked session-id alone is enough
  // to DoS an authenticated user's session.
  const identity = await resolveCallerIdentity(ctx, request, options);
  const result = await ctx.runMutation(component.sessions.deleteSession, {
    sessionId,
    callerIdentitySubject: identity?.subject ?? null,
  });
  if (result === "deleted") return new Response(null, { status: 200 });
  if (result === "not_found") return new Response(null, { status: 404 });
  return new Response(
    "Forbidden: caller identity does not match session owner",
    { status: 403 },
  );
}

async function handlePost(
  ctx: HandlerCtx,
  request: Request,
  component: ComponentApi,
  options: InternalHandleMcpRequestOptions,
): Promise<Response> {
  // MCP 2025-06-18 §"Sending Messages to the Server": clients MUST set
  // Accept to list both application/json and text/event-stream. Enforcing
  // this surfaces interop bugs early instead of silently degrading to
  // JSON-only.
  const accept = (request.headers.get("accept") ?? "").toLowerCase();
  if (
    !accept.includes("application/json") ||
    !accept.includes("text/event-stream")
  ) {
    return new Response(
      "Not Acceptable: Accept header must list both application/json and text/event-stream",
      { status: 406 },
    );
  }

  let message: JsonRpcMessage | JsonRpcMessage[];
  try {
    message = (await request.json()) as JsonRpcMessage | JsonRpcMessage[];
  } catch {
    // Per MCP §"Sending Messages": server SHOULD return an HTTP error
    // status when it cannot accept the input. JSON-RPC body retained
    // for clients that read it.
    return new Response(jsonErrorEnvelope(null, -32700, "Parse error"), {
      status: 400,
      headers: { "content-type": "application/json" },
    });
  }

  // MCP forbids batched requests over Streamable HTTP. Clearer error
  // than the previous "missing method or id" fall-through.
  if (Array.isArray(message)) {
    return new Response(
      jsonErrorEnvelope(
        null,
        -32600,
        "Batched JSON-RPC requests are not supported in MCP Streamable HTTP",
      ),
      { status: 400, headers: { "content-type": "application/json" } },
    );
  }

  const isInitialize = message.method === "initialize";
  const headerProtocolVersion = request.headers.get("mcp-protocol-version");
  const metadataProtocolVersion = statelessProtocolVersion(message);
  // The era is decided by what the request declares, for every method
  // including `initialize`. That revision removed `initialize`, so a
  // request that declares it and calls it anyway is refused below like
  // its four siblings (`ping`, `logging/setLevel`, `resources/subscribe`,
  // `resources/unsubscribe`) rather than being handed a session it did
  // not ask for. A client that declares nothing modern still negotiates
  // exactly as it always has.
  const isStateless =
    headerProtocolVersion === STATELESS_PROTOCOL_VERSION ||
    metadataProtocolVersion !== null;
  // The validated `clientCapabilities` object of a stateless request,
  // hoisted so MRTR and task-augmented `tools/call` can both negotiate
  // against it below.
  let statelessClientCapabilities: Record<string, unknown> | null = null;

  // The 2026 protocol moves protocol negotiation to each request. Check the
  // mirrored routing metadata before identity resolution, catalog writes,
  // authorization, auditing, or tool dispatch.
  if (isStateless) {
    // A request that declares the modern wire and then carries no usable
    // `_meta` is malformed, not mismatched. SEP-2575 asks for `-32602`
    // here, and the distinction is the useful one: a mismatch tells a
    // client its two declarations disagree, which is unhelpful advice
    // when it made only one of them. Checked before the comparison
    // because the comparison is what would otherwise answer, having
    // found a `null` on one side.
    if (metadataProtocolVersion === null) {
      return statelessErrorResponse(
        message.id,
        INVALID_PARAMS,
        isPlainObject(message.params?._meta)
          ? "Invalid params: _meta must carry a string " +
              "io.modelcontextprotocol/protocolVersion"
          : "Invalid params: a stateless request must carry _meta with " +
              "io.modelcontextprotocol/protocolVersion",
      );
    }
    if (headerProtocolVersion !== metadataProtocolVersion) {
      return statelessErrorResponse(
        message.id,
        HEADER_MISMATCH,
        "MCP-Protocol-Version must exactly match request metadata",
      );
    }
    if (metadataProtocolVersion !== STATELESS_PROTOCOL_VERSION) {
      // A session revision named here is supported by this server and
      // still wrong in this position, so say which of the two it is.
      // `supported` lists both eras, and a message that flatly called a
      // listed version unsupported would contradict its own data.
      const isSessionRevision =
        metadataProtocolVersion !== null &&
        (SESSION_PROTOCOL_VERSIONS as readonly string[]).includes(
          metadataProtocolVersion,
        );
      return statelessErrorResponse(
        message.id,
        UNSUPPORTED_PROTOCOL_VERSION,
        isSessionRevision
          ? `MCP protocol version ${metadataProtocolVersion} is session-based ` +
              `and cannot be selected through stateless request metadata; ` +
              `negotiate it with initialize instead`
          : `Unsupported MCP protocol version: ${metadataProtocolVersion}`,
        {
          supported: [STATELESS_PROTOCOL_VERSION, ...SESSION_PROTOCOL_VERSIONS],
          requested: metadataProtocolVersion,
        },
      );
    }
    const metadata = message.params?._meta;
    const clientInfo = metadata?.["io.modelcontextprotocol/clientInfo"];
    const clientCapabilities =
      metadata?.["io.modelcontextprotocol/clientCapabilities"];
    if (
      !isPlainObject(clientCapabilities) ||
      (clientInfo !== undefined &&
        (!isPlainObject(clientInfo) ||
          typeof clientInfo.name !== "string" ||
          typeof clientInfo.version !== "string"))
    ) {
      return statelessErrorResponse(
        message.id,
        INVALID_PARAMS,
        "Invalid required stateless request metadata",
      );
    }
    statelessClientCapabilities = clientCapabilities;
    if (request.headers.get("mcp-method") !== message.method) {
      return statelessErrorResponse(
        message.id,
        HEADER_MISMATCH,
        "Mcp-Method must exactly match the JSON-RPC method",
      );
    }
    if (!statelessNameMatches(message, request)) {
      return statelessErrorResponse(
        message.id,
        HEADER_MISMATCH,
        "Mcp-Name must exactly match the JSON-RPC request name",
      );
    }
  }

  // Resolve identity once at the boundary and reuse it everywhere below
  // (the requireAuth gate, stale-session cleanup, audit subject, the
  // authorize callback input, and the session-binding subject). One
  // resolution avoids a duplicate resolveIdentity/userinfo round-trip.
  const identity = await resolveCallerIdentity(ctx, request, options);
  const auditIdentitySubject = identity?.subject ?? null;
  /**
   * The caller half of every resource audit row, built once so the two
   * fields cannot disagree. They are not the same fact: `anonymous`
   * routes the write (a failing anonymous outcome is dropped), while
   * `identitySubject` is data on the row. A host `resolveIdentity` that
   * returns an object with no `subject` produces an AUTHENTICATED request
   * with a null subject, so deriving one from the other would delete that
   * host's denial trail.
   */
  const auditCaller =
    identity == null
      ? ({ anonymous: true, identitySubject: null } as const)
      : { anonymous: false as const, identitySubject: auditIdentitySubject };

  // requireAuth gate: challenge anonymous POSTs with 401 before session
  // handling / the method switch, so browser MCP clients (claude.ai)
  // get the 401 + WWW-Authenticate they need to begin OAuth discovery
  // instead of a 200 empty tools/list. Opt-in; default behaviour
  // (200 with the filtered catalog) is unchanged. See
  // HandleMcpRequestOptions.requireAuth.
  if (options.requireAuth && identity === null) {
    return await requireAuthChallenge(ctx, request, component, message.id);
  }

  // Declarative catalog synchronization can write component state. Modern
  // traffic must remain stateless at the protocol level, but anonymous
  // requests must not be able to trigger those writes before requireAuth.
  if (isStateless) {
    try {
      await ensureCatalogSynced(options);
    } catch (err) {
      console.error("[mcp-gateway] declarative catalog sync failed", err);
      return statelessErrorResponse(
        message.id,
        INTERNAL_ERROR,
        "Failed to synchronize the declarative catalog",
        {},
        500,
      );
    }
  }

  // MCP-Protocol-Version header: required on post-initialize requests
  // by spec. Missing → silently default to DEFAULT_PROTOCOL_VERSION
  // (legacy clients). Unsupported value → MUST 400 per spec.
  if (!isInitialize && !isStateless) {
    const protoHeader = headerProtocolVersion;
    if (
      protoHeader !== null &&
      !(SESSION_NEGOTIABLE_VERSIONS as readonly string[]).includes(protoHeader)
    ) {
      return new Response(
        `Unsupported MCP-Protocol-Version: ${protoHeader}. ` +
          `Supported: ${SESSION_NEGOTIABLE_VERSIONS.join(", ")}`,
        { status: 400 },
      );
    }
  }

  // Session validation. Initialize creates a fresh session. All other
  // requests must carry a valid Mcp-Session-Id; missing is 400, unknown
  // is 404 (per MCP 2025-06-18 §Session Management).
  let sessionId = "";
  let issueSessionHeader = false;
  // The identity bound to the session at create time (undefined for the
  // initialize path and for legacy pre-binding rows). Used to identity-bind
  // session-scoped mutations like resources/subscribe.
  let sessionOwnerSubject: string | null | undefined;

  if (isStateless) {
    // Modern MCP is deliberately stateless. Ignore a legacy session id rather
    // than looking it up, touching it, or echoing it back to the client.
  } else if (isInitialize) {
    sessionId = generateSessionId();
    issueSessionHeader = true;
    // Best-effort: if the caller `initialize`s while carrying an old
    // session id (buggy client reconnecting without DELETE first),
    // drop the old row so the sessions table doesn't grow unbounded.
    // The deleteSession mutation enforces the identity check, so a
    // mismatched subject (e.g. an attacker who learned someone
    // else's id and tries to re-initialize) cannot evict the
    // original session, only the legitimate owner can.
    const staleSessionId = request.headers.get("mcp-session-id");
    if (staleSessionId) {
      try {
        await ctx.runMutation(component.sessions.deleteSession, {
          sessionId: staleSessionId,
          callerIdentitySubject: identity?.subject ?? null,
        });
      } catch (err) {
        console.warn(
          "[mcp-gateway] failed to clean up stale session on re-initialize",
          err,
        );
      }
    }
  } else {
    const headerSessionId = request.headers.get("mcp-session-id");
    if (!headerSessionId) {
      return new Response("Missing Mcp-Session-Id header", { status: 400 });
    }
    const session = await ctx.runQuery(component.sessions.getSession, {
      sessionId: headerSessionId,
    });
    if (!session) {
      return new Response("Unknown or terminated session", { status: 404 });
    }
    sessionId = headerSessionId;
    sessionOwnerSubject = session.identitySubject;
    try {
      await ctx.runMutation(component.sessions.touchSession, {
        sessionId: headerSessionId,
      });
    } catch (err) {
      // Touch is best-effort; a stuck lastSeenAt only matters for the
      // session pruner, not for the current request. Log so a
      // systematic failure (schema drift, recurring conflict) is
      // discoverable in the deployment log.
      console.warn(
        "[mcp-gateway] touchSession failed (best-effort)",
        headerSessionId,
        err,
      );
    }
  }

  if (isStateless && isJsonRpcNotificationOrResponse(message)) {
    return statelessErrorResponse(
      message.id,
      -32600,
      "Client notifications are not supported by this server",
    );
  }

  // Notifications / responses: 202 Accepted, no body.
  if (isJsonRpcNotificationOrResponse(message)) {
    const headers: Record<string, string> = {};
    if (issueSessionHeader) headers["mcp-session-id"] = sessionId;
    return new Response(null, { status: 202, headers });
  }

  if (!isJsonRpcRequest(message)) {
    return new Response(
      jsonErrorEnvelope(null, -32600, "Invalid Request: missing method or id"),
      { status: 400, headers: { "content-type": "application/json" } },
    );
  }

  let body: string = jsonErrorEnvelope(
    message.id,
    INTERNAL_ERROR,
    "Handler did not produce a response",
  );
  let raw: Response | null = null;
  let responseStatus = 200;

  switch (message.method) {
    case "initialize": {
      if (isStateless) {
        responseStatus = 404;
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `${message.method} is legacy-only; it was removed in ` +
            STATELESS_PROTOCOL_VERSION,
        );
        break;
      }
      // Legacy clients reconcile their declarative catalog when they
      // initialize. Modern requests do the same work before dispatch above.
      try {
        await ensureCatalogSynced(options);
      } catch (err) {
        console.error(
          "[mcp-gateway] declarative catalog sync failed during initialize",
          err,
        );
        throw err;
      }
      const registeredResources = (await ctx.runQuery(
        component.registry.listResources,
        {},
      )) as RegisteredResource[];
      const registeredTemplates = (await ctx.runQuery(
        component.registry.listResourceTemplates,
        {},
      )) as RegisteredResourceTemplate[];
      const requested = message.params?.protocolVersion;
      const negotiated =
        typeof requested === "string" &&
        (SESSION_NEGOTIABLE_VERSIONS as readonly string[]).includes(requested)
          ? requested
          : DEFAULT_PROTOCOL_VERSION;
      await ctx.runMutation(component.sessions.createSession, {
        sessionId,
        protocolVersion: negotiated,
        identitySubject: auditIdentitySubject,
      });
      // Advertise the resources capability when any resource feature is
      // configured. The subscribe/listChanged flags are added only when the
      // host opts in (and thus has a transport that can deliver); otherwise
      // the capability stays `{}`, the historical, accurate default.
      const advertiseResources =
        registeredResources.length > 0 ||
        registeredTemplates.length > 0 ||
        (options.resources ?? []).length > 0 ||
        (options.resourceTemplates ?? []).length > 0 ||
        // A hook-only mount serves reads with no provider and no template
        // (ask, then answer with `completeRead`), so it has the capability
        // even with an empty catalog.
        options.beforeResourceRead != null ||
        Boolean(options.resourceSubscriptions?.subscribe) ||
        Boolean(options.resourceSubscriptions?.listChanged);
      const anonymousOnOptedInMount =
        identity === null && Boolean(options.anonymousResources);
      body = jsonResultEnvelope(message.id, {
        protocolVersion: negotiated,
        serverInfo: options.serverInfo ?? {
          name: SERVER_NAME,
          version: SERVER_VERSION,
        },
        ...(options.initializeInstructions
          ? { instructions: options.initializeInstructions }
          : {}),
        capabilities: {
          tools: {},
          ...(advertiseResources
            ? {
                resources: {
                  // `subscribe` is withheld from an anonymous session on
                  // an opted-in mount: there, resource methods DO serve
                  // it, so advertising a method it will be refused with
                  // -32001 is a promise the mount cannot keep. The
                  // refusal is permanent, not merely current: the session
                  // row binds `identitySubject` here, and
                  // `resources/subscribe` requires both an identity and a
                  // match against that owner, so a session that
                  // initialized anonymously is refused for its whole life
                  // even if the client starts sending a valid token.
                  //
                  // `listChanged` is NOT withheld, because it names no
                  // method the caller invokes. It is a broadcast the host
                  // emits, and an anonymous session on an opted-in mount
                  // can list, so it is exactly the flag still worth
                  // having: withhold it and a spec-compliant client never
                  // re-lists and serves a stale catalog for the life of
                  // the connection.
                  //
                  // Neither is withheld on a mount WITHOUT the option.
                  // There every resource method already refuses this
                  // caller, so the block is moot for it, and the
                  // advertisement is what every release so far sent.
                  // Narrowing that is a separate change.
                  ...(options.resourceSubscriptions?.subscribe &&
                  !anonymousOnOptedInMount
                    ? { subscribe: true }
                    : {}),
                  ...(options.resourceSubscriptions?.listChanged
                    ? { listChanged: true }
                    : {}),
                },
              }
            : {}),
        },
      });
      break;
    }

    case "server/discover": {
      if (!isStateless) {
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `Unsupported method: ${message.method}`,
        );
        break;
      }
      const registeredResources = (await ctx.runQuery(
        component.registry.listResources,
        {},
      )) as RegisteredResource[];
      const registeredTemplates = (await ctx.runQuery(
        component.registry.listResourceTemplates,
        {},
      )) as RegisteredResourceTemplate[];
      const advertiseResources =
        registeredResources.length > 0 ||
        registeredTemplates.length > 0 ||
        (options.resources ?? []).length > 0 ||
        (options.resourceTemplates ?? []).length > 0 ||
        // As on `initialize`: a hook-only mount can serve a read.
        options.beforeResourceRead != null;
      body = jsonResultEnvelope(message.id, {
        resultType: "complete",
        supportedVersions: [
          STATELESS_PROTOCOL_VERSION,
          ...SESSION_PROTOCOL_VERSIONS,
        ],
        ...(options.initializeInstructions
          ? { instructions: options.initializeInstructions }
          : {}),
        capabilities: {
          tools: {},
          ...(advertiseResources ? { resources: {} } : {}),
          // Opt-in negotiation: tasks are advertised only when the host
          // configured task execution, per the extension contract, and
          // under `extensions` where SEP-2663 puts it. The capability
          // object itself is empty: `pollIntervalMs` is a property of
          // each task, and rides on the task.
          ...(options.tasks
            ? { extensions: { [TASKS_CAPABILITY_KEY]: {} } }
            : {}),
        },
      });
      break;
    }

    case "resources/list": {
      const start = Date.now();
      const providers = options.resources ?? [];
      const templates = options.resourceTemplates ?? [];
      const registeredResources = (await ctx.runQuery(
        component.registry.listResources,
        {},
      )) as RegisteredResource[];
      if (
        providers.length === 0 &&
        registeredResources.length === 0 &&
        templates.length === 0 &&
        // A hook-only mount advertises the `resources` capability, and
        // `resources/list` is that capability's base method: refusing it
        // would leave a client that lists on connect with an error for a
        // feature the handshake just promised. Its catalog is genuinely
        // empty, so it lists as empty rather than as unsupported.
        options.beforeResourceRead == null
      ) {
        if (isStateless) responseStatus = 404;
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `Unsupported method: ${message.method}`,
        );
        break;
      }
      if (!identity && !options.anonymousResources) {
        // Intentionally NOT audited on the anonymous deny path. An
        // unauthenticated caller can `initialize` once then spam resource
        // requests with no Bearer; auditing the denials would let them grow
        // the audit table without bound (and `resources/read` carries a
        // caller-controlled `uri`). Mirrors the unknown-tool path in
        // dispatch.ts: only authenticated outcomes are audited.
        //
        // With `anonymousResources` set the caller falls through to the
        // authorizer instead, under `"resource_anonymous"`. The audit rule
        // survives that: `safeRecordResourceAudit` still writes nothing for
        // an anonymous outcome that is not `allowed`.
        body = jsonErrorEnvelope(
          message.id,
          UNAUTHORIZED,
          "Unauthorized: authentication required",
        );
        break;
      }
      try {
        // Isolate each provider: a single provider that throws must not
        // collapse the whole catalog. Mirrors the per-item isolation
        // tools/list uses for authorizer throws: a buggy provider hides
        // only its own resources, the healthy providers still list.
        const providerResources = (
          await Promise.all(
            providers.map(async (provider) => {
              try {
                return await provider.list(ctx, { identity });
              } catch (err) {
                console.error(
                  "[mcp-gateway] resource provider threw during resources/list",
                  provider.name,
                  err,
                );
                return [];
              }
            }),
          )
        ).flat();
        // Validate provider output before it reaches the client. A
        // structurally invalid descriptor is a provider bug, so fail the
        // whole list loudly with a deterministic -32603 (caught below)
        // rather than ship malformed JSON-RPC.
        for (const resource of providerResources) {
          const problem = describeResourceProblem(resource);
          if (problem) {
            throw new ResourceContractError(
              `resources/list provider returned an invalid resource: ${problem}`,
            );
          }
        }
        const metadataByUri = new Map(
          registeredResources.map((resource) => [
            resource.uri,
            resource.metadata ?? null,
          ]),
        );
        const candidates = dedupeResourceCandidates([
          ...registeredResources.map(registeredResourceCandidate),
          ...providerResources.map((resource) => ({
            resource,
            metadata: metadataByUri.get(resource.uri) ?? null,
          })),
        ]);
        const resources = [];
        // Whether the host asked, on any denial, for the caller to
        // authenticate. Same `unauth`-shaped convention `resources/read`
        // uses; on a list the per-candidate reasons are otherwise
        // discarded, so this is the only thing carried out of the loop.
        let authWouldHelp = false;
        for (const candidate of candidates) {
          const { decision, threw } = await safeAuthorizeResource(
            options.authorizeResource,
            ctx,
            resourceAuthorizerArgs(
              "list",
              candidate.resource.uri,
              candidate.metadata,
              identity,
            ),
          );
          // A malformed return is logged alongside a throw, not treated
          // as a policy denial. Both are host faults, and a filtered list
          // has nowhere to report one: the candidate is simply omitted, so
          // a host that forgets a `return` in its `resource_anonymous`
          // branch would otherwise see an empty `resources/list` with no
          // diagnostic anywhere. Per-item isolation still stands and the
          // response shape does not change.
          if (threw || decision.reason === AUTHORIZER_INVALID_SHAPE_REASON) {
            console.error(
              "[mcp-gateway] resource authorizer failed during resources/list for resource",
              candidate.resource.uri,
              decision.reason,
            );
          }
          if (decision.allowed) {
            resources.push(pickResourceFields(candidate.resource));
          } else if (
            !threw &&
            decision.reason !== undefined &&
            /^unauth/i.test(decision.reason)
          ) {
            authWouldHelp = true;
          }
        }
        // An anonymous caller the host granted NOTHING, having said
        // authentication would help, is challenged rather than handed an
        // empty catalog. Before `anonymousResources` existed this method
        // answered `-32001` and the client knew to re-authenticate; a
        // client whose Bearer merely expired would otherwise now get an
        // empty HTTP 200 and no signal at all, and the capability set is
        // fixed at `initialize`, so it could stay that way for the life of
        // the connection. Requiring an EMPTY result keeps a mixed mount
        // quiet: a caller that got the public subset is not told to log in.
        const challengeList =
          identity == null && resources.length === 0 && authWouldHelp;
        if (shouldAuditResource(options.auditResources, "list")) {
          await safeRecordResourceAudit(ctx, component, {
            resourceOperation: "list",
            args: { resourceCount: resources.length },
            // `denied` when an ANONYMOUS caller got nothing it asked
            // for, which is what lets the suppression in
            // `safeRecordResourceAudit` fire: a hardcoded `allowed` never
            // did, so an unauthenticated client looping this method would
            // otherwise write one row per request even when the authorizer
            // granted it nothing. Scoped to anonymous callers, so an authenticated
            // caller's rows keep the outcome every release so far
            // recorded.
            outcome:
              auditCaller.anonymous &&
              candidates.length > 0 &&
              resources.length === 0
                ? "denied"
                : "allowed",
            ...auditCaller,
            durationMs: Date.now() - start,
          });
        }
        if (challengeList) {
          raw = await requireAuthChallenge(ctx, request, component, message.id);
          body = "";
          break;
        }
        if (isStateless) {
          resources.sort((a, b) => a.uri.localeCompare(b.uri));
        }
        body = jsonResultEnvelope(message.id, { resources });
      } catch (err) {
        const { full, wire } = splitErrorText(err, GENERIC_RESOURCE_LIST_ERROR);
        console.error("[mcp-gateway] resources/list failed", err);
        if (shouldAuditResource(options.auditResources, "list")) {
          await safeRecordResourceAudit(ctx, component, {
            resourceOperation: "list",
            args: null,
            outcome: "error",
            ...auditCaller,
            durationMs: Date.now() - start,
            errorCode: INTERNAL_ERROR,
            errorMessage: full,
          });
        }
        body = jsonErrorEnvelope(message.id, INTERNAL_ERROR, wire);
      }
      break;
    }

    case "resources/templates/list": {
      const start = Date.now();
      const providers = options.resourceTemplates ?? [];
      const registeredTemplates = (await ctx.runQuery(
        component.registry.listResourceTemplates,
        {},
      )) as RegisteredResourceTemplate[];
      // Distinct capability surface: unsupported only when NO templates are
      // configured at all (no runtime providers and none registered). A
      // registry-only template catalog is fully supported.
      if (providers.length === 0 && registeredTemplates.length === 0) {
        if (isStateless) responseStatus = 404;
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `Unsupported method: ${message.method}`,
        );
        break;
      }
      if (!identity && !options.anonymousResources) {
        // Not audited on the anonymous deny path, see the resources/list
        // rationale: anonymous spam must never grow the audit table.
        body = jsonErrorEnvelope(
          message.id,
          UNAUTHORIZED,
          "Unauthorized: authentication required",
        );
        break;
      }
      try {
        // Merge registered templates with runtime providers, deduped by
        // uriTemplate; a runtime provider wins (it's live and carries the
        // read handler), mirroring how resources/list prefers providers.
        const byUriTemplate = new Map<string, McpResourceTemplate>();
        for (const row of registeredTemplates) {
          byUriTemplate.set(row.uriTemplate, row);
        }
        for (const provider of providers) {
          byUriTemplate.set(provider.template.uriTemplate, provider.template);
        }
        const resourceTemplates = [];
        // See `resources/list`: the host's `unauth`-shaped denial is the
        // only per-candidate reason that survives the loop.
        let authWouldHelp = false;
        for (const template of byUriTemplate.values()) {
          const problem = describeResourceTemplateProblem(template);
          if (problem) {
            throw new ResourceContractError(
              `resources/templates/list provider returned an invalid template: ${problem}`,
            );
          }
          const { decision, threw } = await safeAuthorizeResource(
            options.authorizeResource,
            ctx,
            resourceAuthorizerArgs(
              "templates_list",
              template.uriTemplate,
              null,
              identity,
            ),
          );
          // Same as `resources/list`: a malformed return is a host fault
          // with nowhere to surface in a filtered list, so log it.
          if (threw || decision.reason === AUTHORIZER_INVALID_SHAPE_REASON) {
            console.error(
              "[mcp-gateway] resource authorizer failed during resources/templates/list for template",
              template.uriTemplate,
              decision.reason,
            );
          }
          if (decision.allowed) {
            resourceTemplates.push(pickTemplateFields(template));
          } else if (
            !threw &&
            decision.reason !== undefined &&
            /^unauth/i.test(decision.reason)
          ) {
            authWouldHelp = true;
          }
        }
        // Same rule and same reasoning as `resources/list`.
        const challengeTemplates =
          identity == null && resourceTemplates.length === 0 && authWouldHelp;
        if (shouldAuditResource(options.auditResources, "templatesList")) {
          await safeRecordResourceAudit(ctx, component, {
            resourceOperation: "templates_list",
            args: { resourceTemplateCount: resourceTemplates.length },
            // Same scoping as `resources/list`.
            outcome:
              auditCaller.anonymous &&
              byUriTemplate.size > 0 &&
              resourceTemplates.length === 0
                ? "denied"
                : "allowed",
            ...auditCaller,
            durationMs: Date.now() - start,
          });
        }
        if (challengeTemplates) {
          raw = await requireAuthChallenge(ctx, request, component, message.id);
          body = "";
          break;
        }
        if (isStateless) {
          resourceTemplates.sort((a, b) =>
            a.uriTemplate.localeCompare(b.uriTemplate),
          );
        }
        body = jsonResultEnvelope(message.id, { resourceTemplates });
      } catch (err) {
        const { full, wire } = splitErrorText(
          err,
          GENERIC_RESOURCE_TEMPLATES_LIST_ERROR,
        );
        console.error("[mcp-gateway] resources/templates/list failed", err);
        if (shouldAuditResource(options.auditResources, "templatesList")) {
          await safeRecordResourceAudit(ctx, component, {
            resourceOperation: "templates_list",
            args: null,
            outcome: "error",
            ...auditCaller,
            durationMs: Date.now() - start,
            errorCode: INTERNAL_ERROR,
            errorMessage: full,
          });
        }
        body = jsonErrorEnvelope(message.id, INTERNAL_ERROR, wire);
      }
      break;
    }

    case "resources/read": {
      const start = Date.now();
      const providers = options.resources ?? [];
      const templates = options.resourceTemplates ?? [];
      const registeredResources = (await ctx.runQuery(
        component.registry.listResources,
        {},
      )) as RegisteredResource[];
      if (
        providers.length === 0 &&
        registeredResources.length === 0 &&
        templates.length === 0 &&
        // A mount whose only read logic is the MRTR hook (ask, then answer
        // with `completeRead`) serves reads with no provider and no
        // template, so "nothing is registered" is not the same as "reads
        // are unsupported here".
        options.beforeResourceRead == null
      ) {
        if (isStateless) responseStatus = 404;
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `Unsupported method: ${message.method}`,
        );
        break;
      }
      if (!identity && !options.anonymousResources) {
        // Not audited on the anonymous deny path, see the resources/list
        // rationale. This matters most here: the denied `read` carries a
        // caller-controlled `uri`, so auditing would let an unauthenticated
        // client grow the table with arbitrary large URIs after one
        // `initialize`.
        body = jsonErrorEnvelope(
          message.id,
          UNAUTHORIZED,
          "Unauthorized: authentication required",
        );
        break;
      }
      const uri = message.params?.uri;
      if (typeof uri !== "string" || uri.length === 0) {
        if (shouldAuditResource(options.auditResources, "read")) {
          await safeRecordResourceAudit(ctx, component, {
            resourceOperation: "read",
            args: null,
            outcome: "error",
            ...auditCaller,
            durationMs: Date.now() - start,
            errorCode: INVALID_PARAMS,
            errorMessage: "Missing resource uri",
          });
        }
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          "Missing resource uri",
        );
        break;
      }
      const metadata =
        registeredResources.find((resource) => resource.uri === uri)
          ?.metadata ?? null;
      const resourceAuthz = await safeAuthorizeResource(
        options.authorizeResource,
        ctx,
        resourceAuthorizerArgs("read", uri, metadata, identity),
      );
      if (!resourceAuthz.decision.allowed) {
        const reason = resourceAuthz.decision.reason ?? "Forbidden";
        // A malformed return is a host BUG, not a policy denial. It is the
        // likeliest first-day failure of a new `resource_anonymous`
        // branch: forget one `return` and this is what arrives.
        const malformed = reason === AUTHORIZER_INVALID_SHAPE_REASON;
        if (resourceAuthz.threw || malformed) {
          // Unconditional, for every caller: the two sibling list
          // branches already log this way, and the read branch relied on
          // the audit row alone, which for an anonymous caller is dropped.
          // So this was the one authorizer fault that could leave no trace
          // anywhere. Adding a log changes no response.
          console.error(
            "[mcp-gateway] resource authorizer failed during resources/read",
            uri,
            reason,
          );
        }
        // Reading a malformed return as a FAULT rather than a denial is
        // scoped to an anonymous caller, which is the case this option
        // creates. For an authenticated one it would move the wire code
        // from -32003 to -32603 and the audit outcome from `denied` to
        // `error` on a mount that opted into nothing, and a host alerting
        // on either would start firing. That the internal shape text
        // still reaches an authenticated caller is a separate,
        // pre-existing issue.
        const faulted =
          resourceAuthz.threw || (malformed && identity == null);
        const code = faulted
          ? INTERNAL_ERROR
          : /^unauth/i.test(reason)
            ? UNAUTHORIZED
            : FORBIDDEN;
        if (shouldAuditResource(options.auditResources, "read")) {
          await safeRecordResourceAudit(ctx, component, {
            resourceUri: uri,
            resourceOperation: "read",
            args: null,
            outcome: faulted ? "error" : "denied",
            ...auditCaller,
            durationMs: Date.now() - start,
            errorCode: code,
            errorMessage: reason,
          });
        }
        // An ANONYMOUS caller denied with an `unauth`-shaped reason gets
        // the same 401 + `WWW-Authenticate` a denied anonymous
        // `tools/call` gets, because a browser MCP client reacts to that
        // status and to nothing else. Without it, a mount that serves
        // some resources publicly and gates the rest could never tell
        // such a client to log in, which is the whole point of mixing
        // the two. Scoped to anonymous callers: an authenticated one
        // keeps the HTTP 200 JSON-RPC body every release so far
        // returned, and changing that is not this option's business.
        if (identity == null && code === UNAUTHORIZED) {
          raw = await requireAuthChallenge(
            ctx,
            request,
            component,
            message.id,
            reason,
          );
          body = "";
          break;
        }
        // Same split as the `tools/call` denial: a returned reason is
        // host-authored and goes to the caller, while a thrown or
        // malformed one carries text the caller must not see. That text
        // goes to the deployment log above, and to the audit row only for
        // an authenticated caller: an anonymous fault is `outcome:
        // "error"`, which the audit rule drops.
        body = jsonErrorEnvelope(
          message.id,
          code,
          faulted ? GENERIC_AUTHORIZER_ERROR : reason,
        );
        break;
      }
      // MRTR for `resources/read`: the spec allows a read to answer with an
      // `InputRequiredResult`. Placed after `authorizeResource` allowed the
      // read and before any provider or template runs, so a hook question
      // never reveals a resource the caller may not read, and a refusal
      // after the answer costs no provider work.
      const requestState = message.params?.requestState;
      const inputResponses = message.params?.inputResponses;
      // `?? undefined` so a config-shaped `null` is absent here exactly as
      // it is at the five `options.beforeResourceRead` predicates above.
      // The advertisement and the -32601 branches read this option too, so
      // one of them disagreeing would advertise, and serve, a resources
      // capability whose only justification was a hook nobody installed.
      const beforeResourceRead = options.beforeResourceRead ?? undefined;
      const hasContinuation =
        requestState !== undefined || inputResponses !== undefined;
      if (hasContinuation && !beforeResourceRead) {
        // Fail closed rather than serve the resource while ignoring the
        // continuation the client believes it is answering.
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          "This gateway does not support MRTR continuations for resources/read",
        );
        break;
      }
      if (beforeResourceRead) {
        if (!identity) {
          // Unreachable: the mount refuses `anonymousResources` together
          // with `beforeResourceRead`. Kept as the fail-closed answer if
          // that ever stops holding, because the hook's contract promises
          // a principal and the chain it can open must bind to one.
          body = jsonErrorEnvelope(
            message.id,
            UNAUTHORIZED,
            "Unauthorized: authentication required",
          );
          break;
        }
        // The chain is keyed on the operation AND the concrete URI, so a
        // tool continuation can never be presented as a read continuation,
        // and a continuation minted for one template expansion cannot be
        // replayed against another (an expansion is a function of the URI).
        const chainName = `resources/read:${uri}`;
        const uriDigest = await sha256Base64Url(stableJson(uri));
        let continuation: VerifiedMrtrState | null = null;
        let chain: MrtrChainState = { status: "open" };
        let requestDigest: string | undefined;
        if (hasContinuation) {
          if (!isStateless) {
            body = jsonErrorEnvelope(
              message.id,
              INVALID_PARAMS,
              "MRTR continuations require MCP protocol " +
                `${STATELESS_PROTOCOL_VERSION} or later`,
            );
            break;
          }
          if (!options.mrtr || requestState === undefined) {
            body = jsonErrorEnvelope(
              message.id,
              INVALID_PARAMS,
              "MRTR retries require configured state verification and requestState",
            );
            break;
          }
          // The secret length and the `inputResponses` shape are both
          // checked inside `verifyMrtrContinuation`, in that order, so both
          // call paths get them and neither can forget one.
          const verified = await verifyMrtrContinuation(
            ctx,
            component,
            options.mrtr,
            {
              chainName,
              requestState,
              inputResponses,
              identitySubject: auditIdentitySubject,
              argsDigest: uriDigest,
            },
          );
          if (verified.status === "error") {
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              verified.message ?? "MRTR verification failed",
            );
            break;
          }
          if (verified.status === "reject") {
            body = jsonErrorEnvelope(
              message.id,
              verified.code,
              verified.message,
            );
            break;
          }
          continuation = verified.continuation;
          requestDigest = verified.requestDigest;
          chain = verified.chain;
        }
        // Refuse a superseded continuation BEFORE running the hook, as the
        // tool path does: every post-hook branch would refuse it anyway, but
        // `beforeResourceRead` is not documented as side-effect-free, and a
        // client holding N sibling continuations could otherwise drive it N
        // times on a decision that is already over.
        if (
          continuation &&
          chain.status === "resolved" &&
          !isChainRepeat(
            chain,
            continuation.jti,
            requestDigest,
            chain.resolution,
          )
        ) {
          body = alreadyResolvedEnvelope(message.id, chainName);
          break;
        }
        let decision: unknown;
        try {
          decision = await beforeResourceRead(ctx, {
            uri,
            resourceMetadata: metadata as Record<string, unknown> | null,
            identity,
            ...(continuation
              ? {
                  state: continuation.state,
                  ...(isPlainObject(inputResponses) ? { inputResponses } : {}),
                  round: continuation.round,
                }
              : {}),
          });
        } catch (err) {
          console.error("[mcp-gateway] beforeResourceRead failed", uri, err);
          body = jsonErrorEnvelope(
            message.id,
            INTERNAL_ERROR,
            "beforeResourceRead failed",
          );
          break;
        }
        let capabilityCheck: MrtrCapabilityCheck | undefined;
        if (isMcpInputRequiredResult(decision)) {
          capabilityCheck = checkMrtrCapabilities(
            decision.inputRequests ?? {},
            isStateless ? statelessClientCapabilities : null,
          ) ?? undefined;
          // Deliberately looser than the substitution below: a hook that
          // supplied any fallback for requests the gateway cannot vouch
          // for is a hook bug worth naming, and this branch only ever
          // errors, so nothing can slip through it.
          if (!capabilityCheck && decision.onUnsupported !== undefined) {
            console.error(
              "[mcp-gateway] beforeResourceRead returned unsupported input requests",
              uri,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "beforeResourceRead returned unsupported input requests",
            );
            break;
          }
          if (
            capabilityCheck &&
            !capabilityCheck.supported &&
            // Nullish means NO fallback, never a pass: `null` is the
            // signal that falls through to the normal read path, so
            // substituting it would drop the gate the hook just asked
            // for and serve the resource unguarded.
            decision.onUnsupported != null
          ) {
            decision = decision.onUnsupported;
          }
        }
        // Validate a hook-supplied payload BEFORE anything is settled, the
        // same order the tool path validates `completeCall`'s result in: a
        // host bug should not consume the chain's single resolution.
        if (isMcpCompleteReadResult(decision)) {
          const problem = describeResourceContentsProblem(decision.contents);
          if (problem) {
            console.error(
              "[mcp-gateway] beforeResourceRead returned invalid contents",
              uri,
              problem,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "beforeResourceRead returned invalid contents",
            );
            break;
          }
        }
        // Serving content and refusing are both terminal, so both resolve
        // the chain, exactly as `completeCall` does on the tool path: a
        // forked branch must not turn a settled refusal into a served read.
        //
        // This settles against the PRE-hook snapshot, where the tool path
        // re-reads after its hook. The difference is safe, not an oversight:
        // `claimMrtrChain` is an atomic insert-or-return-winner, so a chain
        // resolved during the hook's await yields snapshot `open` and then
        // loses the claim. A STALE `resolved` snapshot cannot get here at
        // all, because the pre-hook refusal above already rejected it. Only
        // the ask branch needs a fresher read, because the continuation it
        // would mint must not outlive another branch's claim, and it does
        // re-read.
        const isTerminal =
          isMcpCompleteReadResult(decision) || isMcpDeclineReadResult(decision);
        if (continuation && (isTerminal || decision == null)) {
          const settled = await settleMrtrChain(ctx, component, {
            chainName,
            continuation,
            requestDigest,
            chain,
            resolution: isTerminal ? "completed" : "dispatched",
          });
          if (settled === "error") {
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "MRTR verification failed",
            );
            break;
          }
          if (settled === "lost") {
            body = alreadyResolvedEnvelope(message.id, chainName);
            break;
          }
        }
        if (isMcpDeclineReadResult(decision)) {
          // Same family as an `authorizeResource` denial: the caller asked
          // for a resource and is getting none, so it belongs on the error
          // channel rather than dressed up as content. `reason` is
          // host-authored and reaches the caller verbatim.
          if (shouldAuditResource(options.auditResources, "read")) {
            await safeRecordResourceAudit(ctx, component, {
              resourceUri: uri,
              resourceOperation: "read",
              args: null,
              outcome: "denied",
              ...auditCaller,
              durationMs: Date.now() - start,
              errorCode: FORBIDDEN,
              errorMessage: decision.reason,
            });
          }
          body = jsonErrorEnvelope(message.id, FORBIDDEN, decision.reason);
          break;
        }
        if (isMcpCompleteReadResult(decision)) {
          // Contents were validated above, before the chain was settled.
          if (shouldAuditResource(options.auditResources, "read")) {
            await safeRecordResourceAudit(ctx, component, {
              resourceUri: uri,
              resourceOperation: "read",
              args: null,
              outcome: "allowed",
              ...auditCaller,
              durationMs: Date.now() - start,
            });
          }
          body = jsonResultEnvelope(message.id, {
            contents: decision.contents,
          });
          break;
        }
        if (decision != null) {
          if (!isMcpInputRequiredResult(decision)) {
            console.error(
              "[mcp-gateway] beforeResourceRead returned an invalid result",
              uri,
              isPlainObject(decision)
                ? `object keys: ${Object.keys(decision).join(", ")}`
                : typeof decision,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "beforeResourceRead returned an invalid result",
            );
            break;
          }
          // Asking again on a chain that already resolved would mint a
          // fresh continuation for a decision that is over: exactly the
          // sibling an attacker needs. Re-read rather than trusting the
          // pre-hook snapshot, because the hook is an await point another
          // branch can resolve the chain across, and the continuation
          // minted here would outlive that branch's claim. No repeat case
          // applies: asking again never reproduces a terminal outcome.
          if (continuation) {
            const fresh = await readMrtrChain(
              ctx,
              component,
              chainName,
              continuation.idempotencyKey,
            );
            if (fresh.status === "error") {
              body = jsonErrorEnvelope(
                message.id,
                INTERNAL_ERROR,
                "MRTR verification failed",
              );
              break;
            }
            chain = fresh;
          }
          if (chain.status === "resolved") {
            body = alreadyResolvedEnvelope(message.id, chainName);
            break;
          }
          // Asking is only possible where a continuation can travel: the
          // modern protocol with state verification configured. Fail closed
          // instead of serving a resource whose gate wanted input.
          if (!isStateless) {
            // The client's problem, and actionable: upgrade the protocol.
            // Distinguished from the misconfiguration below exactly as the
            // tool path distinguishes them, rather than telling a legacy
            // client the server broke.
            console.error(
              "[mcp-gateway] beforeResourceRead requested input on a legacy " +
                "request; failing closed for resource",
              uri,
            );
            body = jsonErrorEnvelope(
              message.id,
              -32601,
              "This resource requires multi-round-trip input; connect with " +
                `MCP protocol ${STATELESS_PROTOCOL_VERSION} or later`,
            );
            break;
          }
          if (!options.mrtr) {
            // The host's problem: a hook that asks for input on a mount with
            // no state verification can never be answered.
            console.error(
              "[mcp-gateway] beforeResourceRead requested input but the " +
                "`mrtr` option is not configured; failing closed for resource",
              uri,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "MRTR is not configured on this gateway",
            );
            break;
          }
          const asked = await askMrtrInput(options.mrtr, {
            id: message.id,
            hookLabel: "beforeResourceRead",
            chainName,
            argsDigest: uriDigest,
            identitySubject: auditIdentitySubject,
            continuation,
            // Guaranteed non-null here (validated for every modern
            // request, and non-modern requests broke at the check above);
            // the fallback only satisfies the type system.
            clientCapabilities: statelessClientCapabilities ?? {},
            inputRequests: decision.inputRequests ?? {},
            state: decision.state,
            capabilityCheck,
          });
          if ("response" in asked) return asked.response;
          body = asked.body;
          break;
        }
      }

      try {
        let found = false;
        // Track a provider throw so a buggy provider can't mask a resource
        // a later provider could serve. Providers decline a URI by returning
        // null; a throw must not be *more* powerful than declining, so we
        // isolate it, log it, and keep trying the remaining providers (then
        // the templates).
        //
        // Two texts per throw: `full` for the audit row, `wire` for the
        // caller. They differ unless the provider threw ConvexError on
        // purpose, so an accidental exception can't ship a signed URL or
        // an upstream response body to the LLM.
        let providerError: { full: string; wire: string } | null = null;
        const serveContents = async (contents: McpResourceContent[]) => {
          // Validate handler output before returning it. Invalid contents
          // are a provider bug; throw so the outer catch turns it into a
          // deterministic -32603 instead of shipping malformed JSON-RPC.
          // (Runs outside the per-provider try, so it is a hard error, not a
          // provider-decline that falls through to the next provider.)
          const problem = describeResourceContentsProblem(contents);
          if (problem) {
            throw new ResourceContractError(
              `resources/read provider returned invalid contents: ${problem}`,
            );
          }
          if (shouldAuditResource(options.auditResources, "read")) {
            await safeRecordResourceAudit(ctx, component, {
              resourceUri: uri,
              resourceOperation: "read",
              args: null,
              outcome: "allowed",
              ...auditCaller,
              durationMs: Date.now() - start,
            });
          }
          body = jsonResultEnvelope(message.id, { contents });
          found = true;
        };

        // Concrete providers first: a concrete resource always wins over a
        // template that might also match the same URI, so dispatch stays
        // unambiguous.
        for (const provider of providers) {
          let contents: McpResourceContent[] | null;
          try {
            contents = await provider.read(ctx, { uri, identity });
          } catch (err) {
            providerError = splitErrorText(err, GENERIC_RESOURCE_READ_ERROR);
            console.error(
              "[mcp-gateway] resource provider threw during resources/read",
              provider.name,
              uri,
              err,
            );
            continue;
          }
          // A provider declines a URI by returning null OR an empty array;
          // declining via `[]` keeps it from shipping empty contents and
          // shadowing a later provider/template. Anything else (including a
          // malformed non-array) still goes to serveContents, which validates
          // it and surfaces -32603 on a bad shape.
          if (contents && !(Array.isArray(contents) && contents.length === 0)) {
            await serveContents(contents);
            break;
          }
        }

        // Template-backed resolution, only when no concrete provider served.
        // A template with no `read` handler is listing-only and skipped here.
        if (!found) {
          for (const provider of templates) {
            if (!provider.read) continue;
            const params = provider.match(uri);
            if (!params) continue;
            let contents: McpResourceContent[] | null;
            try {
              contents = await provider.read(ctx, { uri, params, identity });
            } catch (err) {
              providerError = splitErrorText(err, GENERIC_RESOURCE_READ_ERROR);
              console.error(
                "[mcp-gateway] resource template threw during resources/read",
                provider.template.uriTemplate,
                uri,
                err,
              );
              continue;
            }
            if (
              contents &&
              !(Array.isArray(contents) && contents.length === 0)
            ) {
              await serveContents(contents);
              break;
            }
          }
        }

        if (!found) {
          // Distinguish "everything cleanly declined" (a genuine not-found →
          // INVALID_PARAMS) from "a provider/template threw and nothing
          // served" (a real fault → INTERNAL_ERROR), so a bug isn't reported
          // to the client as a benign miss.
          const code = providerError ? INTERNAL_ERROR : INVALID_PARAMS;
          // "Resource not found" is gateway-authored and safe either way;
          // a provider throw splits into audit text and caller text.
          const notFound = `Resource not found: ${uri}`;
          if (shouldAuditResource(options.auditResources, "read")) {
            await safeRecordResourceAudit(ctx, component, {
              resourceUri: uri,
              resourceOperation: "read",
              args: null,
              outcome: "error",
              ...auditCaller,
              durationMs: Date.now() - start,
              errorCode: code,
              errorMessage: providerError?.full ?? notFound,
            });
          }
          // The spec's not-found example carries the offending URI in
          // `data`, so a client can correlate the miss with the request it
          // made instead of parsing our prose. Only on the genuine miss:
          // the provider-error branch is our fault rather than the
          // caller's, and that path exists precisely to keep provider
          // detail off the wire, so it stays message-only.
          // Same condition `code` was derived from, written the same way,
          // so the payload and the code cannot drift apart.
          body = providerError
            ? jsonErrorEnvelope(message.id, code, providerError.wire)
            : jsonErrorEnvelopeWithData(message.id, code, notFound, { uri });
        }
      } catch (err) {
        // Hard faults: invalid provider contents, a throwing audit path,
        // anything the per-provider isolation above didn't catch. The
        // caller gets a generic message, so log the cause here too, the
        // audit row alone would make this hard to trace.
        const { full, wire } = splitErrorText(err, GENERIC_RESOURCE_READ_ERROR);
        console.error("[mcp-gateway] resources/read failed", uri, err);
        if (shouldAuditResource(options.auditResources, "read")) {
          await safeRecordResourceAudit(ctx, component, {
            resourceUri: uri,
            resourceOperation: "read",
            args: null,
            outcome: "error",
            ...auditCaller,
            durationMs: Date.now() - start,
            errorCode: INTERNAL_ERROR,
            errorMessage: full,
          });
        }
        body = jsonErrorEnvelope(message.id, INTERNAL_ERROR, wire);
      }
      break;
    }

    case "ping": {
      if (isStateless) {
        responseStatus = 404;
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `${message.method} is legacy-only; it was removed in ` +
            STATELESS_PROTOCOL_VERSION,
        );
        break;
      }
      // An empty result is the entire contract. Deliberately gated on
      // neither identity nor a capability: the reference SDK answers ping
      // from its `Protocol` constructor, so a client may assume a liveness
      // check works on any connection it already holds. The session lookup
      // above is the only precondition.
      body = jsonResultEnvelope(message.id, {});
      break;
    }

    case "resources/subscribe":
    case "resources/unsubscribe": {
      if (isStateless) {
        responseStatus = 404;
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `${message.method} is legacy-only; use subscriptions/listen when it is supported`,
        );
        break;
      }
      // Off by default: the gateway's HTTP transport can't push, so the
      // capability is unadvertised and these methods report a clear,
      // descriptive -32601 rather than silently accepting a subscription
      // that could never be delivered.
      if (!options.resourceSubscriptions?.subscribe) {
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `${message.method} is not supported: this gateway does not ` +
            `advertise the resources.subscribe capability (its HTTP ` +
            `transport cannot deliver server-initiated notifications). ` +
            `Enable resourceSubscriptions.subscribe only behind a ` +
            `push-capable transport.`,
        );
        break;
      }
      // Subscriptions are identity-scoped, like list/read. (Read-time
      // authorization still governs content: the `updated` notification
      // carries only a URI, and `resources/read` re-checks `resource_read`.)
      //
      // `anonymousResources` deliberately does NOT reach here. Unlike a
      // list or a read, a subscribe is server-side state an anonymous
      // caller accumulates: the rows are capped per session, but nothing
      // caps how many sessions one anonymous client opens. It also buys a
      // client nothing on this transport, which cannot push the
      // `updated` notification a subscription exists to receive.
      if (!identity) {
        body = jsonErrorEnvelope(
          message.id,
          UNAUTHORIZED,
          "Unauthorized: authentication required",
        );
        break;
      }
      // Identity-bound, like DELETE: only the session's owner may mutate its
      // subscription state. Without this, a leaked Mcp-Session-Id plus any
      // valid token would let one user grief another's subscriptions (cap
      // exhaustion, spurious update pushes). Legacy rows with no bound owner
      // (`undefined`) skip the check, matching `deleteSession`.
      if (
        sessionOwnerSubject !== undefined &&
        sessionOwnerSubject !== identity.subject
      ) {
        body = jsonErrorEnvelope(
          message.id,
          FORBIDDEN,
          "Forbidden: caller identity does not match session owner",
        );
        break;
      }
      const uri = message.params?.uri;
      if (typeof uri !== "string" || uri.length === 0) {
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          "Missing resource uri",
        );
        break;
      }
      if (message.method === "resources/subscribe") {
        const result = (await ctx.runMutation(
          component.sessions.subscribeResource,
          { sessionId, uri },
        )) as "subscribed" | "exists" | "limit_exceeded";
        if (result === "limit_exceeded") {
          // A client-induced limit, not a server fault: use FORBIDDEN so it
          // doesn't pollute internal-error signals.
          body = jsonErrorEnvelope(
            message.id,
            FORBIDDEN,
            "Subscription limit reached for this session",
          );
          break;
        }
      } else {
        await ctx.runMutation(component.sessions.unsubscribeResource, {
          sessionId,
          uri,
        });
      }
      // MCP `resources/subscribe` and `resources/unsubscribe` return an
      // empty result object on success.
      body = jsonResultEnvelope(message.id, {});
      break;
    }

    case "tools/list": {
      // Filter the catalog through the authorize callback in mode "list".
      // Throwing authorizers are isolated per tool: a single buggy
      // decision hides only that tool, not the whole list.
      const allTools = (await ctx.runQuery(
        component.registry.listTools,
        {},
      )) as RegisteredTool[];
      const visible = [];
      for (const tool of allTools) {
        const { decision, threw } = await safeAuthorize(
          options.authorize,
          ctx,
          {
            toolName: tool.name,
            toolKind: tool.kind,
            args: {},
            mode: "list",
            toolMetadata: tool.metadata ?? null,
            identity,
          },
        );
        if (threw) {
          // A buggy authorize callback drops only the offending tool,
          // not the whole list. Surface to the deployment log so the
          // shrinking tools/list response is discoverable; the tool
          // stays hidden either way.
          console.error(
            "[mcp-gateway] authorize callback threw during tools/list for tool",
            tool.name,
            decision.reason,
          );
        }
        if (decision.allowed) {
          const advertisedOutputSchema = advertisedSchema(
            tool.authoredOutputSchemaJson,
            tool.outputSchema,
            tool.name,
          );
          visible.push({
            // Spread first so the registry's own columns always win.
            // `protocolMetadata` is stored as `v.any()`, so a caller
            // reaching the component mutation directly could otherwise
            // shadow `name` or `inputSchema` on the wire.
            ...(tool.protocolMetadata ?? {}),
            name: tool.name,
            description: tool.description,
            inputSchema: advertisedSchema(
              tool.authoredInputSchemaJson,
              tool.inputSchema,
              tool.name,
            ),
            // Only emit `outputSchema` when the tool actually declared
            // one, some MCP clients (Inspector older versions) are
            // strict about the field being absent vs null vs {}. A
            // non-object schema is withheld from legacy clients, which
            // would reject this entire response over it.
            ...(mayAdvertiseOutputSchema(advertisedOutputSchema, isStateless)
              ? { outputSchema: advertisedOutputSchema }
              : {}),
            // Advertise task support only when the host actually
            // configured task execution AND the caller speaks the
            // stateless protocol; session-based clients cannot poll tasks.
            ...(mcpTaskSupportLevel(tool) !== "forbidden" &&
            isStateless &&
            options.tasks
              ? { execution: { taskSupport: mcpTaskSupportLevel(tool) } }
              : {}),
          });
        }
      }
      if (isStateless) {
        visible.sort((a, b) => a.name.localeCompare(b.name));
      }
      body = jsonResultEnvelope(message.id, { tools: visible });
      break;
    }

    case "tools/call": {
      const name = message.params?.name;
      if (typeof name !== "string") {
        body = jsonErrorEnvelope(message.id, -32602, "Missing tool name");
        break;
      }
      const args = isPlainObject(message.params?.arguments)
        ? { ...message.params?.arguments }
        : ((message.params?.arguments ?? {}) as Record<string, unknown>);

      const tool = (await ctx.runQuery(component.registry.getTool, {
        name,
      })) as RegisteredTool | null;
      if (!tool) {
        // Anti-DoS: unknown-tool calls are not audited because anonymous
        // callers can spam arbitrary names with arbitrary args.
        body = jsonErrorEnvelope(message.id, -32602, `Unknown tool: ${name}`);
        break;
      }

      // Reserved fields are never client-controlled. Strip them on both
      // first calls and retries before any digest, authorization, or audit.
      if (tool.mrtrArgs) {
        delete args[tool.mrtrArgs.idempotencyKey];
      }

      // Identity is also gateway-owned. Strip it before calculating an MRTR
      // argument digest so a spoofed value cannot invalidate a continuation.
      if (tool.identityArg !== undefined) {
        delete args[tool.identityArg];
      }

      if (isStateless) {
        const headerProblem = validateStatelessToolParameterHeaders(
          request,
          tool.inputSchema,
          message.params?.arguments,
        );
        if (headerProblem) {
          if (headerProblem.kind === "schema") {
            // The tool's own inputSchema is malformed. That is a server
            // configuration error, not a client header mismatch, so it
            // must not be reported as -32020. Registration validates
            // this, so reaching here means a row predates that check or
            // was written past the client API.
            console.error(
              "[mcp-gateway] tool inputSchema has an invalid x-mcp-header annotation",
              tool.name,
              headerProblem.problem,
            );
            return statelessErrorResponse(
              message.id,
              INTERNAL_ERROR,
              "Tool input schema is invalid",
              {},
              500,
            );
          }
          return statelessErrorResponse(
            message.id,
            HEADER_MISMATCH,
            headerProblem.problem,
          );
        }
      }

      const requestState = message.params?.requestState;
      const inputResponses = message.params?.inputResponses;
      const declarativeTool = options.declarativeTools?.find(
        (candidate) => candidate.name === tool.name,
      );
      const beforeCall = declarativeTool?.beforeCall;
      // A registered row is "gated" when it promises a confirmation
      // hook: registered from a declarative catalog with `beforeCall`,
      // or reserving `mrtrArgs` (which is meaningless without one).
      const mrtrGated =
        tool.mrtrGated === true ||
        tool.mrtrArgs !== undefined ||
        beforeCall !== undefined;

      // Fail closed: a gated registry row served by a handler with no
      // matching hook (imperative registration, stale declarative
      // catalog, or a mount without the `tools` option) must never
      // dispatch ungated: that would silently skip the confirmation
      // the row promises, on any transport.
      if (mrtrGated && !beforeCall) {
        console.error(
          "[mcp-gateway] tool is registered as MRTR-gated but this " +
            "handler has no beforeCall for it; failing closed",
          tool.name,
        );
        body = jsonErrorEnvelope(
          message.id,
          INTERNAL_ERROR,
          `Tool "${tool.name}" requires a confirmation hook this ` +
            "deployment did not configure",
        );
        break;
      }

      // The mirror, and the only place both facts are visible: the hook
      // comes from this handler's declarative catalog, the reserved key
      // from the registry row. A hook without the key dispatches a
      // confirmed mutation with nothing to deduplicate on, so the
      // deliberately-allowed replay of the resolving continuation would
      // apply the side effect twice. Take `kind` from the declarative
      // catalog, not from the row: the row is written by the same party
      // this check defends against, so flipping it to "query" would
      // otherwise buy an exemption.
      if (
        beforeCall !== undefined &&
        (declarativeTool?.kind ?? tool.kind) !== "query" &&
        tool.mrtrArgs === undefined
      ) {
        console.error(
          "[mcp-gateway] tool has a confirmation hook but its registry " +
            "row reserves no mrtrArgs key; failing closed rather than " +
            "dispatching a replayable mutation without one",
          tool.name,
        );
        body = jsonErrorEnvelope(
          message.id,
          INTERNAL_ERROR,
          `Tool "${tool.name}" is confirmed but its registration is ` +
            "missing the idempotency key it needs",
        );
        break;
      }

      // Task augmentation (`io.modelcontextprotocol/tasks`). SEP-2663
      // makes the decision the SERVER's: a client opts in once through
      // the extension capability and sends no per-request flag, so a
      // `params.task` from a client is a legacy hint and is ignored
      // rather than refused. On the session era it is still refused,
      // loudly: tasks never existed there, so a client sending one has
      // misread which protocol it is speaking and would otherwise get a
      // synchronous answer to a question it thinks is asynchronous.
      const taskLevel = mcpTaskSupportLevel(tool);
      if (!isStateless && message.params?.task !== undefined) {
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          "Task-augmented calls require MCP protocol " +
            `${STATELESS_PROTOCOL_VERSION} or later`,
        );
        break;
      }
      if (!isStateless && taskLevel === "required") {
        // A `required` tool has no synchronous answer, and the session
        // era has no tasks, so the only honest outcome is a refusal.
        // Without this the era gate below skips the whole block and the
        // call falls through to an ordinary dispatch: the side effect
        // runs inline, which is the one thing the level exists to
        // prevent. `tools/list` does not advertise `execution` on this
        // era either, so the client could not have known.
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          `Tool "${tool.name}" runs only as a task, which requires MCP ` +
            `protocol ${STATELESS_PROTOCOL_VERSION} or later`,
        );
        break;
      }
      // Whether this call may become a task, decided here so the refusals
      // below stay ahead of `authorize` exactly as they were. Only a
      // `required` tool refuses: an `optional` one has nothing to refuse,
      // because nobody asked for a task, so it answers inline.
      let taskEligible = false;
      if (isStateless && taskLevel !== "forbidden") {
        if (!options.tasks) {
          if (taskLevel === "required") {
            // Same "unsupported because unconfigured" shape as the
            // resource catalogs: the capability was never advertised, and
            // this tool cannot run any other way.
            responseStatus = 404;
            body = jsonErrorEnvelope(
              message.id,
              -32601,
              "Tasks are not supported: the host did not configure task " +
                "execution",
            );
            break;
          }
        } else if (!declaresTasksExtension(statelessClientCapabilities)) {
          if (taskLevel === "required") {
            // Naming the capability in machine-readable form is what lets
            // a client fix the request rather than guess.
            raw = statelessErrorResponse(
              message.id,
              MISSING_REQUIRED_CLIENT_CAPABILITY,
              `Tool "${tool.name}" runs only as a task, which requires the ` +
                `${TASKS_CAPABILITY_KEY} client capability`,
              { requiredCapabilities: TASKS_REQUIRED_CAPABILITIES },
            );
            body = "";
            break;
          }
        } else if (!identity) {
          if (taskLevel === "required") {
            // Tasks are owner-bound rows; an anonymous caller could never
            // poll the result, so refuse before creating unpollable
            // state. The real 401 + WWW-Authenticate (same as the
            // authorize denial path) so browser clients begin OAuth
            // discovery instead of seeing an unactionable 200.
            //
            // Audited like every other refusal on this path: this branch
            // sits before `safeAuthorize`, so without the row an
            // anonymous caller could probe which tools require a task and
            // leave no trace at all.
            try {
              await ctx.runMutation(component.dispatch.recordAuthDenial, {
                name: tool.name,
                args,
                auditIdentitySubject: null,
                outcome: "denied",
                errorCode: UNAUTHORIZED,
                errorMessage: "Task-augmented calls require authentication",
                durationMs: 0,
              });
            } catch (err) {
              // Audit must never change the outcome (see safeRecordAudit
              // in dispatch.ts); log so a recurring write failure stays
              // visible.
              console.error(
                "[mcp-gateway] failed to record anonymous task denial",
                tool.name,
                err,
              );
            }
            raw = await requireAuthChallenge(
              ctx,
              request,
              component,
              message.id,
            );
            body = "";
            break;
          }
        } else {
          taskEligible = true;
        }
      }
      if (taskEligible) {
        // Reject a deeply nested args value here, before it reaches the
        // createTask mutation whose arg serialization would overflow the
        // stack and escape as a raw 500.
        if (nestsTooDeep(args)) {
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            "Task arguments nest too deeply",
          );
          break;
        }
        // A non-object `arguments` (string, array, number) reaches a
        // synchronous dispatch as-is and the tool's own validator rejects
        // it. On the task path it would instead be PERSISTED first, and
        // the executor's `{...task.args}` spread would turn a string into
        // a character-indexed object before the tool ever saw it. Reject
        // it while the caller is still here to be told.
        if (!isPlainObject(args)) {
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            "Task arguments must be an object",
          );
          break;
        }
      }

      const start = Date.now();
      const authz = await safeAuthorize(options.authorize, ctx, {
        toolName: tool.name,
        toolKind: tool.kind,
        args,
        mode: "call",
        toolMetadata: tool.metadata ?? null,
        identity,
      });
      const threw = authz.threw;
      let decision = authz.decision;
      // A tool that declares identityArg structurally needs a caller,
      // and so does an MRTR hook (its contract passes a non-null
      // identity, and a continuation must bind to a principal). Deny as
      // Unauthorized regardless of what the host's authorize returned,
      // through this shared path so the client gets the real 401 +
      // WWW-Authenticate challenge and the denial lands in the audit
      // log like every other one.
      if (
        decision.allowed &&
        (tool.identityArg !== undefined || beforeCall !== undefined) &&
        !identity
      ) {
        decision = {
          allowed: false,
          reason: "Unauthorized: tool requires an authenticated caller",
        };
      }

      if (!decision.allowed) {
        const reason = decision.reason ?? "Forbidden";
        const code = threw
          ? INTERNAL_ERROR
          : /^unauth/i.test(reason)
            ? UNAUTHORIZED
            : FORBIDDEN;
        // A returned `reason` is host-authored and meant for the caller;
        // a thrown one is `Authorizer threw: <exception text>` and must
        // not reach the wire. The audit row below keeps the full text.
        const wireReason = threw ? GENERIC_AUTHORIZER_ERROR : reason;
        // Record the rejection in the audit log so operators see who
        // tried what and was denied (or what made the authorizer throw).
        try {
          await ctx.runMutation(component.dispatch.recordAuthDenial, {
            name: tool.name,
            args,
            auditIdentitySubject,
            outcome: threw ? "error" : "denied",
            errorCode: code,
            errorMessage: reason,
            durationMs: Date.now() - start,
          });
        } catch (err) {
          // Match safeRecordAudit's pattern in dispatch.ts: audit must
          // never alter the dispatch outcome, so swallow, but log so
          // a recurring write failure (schema drift, validator
          // mismatch) is visible to operators.
          console.error(
            "[mcp-gateway] failed to record auth denial",
            tool.name,
            err,
          );
        }
        // 401 + WWW-Authenticate per RFC 6750 + RFC 9728 when an OAuth
        // server is configured. Bypasses the JSON-RPC envelope and uses
        // HTTP status semantics so the MCP client begins discovery.
        if (code === UNAUTHORIZED) {
          const oauthConfig = await ctx.runQuery(
            component.registry.getOAuthConfig,
            {},
          );
          if (oauthConfig) {
            const requestUrl = new URL(request.url);
            const mcpPath = requestUrl.pathname.replace(/\/+$/, "") || "/";
            const metadataUrl = buildProtectedResourceMetadataUrl(
              requestUrl.origin,
              mcpPath,
            );
            raw = new Response(
              jsonErrorEnvelope(message.id, code, wireReason),
              {
                status: 401,
                headers: {
                  "content-type": "application/json",
                  "www-authenticate": `Bearer resource_metadata="${metadataUrl}"`,
                  ...(issueSessionHeader
                    ? { "mcp-session-id": sessionId }
                    : {}),
                },
              },
            );
            body = "";
            break;
          }
        }
        body = jsonErrorEnvelope(message.id, code, wireReason);
        break;
      }

      // MRTR continuation verification, after authorization so denials
      // and anonymous callers went through the audited 401 path above.
      let continuation: VerifiedMrtrState | null = null;
      let mrtrArgsDigest: string | null = null;
      // A chain resolves exactly once, and remembers WHICH continuation
      // resolved it. Only that continuation may reproduce the outcome
      // (a client whose response was lost); every sibling is refused,
      // including one re-sent byte-identically, because its hook output
      // is not the settled result. Re-read after `beforeCall`, which is
      // an await point another branch can resolve the chain across.
      let chain: MrtrChainState = { status: "open" };
      // Digest of THIS request's `inputResponses` (absent when it
      // carries none). Compared against the digest the chain recorded
      // when it resolved, so a repeat has to present the same answer.
      let requestDigest: string | undefined;
      if (
        isStateless &&
        (requestState !== undefined || inputResponses !== undefined)
      ) {
        if (!options.mrtr || requestState === undefined) {
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            "MRTR retries require configured state verification and requestState",
          );
          break;
        }
        if (!beforeCall) {
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            "Tool does not support declarative MRTR continuations",
          );
          break;
        }
        // The secret length and the `inputResponses` shape are both checked
        // inside `verifyMrtrContinuation`, in that order, so both call paths
        // get them and neither can forget one.
        mrtrArgsDigest = await sha256Base64Url(stableJson(args));
        const verified = await verifyMrtrContinuation(
          ctx,
          component,
          options.mrtr,
          {
            chainName: tool.name,
            requestState,
            inputResponses,
            identitySubject: auditIdentitySubject,
            argsDigest: mrtrArgsDigest,
          },
        );
        if (verified.status === "error") {
          body = jsonErrorEnvelope(
            message.id,
            INTERNAL_ERROR,
            verified.message ?? "MRTR verification failed",
          );
          break;
        }
        if (verified.status === "reject") {
          body = jsonErrorEnvelope(message.id, verified.code, verified.message);
          break;
        }
        continuation = verified.continuation;
        requestDigest = verified.requestDigest;
        chain = verified.chain;
      }

      // The host-side MRTR state machine. It runs before `runTool` on
      // the first call AND on every verified continuation, so the
      // decision to dispatch (accept), ask again (missing input), or
      // finish without dispatching (decline/cancel) lives in the
      // gateway hook; the underlying Convex function never parses MCP
      // envelopes. It runs on EVERY transport so required input is
      // never silently bypassed: when it demands input and the request
      // cannot carry a continuation (legacy protocol, or `mrtr` not
      // configured), the call fails closed instead of dispatching.
      //
      // A chain resolves exactly once, and `verifyMrtrContinuation` already
      // read that state for us as its last step, so an already-resolved
      // chain refuses the outcomes that would re-open it (another input
      // round, or a dispatch) while still letting an idempotent re-send of a
      // completed call reproduce its result. Re-reading it here would be a
      // second Convex query on every continuation for the same answer; the
      // sites that genuinely need a fresher view (across the hook's await
      // point) re-read for themselves below.
      // Refuse a superseded continuation BEFORE running the host hook.
      // Only the continuation that actually resolved the chain has
      // anything to reproduce; every other branch is answering a
      // decision that is over, and running arbitrary client-chosen
      // `inputResponses` through host code just to discard the result
      // is work nobody asked for. `beforeCall` is not documented as
      // side-effect-free, so do not call it.
      if (
        continuation &&
        chain.status === "resolved" &&
        !(
          chain.resolvedByJti === continuation.jti &&
          chain.resolvedByDigest === requestDigest
        )
      ) {
        body = alreadyResolvedEnvelope(message.id, tool.name);
        break;
      }
      if (beforeCall) {
        // Digest BEFORE the hook, over the client-sent (stripped)
        // arguments: a hook that mutates even a nested value of its
        // (shallow-copied) argument object must not poison the digest
        // the next retry is checked against.
        const argsDigest =
          mrtrArgsDigest ?? (await sha256Base64Url(stableJson(args)));
        let requested: unknown;
        try {
          requested = await beforeCall(ctx, {
            // JSON round-trip: args are JSON by construction, and a
            // deep copy keeps hook-side normalization away from both
            // the digest above and the dispatch below.
            args: JSON.parse(JSON.stringify(args)) as Record<string, unknown>,
            identity: identity!,
            ...(continuation
              ? {
                  state: continuation.state,
                  ...(isPlainObject(inputResponses) ? { inputResponses } : {}),
                  idempotencyKey: continuation.idempotencyKey,
                  round: continuation.round,
                }
              : {}),
          });
        } catch (err) {
          console.error("[mcp-gateway] MRTR beforeCall failed", tool.name, err);
          body = jsonErrorEnvelope(
            message.id,
            INTERNAL_ERROR,
            "MRTR beforeCall failed",
          );
          break;
        }
        let capabilityCheck: MrtrCapabilityCheck | undefined;
        if (isMcpInputRequiredResult(requested)) {
          capabilityCheck = checkMrtrCapabilities(
            requested.inputRequests ?? {},
            isStateless ? statelessClientCapabilities : null,
          ) ?? undefined;
          // Deliberately looser than the substitution below: a hook that
          // supplied any fallback for requests the gateway cannot vouch
          // for is a hook bug worth naming, and this branch only ever
          // errors, so nothing can slip through it.
          if (!capabilityCheck && requested.onUnsupported !== undefined) {
            console.error(
              "[mcp-gateway] MRTR beforeCall returned unsupported input requests",
              tool.name,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "MRTR beforeCall returned unsupported input requests",
            );
            break;
          }
          if (
            capabilityCheck &&
            !capabilityCheck.supported &&
            // Nullish means NO fallback, never a pass: `null` is the
            // signal that falls through to the dispatch below, so
            // substituting it would drop the gate the hook just asked
            // for and run the tool unconfirmed.
            requested.onUnsupported != null
          ) {
            requested = requested.onUnsupported;
          }
        }
        if (isMcpCompleteCallResult(requested)) {
          // Terminal without dispatch, e.g. a declined confirmation.
          // Valid on both eras: it is an ordinary tools/call result.
          // Validate the shape rather than forward a malformed result a
          // spec-compliant client would reject, matching how every other
          // hook output is checked.
          const problem = describeCompleteCallResultProblem(
            requested.result,
            isStateless,
          );
          if (problem) {
            console.error(
              "[mcp-gateway] MRTR beforeCall completeCall() returned a " +
                "malformed tools/call result",
              tool.name,
              problem,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "MRTR beforeCall returned an invalid result",
            );
            break;
          }
          // Terminal for the whole chain, so claim it before answering.
          // Only a continuation can be forked; a first call has no
          // earlier round to replay.
          //
          // A chain already resolved by a *completion* may reproduce
          // that result: the hook is deterministic over the same
          // inputs, so an idempotent re-send (a client whose response
          // was lost) legitimately gets its answer again. One resolved
          // by a dispatch may not: the tool has run, and telling the
          // caller "declined" afterwards would be a lie.
          if (continuation) {
            const fresh = await readMrtrChain(
              ctx,
              component,
              tool.name,
              continuation.idempotencyKey,
            );
            if (fresh.status === "error") {
              body = jsonErrorEnvelope(
                message.id,
                INTERNAL_ERROR,
                "MRTR verification failed",
              );
              break;
            }
            chain = fresh;
          }
          if (continuation) {
            // Finishing the call resolves the chain, so claim it. Only the
            // continuation that resolved it may reproduce the outcome (a
            // client whose response was lost); any sibling would be handed
            // its own hook output as the settled result.
            const settled = await settleMrtrChain(ctx, component, {
              chainName: tool.name,
              continuation,
              requestDigest,
              chain,
              resolution: "completed",
            });
            if (settled === "error") {
              body = jsonErrorEnvelope(
                message.id,
                INTERNAL_ERROR,
                "MRTR verification failed",
              );
              break;
            }
            if (settled === "lost") {
              body = alreadyResolvedEnvelope(message.id, tool.name);
              break;
            }
          }
          // The MRTR chain claim above is already written, so an escape
          // here would leave the client with a 500 it cannot even retry.
          body = hostResultEnvelope(message.id, requested.result, tool.name);
          break;
        }
        if (requested !== null && requested !== undefined) {
          if (!isMcpInputRequiredResult(requested)) {
            // A host-side hook bug (e.g. completeCall with a non-object
            // result). Leave the operator a breadcrumb: tool name plus
            // the returned shape, never the value itself (it may carry
            // host-private state).
            console.error(
              "[mcp-gateway] MRTR beforeCall returned an invalid result",
              tool.name,
              isPlainObject(requested)
                ? `object keys: ${Object.keys(requested).join(", ")}`
                : typeof requested,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "MRTR beforeCall returned an invalid result",
            );
            break;
          }
          // Asking again on a chain that already resolved would mint a
          // fresh continuation for a decision that is over: exactly the
          // sibling an attacker needs. Re-read rather than trusting the
          // pre-hook snapshot, because `beforeCall` is an await point
          // another branch can resolve the chain across, and the
          // continuation minted here would outlive that branch's claim.
          // No repeat case applies: asking again is never a
          // reproduction of a terminal outcome.
          if (continuation) {
            const fresh = await readMrtrChain(
              ctx,
              component,
              tool.name,
              continuation.idempotencyKey,
            );
            if (fresh.status === "error") {
              body = jsonErrorEnvelope(
                message.id,
                INTERNAL_ERROR,
                "MRTR verification failed",
              );
              break;
            }
            chain = fresh;
          }
          if (chain.status === "resolved") {
            body = alreadyResolvedEnvelope(message.id, tool.name);
            break;
          }
          if (!isStateless) {
            body = jsonErrorEnvelope(
              message.id,
              -32601,
              `Tool "${tool.name}" requires multi-round-trip input; ` +
                `connect with MCP protocol ${STATELESS_PROTOCOL_VERSION} or later`,
            );
            break;
          }
          if (!options.mrtr) {
            console.error(
              "[mcp-gateway] beforeCall requested input but the `mrtr` " +
                "option is not configured; failing closed for tool",
              tool.name,
            );
            body = jsonErrorEnvelope(
              message.id,
              INTERNAL_ERROR,
              "MRTR is not configured on this gateway",
            );
            break;
          }
          const asked = await askMrtrInput(options.mrtr, {
            id: message.id,
            hookLabel: "MRTR beforeCall",
            chainName: tool.name,
            argsDigest,
            identitySubject: auditIdentitySubject,
            continuation,
            // Guaranteed non-null here (validated for every stateless
            // request, and session-era requests broke at the -32601 above);
            // the fallback only satisfies the type system.
            clientCapabilities: statelessClientCapabilities ?? {},
            inputRequests: requested.inputRequests ?? {},
            state: requested.state,
            capabilityCheck,
          });
          if ("response" in asked) return asked.response;
          body = asked.body;
          break;
        }
      }

      // Dispatch resolves the chain, so claim it first: the claim is
      // the decision, not a record of one. Without this, a replay that
      // re-runs the hook mints an unpinned sibling continuation that
      // stays answerable after another branch already resolved, and a
      // captured `requestState` could still turn a decline into a
      // dispatch. Claiming also makes dispatch at-most-once per chain
      // gateway-side, ahead of the tool's own idempotency key.
      // A chain resolved by a dispatch may dispatch again: that is the
      // idempotent replay of an accept whose response was lost, and the
      // mandatory `mrtrArgs` key is what stops the tool double-applying
      // it. A chain resolved by a *completion* may not: turning a
      // settled decline into a run is precisely the flip being
      // prevented.
      if (beforeCall && continuation) {
        const fresh = await readMrtrChain(
          ctx,
          component,
          tool.name,
          continuation.idempotencyKey,
        );
        if (fresh.status === "error") {
          body = jsonErrorEnvelope(
            message.id,
            INTERNAL_ERROR,
            "MRTR verification failed",
          );
          break;
        }
        chain = fresh;
        // Only the continuation that dispatched may dispatch again, which
        // is the lost-response retry the tool deduplicates via its
        // injected idempotency key. Everything else claims, and losing the
        // claim means another branch settled this chain first: there is no
        // idempotent case for a second dispatch.
        const settled = await settleMrtrChain(ctx, component, {
          chainName: tool.name,
          continuation,
          requestDigest,
          chain,
          resolution: "dispatched",
        });
        if (settled === "error") {
          body = jsonErrorEnvelope(
            message.id,
            INTERNAL_ERROR,
            "MRTR verification failed",
          );
          break;
        }
        if (settled === "lost") {
          body = alreadyResolvedEnvelope(message.id, tool.name);
          break;
        }
      }

      // Task-augmented call: create the durable task row and return the
      // handle immediately; the tool runs after this request completes
      // (built-in scheduled executor) or inside the host's own durable
      // execution.
      //
      // Ordering matters, and this is deliberately the LAST gate before
      // dispatch would have happened:
      //   - after authorize, so a denied caller cannot create tasks;
      //   - after the identityArg strip, so the stored args snapshot is
      //     the public argument set;
      //   - after the MRTR `beforeCall` hook, so no durable row exists
      //     until the hook approved the call. A hook that demands input
      //     answered with the `input_required` envelope above and created
      //     nothing, which keeps MRTR's one negotiation channel intact:
      //     a task-augmented MRTR tool negotiates over `requestState`
      //     first and only then becomes a task.
      //   - after the chain claim above, which is the load-bearing one.
      //     Creating a task is a TERMINAL resolution of the chain, not a
      //     deferral of one: it returns a handle INSTEAD of dispatching,
      //     and once the row exists the executor runs the tool with no
      //     further gate. So it must be covered by the same
      //     `resolution: "dispatched"` claim as a synchronous dispatch;
      //     otherwise a task-augmented continuation of a chain another
      //     branch already settled (a decline, say) would create a row
      //     that quietly runs the tool. A lost claim broke above with
      //     `alreadyResolvedEnvelope` and created nothing.
      // `shouldCreate` runs here rather than beside the gates above, so
      // host policy never sees a call this mount's `authorize` denied.
      const createTask =
        taskEligible && options.tasks && identity
          ? taskLevel === "required" ||
            (await safeShouldCreateTask(options.tasks.shouldCreate, ctx, {
              toolName: tool.name,
              toolKind: tool.kind,
              args: args as Record<string, unknown>,
              identity,
            }))
          : false;
      if (createTask && options.tasks && identity) {
        const taskId = generateSessionId();
        // A verified continuation carries the chain's idempotency key, and
        // the executor injects the task row's key into `mrtrArgs`. Reusing
        // it means a replayed continuation (`redemption === "replay"`)
        // that lands as a second task still dedupes inside the tool,
        // exactly as a replayed synchronous continuation does.
        const idempotencyKey =
          continuation?.idempotencyKey ?? crypto.randomUUID();
        const executor = options.tasks.execute ? "host" : "component";
        // The host's ceiling (or the 24h default, mirroring the
        // component's TASK_DEFAULT_TTL_MS). The component additionally
        // clamps to the global [1 minute, 7 days] bounds.
        const hostRetentionMs =
          options.tasks.retentionMs ?? TASK_DEFAULT_RETENTION_MS;
        // The client no longer has a say: SEP-2663 removed the request
        // shape it used to shorten this with, so the host's ceiling is
        // the whole answer.
        const requestedTtlMs = hostRetentionMs;
        // A task created from a continuation must outlive every
        // continuation that could ask for it again. A mount whose
        // retention is shorter than its MRTR TTL would otherwise have the
        // reuse lookup miss (the row is expired, so it is not reusable)
        // and every replay would mint another task and another tool run
        // from one chain: the exact duplication the shared chain key
        // exists to prevent.
        const ttlMs =
          continuation !== null
            ? Math.max(requestedTtlMs, continuation.exp - Date.now())
            : requestedTtlMs;
        const created = await ctx.runMutation(component.tasks.createTask, {
          taskId,
          ownerSubject: identity.subject,
          ...(options.tasks.scope !== undefined
            ? { scope: options.tasks.scope }
            : {}),
          toolName: tool.name,
          toolKind: tool.kind,
          args,
          caller: identity,
          idempotencyKey,
          executor,
          // The hook ran above and returned null (approve); record it so
          // the executor can tell "this call was confirmed" from "this
          // task predates the tool becoming gated".
          ...(beforeCall !== undefined ? { mrtrApproved: true } : {}),
          ...(ttlMs !== undefined ? { ttlMs } : {}),
        });
        if (!created.created) {
          // Client-caused rejections map to INVALID_PARAMS with a clear
          // message; a duplicate 128-bit id is genuinely-broken and logs.
          const clientReason: Record<string, string | undefined> = {
            args_too_large:
              "Task arguments exceed the permitted serialized size",
            caller_too_large:
              "The caller identity is too large to snapshot for a task",
            limit_exceeded:
              "Too many active tasks for this caller; poll or cancel " +
              "existing tasks before creating more",
          };
          const message_ = clientReason[created.reason];
          if (message_ === undefined) {
            console.error(
              "[mcp-gateway] task creation failed",
              created.reason,
              tool.name,
            );
          }
          body = jsonErrorEnvelope(
            message.id,
            message_ !== undefined ? INVALID_PARAMS : INTERNAL_ERROR,
            message_ ?? "Failed to create task",
          );
          break;
        }
        // The id the CLIENT gets. Normally the one just generated, but a
        // replayed continuation is answered with the task its chain key
        // already owns, so every step below must speak about that row, not
        // the id this request happened to mint and never used.
        const effectiveTaskId = created.task.taskId;
        // A reused row is normally the task the original request already
        // started, and starting again would run the host's workflow twice
        // for one task. `startPending` is the exception the component
        // reports: the row is host-executed, non-terminal, and was never
        // marked started, so its original request died between creating it
        // and getting execution going. This retry is that row's only
        // chance, skipping it would hand back a handle nothing advances.
        const mustStart =
          created.reused !== true || created.startPending === true;
        if (created.startPending === true) {
          console.warn(
            "[mcp-gateway] starting a reused task whose original request " +
              "never recorded a start",
            effectiveTaskId,
            tool.name,
          );
        }
        if (options.tasks.execute && mustStart) {
          try {
            await options.tasks.execute(ctx, {
              taskId: effectiveTaskId,
              toolName: tool.name,
              toolKind: tool.kind,
              args,
              identity,
              idempotencyKey,
              expiresAt: created.task.expiresAt,
            });
            // Durable execution is going; record it so a replay of this
            // request reuses the row without starting a second run, and so
            // a replay after a FAILED start still starts one.
            try {
              await ctx.runMutation(component.tasks.markTaskStarted, {
                taskId: effectiveTaskId,
              });
            } catch (markErr) {
              // Losing the marker only costs an extra start on a replay,
              // which the host's idempotency key already absorbs. Do not
              // fail a task whose execution is running.
              console.warn(
                "[mcp-gateway] could not record that task execution started",
                effectiveTaskId,
                markErr,
              );
            }
          } catch (err) {
            // The executor could not start durable execution; fail the
            // task so the client is not left polling a dead handle.
            console.error(
              "[mcp-gateway] task executor threw during start",
              tool.name,
              err,
            );
            let failed: string | null = null;
            try {
              failed = await ctx.runMutation(component.tasks.failTask, {
                taskId: effectiveTaskId,
                error: {
                  code: INTERNAL_ERROR,
                  message: "Task failed to start",
                },
                auditErrorMessage:
                  err instanceof Error ? err.message : String(err),
              });
            } catch (failErr) {
              // Double fault: the row could not be failed either. The
              // client still gets a clean error (not a raw 500); the
              // orphaned working row is bounded by its TTL.
              console.error(
                "[mcp-gateway] failed to mark task as failed after " +
                  "executor start error",
                effectiveTaskId,
                failErr,
              );
            }
            if (failed !== null && failed !== "finalized") {
              // The throw came AFTER the executor advanced the task: it
              // already completed, was cancelled, or is awaiting input
              // (a `requireTaskInput` that succeeded before the hook
              // threw). Reporting "failed to start" would strand a live
              // task behind an error envelope with no handle, so hand
              // back the handle and let the recorded state speak.
              console.error(
                "[mcp-gateway] task executor threw after the task had " +
                  "already advanced (" +
                  failed +
                  "); durable execution may still be running, so the " +
                  "recorded task state is authoritative, not this error",
                effectiveTaskId,
              );
            } else {
              body = jsonErrorEnvelope(
                message.id,
                INTERNAL_ERROR,
                "Task failed to start",
              );
              break;
            }
          }
        }
        // `created.task` is a snapshot from before `execute` ran, and a
        // host executor may legitimately advance the row inside this same
        // request (a `requireTaskInput` that puts it straight into
        // `input_required`). Re-read so the handle we return does not
        // advertise a status the database no longer holds; fall back to
        // the snapshot if the read fails, since the row demonstrably
        // exists and the client can just poll.
        let descriptor_ = created.task;
        if (options.tasks.execute && mustStart) {
          try {
            const fresh = await ctx.runQuery(component.tasks.getTaskForOwner, {
              taskId: effectiveTaskId,
              ownerSubject: identity.subject,
              ...(options.tasks.scope !== undefined
                ? { scope: options.tasks.scope }
                : {}),
            });
            if (fresh) descriptor_ = fresh;
          } catch (err) {
            console.warn(
              "[mcp-gateway] could not re-read the created task; returning " +
                "the pre-execution snapshot",
              effectiveTaskId,
              err,
            );
          }
        }
        body = jsonResultEnvelope(message.id, {
          resultType: "task",
          ...taskWireFields(
            descriptor_,
            options.tasks.pollIntervalMs ?? TASK_POLL_INTERVAL_MS,
          ),
        });
        break;
      }

      // Allowed (and hook-approved, when one exists): dispatch via the
      // component. Only the chain's idempotency key is ever injected;
      // continuation state and input responses stayed in the hook, so
      // the Convex function remains MCP-unaware.
      const dispatchArgs =
        continuation && tool.mrtrArgs
          ? {
              ...args,
              [tool.mrtrArgs.idempotencyKey]: continuation.idempotencyKey,
            }
          : args;
      const dispatched = await ctx.runAction(component.dispatch.runTool, {
        name,
        args: dispatchArgs,
        auditIdentitySubject,
        identity,
      });
      if (!dispatched.ok) {
        // MCP 2025-06-18 §tools/call distinguishes:
        //   - Protocol errors (unknown tool, invalid args) → JSON-RPC error
        //   - Tool execution errors                        → result with isError:true
        // The model uses the latter to reason about retries; protocol
        // errors abort the call. Keep -32602 (unknown tool) as a
        // JSON-RPC error; everything else is an execution error and
        // surfaces as a tool result so the LLM can react.
        if (dispatched.error.code === -32602) {
          body = jsonErrorEnvelope(
            message.id,
            dispatched.error.code,
            dispatched.error.message,
          );
        } else {
          body = jsonResultEnvelope(message.id, {
            content: [{ type: "text", text: dispatched.error.message }],
            isError: true,
          });
        }
        break;
      }
      // Always ship the text-JSON `content` block for backwards-compat
      // with clients that don't know `structuredContent`. When the tool
      // declared an `outputSchema` (via `defineMcp*({ returns })`), MCP
      // 2025-06-18 §tools/call mandates ALSO sending the typed value
      // as `structuredContent`. Spec-compliant clients (claude.ai,
      // recent Inspector) prefer the structured form when present.
      // `JSON.stringify` throws on a bigint, and `v.int64()` is a
      // supported `returns` validator, so a tool declaring one would
      // otherwise escape the switch as a raw 500: no JSON-RPC envelope,
      // no CORS headers, nothing the client can act on.
      let serializedResult: string;
      try {
        serializedResult = JSON.stringify(dispatched.data, null, 2);
      } catch (err) {
        console.error(
          "[mcp-gateway] tool result cannot be serialized for the wire",
          tool.name,
          err,
        );
        body = jsonResultEnvelope(message.id, {
          content: [
            {
              type: "text",
              text: `Tool "${tool.name}" returned a value that cannot be represented on the wire`,
            },
          ],
          isError: true,
        });
        break;
      }
      body = jsonResultEnvelope(message.id, {
        content: [
          {
            type: "text",
            text: serializedResult,
          },
        ],
        // Tied to the SAME condition as the advertisement in
        // `tools/list`: a client that was shown an `outputSchema` treats
        // a missing `structuredContent` as a protocol error, and one
        // that was not shown a schema rejects a scalar block against its
        // own revision's type. Deciding both with one predicate is what
        // keeps the two halves from contradicting each other.
        ...(mayAdvertiseOutputSchema(
          advertisedSchema(
            tool.authoredOutputSchemaJson,
            tool.outputSchema,
            tool.name,
          ),
          isStateless,
        )
          ? { structuredContent: dispatched.data }
          : {}),
        isError: false,
      });
      break;
    }

    case "tasks/get":
    case "tasks/cancel":
    case "tasks/update": {
      // Task methods exist only on the stateless protocol: a session-based client
      // could never have created a task in the first place.
      if (!isStateless) {
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `Unsupported method: ${message.method}`,
        );
        break;
      }
      if (!options.tasks) {
        // Never advertised (see server/discover): unknown method.
        responseStatus = 404;
        body = jsonErrorEnvelope(
          message.id,
          -32601,
          `Unsupported method: ${message.method}`,
        );
        break;
      }
      // Before the identity gate and before the lookup: a client that
      // never negotiated the extension is told what it is missing rather
      // than being challenged to authenticate for a feature it has not
      // asked for. SEP-2663 names `-32021` for exactly this.
      if (!declaresTasksExtension(statelessClientCapabilities)) {
        raw = statelessErrorResponse(
          message.id,
          MISSING_REQUIRED_CLIENT_CAPABILITY,
          `${message.method} requires the ${TASKS_CAPABILITY_KEY} client ` +
            "capability",
          { requiredCapabilities: TASKS_REQUIRED_CAPABILITIES },
        );
        body = "";
        break;
      }
      if (!identity) {
        // Tasks are owner-bound; without an identity there is nothing a
        // task lookup could legally return. 401 + WWW-Authenticate so
        // browser clients begin OAuth discovery.
        raw = await requireAuthChallenge(ctx, request, component, message.id);
        body = "";
        break;
      }
      const taskId = message.params?.taskId;
      if (typeof taskId !== "string" || taskId.length === 0) {
        body = jsonErrorEnvelope(message.id, INVALID_PARAMS, "Missing task id");
        break;
      }
      const pollIntervalMs =
        options.tasks.pollIntervalMs ?? TASK_POLL_INTERVAL_MS;

      if (message.method === "tasks/cancel") {
        const cancelled = await ctx.runMutation(
          component.tasks.cancelTaskForOwner,
          {
            taskId,
            ownerSubject: identity.subject,
            ...(options.tasks.scope !== undefined
              ? { scope: options.tasks.scope }
              : {}),
          },
        );
        if (cancelled.outcome === "not_found") {
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            `Unknown task: ${taskId}`,
          );
          break;
        }
        if (cancelled.outcome === "conflict") {
          // Already terminal. SEP-2663 makes cancellation idempotent and
          // reserves an error for an id the server does not know, so this
          // answers the same empty ack a live task does. The settled
          // status is observed on the next `tasks/get`, which is where
          // the spec puts it either way.
          body = jsonResultEnvelope(message.id, { resultType: "complete" });
          break;
        }
        // Notify the host so it can stop its workflow run. Idempotent
        // repeats re-fire the hook on purpose: the hook is best-effort,
        // so re-sending the cancel is the client's (and operator's) way
        // to retry a notification that previously threw. Hooks must
        // therefore be idempotent (see docs/tasks.md).
        if (options.tasks.onCancel) {
          try {
            await options.tasks.onCancel(ctx, {
              taskId,
              toolName: cancelled.task.toolName,
            });
          } catch (err) {
            console.error(
              "[mcp-gateway] tasks onCancel hook threw; the task is " +
                "cancelled but the host workflow may still be running. " +
                "Re-sending the cancel retries the notification.",
              taskId,
              err,
            );
          }
        } else if (cancelled.executor === "host") {
          // The row is cancelled either way, so the wire answer is
          // correct, but nothing here stops the durable run. Keyed on the
          // ROW's executor, not this mount's config: the task table is
          // component-wide, so a host-executed task can legitimately be
          // cancelled through a mount that runs the built-in executor and
          // has no hook, which is exactly the case worth reporting.
          console.warn(
            "[mcp-gateway] host-executed task cancelled on a mount with no " +
              "onCancel hook; the durable run will not be stopped",
            taskId,
            cancelled.task.toolName,
          );
        }
        // An empty ack, not a task envelope: the discriminator belongs to
        // the creating call, and cancellation is cooperative, so the
        // status this settles to is whatever the next `tasks/get` reports.
        body = jsonResultEnvelope(message.id, { resultType: "complete" });
        break;
      }

      if (message.method === "tasks/get") {
        const task = await ctx.runQuery(component.tasks.getTaskForOwner, {
          taskId,
          ownerSubject: identity.subject,
          // Scope binds the task to the mount that created it: without
          // it, a mount with a narrower policy can serve a task the
          // caller started on a broader one.
          ...(options.tasks.scope !== undefined
            ? { scope: options.tasks.scope }
            : {}),
        });
        if (!task) {
          // Unknown, foreign, and expired ids answer identically so a
          // task's existence never leaks across callers.
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            `Unknown task: ${taskId}`,
          );
          break;
        }
        // `tasks/get` is an ordinary answer about a task, not the
        // creation of one, so it carries `complete` like every other
        // non-creating result. Only `tools/call` mints `resultType:
        // "task"`.
        body = jsonResultEnvelope(message.id, {
          resultType: "complete",
          ...taskWireFields(task, pollIntervalMs),
        });
        break;
      }

      // tasks/update answers an input round, and nothing else.
      // Cancellation is `tasks/cancel` above, which SEP-2663 gives its
      // own method rather than an action on this one.
      const inputResponses = message.params?.inputResponses;
      // The round the client is answering, echoed from the descriptor it
      // polled. Optional, but a malformed value is rejected rather than
      // coerced to "absent": absence is itself meaningful (it means
      // "answering a task that never asked a round"), so coercing `"1"`
      // or `1.5` would answer a client's type bug with the factually
      // wrong "these responses answer a superseded input round".
      const inputRoundRaw = message.params?.inputRound;
      const inputRound =
        inputRoundRaw === undefined ? undefined : Number(inputRoundRaw);
      if (
        inputRoundRaw !== undefined &&
        !(
          typeof inputRoundRaw === "number" &&
          Number.isSafeInteger(inputRoundRaw) &&
          inputRoundRaw >= 0
        )
      ) {
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          "inputRound must be a non-negative integer",
        );
        break;
      }
      if (inputResponses === undefined) {
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          "tasks/update requires inputResponses",
        );
        break;
      }
      // Reject a deeply nested inputResponses before it reaches the
      // submit mutation's serialization (would overflow into a raw 500).
      if (inputResponses !== undefined && nestsTooDeep(inputResponses)) {
        body = jsonErrorEnvelope(
          message.id,
          INVALID_PARAMS,
          "inputResponses nest too deeply",
        );
        break;
      }


      const submitted = await ctx.runMutation(
        component.tasks.submitInputResponsesForOwner,
        {
          taskId,
          ownerSubject: identity.subject,
          ...(options.tasks.scope !== undefined
            ? { scope: options.tasks.scope }
            : {}),
          inputResponses,
          ...(inputRound !== undefined ? { inputRound } : {}),
        },
      );
      switch (submitted.outcome) {
        case "not_found":
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            `Unknown task: ${taskId}`,
          );
          break;
        case "stale_round":
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            `These responses answer a superseded input round; the task is ` +
              `now awaiting round ${submitted.expectedRound}`,
          );
          break;
        case "conflict":
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            `Task is ${submitted.status} and does not accept input responses`,
          );
          break;
        case "mismatch":
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            "inputResponses must answer exactly the requested input keys " +
              'with an action of "accept", "decline", or "cancel"',
          );
          break;
        case "too_large":
          body = jsonErrorEnvelope(
            message.id,
            INVALID_PARAMS,
            "inputResponses exceed the permitted serialized size",
          );
          break;
        case "cancelled":
          // Every response carried action "cancel": treated as an owner
          // cancellation, including the host notification.
          if (options.tasks.onCancel) {
            try {
              await options.tasks.onCancel(ctx, {
                taskId,
                toolName: submitted.task.toolName,
              });
            } catch (err) {
              console.error(
                "[mcp-gateway] tasks onCancel hook threw; the task is " +
                  "cancelled but the host workflow may still be running. " +
                  "Re-sending the same responses retries the notification.",
                taskId,
                err,
              );
            }
          } else if (submitted.executor === "host") {
            console.warn(
              "[mcp-gateway] input responses cancelled a host-executed task " +
                "on a mount with no onCancel hook; the durable run will not " +
                "be stopped",
              taskId,
              submitted.task.toolName,
            );
          }
          body = jsonResultEnvelope(message.id, { resultType: "complete" });
          break;
        case "duplicate":
        case "accepted": {
          // Duplicates re-fire the hook on purpose: the hook is the only
          // signal that resumes a paused host workflow, and it is
          // best-effort. If it threw on the fresh acceptance, re-sending
          // the same responses is the client's recovery path, so it must
          // reach the host again. Hooks are required to be idempotent
          // (see docs/tasks.md). Re-sent all-cancel responses do NOT
          // arrive here: the component reports them as `cancelled` above,
          // which is what re-fires `onCancel`.
          if (options.tasks.onInputResponses) {
            try {
              await options.tasks.onInputResponses(ctx, {
                taskId,
                toolName: submitted.task.toolName,
                inputResponses: inputResponses as Record<string, unknown>,
              });
            } catch (err) {
              // The responses are durably stored on the row and the
              // client can retry the notification by re-sending them.
              console.error(
                "[mcp-gateway] tasks update hook threw; the responses are " +
                  "stored but the host workflow was not notified. " +
                  "Re-sending the same tasks/update retries the hook.",
                taskId,
                err,
              );
            }
          } else if (submitted.executor === "host") {
            // Storing the responses is not the point: resuming the paused
            // execution is, and only the hook can do that. The client sees
            // success either way, so this is the operator's only signal.
            // Keyed on the ROW: only a host-executed task can be paused,
            // and it can be answered through any mount that resolves its
            // owner, including one with no hook to resume it.
            console.warn(
              "[mcp-gateway] accepted input responses for a host-executed " +
                "task on a mount with no onInputResponses hook; the paused " +
                "task will not resume",
              taskId,
              submitted.task.toolName,
            );
          }
          body = jsonResultEnvelope(message.id, { resultType: "complete" });
          break;
        }
      }
      break;
    }

    default:
      if (isStateless) responseStatus = 404;
      body = jsonErrorEnvelope(
        message.id,
        -32601,
        `Unsupported method: ${message.method}`,
      );
  }

  if (raw) return raw;

  if (isStateless) {
    body = finalizeStatelessResult(body, message.method!, options);
  }

  if (responseStatus !== 200) {
    return new Response(body, {
      status: responseStatus,
      headers: { "content-type": "application/json" },
    });
  }

  const headers: Record<string, string> = {};
  if (issueSessionHeader) headers["mcp-session-id"] = sessionId;

  if (clientWantsSse(request)) {
    const stream = new ReadableStream<Uint8Array>({
      start(controller) {
        const encoder = new TextEncoder();
        // 2026-07-28 dropped `Last-Event-ID` resumability, so an event id
        // carries no meaning there. Legacy frames keep id 1, identical
        // across every session-based revision (see sseResponseFrame).
        controller.enqueue(encoder.encode(sseResponseFrame(body, isStateless)));
        controller.close();
      },
    });
    return new Response(stream, {
      status: 200,
      headers: {
        ...headers,
        "content-type": "text/event-stream",
        "cache-control": "no-cache, no-transform",
        // Spec SHOULD: tell reverse proxies (nginx) not to buffer the
        // stream, otherwise events are held back until the response ends.
        "x-accel-buffering": "no",
      },
    });
  }

  return new Response(body, {
    status: 200,
    headers: {
      ...headers,
      "content-type": "application/json",
    },
  });
}