import path from "node:path"
import fs from "node:fs/promises"
import os from "node:os"
import { describe, expect, test } from "bun:test"
import {
  Client,
  InMemoryTransport,
  StreamableHTTPClientTransport,
  isJSONRPCRequest,
  type ListToolsResult,
} from "@modelcontextprotocol/client"
import {
  createMcpHandler,
  inputRequired,
  inputResponse,
  Server,
  WebStandardStreamableHTTPServerTransport,
} from "@modelcontextprotocol/server"
import { Document, Event, Info } from "@opencode/schema/config"
import { ConfigMCP } from "@opencode/schema/config/mcp"
import { McpEvent } from "@opencode/schema/mcp-event"
import { ConfigPolicy } from "@opencode/schema/config/policy"
import { Config } from "@opencode/core/config"
import { ConfigMcpPlugin } from "@opencode/core/config/plugin/mcp"
import { Credential } from "@opencode/core/credential"
import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
import { LayerNode } from "@opencode/util/effect/layer-node"
import { Bus } from "@opencode/core/bus"
import { ID, type Payload } from "@opencode/schema/event"
import { Form } from "@opencode/core/form"
import { Integration } from "@opencode/core/integration"
import { Environment } from "@opencode/core/environment/index"
import { EnvironmentUnavailable } from "@opencode/core/environment/unavailable"
import { Location } from "@opencode/core/location"
import { Mcp } from "@opencode/core/mcp/index"
import { McpClient } from "@opencode/core/mcp/client"
import { McpPool } from "@opencode/core/mcp/pool"
import { McpStdio } from "@opencode/core/mcp/stdio"
import { Permission } from "@opencode/core/permission"
import { AbsolutePath } from "@opencode/core/schema"
import { Session } from "@opencode/core/session"
import { State } from "@opencode/core/state"
import { McpTool } from "@opencode/core/tool/mcp"
import { Command } from "@opencode/core/command"
import { CommandPlugin } from "@opencode/core/plugin/command"
import { McpResourceTools } from "@opencode/core/tool/plugin/mcp-resource"
import { Tool } from "@opencode/core/tool"
import { ToolOutput } from "@opencode/core/tool-output"
import {
  Context,
  Deferred,
  Effect,
  Exit,
  Fiber,
  FiberSet,
  Layer,
  Logger,
  Option,
  PubSub,
  Ref,
  Schedule,
  Schema,
  Scope,
  Semaphore,
  Sink,
  Stream,
} from "effect"
import { TestClock } from "effect/testing"
import { ChildProcess, ChildProcessSpawner } from "effect/process"
import { ExitCode, makeHandle, ProcessId } from "effect/process/ChildProcessSpawner"
import { Image } from "@opencode/core/image"
import { Global } from "@opencode/util/global"
import { isRecord } from "@opencode/util/record"
import { CrossSpawnSpawner } from "@opencode/util/cross-spawn-spawner"
import { advance, drain } from "./lib/clock"
import { testEffect } from "./lib/effect"
import { registerIntegrationPolicy, setOrganizationPolicy } from "./fixture/policy"
import { imagePassthrough } from "./lib/image"
import { location } from "./fixture/location"
import { host } from "./plugin/host"
import { tmpdirScoped } from "./fixture/tmpdir"
import { hostEnvironmentLayer, recordingEnvironmentLayer } from "./fixture/environment"
import {
  codeModeListings,
  executeTool,
  registerToolPlugin,
  toolDefinitions,
  toolIdentity,
  waitForTool,
} from "./lib/tool"

let assertion: Deferred.Deferred<Permission.AssertInput> | undefined
let decision: Effect.Effect<void, Permission.Error> = Effect.void
let calls = 0
let invocations: Array<Parameters<Mcp.Interface["callTool"]>[0]> = []

type ResourcePage = {
  items: Array<{ name: string; uri: string; description?: string; mimeType?: string }>
  nextCursor?: string
}

type ResourceTemplatePage = {
  items: Array<{ name: string; uriTemplate: string; description?: string; mimeType?: string }>
  nextCursor?: string
}

function resourceServer(
  input: {
    /** Serve 2026-07-28 only through createMcpHandler; the default is a sessionful legacy transport. */
    modern?: boolean
    resources?: boolean
    listChanged?: boolean
    emptyElicitation?: boolean
    urlElicitation?: boolean
    /**
     * Ahead of the valid tools, list entries that cannot be used (null, not an object, no name, an input
     * schema that is not an object), one whose input schema has a string `required`, and one with a
     * malformed optional field of each kind, then one with every optional field valid.
     */
    malformedTool?: boolean
    /** Listed after the default tools. A call's `structured` argument comes back as structured content. */
    tools?: ListToolsResult["tools"]
    /** Legacy servers give up on the URL elicitation after this many milliseconds. */
    elicitationTimeout?: number
    /** Lists a `roots` tool that asks for the client's roots (2026-07-28 only) and returns them. */
    roots?: boolean
    respond?: (request: Request) => Response | undefined | Promise<Response | undefined>
  } = {},
) {
  return Effect.acquireRelease(
    Effect.promise(async () => {
      const state = {
        resources: [] as ResourcePage["items"],
        templates: [] as ResourceTemplatePage["items"],
        resourcePages: undefined as Record<string, ResourcePage> | undefined,
        templatePages: undefined as Record<string, ResourceTemplatePage> | undefined,
        contents: [
          { uri: "docs://readme", text: "hello", mimeType: "text/plain" },
          { uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" },
        ] as Array<{ uri: string; text: string; mimeType?: string } | { uri: string; blob: string; mimeType?: string }>,
        resourceLists: 0,
        resourceReads: [] as string[],
        templatesUnsupported: false,
        missing: [] as string[],
        templateLists: 0,
        toolLists: 0,
        toolCalls: [] as Array<{
          name: string
          arguments: Record<string, unknown> | undefined
          sessionID: unknown
          progressToken: unknown
        }>,
        initializations: 0,
        /** Listed after the configured tools; change it, then announce the change. */
        extraTools: [] as ListToolsResult["tools"],
        urls: [] as string[],
        sessions: [] as string[],
      }
      // One Server speaks one session, so a restart is a fresh Server and transport. Requests that
      // still carry the previous session id are then unknown to the new transport.
      const server = () => {
        const protocol = new Server(
          { name: "mcp-resources", title: "MCP Resources", version: "1.0.0" },
          {
            capabilities: {
              tools: { listChanged: input.listChanged },
              prompts: {},
              ...(input.resources === false ? {} : { resources: { listChanged: input.listChanged } }),
            },
            instructions: "Use the resources tools.",
          },
        )
        protocol.setRequestHandler("tools/list", () => {
          state.toolLists += 1
          return Promise.resolve({
            tools: [
              // Servers send raw JSON; the cast lets the fixture send a tool the schema rejects.
              ...(input.malformedTool
                ? ([
                    null,
                    "not-a-tool",
                    { inputSchema: { type: "object" } },
                    { name: "no-schema" },
                    { name: "malformed", inputSchema: { type: "string" } },
                    { name: "bad-required", inputSchema: { type: "object", required: "text" } },
                    {
                      name: "bad-fields",
                      title: 1,
                      description: 42,
                      outputSchema: "none",
                      annotations: "read-only",
                      execution: true,
                      _meta: [],
                      inputSchema: { type: "object" },
                    },
                    {
                      name: "full",
                      title: "Full",
                      description: "Every field",
                      inputSchema: { type: "object" },
                      annotations: { readOnlyHint: true },
                      execution: { taskSupport: "optional" },
                      _meta: { "example/key": 1 },
                      icons: [{ src: "https://example.com/icon.png" }],
                    },
                  ] as unknown as ListToolsResult["tools"])
                : []),
              ...(input.emptyElicitation
                ? [{ name: "empty-elicitation", inputSchema: { type: "object" as const, properties: {} } }]
                : input.urlElicitation
                  ? [{ name: "url-elicitation", inputSchema: { type: "object" as const, properties: {} } }]
                  : [{ name: "echo", inputSchema: { type: "object" as const, properties: {} } }]),
              ...(input.tools ?? []),
              ...(input.roots ? [{ name: "roots", inputSchema: { type: "object" as const, properties: {} } }] : []),
              ...state.extraTools,
            ],
          })
        })
        if (input.emptyElicitation) {
          protocol.setRequestHandler("tools/call", async () => {
            const result = await protocol.elicitInput({
              mode: "form",
              message: "Confirm",
              requestedSchema: { type: "object", properties: {} },
            })
            return {
              content: [{ type: "text", text: JSON.stringify(result) }],
              structuredContent: result,
            }
          })
        }
        if (input.urlElicitation) {
          const url = "https://example.com/authorize"
          // Modern servers cannot call elicitInput; they return input_required and the client retries.
          protocol.setRequestHandler("tools/call", async (request, ctx) => {
            const responses = ctx.mcpReq.inputResponses
            if (input.modern && !responses)
              return inputRequired({ inputRequests: { auth: inputRequired.elicitUrl({ message: "Authorize", url }) } })
            const response = inputResponse(responses, "auth")
            const result = input.modern
              ? { action: response.kind === "elicit" ? response.action : "cancel" }
              : await protocol.elicitInput(
                  {
                    mode: "url",
                    message: "Authorize access",
                    url,
                    elicitationId: "elicitation-test",
                  },
                  input.elicitationTimeout === undefined ? undefined : { timeout: input.elicitationTimeout },
                )
            return {
              content: [{ type: "text", text: JSON.stringify(result) }],
              structuredContent: result,
            }
          })
        }
        if (!input.emptyElicitation && !input.urlElicitation) {
          protocol.setRequestHandler("tools/call", (request, ctx) => {
            if (request.params.name === "roots") {
              const responses = ctx.mcpReq.inputResponses
              if (!responses)
                return Promise.resolve(inputRequired({ inputRequests: { roots: inputRequired.listRoots() } }))
              return Promise.resolve({ content: [], structuredContent: inputResponse(responses, "roots") })
            }
            state.toolCalls.push({
              name: request.params.name,
              arguments: request.params.arguments,
              sessionID: request.params._meta?.["ai.opencode/sessionID"],
              progressToken: request.params._meta?.progressToken,
            })
            const structured = request.params.arguments?.structured
            return Promise.resolve(
              structured === undefined ? { content: [] } : { content: [], structuredContent: structured },
            )
          })
        }
        protocol.setRequestHandler("prompts/list", () => Promise.resolve({ prompts: [{ name: "greet" }] }))
        protocol.setRequestHandler("prompts/get", (request) =>
          Promise.resolve({
            messages: [{ role: "user", content: { type: "text", text: `hi ${request.params.arguments?.name}` } }],
          }),
        )
        if (input.resources !== false) {
          protocol.setRequestHandler("resources/list", (request) => {
            state.resourceLists += 1
            const page = state.resourcePages?.[request.params?.cursor ?? "initial"]
            return Promise.resolve({ resources: page?.items ?? state.resources, nextCursor: page?.nextCursor })
          })
          protocol.setRequestHandler("resources/templates/list", (request) => {
            state.templateLists += 1
            if (state.templatesUnsupported) return Promise.reject(new Error("Method not found"))
            const page = state.templatePages?.[request.params?.cursor ?? "initial"]
            return Promise.resolve({ resourceTemplates: page?.items ?? state.templates, nextCursor: page?.nextCursor })
          })
          protocol.setRequestHandler("resources/read", (request) => {
            state.resourceReads.push(request.params.uri)
            if (state.missing.includes(request.params.uri)) return Promise.reject(new Error("Resource not found"))
            return Promise.resolve({ contents: state.contents })
          })
        }
        return protocol
      }
      const build = async () => {
        const protocol = server()
        const transport = new WebStandardStreamableHTTPServerTransport({
          sessionIdGenerator: () => crypto.randomUUID(),
          enableJsonResponse: true,
        })
        await protocol.connect(transport)
        return { protocol, transport }
      }
      let current = await build()
      const modern = input.modern ? createMcpHandler(server, { legacy: "reject" }) : undefined
      const http = Bun.serve({
        port: 0,
        fetch: async (request) => {
          state.urls.push(request.url)
          const session = request.headers.get("mcp-session-id")
          if (session !== null && !state.sessions.includes(session)) state.sessions.push(session)
          const body: unknown = request.method === "POST" ? await request.clone().json() : undefined
          if (typeof body === "object" && body !== null && "method" in body && body.method === "initialize") {
            state.initializations += 1
          }
          return (await input.respond?.(request)) ?? modern?.fetch(request) ?? current.transport.handleRequest(request)
        },
      })
      return {
        state,
        url: http.url.toString(),
        clientVersion: () => current.protocol.getClientVersion(),
        sendToolListChanged: () =>
          modern ? Promise.resolve(modern.notify.toolsChanged()) : current.protocol.sendToolListChanged(),
        sendResourceListChanged: () =>
          modern ? Promise.resolve(modern.notify.resourcesChanged()) : current.protocol.sendResourceListChanged(),
        completeElicitation: () => current.protocol.createElicitationCompletionNotifier("elicitation-test")(),
        restart: async () => {
          await current.protocol.close().catch(() => {})
          current = await build()
        },
        close: async () => {
          await current.protocol.close().catch(() => {})
          await modern?.close()
          await http.stop(true)
        },
      }
    }),
    (server) => Effect.promise(server.close),
  )
}

const poolLayer = McpPool.layer.pipe(Layer.provide(LayerNode.compile(CrossSpawnSpawner.node)))

function resourceMcpLayer(
  server: string | typeof ConfigMCP.Server.Type,
  onFormCreated?: (form: Form.Info) => Effect.Effect<void>,
  options?: Mcp.Options,
  overrides?: {
    policies?: readonly ConfigPolicy.Info[]
    entries?: Config.Interface["entries"]
    subscribe?: Bus.Interface["subscribe"]
    environment?: Layer.Layer<Environment.Service>
    published?: string[]
    pool?: Layer.Layer<McpPool.Service>
    directory?: string
    /** The configured server's name; "resources" unless given. */
    name?: string
    credentials?: Partial<Credential.Interface>
    bus?: Layer.Layer<Bus.Service>
  },
) {
  const directory = AbsolutePath.make(overrides?.directory ?? import.meta.dir)
  const unusedIntegration = () => Effect.die("unused integration service")
  return Layer.effectDiscard(
    Effect.gen(function* () {
      const bus = yield* Bus.Service
      const mcp = yield* Mcp.Service
      yield* State.batch(
        Effect.gen(function* () {
          yield* ConfigMcpPlugin.register(bus.subscribe())
          yield* registerIntegrationPolicy({ mcp, events: bus.subscribe() })
        }),
      )
    }),
  ).pipe(
    Layer.provideMerge(Mcp.layer(options)),
    Layer.provideMerge(Form.layer),
    Layer.provide(
      Layer.mergeAll(
        overrides?.pool ?? poolLayer,
        overrides?.entries
          ? Layer.succeed(
              Config.Service,
              Config.Service.of({
                entries: overrides.entries,
                changes: () => Stream.never,
              }),
            )
          : Config.testLayer([
              new Document({
                type: "document",
                info: new Info({
                  policies: overrides?.policies ?? [],
                  mcp: new ConfigMCP.Info({
                    servers: {
                      [overrides?.name ?? "resources"]:
                        typeof server === "string"
                          ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false })
                          : server,
                    },
                  }),
                }),
              }),
            ]),
        Layer.succeed(Location.Service, Location.Service.of(location({ directory }))),
        overrides?.bus ??
          Layer.mock(Bus.Service, {
            subscribe: overrides?.subscribe ?? (() => Stream.never),
            publish: (definition, data) =>
              Effect.suspend(() => {
                const event = {
                  id: ID.create(),
                  type: definition.type,
                  data,
                } as Payload<typeof definition>
                overrides?.published?.push(event.type)
                if (event.type !== Form.Event.Created.type || !onFormCreated) return Effect.succeed(event)
                return onFormCreated(Schema.decodeUnknownSync(Form.Event.Created.data)(data).form).pipe(
                  Effect.as(event),
                )
              }),
          }),
        Layer.mock(Integration.Service, {
          revision: () => 0,
          transform: () => Effect.succeed({ dispose: Effect.void }),
          connection: {
            active: unusedIntegration,
            resolve: unusedIntegration,
            key: unusedIntegration,
            external: unusedIntegration,
            activate: unusedIntegration,
            update: unusedIntegration,
            remove: unusedIntegration,
            status: unusedIntegration,
          },
          oauth: {
            connect: unusedIntegration,
            status: unusedIntegration,
            complete: unusedIntegration,
            cancel: unusedIntegration,
          },
          command: {
            connect: unusedIntegration,
            status: unusedIntegration,
            cancel: unusedIntegration,
          },
        }),
        Layer.mock(Credential.Service, { list: () => Effect.succeed([]), ...overrides?.credentials }),
        overrides?.environment ?? hostEnvironmentLayer,
      ),
    ),
  )
}

const connect = (server: string, config: typeof ConfigMCP.Server.Type, directory: string) =>
  McpClient.connect({ server, config, directory }).pipe(Effect.provide(hostEnvironmentLayer))

// Reads no longer wait for startup, so tests that assert on a connected server settle it first.
const settled = (service: Mcp.Interface, name = "resources") =>
  Effect.gen(function* () {
    const status = (yield* service.servers()).find((server) => server.name === name)?.status
    if (status?.status === "pending") return yield* Effect.fail(status)
    return status
  }).pipe(Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }))

const mcp = Layer.mock(Mcp.Service, {
  awaitStartup: () => Effect.void,
  tools: () =>
    Effect.succeed([
      {
        server: Mcp.ServerName.make("demo"),
        name: "search",
        description: "Search",
        inputSchema: { type: "object", properties: {} },
        outputSchema: {
          type: "object",
          properties: { ok: { type: "boolean" } },
          required: ["ok"],
        },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("demo"),
        name: "status",
        description: "Status",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("demo"),
        name: "issues",
        description: "Returns JSON as text",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("demo"),
        name: "count",
        description: "Returns a number as text",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("demo"),
        name: "typed",
        description: "Declares a string output and returns JSON as text",
        inputSchema: { type: "object", properties: {} },
        outputSchema: { type: "string" },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("direct"),
        name: "issues",
        codemode: false,
        description: "Returns JSON as text",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("direct"),
        name: "lookup",
        codemode: false,
        description: "Lookup",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("direct"),
        name: "fail",
        codemode: false,
        description: "Always fails",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("direct"),
        name: "media",
        codemode: false,
        description: "Returns text and an image",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
      {
        server: Mcp.ServerName.make("direct"),
        name: "attachments",
        codemode: false,
        description: "Returns resource blobs that cannot be sent inline",
        inputSchema: { type: "object", properties: {} },
      } satisfies Mcp.Tool,
    ]),
  callTool: (input) =>
    Effect.sync(() => {
      calls += 1
      invocations.push(input)
      if (input.name === "fail")
        return {
          isError: true,
          content: [{ type: "text", text: "search index unavailable" }],
        } satisfies Mcp.ToolResult
      if (input.name === "media")
        return {
          isError: false,
          content: [
            { type: "text", text: "rendered chart" },
            { type: "media", data: "aGVsbG8=", mimeType: "image/png" },
          ],
        } satisfies Mcp.ToolResult
      if (input.name === "attachments")
        return {
          isError: false,
          content: [
            {
              type: "media",
              data: Buffer.from("# Notes").toString("base64"),
              mimeType: "text/markdown",
              uri: "file:///README.md",
            },
            {
              type: "media",
              data: "UEFSMQ==",
              mimeType: "application/vnd.apache.parquet",
              uri: "file:///data.parquet",
            },
          ],
        } satisfies Mcp.ToolResult
      if (input.name === "status")
        return {
          isError: false,
          content: [{ type: "text", text: "hello" }],
        } satisfies Mcp.ToolResult
      if (input.name === "issues" || input.name === "typed")
        return {
          isError: false,
          content: [{ type: "text", text: '{"issues":[{"id":1}]}' }],
        } satisfies Mcp.ToolResult
      if (input.name === "count")
        return {
          isError: false,
          content: [{ type: "text", text: "42" }],
        } satisfies Mcp.ToolResult
      return {
        isError: false,
        structured: { ok: true },
        content: [],
      } satisfies Mcp.ToolResult
    }),
})
const permissions = Layer.mock(Permission.Service, {
  assert: (input) =>
    Effect.gen(function* () {
      if (!assertion) return yield* Effect.die("Permission test is not initialized")
      yield* Deferred.succeed(assertion, input)
      yield* decision
    }),
})
const events = Layer.mock(Bus.Service, { subscribe: () => Stream.never })
const outputData = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-mcp-output-"))
const it = testEffect(
  AppNodeBuilder.build(LayerNode.group([Tool.node, McpTool.node]), [
    Mcp.node.replace(mcp),
    Permission.node.replace(permissions),
    Bus.node.replace(events),
    Image.node.replace(imagePassthrough),
    Global.node.replace(Global.layerWith({ data: outputData })),
  ]),
)

describe("MCP errors", () => {
  test("expose useful messages", () => {
    expect(new Mcp.NotFoundError({ server: Mcp.ServerName.make("demo") }).message).toBe("MCP server not found: demo")
    expect(
      new Mcp.ToolCallError({ server: Mcp.ServerName.make("demo"), tool: "search", message: "failed" }).message,
    ).toBe("failed")
    expect(new McpClient.NeedsAuthError({ server: "demo", message: "Unauthorized" }).message).toBe("Unauthorized")
    expect(new McpClient.ConnectError({ server: "demo", message: "offline" }).message).toBe("offline")
  })
})

test("MCP tool names match V1 sanitization", () => {
  expect(McpTool.namespace("context 7")).toBe("context_7")
  expect(McpTool.name("context 7", "resolve.library/id")).toBe("context_7_resolve_library_id")
})

test("passes session IDs as MCP request metadata", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer()
        const connection = yield* connect(
          "session-metadata",
          new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }),
          import.meta.dir,
        )
        yield* connection.callTool({
          name: "echo",
          args: { text: "hello" },
          sessionID: Session.ID.make("ses_mcp_metadata"),
        })
        yield* connection.callTool({ name: "echo" })

        expect(server.state.toolCalls).toEqual([
          {
            name: "echo",
            arguments: { text: "hello" },
            sessionID: "ses_mcp_metadata",
            progressToken: expect.any(Number),
          },
          {
            name: "echo",
            arguments: {},
            sessionID: undefined,
            progressToken: expect.any(Number),
          },
        ])
        expect(server.state.toolCalls[0]?.progressToken).not.toBe(server.state.toolCalls[1]?.progressToken)
      }),
    ),
  )
})

test("preserves output schema validation across paginated tool discovery", async () => {
  const server = new Server({ name: "pagination", version: "1.0.0" }, { capabilities: { tools: {} } })
  server.setRequestHandler("tools/list", ({ params }) =>
    Promise.resolve(
      params?.cursor === "page-2"
        ? {
            tools: [
              {
                name: "second",
                inputSchema: { type: "object" },
                outputSchema: {
                  type: "object",
                  properties: { value: { type: "number" } },
                  required: ["value"],
                },
              },
            ],
          }
        : {
            tools: [
              {
                name: "first",
                inputSchema: { type: "object" },
                outputSchema: {
                  type: "object",
                  properties: { value: { type: "string" } },
                  required: ["value"],
                },
              },
            ],
            nextCursor: "page-2",
          },
    ),
  )
  server.setRequestHandler("tools/call", ({ params }) =>
    Promise.resolve({
      content: [],
      structuredContent: { value: params.name === "first" ? 42 : 1 },
    }),
  )

  const client = new Client({ name: "pagination-test", version: "1.0.0" })
  const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair()
  await Promise.all([client.connect(clientTransport), server.connect(serverTransport)])

  try {
    // Without a cursor the SDK walks every page; the page-1 validator must survive the page-2 fetch.
    const listed = await client.listTools()
    expect(listed.tools.map((tool) => tool.name)).toEqual(["first", "second"])
    expect(listed.nextCursor).toBeUndefined()
    await expect(client.callTool({ name: "first", arguments: {} })).rejects.toThrow(
      "Structured content does not match the tool's output schema",
    )
    await expect(client.callTool({ name: "second", arguments: {} })).resolves.toMatchObject({
      structuredContent: { value: 1 },
    })
  } finally {
    await Promise.all([client.close(), server.close()])
  }
})

test("retains output schemas across paginated MCP discovery", async () => {
  const tools = await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "pagination",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
          }),
          import.meta.dir,
        )
        return yield* connection.tools()
      }),
    ),
  )

  expect(tools.map((tool) => ({ name: tool.name, outputSchema: tool.outputSchema }))).toEqual([
    {
      name: "first",
      outputSchema: {
        type: "object",
        properties: { value: { type: "string" } },
        required: ["value"],
      },
    },
    {
      name: "second",
      outputSchema: {
        type: "object",
        properties: { value: { type: "number" } },
        required: ["value"],
      },
    },
  ])
})

test("accepts legacy MCP tools whose output schema is not a root object", async () => {
  const result = await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "union",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-union-output-schema.ts")],
          }),
          import.meta.dir,
        )
        return {
          tools: yield* connection.tools(),
          call: yield* connection.callTool({ name: "union" }),
        }
      }),
    ),
  )

  expect(result.tools.map((tool) => tool.name)).toEqual(["union"])
  expect(result.tools[0]?.outputSchema).toHaveProperty("anyOf")
  expect(result.call.structured).toEqual({ kind: "struct" })
})

test("lists paginated prompts and invokes them through the MCP client", async () => {
  const result = await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "prompts",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-prompts.ts")],
          }),
          import.meta.dir,
        )
        return {
          prompts: yield* connection.prompts(),
          result: yield* connection.prompt({ name: "first", args: { topic: "Effect" } }),
        }
      }),
    ),
  )

  expect(result.prompts).toEqual([
    {
      name: "first",
      description: "First prompt",
      arguments: [{ name: "topic", description: "Topic to explain", required: true }],
    },
    { name: "second", description: "Second prompt", arguments: undefined },
  ])
  expect(result.result).toEqual({ messages: [{ role: "user", content: { type: "text", text: "Effect" } }] })
})

test("spawns local MCP servers through the location environment", async () => {
  const spawns: Array<ChildProcess.Command> = []
  const cwd = path.join(import.meta.dir, "fixture")
  const config = new ConfigMCP.Local({
    type: "local",
    command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
    cwd: "fixture",
    environment: { MCP_LOCATION_TEST: "configured" },
  })

  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* McpClient.connect({ server: "environment", config, directory: import.meta.dir })
        yield* connection.tools()
      }),
    ).pipe(Effect.provide(recordingEnvironmentLayer(spawns))),
  )

  expect(spawns).toHaveLength(1)
  const command = spawns[0]
  if (!command || !ChildProcess.isStandardCommand(command)) throw new Error("Expected a standard process command")
  expect(command.command).toBe(process.execPath)
  expect(command.options.cwd).toBe(cwd)
  expect(command.options.extendEnv).toBe(true)
  expect(command.options.env).toEqual({ MCP_LOCATION_TEST: "configured" })
})

test("reports a local MCP server as failed when the location has no execution plane", async () => {
  const config = new ConfigMCP.Local({ type: "local", command: ["example-mcp"] })
  const driver = Environment.makeMemoryDriver()
  const environment = Layer.succeed(
    Environment.Service,
    Environment.Service.of({ files: Environment.makeFiles(driver), spawner: EnvironmentUnavailable.spawner }),
  )

  await Effect.runPromise(
    Effect.gen(function* () {
      const service = yield* Mcp.Service
      const status = yield* settled(service)
      expect(status).toEqual({
        status: "failed",
        error: expect.stringContaining("location has no execution plane"),
      })
    }).pipe(Effect.provide(resourceMcpLayer(config, undefined, undefined, { environment }))),
  )
})

test("rejects sends before the stdio transport is started", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const transport = yield* McpStdio.make({
          server: "not-started",
          command: process.execPath,
          args: [path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
          cwd: import.meta.dir,
          environment: {},
        })
        yield* Effect.tryPromise({
          try: () => transport.send({ jsonrpc: "2.0", method: "notifications/initialized" }),
          catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
        }).pipe(
          Effect.flip,
          Effect.tap((error) => Effect.sync(() => expect(error.message).toBe("Not connected"))),
        )
      }).pipe(Effect.provide(hostEnvironmentLayer)),
    ),
  )
})

test("joins concurrent stdio transport closes", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const transport = yield* McpStdio.make({
          server: "concurrent-close",
          command: "unused",
          args: [],
          cwd: import.meta.dir,
          environment: {},
        })
        const first = transport.close()
        expect(transport.close()).toBe(first)
        yield* Effect.promise(() => first)
      }).pipe(Effect.provide(hostEnvironmentLayer)),
    ),
  )
})

const testMcpDescendants =
  process.platform === "win32" ? testEffect(hostEnvironmentLayer).live.skip : testEffect(hostEnvironmentLayer).live
testMcpDescendants(
  "terminates MCP descendants after the wrapper exits successfully",
  Effect.gen(function* () {
    const tmp = yield* tmpdirScoped()
    const pidFile = path.join(tmp.path, "child.pid")
    yield* Effect.addFinalizer(() =>
      Effect.tryPromise(async () => process.kill(Number(await fs.readFile(pidFile, "utf8")), "SIGKILL")).pipe(
        Effect.ignore,
      ),
    )
    const ready = yield* Deferred.make<void>()
    // Stdin EOF exits the wrapper, but its ready descendant ignores SIGTERM and retains both pipes.
    const transport = yield* McpStdio.make({
      server: "held-stdio",
      command: "node",
      args: [path.join(import.meta.dir, "fixture/held-stdio.cjs"), "mcp", pidFile],
      cwd: tmp.path,
      environment: {},
    })
    transport.onmessage = () => {
      Deferred.doneUnsafe(ready, Exit.void)
    }
    yield* Effect.promise(() => transport.start())
    yield* Deferred.await(ready).pipe(Effect.timeout("3 seconds"))
    const pid = Number(yield* Effect.promise(() => fs.readFile(pidFile, "utf8")))

    yield* Effect.promise(() => transport.close()).pipe(Effect.timeout("6 seconds"))

    // close() must kill the descendant, not merely observe the wrapper's bounded exitCode.
    const stopped = yield* Effect.try(() => process.kill(pid, 0)).pipe(
      Effect.exit,
      Effect.repeat({ while: Exit.isSuccess, schedule: Schedule.spaced("25 millis") }),
      Effect.timeout("2 seconds"),
    )
    expect(Exit.isFailure(stopped)).toBe(true)
  }),
  15_000,
)

test("closes a stdio process that finishes spawning after close", async () => {
  const spawning = Deferred.makeUnsafe<void>()
  const release = Deferred.makeUnsafe<void>()
  const exited = Deferred.makeUnsafe<ExitCode>()
  const signals: Array<string> = []
  const driver = Environment.makeMemoryDriver()
  const environment = Layer.succeed(
    Environment.Service,
    Environment.Service.of({
      files: Environment.makeFiles(driver),
      spawner: ChildProcessSpawner.make(() =>
        Effect.gen(function* () {
          yield* Deferred.succeed(spawning, undefined)
          yield* Deferred.await(release)
          return makeHandle({
            pid: ProcessId(1),
            exitCode: Deferred.await(exited),
            isRunning: Deferred.isDone(exited).pipe(Effect.map((done) => !done)),
            kill: (options) =>
              Effect.gen(function* () {
                signals.push(options?.killSignal ?? "SIGTERM")
                yield* Deferred.succeed(exited, ExitCode(143))
              }),
            stdin: Sink.drain,
            stdout: Stream.never,
            stderr: Stream.empty,
            all: Stream.never,
            getInputFd: () => Sink.drain,
            getOutputFd: () => Stream.empty,
            unref: Effect.succeed(Effect.void),
          })
        }),
      ),
    }),
  )

  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const transport = yield* McpStdio.make({
          server: "close-during-spawn",
          command: "unused",
          args: [],
          cwd: import.meta.dir,
          environment: {},
        })
        const start = transport.start()
        yield* Deferred.await(spawning)
        const close = transport.close()
        yield* Deferred.succeed(release, undefined)
        yield* Effect.promise(() => Promise.all([start, close]))
      }).pipe(Effect.provide(environment)),
    ),
  )

  expect(signals).toEqual(["SIGTERM"])
})

test("falls back to legacy when a stdio server ignores the version probe", async () => {
  const result = await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "legacy-silent",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-legacy-silent.ts")],
            protocol: "auto",
            // The probe waits out the startup timeout before falling back; Windows needs room to start Bun.
            timeout: new ConfigMCP.Timeout({ startup: 1_000 }),
          }),
          import.meta.dir,
        )
        return { modern: connection.modern, tools: yield* connection.tools() }
      }),
    ),
  )

  expect(result.modern).toBe(false)
  expect(result.tools.map((tool) => tool.name)).toEqual(["echo"])
})

test("applies the configured MCP catalog timeout", async () => {
  const result = Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "catalog-timeout",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-timeout.ts")],
            environment: { MCP_TIMEOUT_TARGET: "catalog" },
            timeout: new ConfigMCP.Timeout({ catalog: 10 }),
          }),
          import.meta.dir,
        )
        return yield* connection.tools()
      }),
    ),
  )

  await expect(result).rejects.toThrow("Request timed out")
})

test("applies the configured MCP execution timeout", async () => {
  const result = Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "execution-timeout",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-timeout.ts")],
            timeout: new ConfigMCP.Timeout({ execution: 10 }),
          }),
          import.meta.dir,
        )
        return yield* connection.callTool({ name: "slow" })
      }),
    ),
  )

  await expect(result).rejects.toThrow("Request timed out")
})

test("applies the configured MCP execution timeout to prompts", async () => {
  const result = Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "prompt-timeout",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-timeout.ts")],
            timeout: new ConfigMCP.Timeout({ execution: 10 }),
          }),
          import.meta.dir,
        )
        return yield* connection.prompt({ name: "slow" })
      }),
    ),
  )

  await expect(result).rejects.toThrow("Request timed out")
})

test("applies configured MCP timeouts to resource operations", async () => {
  const catalog = Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "resource-catalog-timeout",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-timeout.ts")],
            environment: { MCP_TIMEOUT_TARGET: "resource-catalog" },
            timeout: new ConfigMCP.Timeout({ catalog: 10 }),
          }),
          import.meta.dir,
        )
        return yield* connection.resources()
      }),
    ),
  )
  await expect(catalog).rejects.toThrow("Request timed out")

  const read = Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const connection = yield* connect(
          "resource-read-timeout",
          new ConfigMCP.Local({
            type: "local",
            command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-timeout.ts")],
            timeout: new ConfigMCP.Timeout({ execution: 10 }),
          }),
          import.meta.dir,
        )
        return yield* connection.readResource({ uri: "test://slow" })
      }),
    ),
  )
  await expect(read).rejects.toThrow("Request timed out")
})

for (const entry of [
  { name: "direct tools", query: "", codemode: false, expected: "" },
  { name: "explicit remote code mode", query: "?codemode=true", codemode: undefined, expected: "?codemode=true" },
  { name: "explicit remote opt-out", query: "?codemode=false", codemode: undefined, expected: "?codemode=false" },
  { name: "portal opt-out", query: "?codemode=off", codemode: undefined, expected: "?codemode=off" },
]) {
  testEffect(Layer.empty).live(`remote MCP code mode preference: ${entry.name}`, () =>
    Effect.gen(function* () {
      const server = yield* resourceServer()
      const config = new ConfigMCP.Remote({
        type: "remote",
        url: server.url + entry.query,
        oauth: false,
        codemode: entry.codemode,
      })
      const connection = yield* connect("resources", config, import.meta.dir)
      yield* connection.tools()
      expect(server.state.initializations).toBe(1)
      expect(server.state.toolLists).toBe(1)
      expect(server.state.urls.length).toBeGreaterThanOrEqual(3)
      expect(new Set(server.state.urls)).toEqual(new Set([server.url + entry.expected]))
      expect(config.url).toBe(server.url + entry.query)
    }),
  )
}

for (const entry of [
  { name: "401", status: 401, query: "", codemode: undefined, attempts: 1 },
  { name: "403", status: 403, query: "", codemode: undefined, attempts: 1 },
  { name: "501", status: 501, query: "", codemode: undefined, attempts: 1 },
  { name: "user codemode=true", status: 404, query: "?codemode=true", codemode: undefined, attempts: 1 },
  { name: "user codemode=false", status: 404, query: "?codemode=false", codemode: undefined, attempts: 1 },
  { name: "empty user codemode", status: 404, query: "?codemode=", codemode: undefined, attempts: 1 },
  { name: "direct tools", status: 404, query: "", codemode: false, attempts: 1 },
  { name: "400 with user codemode=true", status: 400, query: "?codemode=true", codemode: undefined, attempts: 1 },
  { name: "400 with user codemode=false", status: 400, query: "?codemode=false", codemode: undefined, attempts: 1 },
  { name: "400 with empty user codemode", status: 400, query: "?codemode=", codemode: undefined, attempts: 1 },
  { name: "400 with direct tools", status: 400, query: "", codemode: false, attempts: 1 },
]) {
  testEffect(Layer.empty).live(`does not retry MCP beyond the query fallback: ${entry.name}`, () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({
        respond: () => new Response(null, { status: entry.status }),
      })
      const config = new ConfigMCP.Remote({
        type: "remote",
        url: server.url + entry.query,
        codemode: entry.codemode,
        oauth: false,
      })
      const error = yield* connect("resources", config, import.meta.dir).pipe(Effect.flip)

      expect(error).toBeInstanceOf(McpClient.ConnectError)
      expect(server.state.initializations).toBe(entry.attempts)
      expect(server.state.urls).toHaveLength(entry.attempts)
      if (entry.query || entry.codemode === false) expect(server.state.urls).toEqual([config.url])
    }),
  )
}

test("reconnects and retries a tool call after the MCP session expires", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer()
        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          yield* service.callTool({ server: "resources", name: "echo", args: { n: 1 } })
          expect(server.state.toolCalls).toHaveLength(1)
          expect(server.state.initializations).toBe(1)

          // The restarted server does not know the client's session, so the next request 404s.
          yield* Effect.promise(server.restart)
          const result = yield* service.callTool({ server: "resources", name: "echo", args: { n: 2 } })

          expect(result.isError).toBe(false)
          expect(server.state.toolCalls.map((call) => call.arguments)).toEqual([{ n: 1 }, { n: 2 }])
          expect(server.state.initializations).toBe(2)
          expect(server.state.sessions).toHaveLength(2)
          expect((yield* service.servers()).find((entry) => entry.name === "resources")).toMatchObject({
            status: { status: "connected" },
            title: "MCP Resources",
          })
        }).pipe(Effect.provide(resourceMcpLayer(server.url)))
      }),
    ),
  )
})

describe.each([
  ["legacy", undefined],
  ["modern", "2026-07-28"],
] as const)("MCP connection over the %s protocol", (era, protocol) => {
  test("exposes every connection operation", async () => {
    await Effect.runPromise(
      Effect.scoped(
        Effect.gen(function* () {
          const server = yield* resourceServer({ modern: era === "modern", listChanged: true })
          server.state.resources = [{ name: "Readme", uri: "docs://readme" }]
          server.state.templates = [{ name: "File", uriTemplate: "docs://{path}" }]
          const connection = yield* connect(
            "resources",
            new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol }),
            import.meta.dir,
          )

          expect(connection.modern).toBe(era === "modern")
          expect(connection.title).toBe("MCP Resources")
          expect(connection.instructions).toBe("Use the resources tools.")
          expect((yield* connection.tools()).map((tool) => tool.name)).toEqual(["echo"])
          expect((yield* connection.prompts()).map((prompt) => prompt.name)).toEqual(["greet"])
          expect((yield* connection.resources()).map((resource) => resource.uri)).toEqual(["docs://readme"])
          expect((yield* connection.resourceTemplates()).map((template) => template.uriTemplate)).toEqual([
            "docs://{path}",
          ])
          expect((yield* connection.readResource({ uri: "docs://readme" }))?.contents).toHaveLength(2)
          expect((yield* connection.prompt({ name: "greet", args: { name: "bob" } })).messages[0]?.content).toEqual({
            type: "text",
            text: "hi bob",
          })
          const sessionID = Session.ID.make("ses_mcp_era")
          yield* connection.callTool({ name: "echo", args: { text: "hi" }, sessionID })
          expect(server.state.toolCalls.at(-1)).toMatchObject({ name: "echo", sessionID })

          const changed = yield* Deferred.make<void>()
          connection.onResourcesChanged(() => Deferred.doneUnsafe(changed, Exit.void))
          yield* Effect.promise(server.sendResourceListChanged)
          yield* Deferred.await(changed)
        }),
      ),
    )
  })

  test("skips unusable tools and keeps the rest of the server's tools", async () => {
    await Effect.runPromise(
      Effect.scoped(
        Effect.gen(function* () {
          const server = yield* resourceServer({ modern: era === "modern", resources: false, malformedTool: true })
          const connection = yield* connect(
            "malformed",
            new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol }),
            import.meta.dir,
          )

          const tools = yield* connection.tools()
          expect(tools.map((tool) => tool.name)).toEqual(["bad-required", "bad-fields", "full", "echo"])
          // A malformed optional field or top-level `required` is dropped, not the tool. Valid fields are
          // kept except icons; a modern server's SDK strips `execution` before sending.
          expect(tools.slice(0, 3)).toEqual([
            { name: "bad-required", inputSchema: { type: "object" } },
            { name: "bad-fields", inputSchema: { type: "object" } },
            {
              name: "full",
              title: "Full",
              description: "Every field",
              inputSchema: { type: "object" },
              annotations: { readOnlyHint: true },
              ...(era === "legacy" ? { execution: { taskSupport: "optional" } } : {}),
              _meta: { "example/key": 1 },
            },
          ])
          for (const name of ["bad-required", "bad-fields", "full", "echo"]) {
            yield* connection.callTool({ name, args: { text: "hi" } })
            expect(server.state.toolCalls.at(-1)).toMatchObject({ name, arguments: { text: "hi" } })
          }
        }),
      ),
    )
  })

  test("validates structured content against the called tool's output schema", async () => {
    await Effect.runPromise(
      Effect.scoped(
        Effect.gen(function* () {
          const server = yield* resourceServer({
            modern: era === "modern",
            resources: false,
            tools: [
              {
                name: "typed",
                inputSchema: { type: "object" },
                outputSchema: { type: "object", properties: { value: { type: "string" } }, required: ["value"] },
              },
            ],
          })
          const connection = yield* connect(
            "typed",
            new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol }),
            import.meta.dir,
          )
          yield* connection.tools()

          expect(yield* connection.callTool({ name: "typed", args: { structured: { value: "ok" } } })).toMatchObject({
            structured: { value: "ok" },
          })
          expect(
            (yield* connection.callTool({ name: "typed", args: { structured: { value: 42 } } }).pipe(Effect.flip))
              .message,
          ).toContain("Structured content does not match the tool's output schema")
          expect((yield* connection.callTool({ name: "typed" }).pipe(Effect.flip)).message).toContain(
            "has an output schema but did not return structured content",
          )
          expect(yield* connection.callTool({ name: "echo", args: { structured: { value: 42 } } })).toMatchObject({
            structured: { value: 42 },
          })
        }),
      ),
    )
  })
})

test("retries a modern MCP tool call whose listed header declaration went stale", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const region = { type: "string" as const, "x-mcp-header": "Region" }
        const rejected: Array<string | null> = []
        const server = yield* resourceServer({
          modern: true,
          resources: false,
          tools: [{ name: "regional", inputSchema: { type: "object", properties: { region } } }],
          // Stands in for the server-side check: the region must arrive in the currently declared header.
          respond: async (request) => {
            const body: unknown = request.method === "POST" ? await request.clone().json() : undefined
            if (!isJSONRPCRequest(body) || body.method !== "tools/call") return undefined
            if (request.headers.get(`mcp-param-${region["x-mcp-header"]}`) === "eu") return undefined
            rejected.push(request.headers.get("mcp-param-region"))
            return Response.json(
              { jsonrpc: "2.0", id: body.id, error: { code: -32020, message: "Header mismatch" } },
              { status: 400 },
            )
          },
        })
        const connection = yield* connect(
          "regional",
          new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" }),
          import.meta.dir,
        )
        yield* connection.tools()
        region["x-mcp-header"] = "Zone"

        expect((yield* connection.callTool({ name: "regional", args: { region: "eu" } })).isError).toBe(false)
        expect(server.state.toolCalls.map((call) => call.arguments)).toEqual([{ region: "eu" }])
        // The stale Region header is rejected once; the relisted definition then sends Zone.
        expect(rejected).toEqual(["eu"])
        expect(server.state.toolLists).toBe(2)
      }),
    ),
  )
})

test("lists, reads, and reports MCP resource changes", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ listChanged: true })
        server.state.resourcePages = {
          initial: {
            items: [{ name: "Readme", uri: "docs://readme", description: "Project docs" }],
            nextCursor: "resources-2",
          },
          "resources-2": { items: [{ name: "Logo", uri: "docs://logo", mimeType: "image/png" }] },
        }
        server.state.templatePages = {
          initial: {
            items: [{ name: "File", uriTemplate: "docs://{path}" }],
            nextCursor: "templates-2",
          },
          "templates-2": { items: [{ name: "Issue", uriTemplate: "issue://{id}", description: "Issue" }] },
        }
        const connection = yield* connect(
          "resources",
          new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }),
          import.meta.dir,
        )

        expect(yield* connection.resources()).toEqual([
          { name: "Readme", uri: "docs://readme", description: "Project docs", mimeType: undefined },
          { name: "Logo", uri: "docs://logo", description: undefined, mimeType: "image/png" },
        ])
        expect(yield* connection.resourceTemplates()).toEqual([
          { name: "File", uriTemplate: "docs://{path}", description: undefined, mimeType: undefined },
          { name: "Issue", uriTemplate: "issue://{id}", description: "Issue", mimeType: undefined },
        ])
        expect(yield* connection.readResource({ uri: "docs://readme" })).toEqual({
          contents: [
            { uri: "docs://readme", text: "hello", mimeType: "text/plain" },
            { uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" },
          ],
        })

        const changed = yield* Deferred.make<void>()
        connection.onResourcesChanged(() => Deferred.doneUnsafe(changed, Exit.void))
        yield* Effect.promise(server.sendResourceListChanged)
        yield* Deferred.await(changed)
      }),
    ),
  )
})

test("does not reconnect an SSE stream after a JSON-RPC error response", async () => {
  let requests = 0
  const transport = new StreamableHTTPClientTransport(new URL("http://mcp.invalid"), {
    fetch: async () => {
      requests += 1
      return new Response(
        new ReadableStream({
          start(controller) {
            controller.enqueue(new TextEncoder().encode("id: prime\nretry: 1\ndata:\n\n"))
            controller.enqueue(
              new TextEncoder().encode(
                'id: error\ndata: {"jsonrpc":"2.0","error":{"code":-32601,"message":"Method not found"},"id":1}\n\n',
              ),
            )
            controller.close()
          },
        }),
        { status: 200, headers: { "content-type": "text/event-stream" } },
      )
    },
    reconnectionOptions: {
      initialReconnectionDelay: 1,
      maxReconnectionDelay: 1,
      reconnectionDelayGrowFactor: 1,
      maxRetries: 2,
    },
  })

  await transport.start()
  await transport.send({ jsonrpc: "2.0", method: "resources/list", id: 1 })
  await Bun.sleep(25)
  await transport.close()

  expect(requests).toBe(1)
})

test("skips MCP resource requests when the capability is absent", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ resources: false })
        const connection = yield* connect(
          "resources",
          new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }),
          import.meta.dir,
        )
        expect(yield* connection.resources()).toEqual([])
        expect(yield* connection.resourceTemplates()).toEqual([])
        expect(yield* connection.readResource({ uri: "docs://readme" })).toBeUndefined()
        expect({ resources: server.state.resourceLists, templates: server.state.templateLists }).toEqual({
          resources: 0,
          templates: 0,
        })
      }),
    ),
  )
})

type CreatedForm = { readonly form: Form.Info; readonly directory: string | undefined }

// Clients drop `form.created` events without a Location, so the form must be published inside the server's Location.
const recordForm = (created: Deferred.Deferred<CreatedForm>, form: Form.Info) =>
  Effect.serviceOption(Location.Service).pipe(
    Effect.flatMap((location) =>
      Deferred.succeed(created, { form, directory: Option.getOrUndefined(location)?.directory }),
    ),
    Effect.asVoid,
  )

test("accepts empty MCP elicitations without creating forms", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ resources: false, emptyElicitation: true })
        const result = yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          const forms = yield* Form.Service
          const result = yield* service.callTool({ server: "resources", name: "empty-elicitation" })
          expect(yield* forms.list()).toEqual([])
          return result
        }).pipe(Effect.provide(resourceMcpLayer(server.url)))

        expect(result.structured).toEqual({ action: "accept", content: {} })
      }),
    ),
  )
})

test("acknowledges completed MCP URL elicitations without returning internal content", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ resources: false, urlElicitation: true })
        const created = yield* Deferred.make<CreatedForm>()
        const result = yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          const forms = yield* Form.Service
          const call = yield* service.callTool({ server: "resources", name: "url-elicitation" }).pipe(Effect.forkScoped)

          const { form, directory } = yield* Deferred.await(created)
          expect(directory).toBe(import.meta.dir)
          expect(form.fields).toEqual([{ key: "elicitation", type: "external", url: "https://example.com/authorize" }])

          yield* Effect.promise(server.completeElicitation)
          const result = yield* Fiber.join(call)
          expect(yield* forms.state(form.id)).toEqual({ status: "answered", answer: { elicitation: true } })
          return result
        }).pipe(
          Effect.provide(resourceMcpLayer(server.url, (form) => recordForm(created, form))),
        )

        expect(result.structured).toEqual({ action: "accept" })
      }),
    ),
  )
})

test("cancels MCP elicitation forms when the server gives up on them", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ resources: false, urlElicitation: true, elicitationTimeout: 200 })
        const created = yield* Deferred.make<CreatedForm>()
        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          const forms = yield* Form.Service
          const call = yield* service.callTool({ server: "resources", name: "url-elicitation" }).pipe(Effect.forkScoped)
          const { form } = yield* Deferred.await(created)
          yield* Fiber.await(call)
          // The server's cancellation and its tool error race; the form settles shortly after either.
          const state = yield* forms.state(form.id).pipe(
            Effect.repeat({ until: (state) => state.status !== "pending", schedule: Schedule.spaced("10 millis") }),
            Effect.timeout("2 seconds"),
          )
          expect(state.status).toBe("cancelled")
        }).pipe(Effect.provide(resourceMcpLayer(server.url, (form) => recordForm(created, form))))
      }),
    ),
  )
})

test("settles modern MCP URL elicitations when the user confirms", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ modern: true, resources: false, urlElicitation: true })
        const created = yield* Deferred.make<CreatedForm>()
        const result = yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          const forms = yield* Form.Service
          const call = yield* service.callTool({ server: "resources", name: "url-elicitation" }).pipe(Effect.forkScoped)

          const { form, directory } = yield* Deferred.await(created)
          expect(directory).toBe(import.meta.dir)
          expect(form.metadata).not.toHaveProperty("elicitationID")
          yield* forms.reply({ id: form.id, answer: { elicitation: true } })
          return yield* Fiber.join(call)
        }).pipe(
          Effect.provide(
            resourceMcpLayer(
              new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" }),
              (form) => recordForm(created, form),
            ),
          ),
        )

        expect(result.structured).toEqual({ action: "accept" })
      }),
    ),
  )
})

test("loads and reads MCP resources", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer()
        server.state.resources = [{ name: "Readme", uri: "docs://readme" }]
        server.state.templates = [{ name: "File", uriTemplate: "docs://{path}" }]

        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          yield* settled(service)
          expect(yield* service.resourceCatalog()).toEqual({
            resources: [
              {
                server: "resources",
                name: "Readme",
                uri: "docs://readme",
                description: undefined,
                mimeType: undefined,
              },
            ],
            templates: [
              {
                server: "resources",
                name: "File",
                uriTemplate: "docs://{path}",
                description: undefined,
                mimeType: undefined,
              },
            ],
          })

          server.state.resources = [{ name: "Guide", uri: "docs://guide" }]
          expect((yield* service.resourceCatalog()).resources.map((resource) => resource.uri)).toEqual(["docs://guide"])
          expect(yield* service.readResource({ server: "resources", uri: "docs://readme" })).toEqual({
            server: "resources",
            uri: "docs://readme",
            contents: [
              { type: "text", uri: "docs://readme", text: "hello", mimeType: "text/plain" },
              { type: "blob", uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" },
            ],
          })
          expect(server.clientVersion()).toMatchObject({ name: "sdk", version: "1.2.3" })
        }).pipe(
          Effect.provide(resourceMcpLayer(server.url, undefined, { clientInfo: { name: "sdk", version: "1.2.3" } })),
        )
      }),
    ),
  )
})

it.live("discovers and reads MCP resources through Code Mode", () =>
  Effect.gen(function* () {
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.void
    const server = yield* resourceServer()
    server.state.resourcePages = {
      initial: { items: [{ name: "Readme", uri: "docs://readme" }], nextCursor: "resources-2" },
      "resources-2": { items: [{ name: "Guide", uri: "docs://guide" }] },
    }
    server.state.templatePages = {
      initial: { items: [{ name: "File", uriTemplate: "docs://{path}" }], nextCursor: "templates-2" },
      "templates-2": { items: [{ name: "Issue", uriTemplate: "issue://{id}" }] },
    }

    yield* Effect.gen(function* () {
      const mcp = yield* Mcp.Service
      yield* settled(mcp)
      yield* registerToolPlugin(McpResourceTools.Plugin).pipe(Effect.provide(permissions))
      const tools = yield* Tool.Service
      const snapshot = yield* tools.snapshot()
      const sessionID = Session.ID.make("ses_resource_tools")
      const run = (code: string) =>
        snapshot.execute({
          sessionID,
          ...toolIdentity,
          call: { type: "tool-call", id: "call_resource", name: "execute", input: { code } },
        })

      // The SDK walks every page, so one call returns the full catalog.
      const listed = yield* run('return await tools.opencode.list_mcp_resources({ server: "resources" })')
      expect(JSON.parse(listed.output.output)).toEqual({
        resources: [
          { server: "resources", name: "Guide", uri: "docs://guide" },
          { server: "resources", name: "Readme", uri: "docs://readme" },
        ],
        templates: [
          { server: "resources", name: "File", uriTemplate: "docs://{path}" },
          { server: "resources", name: "Issue", uriTemplate: "issue://{id}" },
        ],
      })
      expect(server.state.resourceReads).toEqual([])

      // Omitting the server lists every server, so the model can find which one owns a URI.
      assertion = yield* Deferred.make<Permission.AssertInput>()
      const everywhere = yield* run("return await tools.opencode.list_mcp_resources({})")
      expect(yield* Deferred.await(assertion)).toMatchObject({
        action: "opencode_list_mcp_resources",
        resources: ["resources"],
        save: ["resources"],
      })
      expect(JSON.parse(everywhere.output.output)).toEqual({
        resources: [
          { server: "resources", name: "Guide", uri: "docs://guide" },
          { server: "resources", name: "Readme", uri: "docs://readme" },
        ],
        templates: [
          { server: "resources", name: "File", uriTemplate: "docs://{path}" },
          { server: "resources", name: "Issue", uriTemplate: "issue://{id}" },
        ],
      })

      // A server may declare resources without implementing template listing.
      server.state.templatesUnsupported = true
      const untemplated = yield* run('return await tools.opencode.list_mcp_resources({ server: "resources" })')
      expect(JSON.parse(untemplated.output.output)).toEqual({
        resources: [
          { server: "resources", name: "Guide", uri: "docs://guide" },
          { server: "resources", name: "Readme", uri: "docs://readme" },
        ],
        templates: [],
      })
      server.state.templatesUnsupported = false

      assertion = yield* Deferred.make<Permission.AssertInput>()
      const read = yield* run(
        'const resource = await tools.opencode.read_mcp_resource({ server: "resources", uri: "docs://readme" }); return resource.contents.filter(part => part.type === "text").map(part => part.text).join("\\n")',
      )
      expect(read.content).toEqual([
        { type: "text", text: "hello" },
        { type: "file", uri: "data:image/png;base64,aGVsbG8=", mime: "image/png" },
      ])
      expect(read.metadata?.toolCalls).toMatchObject([
        {
          tool: "opencode.read_mcp_resource",
          status: "completed",
          input: { server: "resources", uri: "docs://readme" },
        },
      ])
      expect(server.state.resourceReads).toEqual(["docs://readme"])
      expect(yield* Deferred.await(assertion)).toEqual({
        action: "opencode_read_mcp_resource",
        resources: ["resources:docs://readme"],
        save: ["resources:*"],
        metadata: { server: "resources", uri: "docs://readme" },
        sessionID,
        agent: toolIdentity.agent,
        source: { type: "tool", messageID: toolIdentity.messageID, id: "call_resource" },
      })

      server.state.contents = [{ uri: "docs://readme", text: "line\n".repeat(20_000), mimeType: "text/plain" }]
      const large = yield* run(
        'const resource = await tools.opencode.read_mcp_resource({ server: "resources", uri: "docs://readme" }); return resource.contents[0].text',
      )
      const bounded = yield* ToolOutput.Service.use((output) => output.truncate(large)).pipe(
        Effect.provide(AppNodeBuilder.build(ToolOutput.node)),
      )
      expect(bounded.metadata?.truncated).toBe(true)
      expect(bounded.content[0]).toMatchObject({ type: "text" })
      const outputPath = bounded.metadata?.outputPath
      expect(typeof outputPath).toBe("string")
      if (typeof outputPath !== "string") throw new Error("Missing full resource output")
      expect(yield* Effect.promise(() => Bun.file(outputPath).text())).toBe("line\n".repeat(20_000))

      // An empty contents array means the resource exists without content, not that it is missing.
      server.state.contents = []
      const empty = yield* run(
        'return await tools.opencode.read_mcp_resource({ server: "resources", uri: "docs://empty" })',
      )
      expect(empty.metadata?.error).toBeUndefined()
      expect(JSON.parse(empty.output.output)).toEqual({ server: "resources", uri: "docs://empty", contents: [] })
      server.state.missing = ["docs://gone"]
      const gone = yield* run(
        'return await tools.opencode.read_mcp_resource({ server: "resources", uri: "docs://gone" })',
      )
      expect(gone.metadata?.error).toBe(true)
      expect(gone.output.output).toContain("Resource not found")
      const missing = yield* run(
        'return await tools.opencode.read_mcp_resource({ server: "missing", uri: "docs://readme" })',
      )
      expect(missing.metadata?.error).toBe(true)
      expect(missing.output.output).toContain("MCP server not found: missing")

      const reads = server.state.resourceReads.length
      decision = Effect.fail(
        new Permission.BlockedError({ rules: [], permission: "opencode_read_mcp_resource", resources: ["*"] }),
      )
      const denied = yield* run(
        'return await tools.opencode.read_mcp_resource({ server: "resources", uri: "docs://denied" })',
      )
      expect(denied.metadata?.error).toBe(true)
      expect(server.state.resourceReads).toHaveLength(reads)
    }).pipe(Effect.provide(resourceMcpLayer(server.url)))
  }),
)

test("adds, disconnects, and reconnects MCP servers at runtime", async () => {
  const published: string[] = []
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ modern: true, resources: false })
        const config = { type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" } as const
        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service

          expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" })
          expect(published).toContain(McpEvent.StatusChanged.type)
          expect(yield* service.connect("missing").pipe(Effect.flip)).toBeInstanceOf(Mcp.NotFoundError)
          expect(yield* service.disconnect("missing").pipe(Effect.flip)).toBeInstanceOf(Mcp.NotFoundError)
          yield* service.add("dynamic", new ConfigMCP.Remote(config))
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({
            status: "connected",
          })

          yield* service.add("dynamic", new ConfigMCP.Remote({ ...config, disabled: true }))
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({
            status: "disabled",
          })
          expect(yield* service.tools()).toEqual([])

          yield* service.connect("dynamic")
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({
            status: "connected",
          })
          yield* service.disconnect("dynamic")
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({
            status: "disabled",
          })
          expect(yield* service.tools()).toEqual([])

          yield* service.connect("dynamic")
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({
            status: "connected",
          })

          yield* service.remove("dynamic")
          expect((yield* service.servers()).some((server) => server.name === "dynamic")).toBe(false)
          expect(yield* service.tools()).toEqual([])
          expect(yield* service.remove("dynamic").pipe(Effect.flip)).toBeInstanceOf(Mcp.NotFoundError)
        }).pipe(
          Effect.provide(
            resourceMcpLayer(
              new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }),
              undefined,
              undefined,
              { published },
            ),
          ),
        )
      }),
    ),
  )
})

test("marks a remote MCP server with an invalid URL as failed", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ modern: true, resources: false })
        const config = { type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" } as const
        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          const status = (name: string) =>
            service.servers().pipe(Effect.map((servers) => servers.find((server) => server.name === name)?.status))
          expect(yield* settled(service)).toEqual({ status: "connected" })

          yield* service.add("broken", new ConfigMCP.Remote({ type: "remote", url: "not a url" }))
          expect(yield* status("broken")).toEqual({ status: "failed", error: "Invalid MCP server URL: not a url" })
          expect(yield* status("resources")).toEqual({ status: "connected" })

          yield* service.add("broken", new ConfigMCP.Remote(config))
          expect(yield* status("broken")).toEqual({ status: "connected" })
        }).pipe(Effect.provide(resourceMcpLayer(new ConfigMCP.Remote(config))))
      }),
    ),
  )
})

testEffect(Layer.empty).live(
  "merges MCP defaults into the winning configured server without changing runtime overrides",
  () =>
    Effect.gen(function* () {
      const entries = [
        new Document({
          type: "document",
          info: new Info({
            mcp: new ConfigMCP.Info({
              timeout: { startup: 10, catalog: 20, execution: 30 },
              servers: {
                resources: { type: "local", command: ["earlier"], disabled: true, timeout: { execution: 90 } },
                pinned: { type: "local", command: ["pinned"], disabled: true, protocol: "2026-07-28" },
              },
            }),
          }),
        }),
        new Document({
          type: "document",
          info: new Info({
            mcp: new ConfigMCP.Info({
              timeout: { catalog: 40 },
              servers: {
                resources: { type: "local", command: ["later"], disabled: true, timeout: { startup: 50 } },
              },
            }),
          }),
        }),
      ]
      const original = JSON.stringify(entries)
      yield* Effect.gen(function* () {
        const service = yield* Mcp.Service
        const check = yield* service.transform((editor) => {
          expect(editor.get("resources")).toEqual({
            type: "local",
            command: ["later"],
            disabled: true,
            timeout: { startup: 50, catalog: 40, execution: 30 },
          })
          expect(editor.get("pinned")).toEqual({
            type: "local",
            command: ["pinned"],
            disabled: true,
            timeout: { startup: 10, catalog: 40, execution: 30 },
            protocol: "2026-07-28",
          })
        })
        yield* check.dispose
        const runtime = {
          type: "local",
          command: ["runtime"],
          disabled: true,
          timeout: { catalog: 60 },
        } satisfies ConfigMCP.Local
        yield* service.add("resources", runtime)
        yield* service.reload()
        yield* service.transform((editor) => {
          expect(editor.get("resources")).toEqual(runtime)
        })
      }).pipe(
        Effect.provide(
          resourceMcpLayer("https://unused.example", undefined, undefined, {
            entries: () => Effect.succeed(entries),
          }),
        ),
      )
      expect(JSON.stringify(entries)).toBe(original)
    }),
)

testEffect(resourceMcpLayer(new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }))).live(
  "manages live MCP servers entirely through scoped transforms",
  () =>
    Effect.gen(function* () {
      const service = yield* Mcp.Service
      const server = yield* resourceServer({ modern: true, resources: false })

      yield* Effect.scoped(
        Effect.gen(function* () {
          yield* service.transform((editor) => {
            editor.set("dynamic", { type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" })
          })
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status.status).toBe(
            "connected",
          )
          expect(yield* service.tools()).toHaveLength(1)

          yield* service.transform((editor) => editor.update("dynamic", (server) => (server.codemode = false)))
          expect((yield* service.tools()).map((tool) => tool.codemode)).toEqual([false])

          const settings = { disabled: true }
          yield* service.transform((editor) => {
            editor.update("dynamic", (server) => {
              server.disabled = settings.disabled
            })
          })
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status.status).toBe("disabled")
          expect(yield* service.tools()).toEqual([])

          settings.disabled = false
          yield* service.reload()
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status.status).toBe(
            "connected",
          )
          expect(yield* service.tools()).toHaveLength(1)

          const removed = yield* service.transform((editor) => editor.remove("dynamic"))
          expect((yield* service.servers()).some((server) => server.name === "dynamic")).toBe(false)
          expect(yield* service.tools()).toEqual([])

          yield* removed.dispose
          expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status.status).toBe(
            "connected",
          )
          expect(yield* service.tools()).toHaveLength(1)
        }),
      )

      expect((yield* service.servers()).map((server) => server.name)).toEqual([Mcp.ServerName.make("resources")])
      expect(yield* service.tools()).toEqual([])
    }),
)

test("restores runtime MCP config when a transform is disposed", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const service = yield* Mcp.Service
        const config = new ConfigMCP.Remote({
          type: "remote",
          url: "https://example.com/mcp",
          headers: { Authorization: "original" },
          oauth: false,
          disabled: true,
        })
        yield* service.add("dynamic", config)
        const transformed = yield* service.transform((editor) =>
          editor.update("dynamic", (server) => {
            if (server.type === "remote") server.headers = { Authorization: "transformed" }
          }),
        )
        let observed: string | undefined
        yield* service.transform((editor) => {
          const server = editor.get("dynamic")
          observed = server?.type === "remote" ? server.headers?.Authorization : undefined
        })

        expect(observed).toBe("transformed")
        expect(config.headers?.Authorization).toBe("original")
        yield* transformed.dispose
        expect(observed).toBe("original")
      }).pipe(
        Effect.provide(resourceMcpLayer(new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }))),
      ),
    ),
  )
})

test("isolates nested configured MCP mutations and reconciles them", async () => {
  const published: string[] = []
  const config = new ConfigMCP.Remote({
    type: "remote",
    url: "https://example.com/mcp",
    headers: { Authorization: "original" },
    oauth: false,
    disabled: true,
  })
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const service = yield* Mcp.Service
        expect(published.filter((type) => type === McpEvent.StatusChanged.type)).toHaveLength(1)
        yield* service.transform((editor) =>
          editor.update("resources", (server) => {
            if (server.type === "remote" && server.headers) server.headers.Authorization = "transformed"
          }),
        )

        expect(config.headers?.Authorization).toBe("original")
        expect(published.filter((type) => type === McpEvent.StatusChanged.type)).toHaveLength(2)
      }).pipe(Effect.provide(resourceMcpLayer(config, undefined, undefined, { published }))),
    ),
  )
})

test("publishes MCP changes while live but not while the location shuts down", async () => {
  const published: string[] = []
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ listChanged: true })
        const scope = yield* Scope.make()
        const context = yield* Layer.buildWithScope(
          resourceMcpLayer(server.url, undefined, undefined, { published }),
          scope,
        )
        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          expect(yield* settled(service)).toEqual({ status: "connected" })
          published.length = 0
          yield* Effect.promise(server.sendResourceListChanged)
          yield* Effect.sync(() => published).pipe(
            Effect.filterOrFail(
              (types) => types.includes(McpEvent.ResourcesChanged.type),
              () => new Error("resource change was not published"),
            ),
            Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }),
          )
        }).pipe(Effect.provide(context))
        published.length = 0
        // Location shutdown closes plugin registrations under State.shutdown, so config does not reconcile.
        yield* State.shutdown(Scope.close(scope, Exit.void))
        expect(published.filter((type) => type.startsWith("mcp."))).toEqual([])
      }),
    ),
  )
})

testEffect(Layer.empty).live("batches MCP transforms without connecting intermediate configurations", () =>
  Effect.gen(function* () {
    const server = yield* resourceServer()
    yield* Effect.gen(function* () {
      const service = yield* Mcp.Service
      const registrations = yield* State.batch(
        Effect.gen(function* () {
          const added = yield* service.transform((editor) =>
            editor.set("dynamic", {
              type: "remote",
              url: server.url,
              oauth: false,
            }),
          )
          expect((yield* service.servers()).some((server) => server.name === "dynamic")).toBe(false)
          const disabled = yield* service.transform((editor) =>
            editor.update("dynamic", (config) => (config.disabled = true)),
          )
          return [added, disabled]
        }),
      )

      expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({
        status: "disabled",
      })
      expect(yield* service.tools()).toEqual([])
      yield* State.batch(Effect.forEach(registrations, (registration) => registration.dispose))
      expect((yield* service.servers()).some((server) => server.name === "dynamic")).toBe(false)
      expect(server.state.initializations).toBe(0)
    }).pipe(
      Effect.provide(resourceMcpLayer(new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }))),
    )
  }),
)

test("reconciles only changed MCP server config", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer()
        const updates = yield* PubSub.unbounded<Payload>()
        const resources = (codemode?: boolean) =>
          new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, codemode })
        const added = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true })
        const dynamic = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true })
        const document = (servers: Record<string, typeof ConfigMCP.Server.Type>, username?: string) =>
          new Document({
            type: "document",
            info: new Info({
              username,
              mcp: new ConfigMCP.Info({ servers }),
            }),
          })
        let entries = [document({ resources: resources() })]
        const publishUpdate = () =>
          PubSub.publish(updates, {
            id: ID.create(),
            created: 0,
            type: Event.Updated.type,
            data: {},
          } satisfies Payload<typeof Event.Updated>)

        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          yield* settled(service)
          expect(server.state.toolLists).toBe(1)
          expect(server.state.initializations).toBe(1)

          yield* service.add("dynamic", dynamic)
          entries = [document({ resources: resources() }, "unrelated")]
          yield* publishUpdate()
          entries = [document({ resources: resources(), added }, "unrelated")]
          yield* publishUpdate()
          const appended = yield* service.servers().pipe(
            Effect.filterOrFail(
              (items) => items.some((item) => item.name === "added"),
              () => new Error("MCP config addition was not applied"),
            ),
            Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }),
          )
          expect(appended.map((item) => String(item.name)).toSorted()).toEqual(["added", "dynamic", "resources"])
          expect(server.state.toolLists).toBe(1)
          expect(server.state.initializations).toBe(1)

          entries = [
            document(
              {
                resources: resources(false),
                added,
              },
              "unrelated",
            ),
          ]
          yield* publishUpdate()
          yield* Effect.sync(() => server.state.initializations).pipe(
            Effect.filterOrFail(
              (count) => count === 2,
              () => new Error("MCP config change did not reconnect the server"),
            ),
            Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }),
          )

          entries = [document({ added }, "unrelated")]
          yield* publishUpdate()
          const removed = yield* service.servers().pipe(
            Effect.filterOrFail(
              (items) => !items.some((item) => item.name === "resources"),
              () => new Error("MCP config removal was not applied"),
            ),
            Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }),
          )
          expect(removed.map((item) => String(item.name)).toSorted()).toEqual(["added", "dynamic"])
        }).pipe(
          Effect.provide(
            resourceMcpLayer(resources(), undefined, undefined, {
              entries: () => Effect.sync(() => entries),
              subscribe: ((definition?: { readonly type: string }) =>
                Stream.fromPubSub(updates).pipe(
                  Stream.filter((event) => !definition || event.type === definition.type),
                )) as Bus.Interface["subscribe"],
            }),
          ),
        )
      }),
    ),
  )
})

testEffect(Layer.empty).live("keeps MCP config snapshots stable during an in-flight replacement", () =>
  Effect.gen(function* () {
    const started = yield* Deferred.make<void>()
    const release = yield* Deferred.make<void>()
    const accepted = yield* Deferred.make<void>()
    const server = yield* resourceServer({
      respond: (request) =>
        request.method !== "POST"
          ? undefined
          : Effect.runPromise(
              Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release)), Effect.as(undefined)),
            ),
    })

    yield* Effect.gen(function* () {
      const service = yield* Mcp.Service
      const replacing = yield* service
        .transform((editor) => editor.update("resources", (config) => (config.disabled = false)))
        .pipe(Effect.forkScoped({ startImmediately: true }))
      yield* Deferred.await(started)

      const restoring = yield* State.batch(
        Effect.gen(function* () {
          yield* service.transform((editor) => editor.update("resources", (config) => (config.disabled = true)))
          yield* Deferred.succeed(accepted, undefined)
        }),
      ).pipe(Effect.forkScoped({ startImmediately: true }))
      yield* Deferred.await(accepted)
      expect((yield* service.servers())[0]?.status).toEqual({ status: "pending" })

      yield* Deferred.succeed(release, undefined)
      yield* Fiber.join(replacing)
      yield* Fiber.join(restoring)
      expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" })
      expect(yield* service.tools()).toEqual([])
      expect(server.state.initializations).toBe(1)
    }).pipe(
      Effect.ensuring(Deferred.succeed(release, undefined)),
      Effect.provide(
        resourceMcpLayer(new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, disabled: true })),
      ),
    )
  }),
)

const shutdownIt = testEffect(
  AppNodeBuilder.build(
    LayerNode.group([
      Config.node,
      Bus.node,
      Integration.node,
      Credential.node,
      Form.node,
      Environment.node,
      Location.node,
      McpPool.node,
    ]),
    [
      Location.node.replace(
        Layer.succeed(
          Location.Service,
          Location.Service.of(location({ directory: AbsolutePath.make(import.meta.dir) })),
        ),
      ),
      Environment.node.replace(hostEnvironmentLayer),
    ],
  ),
)
shutdownIt.effect("discards in-flight and queued MCP notifications after its layer closes", () =>
  Effect.gen(function* () {
    const bus = yield* Bus.Service
    const entered = yield* Deferred.make<void>()
    const release = yield* Deferred.make<void>()
    const root = yield* Scope.make()
    yield* Effect.addFinalizer(() =>
      Deferred.succeed(release, undefined).pipe(Effect.andThen(State.shutdown(Scope.close(root, Exit.void)))),
    )
    const context = yield* Layer.buildWithScope(Mcp.layer(), root)
    const service = Context.get(context, Mcp.Service)
    const observed: string[] = []
    let block = false
    yield* Effect.acquireRelease(
      bus.listen((event) =>
        Effect.gen(function* () {
          if (event.type !== McpEvent.StatusChanged.type) return
          observed.push(Schema.decodeUnknownSync(McpEvent.StatusChanged.data)(event.data).server)
          if (!block) return
          block = false
          yield* Deferred.succeed(entered, undefined)
          yield* Deferred.await(release)
        }),
      ),
      (unsubscribe) => unsubscribe,
    )
    const source = { url: "https://example.com/initial", added: false }
    yield* service
      .transform((editor) => {
        editor.set("fixture", { type: "remote", url: source.url, oauth: false, disabled: true })
        if (source.added) editor.set("queued", { type: "local", command: ["unused"], disabled: true })
      })
      .pipe(Scope.provide(root))

    block = true
    source.url = "https://example.com/first"
    source.added = true
    const first = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))
    yield* Deferred.await(entered)
    source.url = "https://example.com/second"
    const second = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))

    const shutdown = yield* State.shutdown(Scope.close(root, Exit.void)).pipe(
      Effect.forkChild({ startImmediately: true }),
    )
    yield* TestClock.adjust("1 millis")
    expect(shutdown.pollUnsafe()).toBeDefined()
    expect(first.pollUnsafe()).toBeDefined()
    expect(second.pollUnsafe()).toBeDefined()
    expect(yield* Deferred.isDone(release)).toBe(false)
    yield* Fiber.join(shutdown)
    observed.length = 0
    yield* Deferred.succeed(release, undefined)
    yield* Fiber.join(first)
    yield* Fiber.join(second)
    expect(observed).toEqual([])
    expect((yield* service.servers()).map((server) => server.name)).toEqual([Mcp.ServerName.make("fixture")])
  }),
)

describe.each(["connect", "disconnect"] as const)("%s during initial MCP startup", (operation) => {
  shutdownIt.live("is not undone by the queued startup", () =>
    Effect.gen(function* () {
      const bus = yield* Bus.Service
      const server = yield* resourceServer()
      const entered = yield* Deferred.make<void>()
      const release = yield* Deferred.make<void>()
      const root = yield* Scope.make()
      yield* Effect.addFinalizer(() =>
        Deferred.succeed(release, undefined).pipe(Effect.andThen(State.shutdown(Scope.close(root, Exit.void)))),
      )
      const service = Context.get(yield* Layer.buildWithScope(Mcp.layer(), root), Mcp.Service)
      // Pausing the disabled server's status event holds initial reconciliation after every entry exists
      // but before the enabled server's startup is scheduled.
      yield* Effect.acquireRelease(
        bus.listen((event) =>
          Effect.gen(function* () {
            if (event.type !== McpEvent.StatusChanged.type) return
            if (Schema.decodeUnknownSync(McpEvent.StatusChanged.data)(event.data).server !== "gate") return
            yield* Deferred.succeed(entered, undefined)
            yield* Deferred.await(release)
          }),
        ),
        (unsubscribe) => unsubscribe,
      )
      const reconciling = yield* service
        .transform((editor) => {
          editor.set("gate", { type: "local", command: ["unused"], disabled: true })
          editor.set("resources", { type: "remote", url: server.url, oauth: false })
        })
        .pipe(Scope.provide(root), Effect.forkChild({ startImmediately: true }))
      yield* Deferred.await(entered)

      yield* service[operation]("resources")
      yield* Deferred.succeed(release, undefined)
      yield* Fiber.join(reconciling)
      // The queued startup already holds or awaits the server lock, so this waits for it to finish.
      yield* service.disconnect("resources")

      expect(server.state.initializations).toBe(operation === "connect" ? 1 : 0)
      expect((yield* service.servers()).find((item) => item.name === "resources")?.status).toEqual({
        status: "disabled",
      })
    }),
  )
})

test("serializes concurrent MCP lifecycle operations", async () => {
  await Effect.runPromise(
    Effect.scoped(
      Effect.gen(function* () {
        const server = yield* resourceServer({ modern: true, resources: false })
        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service

          // Whatever order the racing operations land in, the resulting state must be consistent.
          yield* Effect.all(
            [
              service.connect("resources"),
              service.connect("resources"),
              service.disconnect("resources"),
              service.connect("resources"),
            ],
            { concurrency: "unbounded", discard: true },
          )
          const status = (yield* service.servers()).find((server) => server.name === "resources")?.status
          const tools = yield* service.tools()
          expect(status?.status === "connected" || status?.status === "disabled").toBe(true)
          if (status?.status === "disabled") expect(tools).toEqual([])
          if (status?.status === "connected") expect(tools.length).toBeGreaterThan(0)

          yield* service.disconnect("resources")
          expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" })
          expect(yield* service.tools()).toEqual([])
          yield* service.connect("resources")
          expect((yield* service.servers())[0]?.status).toEqual({ status: "connected" })
          expect((yield* service.tools()).length).toBeGreaterThan(0)
        }).pipe(
          Effect.provide(
            resourceMcpLayer(
              new ConfigMCP.Remote({
                type: "remote",
                url: server.url,
                oauth: false,
                protocol: "2026-07-28",
                disabled: true,
              }),
            ),
          ),
        )
      }),
    ),
  )
})

const awaitStatus = (service: Mcp.Interface, status: Mcp.Status["status"]) =>
  service.servers().pipe(
    Effect.map((servers) => servers.find((server) => server.name === "resources")?.status),
    Effect.filterOrFail(
      (current) => current?.status === status,
      (current) => new Error(`MCP server status is ${JSON.stringify(current)}, expected ${status}`),
    ),
    Effect.retry({ schedule: Schedule.spaced("10 millis") }),
    Effect.timeout("5 seconds"),
  )

testEffect(Layer.empty).live("finishes an MCP connect whose caller is interrupted during the handshake", () =>
  Effect.gen(function* () {
    const started = yield* Deferred.make<void>()
    const release = yield* Deferred.make<void>()
    const server = yield* resourceServer({
      respond: async (request) => {
        if (request.method !== "POST") return undefined
        const body: unknown = await request.clone().json()
        if (typeof body !== "object" || body === null || !("method" in body) || body.method !== "initialize")
          return undefined
        await Effect.runPromise(Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release))))
        return undefined
      },
    })
    yield* Effect.gen(function* () {
      const service = yield* Mcp.Service
      const connecting = yield* service.connect("resources").pipe(Effect.forkChild({ startImmediately: true }))
      yield* Deferred.await(started)
      yield* Fiber.interrupt(connecting)
      yield* Deferred.succeed(release, undefined)

      yield* awaitStatus(service, "connected")
      expect((yield* service.tools()).length).toBeGreaterThan(0)
    }).pipe(
      Effect.ensuring(Deferred.succeed(release, undefined)),
      Effect.provide(
        resourceMcpLayer(new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, disabled: true })),
      ),
    )
  }),
)

testEffect(Layer.empty).live("finishes an MCP disconnect whose caller is interrupted while the session ends", () =>
  Effect.gen(function* () {
    const started = yield* Deferred.make<void>()
    const release = yield* Deferred.make<void>()
    const server = yield* resourceServer({
      respond: async (request) => {
        if (request.method !== "DELETE") return undefined
        await Effect.runPromise(Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release))))
        return undefined
      },
    })
    yield* Effect.gen(function* () {
      const service = yield* Mcp.Service
      expect(yield* settled(service)).toEqual({ status: "connected" })
      const disconnecting = yield* service.disconnect("resources").pipe(Effect.forkChild({ startImmediately: true }))
      yield* Deferred.await(started)
      yield* Fiber.interrupt(disconnecting)
      yield* Deferred.succeed(release, undefined)

      yield* awaitStatus(service, "disabled")
      expect(yield* service.tools()).toEqual([])
    }).pipe(Effect.ensuring(Deferred.succeed(release, undefined)), Effect.provide(resourceMcpLayer(server.url)))
  }),
)

testEffect(Layer.empty).live("preserves plugin transforms through MCP catalog updates", () =>
  Effect.gen(function* () {
    const tool = (server: string, name: string, description = name) =>
      ({
        server: Mcp.ServerName.make(server),
        name,
        description,
        codemode: false,
        inputSchema: { type: "object", properties: {} },
      }) satisfies Mcp.Tool
    const healthy = [tool("demo", "search"), tool("other", "lookup")]
    const catalog = yield* Ref.make(healthy)

    yield* Effect.gen(function* () {
      const registry = yield* Tool.Service
      const registration = yield* McpTool.Service
      const bus = yield* Bus.Service
      yield* registration.flush
      expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toEqual([
        "demo_search",
        "other_lookup",
        "execute",
      ])
      const override = yield* registry.transform((editor) => {
        editor.add({
          name: "search",
          options: { namespace: "demo", codemode: false },
          description: "Override search",
          input: Schema.Struct({}),
          output: Schema.String,
          execute: () => Effect.succeed({ output: "override" }),
        })
      })
      const mutation = yield* registry.transform((editor) => {
        editor.update("other_lookup", (tool) => {
          tool.description += " updated"
        })
        editor.remove("repaired_lookup")
      })

      yield* Ref.set(catalog, [...healthy, tool("demo", "added")])
      yield* bus.publish(McpEvent.ToolsChanged, { server: "demo" })
      yield* waitForTool(registry, "demo_added")
      expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toEqual([
        "demo_added",
        "demo_search",
        "other_lookup",
        "execute",
      ])
      expect((yield* toolDefinitions(registry)).find((tool) => tool.name === "demo_search")?.description).toBe(
        "Override search",
      )
      expect((yield* toolDefinitions(registry)).find((tool) => tool.name === "other_lookup")?.description).toBe(
        "lookup updated",
      )
      yield* Effect.forEach(["demo_search", "other_lookup"], (name) =>
        executeTool(registry, {
          sessionID: Session.ID.make("ses_mcp_invalid_catalog"),
          ...toolIdentity,
          call: { type: "tool-call", id: `call_${name}`, name, input: {} },
        }).pipe(
          Effect.tap((result) =>
            Effect.sync(() =>
              expect(result).toMatchObject({
                status: "completed",
                output: name === "demo_search" ? "override" : "healthy",
              }),
            ),
          ),
        ),
      )

      yield* Ref.set(catalog, [
        tool("demo", "status"),
        tool("other", "lookup"),
        tool("demo", "added"),
        tool("repaired", "lookup"),
      ])
      yield* bus.publish(McpEvent.ToolsChanged, { server: "demo" })
      yield* waitForTool(registry, "demo_status")
      expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toEqual([
        "demo_added",
        "demo_search",
        "demo_status",
        "other_lookup",
        "execute",
      ])
      expect((yield* toolDefinitions(registry)).find((tool) => tool.name === "demo_search")?.description).toBe(
        "Override search",
      )
      expect((yield* toolDefinitions(registry)).find((tool) => tool.name === "other_lookup")?.description).toBe(
        "lookup updated",
      )
      yield* mutation.dispose
      expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toContain("repaired_lookup")
      expect((yield* toolDefinitions(registry)).find((tool) => tool.name === "other_lookup")?.description).toBe(
        "lookup",
      )

      yield* Ref.set(catalog, [tool("demo", "search", "Latest search"), tool("demo", "refreshed")])
      yield* bus.publish(McpEvent.ToolsChanged, { server: "demo" })
      yield* waitForTool(registry, "demo_refreshed")
      expect((yield* toolDefinitions(registry)).find((tool) => tool.name === "demo_search")?.description).toBe(
        "Override search",
      )

      yield* override.dispose
      expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toEqual([
        "demo_refreshed",
        "demo_search",
        "execute",
      ])
      expect((yield* toolDefinitions(registry)).find((tool) => tool.name === "demo_search")?.description).toBe(
        "Latest search",
      )
      expect(
        yield* executeTool(registry, {
          sessionID: Session.ID.make("ses_mcp_invalid_catalog"),
          ...toolIdentity,
          call: { type: "tool-call", id: "call_restored_search", name: "demo_search", input: {} },
        }),
      ).toMatchObject({ status: "completed", output: "healthy" })
    }).pipe(
      Effect.provide(
        Layer.fresh(
          AppNodeBuilder.build(LayerNode.group([Tool.node, McpTool.node, Bus.node]), [
            Mcp.node.replace(
              Layer.mock(Mcp.Service, {
                awaitStartup: () => Effect.void,
                tools: () => Ref.get(catalog),
                callTool: (input) =>
                  Effect.succeed({
                    isError: false,
                    content: [{ type: "text", text: "healthy" }],
                  } satisfies Mcp.ToolResult),
              }),
            ),
            Permission.node.replace(Layer.mock(Permission.Service, { assert: () => Effect.void })),
            Image.node.replace(imagePassthrough),
          ]),
        ),
      ),
    )
  }),
)

testEffect(Layer.empty).effect("coalesces queued MCP tool notifications after initial registration", () => {
  const tool = (name: string) =>
    ({
      server: Mcp.ServerName.make("demo"),
      name,
      codemode: false,
      inputSchema: { type: "object", properties: {} },
    }) satisfies Mcp.Tool
  let reads = 0
  let catalog = [tool("read_1")]
  return Effect.gen(function* () {
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    const bus = yield* Bus.Service
    yield* registration.flush
    // Initial registration plus the flush's check that the registry is current.
    expect(reads).toBe(2)
    expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toEqual(["demo_read_1", "execute"])

    catalog = [tool("read_2")]
    yield* bus.publish(McpEvent.ToolsChanged, { server: "demo" })
    yield* advance(() => reads >= 3)
    expect(reads).toBe(3)
    catalog = [tool("read_3")]
    yield* Effect.forEach(Array.from({ length: 20 }), () => bus.publish(McpEvent.ToolsChanged, { server: "demo" }))
    yield* advance(() => reads >= 4)
    yield* drain
    expect(reads).toBe(4)
    expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toEqual(["demo_read_3", "execute"])
  }).pipe(
    Effect.provide(
      AppNodeBuilder.build(LayerNode.group([Tool.node, McpTool.node, Bus.node]), [
        Mcp.node.replace(
          Layer.mock(Mcp.Service, {
            awaitStartup: () => Effect.void,
            tools: () =>
              Effect.sync(() => {
                reads++
                return catalog
              }),
          }),
        ),
        Permission.node.replace(Layer.mock(Permission.Service, { assert: () => Effect.void })),
        Image.node.replace(imagePassthrough),
      ]),
    ),
  )
})

// Holds every MCP POST, including the handshake, until the test releases it.
const gatedServer = Effect.gen(function* () {
  const started = yield* Deferred.make<void>()
  const release = yield* Deferred.make<void>()
  const server = yield* resourceServer({
    respond: (request) =>
      request.method !== "POST"
        ? undefined
        : Effect.runPromise(
            Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release)), Effect.as(undefined)),
          ),
  })
  return { server, started, release }
})

// The real Mcp layer behind the tool registry; the bus never delivers, so only flush can register tools.
const startupRegistry = (url: string) =>
  AppNodeBuilder.build(LayerNode.group([Mcp.node, Tool.node, McpTool.node]), [
    Mcp.node.replace(resourceMcpLayer(new ConfigMCP.Remote({ type: "remote", url, oauth: false, codemode: false }))),
    Permission.node.replace(permissions),
    Bus.node.replace(events),
    Image.node.replace(imagePassthrough),
    Global.node.replace(Global.layerWith({ data: outputData })),
  ])

testEffect(Layer.empty).effect("flush waits for a server that connects within its startup window", () =>
  Effect.gen(function* () {
    const gated = yield* gatedServer
    yield* Effect.gen(function* () {
      const registry = yield* Tool.Service
      const registration = yield* McpTool.Service
      yield* Deferred.await(gated.started)
      const flush = yield* registration.flush.pipe(Effect.forkChild({ startImmediately: true }))
      yield* Effect.yieldNow
      expect(flush.pollUnsafe()).toBeUndefined()

      yield* Deferred.succeed(gated.release, undefined)
      yield* Fiber.join(flush)
      expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toContain("resources_echo")
    }).pipe(
      Effect.ensuring(Deferred.succeed(gated.release, undefined)),
      Effect.provide(startupRegistry(gated.server.url)),
    )
  }),
)

testEffect(Layer.empty).effect("flush stops waiting for a hung server when its startup window ends", () =>
  Effect.gen(function* () {
    const gated = yield* gatedServer
    yield* Effect.gen(function* () {
      const mcp = yield* Mcp.Service
      const registration = yield* McpTool.Service
      yield* Deferred.await(gated.started)
      const flush = yield* registration.flush.pipe(Effect.forkChild({ startImmediately: true }))
      yield* TestClock.adjust("1999 millis")
      expect(flush.pollUnsafe()).toBeUndefined()

      yield* TestClock.adjust("1 millis")
      yield* Fiber.join(flush)
      // Past its window, the still-pending server no longer holds up a model call.
      yield* registration.flush
      expect((yield* mcp.servers())[0]?.status).toEqual({ status: "pending" })
    }).pipe(
      // Shutting down mid-handshake would wait on the held request.
      Effect.ensuring(Deferred.succeed(gated.release, undefined)),
      Effect.provide(startupRegistry(gated.server.url)),
    )
  }),
)

it.effect("advertises MCP output schemas to Code Mode", () =>
  Effect.gen(function* () {
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const toolSet = yield* registry.snapshot()
    const execute = toolSet.definitions.find((tool) => tool.name === "execute")

    expect(toolSet.definitions.map((tool) => tool.name)).toEqual([
      "direct_attachments",
      "direct_fail",
      "direct_issues",
      "direct_lookup",
      "direct_media",
      "execute",
    ])
    expect(codeModeListings(toolSet.codeModeCatalog!).find((tool) => tool.path === "demo.search")?.line).toContain(
      "ok: boolean",
    )
    expect(execute?.description).not.toContain("tools.demo.search")
  }),
)

it.effect("forwards the invoking session through direct and Code Mode MCP tools", () =>
  Effect.gen(function* () {
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.void
    invocations = []
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const toolSet = yield* registry.snapshot()

    expect(toolSet.definitions.find((tool) => tool.name === "direct_lookup")?.inputSchema).not.toHaveProperty(
      "properties.sessionID",
    )
    expect(codeModeListings(toolSet.codeModeCatalog!).find((tool) => tool.path === "demo.search")?.line).not.toContain(
      "sessionID",
    )

    const directSessionID = Session.ID.make("ses_mcp_direct")
    yield* toolSet.execute({
      sessionID: directSessionID,
      ...toolIdentity,
      call: { type: "tool-call", id: "call_mcp_direct", name: "direct_lookup", input: {} },
    })
    expect(invocations[0]).toEqual({
      server: "direct",
      name: "lookup",
      args: {},
      sessionID: directSessionID,
    })

    const codeModeSessionID = Session.ID.make("ses_mcp_codemode")
    yield* toolSet.execute({
      sessionID: codeModeSessionID,
      ...toolIdentity,
      call: {
        type: "tool-call",
        id: "call_mcp_codemode",
        name: "execute",
        input: { code: "return await tools.demo.search({})" },
      },
    })
    expect(invocations[1]).toEqual({
      server: "demo",
      name: "search",
      args: {},
      sessionID: codeModeSessionID,
    })
  }),
)

it.effect("returns content-only MCP results through Code Mode", () =>
  Effect.gen(function* () {
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.void
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const toolSet = yield* registry.snapshot()

    expect(codeModeListings(toolSet.codeModeCatalog!).some((tool) => tool.path === "demo.status")).toBe(true)

    const execution = yield* toolSet.execute({
      sessionID: Session.ID.make("ses_mcp_content_only"),
      ...toolIdentity,
      call: {
        type: "tool-call",
        id: "call_mcp_content_only",
        name: "execute",
        input: { code: "return await tools.demo.status({})" },
      },
    })

    expect(execution).toMatchObject({
      output: { output: "hello", toolCalls: [{ tool: "demo.status", status: "completed" }] },
      content: [{ type: "text", text: "hello" }],
    })
  }),
)

it.effect("parses JSON text results from MCP tools without an output schema", () =>
  Effect.gen(function* () {
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.void
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const toolSet = yield* registry.snapshot()

    const run = (code: string) =>
      toolSet
        .execute({
          sessionID: Session.ID.make("ses_mcp_json_text"),
          ...toolIdentity,
          call: { type: "tool-call", id: `call_${code.length}`, name: "execute", input: { code } },
        })
        .pipe(Effect.map((execution) => execution.output.output))

    expect(yield* run("return (await tools.demo.issues({})).issues[0].id")).toBe("1")
    expect(yield* run("return typeof (await tools.demo.count({}))")).toBe("string")
    expect(yield* run("return typeof (await tools.demo.typed({}))")).toBe("string")

    // Outside Code Mode the content the model reads is the original text.
    expect(
      yield* toolSet.execute({
        sessionID: Session.ID.make("ses_mcp_json_text"),
        ...toolIdentity,
        call: { type: "tool-call", id: "call_direct_issues", name: "direct_issues", input: {} },
      }),
    ).toMatchObject({ output: { issues: [{ id: 1 }] }, content: [{ type: "text", text: '{"issues":[{"id":1}]}' }] })
  }),
)

it.effect("advertises MCP tools directly when Code Mode is disabled for the server", () =>
  Effect.gen(function* () {
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const definitions = yield* toolDefinitions(registry)
    const execute = definitions.find((tool) => tool.name === "execute")

    expect(definitions.some((tool) => tool.name === "direct_lookup")).toBe(true)
    expect(execute?.description).not.toContain("tools.direct.lookup")
  }),
)

// Baseline (PLAN.md step 1): MCP isError must become one failed tool call, not a
// success whose text happens to describe an error.
it.effect("fails the call when MCP reports isError", () =>
  Effect.gen(function* () {
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.void
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush

    const execution = yield* executeTool(registry, {
      sessionID: Session.ID.make("ses_mcp_is_error"),
      ...toolIdentity,
      call: { type: "tool-call", id: "call_mcp_is_error", name: "direct_fail", input: {} },
    })

    expect(execution).toMatchObject({ status: "error", error: { message: "search index unavailable" } })
  }),
)

// Baseline (PLAN.md step 1): mixed MCP text and media content must reach the model intact.
it.effect("preserves MCP text and media content for the model", () =>
  Effect.gen(function* () {
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.void
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush

    const execution = yield* executeTool(registry, {
      sessionID: Session.ID.make("ses_mcp_media"),
      ...toolIdentity,
      call: { type: "tool-call", id: "call_mcp_media", name: "direct_media", input: {} },
    })

    expect(execution.output).toBe("rendered chart")
    expect(execution.content).toMatchObject([
      { type: "text", text: "rendered chart" },
      { type: "file", mime: "image/png" },
    ])
  }),
)

it.effect("decodes text blobs and saves other MCP blobs instead of attaching them", () =>
  Effect.gen(function* () {
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.void
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush

    const execution = yield* executeTool(registry, {
      sessionID: Session.ID.make("ses_mcp_attachments"),
      ...toolIdentity,
      call: { type: "tool-call", id: "call_mcp_attachments", name: "direct_attachments", input: {} },
    })

    const text = execution.content?.map((part) => (part.type === "text" ? part.text : part.mime)) ?? []
    const saved =
      /^\[Binary resource file:\/\/\/data\.parquet \(application\/vnd\.apache\.parquet, 4 bytes\) saved to (.+)\]$/.exec(
        text[1] ?? "",
      )?.[1]
    expect(text[0]).toBe("[Resource file:///README.md (text/markdown)]\n# Notes")
    expect(saved?.startsWith(path.join(outputData, ToolOutput.DIRECTORY, "tool_"))).toBe(true)
    expect(yield* Effect.promise(() => fs.readFile(saved ?? "", "utf8"))).toBe("PAR1")
  }),
)

it.effect("waits for permission before calling an MCP tool", () =>
  Effect.gen(function* () {
    calls = 0
    assertion = yield* Deferred.make<Permission.AssertInput>()
    const permission = yield* Deferred.make<void>()
    decision = Deferred.await(permission)
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const toolSet = yield* registry.snapshot()
    expect(codeModeListings(toolSet.codeModeCatalog!).some((tool) => tool.path === "demo.search")).toBe(true)

    const fiber = yield* toolSet
      .execute({
        sessionID: Session.ID.make("ses_mcp_permission"),
        ...toolIdentity,
        call: {
          type: "tool-call",
          id: "call_mcp_permission",
          name: "execute",
          input: { code: 'return await tools.demo.search({ query: "open issues" })' },
        },
      })
      .pipe(Effect.forkScoped)
    expect(yield* Deferred.await(assertion)).toEqual({
      action: "demo_search",
      resources: ["*"],
      save: ["*"],
      metadata: { input: { query: "open issues" } },
      sessionID: Session.ID.make("ses_mcp_permission"),
      agent: toolIdentity.agent,
      source: {
        type: "tool",
        messageID: toolIdentity.messageID,
        id: "call_mcp_permission",
      },
    })
    expect(calls).toBe(0)

    yield* Deferred.succeed(permission, undefined)
    yield* Fiber.join(fiber)
    expect(calls).toBe(1)
  }),
)

it.effect("does not call MCP when permission is blocked", () =>
  Effect.gen(function* () {
    calls = 0
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.fail(
      new Permission.BlockedError({
        rules: [],
        permission: "demo_search",
        resources: ["*"],
        reason: "Search is disabled in this workspace",
      }),
    )
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const toolSet = yield* registry.snapshot()
    expect(codeModeListings(toolSet.codeModeCatalog!).some((tool) => tool.path === "demo.search")).toBe(true)

    const execution = yield* toolSet.execute({
      sessionID: Session.ID.make("ses_mcp_blocked"),
      ...toolIdentity,
      call: {
        type: "tool-call",
        id: "call_mcp_blocked",
        name: "execute",
        input: { code: "return await tools.demo.search({})" },
      },
    })
    expect(execution.content).toEqual([{ type: "text", text: "Search is disabled in this workspace" }])
    expect(execution.metadata).toEqual({
      toolCalls: [{ tool: "demo.search", status: "error" }],
      error: true,
    })
    expect(
      yield* executeTool(registry, {
        sessionID: Session.ID.make("ses_mcp_blocked"),
        ...toolIdentity,
        call: { type: "tool-call", id: "call_mcp_blocked_direct", name: "direct_lookup", input: {} },
      }),
    ).toEqual({
      status: "error",
      error: { type: "permission.rejected", message: "Search is disabled in this workspace" },
    })
    expect(calls).toBe(0)
  }),
)

it.effect("returns permission feedback instead of calling MCP", () =>
  Effect.gen(function* () {
    calls = 0
    assertion = yield* Deferred.make<Permission.AssertInput>()
    decision = Effect.fail(new Permission.CorrectedError({ feedback: "Search the docs folder instead" }))
    const registry = yield* Tool.Service
    const registration = yield* McpTool.Service
    yield* registration.flush
    const toolSet = yield* registry.snapshot()

    const execution = yield* toolSet.execute({
      sessionID: Session.ID.make("ses_mcp_corrected"),
      ...toolIdentity,
      call: {
        type: "tool-call",
        id: "call_mcp_corrected",
        name: "execute",
        input: { code: "return await tools.demo.search({})" },
      },
    })
    expect(execution.content).toEqual([{ type: "text", text: "Search the docs folder instead" }])
    expect(
      yield* executeTool(registry, {
        sessionID: Session.ID.make("ses_mcp_corrected"),
        ...toolIdentity,
        call: { type: "tool-call", id: "call_mcp_corrected_direct", name: "direct_lookup", input: {} },
      }),
    ).toEqual({
      status: "error",
      error: { type: "permission.rejected", message: "Search the docs folder instead" },
    })
    expect(calls).toBe(0)
  }),
)

for (const modern of [false, true]) {
  testEffect(Layer.empty).live(`integration policy denies MCP startup (${modern ? "modern" : "legacy"})`, () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern })
      yield* Effect.gen(function* () {
        const service = yield* Mcp.Service
        expect(yield* service.servers()).toEqual([])
        expect(Exit.isFailure(yield* service.connect("resources").pipe(Effect.exit))).toBe(true)
        yield* service.add("resources", { type: "remote", url: server.url, oauth: false })
        expect(yield* service.servers()).toEqual([])
        expect(yield* service.tools()).toEqual([])
        expect(yield* service.instructions()).toEqual([])
        expect(yield* service.prompts()).toEqual([])
        expect(yield* service.resourceCatalog()).toEqual({ resources: [], templates: [] })
        expect(Exit.isFailure(yield* service.resources({ server: "resources" }).pipe(Effect.exit))).toBe(true)
        expect(Exit.isFailure(yield* service.prompt({ server: "resources", name: "greet" }).pipe(Effect.exit))).toBe(
          true,
        )
        expect(
          Exit.isFailure(yield* service.readResource({ server: "resources", uri: "docs://readme" }).pipe(Effect.exit)),
        ).toBe(true)
        expect(Exit.isFailure(yield* service.callTool({ server: "resources", name: "echo" }).pipe(Effect.exit))).toBe(
          true,
        )
        expect(server.state.initializations).toBe(0)
      }).pipe(
        Effect.provide(
          resourceMcpLayer(server.url, undefined, undefined, {
            policies: [{ action: "mcp.use", resource: "*", effect: "deny" }],
          }),
        ),
      )
    }),
  )
}

for (const source of ["config", "organization"] as const) {
  testEffect(Config.testLayer()).live(
    `revokes and restores a connected MCP through ${source} policy`,
    () =>
      Effect.gen(function* () {
        const server = yield* resourceServer()
        const config = yield* Config.Service
        const test = yield* Config.Test
        const updates = yield* PubSub.unbounded<{ readonly type: string }>()
        const document = (effect: ConfigPolicy.Effect) =>
          new Document({
            type: "document",
            info: new Info({
              mcp: new ConfigMCP.Info({
                servers: { resources: new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }) },
              }),
              policies: [{ action: "mcp.use", resource: "*", effect }],
            }),
          })
        yield* test.setEntries([document("allow")])
        yield* Effect.gen(function* () {
          const service = yield* Mcp.Service
          expect(yield* settled(service)).toEqual({ status: "connected" })
          expect((yield* service.tools()).length).toBeGreaterThan(0)
          if (source === "config") yield* test.setEntries([document("deny")])
          if (source === "organization")
            yield* setOrganizationPolicy([{ action: "mcp.use", resource: "*", effect: "deny" }])
          yield* PubSub.publish(updates, { type: Event.Updated.type })
          yield* service.servers().pipe(
            Effect.filterOrFail(
              (servers) => servers.length === 0,
              () => new Error("MCP policy was not applied"),
            ),
            Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }),
          )
          expect(yield* service.servers()).toEqual([])
          expect(yield* service.tools()).toEqual([])
          expect(yield* service.instructions()).toEqual([])
          expect(yield* service.prompts()).toEqual([])
          expect(yield* service.resourceCatalog()).toEqual({ resources: [], templates: [] })
          expect(Exit.isFailure(yield* service.resources({ server: "resources" }).pipe(Effect.exit))).toBe(true)
          expect(Exit.isFailure(yield* service.prompt({ server: "resources", name: "greet" }).pipe(Effect.exit))).toBe(
            true,
          )
          expect(
            Exit.isFailure(
              yield* service.readResource({ server: "resources", uri: "docs://readme" }).pipe(Effect.exit),
            ),
          ).toBe(true)
          expect(Exit.isFailure(yield* service.connect("resources").pipe(Effect.exit))).toBe(true)
          expect(Exit.isFailure(yield* service.callTool({ server: "resources", name: "echo" }).pipe(Effect.exit))).toBe(
            true,
          )
          expect(server.state.toolCalls).toEqual([])
          expect(server.state.resourceReads).toEqual([])
          yield* Effect.promise(server.restart)
          if (source === "config") yield* test.setEntries([document("allow")])
          if (source === "organization") yield* setOrganizationPolicy(undefined)
          yield* PubSub.publish(updates, { type: Event.Updated.type })
          yield* service.servers().pipe(
            Effect.filterOrFail(
              (servers) => servers.some((server) => server.status.status === "connected"),
              () => new Error("MCP policy was not restored"),
            ),
            Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }),
          )
          expect(yield* settled(service)).toEqual({ status: "connected" })
          expect((yield* service.tools()).length).toBeGreaterThan(0)
        }).pipe(
          Effect.provide(
            resourceMcpLayer(server.url, undefined, undefined, {
              entries: config.entries,
              subscribe: ((definition?: { readonly type: string }) =>
                Stream.fromPubSub(updates).pipe(
                  Stream.filter((event) => !definition || event.type === definition.type),
                )) as Bus.Interface["subscribe"],
            }),
          ),
        )
      }),
  )
}

describe("MCP pool", () => {
  const live = testEffect(Layer.empty).live
  const pooled = (url: string, fields: Partial<ConfigMCP.Remote> = {}) =>
    new ConfigMCP.Remote({ type: "remote", url, oauth: false, pool: true, ...fields })
  const makePool = Layer.build(poolLayer).pipe(Effect.map((context) => Context.get(context, McpPool.Service)))
  const directories = [import.meta.dir, path.join(import.meta.dir, "fixture")]

  const status = (mcp: Mcp.Interface) => mcp.servers().pipe(Effect.map((servers) => servers[0]?.status))

  // Collects the pool's "not shared" messages with their details.
  const poolLogs = () => {
    const logs: Array<readonly [string, unknown]> = []
    const logger = Logger.map(Logger.formatStructured, (entry) => {
      if (Array.isArray(entry.message) && String(entry.message[0]).startsWith("mcp pool ignored"))
        logs.push([entry.message[0], entry.message[1]])
    })
    return { logs, layer: Logger.layer([logger]) }
  }

  // A pool whose own spawns, the attempts at a shared local connection, are recorded.
  const recordingPool = (spawns: Array<ChildProcess.Command>) =>
    Layer.build(
      McpPool.layer.pipe(
        Layer.provide(
          Layer.effect(
            ChildProcessSpawner.ChildProcessSpawner,
            Effect.gen(function* () {
              const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
              return ChildProcessSpawner.make((command) => {
                spawns.push(command)
                return spawner.spawn(command)
              })
            }),
          ).pipe(Layer.provide(LayerNode.compile(CrossSpawnSpawner.node))),
        ),
      ),
    ).pipe(Effect.map((context) => Context.get(context, McpPool.Service)))

  // One Location's MCP service on the given pool; `shutdown` closes the Location as eviction would.
  const open = (
    config: typeof ConfigMCP.Server.Type,
    pool: McpPool.Interface,
    directory: string,
    overrides: {
      onFormCreated?: (form: Form.Info) => Effect.Effect<void>
      published?: string[]
      environment?: Layer.Layer<Environment.Service>
      name?: string
      credentials?: Partial<Credential.Interface>
      bus?: Layer.Layer<Bus.Service>
    } = {},
  ) =>
    Effect.gen(function* () {
      const scope = yield* Scope.make()
      yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
      const context = yield* Layer.buildWithScope(
        resourceMcpLayer(config, overrides.onFormCreated, undefined, {
          pool: Layer.succeed(McpPool.Service, pool),
          directory,
          published: overrides.published,
          environment: overrides.environment,
          name: overrides.name,
          credentials: overrides.credentials,
          bus: overrides.bus,
        }),
        scope,
      )
      const mcp = Context.get(context, Mcp.Service)
      // Up to 10 seconds: a server that is down is retried for a while before it shows failed.
      yield* status(mcp).pipe(
        Effect.filterOrFail((status) => status?.status !== "pending"),
        Effect.retry({ times: 1_000, schedule: Schedule.spaced("10 millis") }),
      )
      return { mcp, forms: Context.get(context, Form.Service), shutdown: Scope.close(scope, Exit.void) }
    })

  live("shares one connection and one tool list between Locations", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false })
      const pool = yield* makePool
      const a = yield* open(pooled(server.url), pool, directories[0])
      const b = yield* open(pooled(server.url), pool, directories[1])

      const [toolsA, toolsB] = [yield* a.mcp.tools(), yield* b.mcp.tools()]
      expect(toolsA.map((tool) => tool.name)).toEqual(["echo"])
      expect(toolsB.map((tool) => tool.name)).toEqual(["echo"])
      // Each Location tags its own copy of the one listed tool.
      expect(toolsA[0]?.inputSchema).toBe(toolsB[0]?.inputSchema)
      expect(server.state.toolLists).toBe(1)
      yield* a.mcp.callTool({ server: "resources", name: "echo" })
      yield* b.mcp.callTool({ server: "resources", name: "echo" })
      expect(server.state.toolCalls).toHaveLength(2)
    }),
  )

  live("keeps connections apart when the configs differ", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false })
      const pool = yield* makePool
      const a = yield* open(pooled(server.url), pool, directories[0])
      const b = yield* open(pooled(server.url, { headers: { "x-account": "b" } }), pool, directories[0])
      expect((yield* a.mcp.tools())[0]?.inputSchema).not.toBe((yield* b.mcp.tools())[0]?.inputSchema)
      expect(server.state.toolLists).toBe(2)
    }),
  )

  test("keys connections on everything that changes them, and on no secret or name", () => {
    const remote = new ConfigMCP.Remote({
      type: "remote",
      url: "https://mcp.example.com",
      headers: { authorization: "secret" },
    })
    const local = new ConfigMCP.Local({ type: "local", command: ["server"], environment: { TOKEN: "secret" } })
    const key = (input: Partial<Parameters<typeof McpPool.key>[0]> = {}) => McpPool.key({ config: remote, ...input })
    const localKey = (input: Partial<Parameters<typeof McpPool.key>[0]> = {}) =>
      McpPool.key({ config: local, cwd: "/a", ...input })

    expect(key()).not.toContain("secret")
    expect(localKey()).not.toContain("secret")
    expect(key({ credential: "cred_a" })).not.toBe(key({ credential: "cred_b" }))
    expect(key({ config: new ConfigMCP.Remote({ ...remote, headers: { authorization: "other" } }) })).not.toBe(key())
    expect(key({ config: new ConfigMCP.Remote({ ...remote, codemode: false }) })).not.toBe(key())
    expect(key({ config: new ConfigMCP.Remote({ ...remote, disabled: false }) })).toBe(key())
    expect(localKey()).not.toBe(localKey({ cwd: "/b" }))
    expect(localKey({ config: new ConfigMCP.Local({ ...local, environment: { TOKEN: "other" } }) })).not.toBe(
      localKey(),
    )
    expect(localKey({ clientInfo: { name: "other", version: "1" } })).not.toBe(localKey())
  })

  live("keeps a server that negotiates legacy private to each Location", () => {
    const captured = poolLogs()
    return Effect.gen(function* () {
      const probes: Array<ChildProcess.Command> = []
      const spawns: Array<ChildProcess.Command> = []
      const pool = yield* recordingPool(probes)
      const config = new ConfigMCP.Local({
        type: "local",
        command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
        cwd: directories[0],
        pool: true,
      })
      const a = yield* open(config, pool, directories[0], { environment: recordingEnvironmentLayer(spawns) })
      const b = yield* open(config, pool, directories[0], { environment: recordingEnvironmentLayer(spawns) })
      expect(yield* status(a.mcp)).toEqual({ status: "connected" })
      expect(yield* status(b.mcp)).toEqual({ status: "connected" })
      // One discovery attempt through the pool, then each Location spawns its own, as it would unpooled.
      expect(probes).toHaveLength(1)
      expect(spawns).toHaveLength(2)
      // Once, when the verdict is recorded; the second Location skips straight to its own connection.
      expect(captured.logs).toEqual([
        ["mcp pool ignored: server speaks an older protocol", { server: "resources", version: "2025-11-25" }],
      ])
    }).pipe(Effect.provide(captured.layer))
  })

  live("connects on its own when a server refuses the pooled handshake", () => {
    const captured = poolLogs()
    return Effect.gen(function* () {
      const probes: Array<ChildProcess.Command> = []
      const spawns: Array<ChildProcess.Command> = []
      const pool = yield* recordingPool(probes)
      // Exits during initialize unless the client declares roots, which shared connections never do.
      const config = new ConfigMCP.Local({
        type: "local",
        command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
        cwd: directories[0],
        environment: { MCP_REQUIRE_ROOTS: "1" },
        pool: true,
      })
      const a = yield* open(config, pool, directories[0], { environment: recordingEnvironmentLayer(spawns) })
      expect(yield* status(a.mcp)).toEqual({ status: "connected" })
      const b = yield* open(config, pool, directories[0], { environment: recordingEnvironmentLayer(spawns) })
      expect(yield* status(b.mcp)).toEqual({ status: "connected" })
      expect(probes).toHaveLength(1)
      expect(spawns).toHaveLength(2)
      expect(captured.logs).toEqual([
        [
          "mcp pool ignored: server refused the shared connection",
          { server: "resources", error: expect.stringContaining("Connection closed") },
        ],
      ])
    }).pipe(Effect.provide(captured.layer))
  })

  // A server that is down, or that refuses this client, would refuse a private connection too.
  for (const [code, what, attempts] of [
    [503, "was down", 3],
    [401, "rejected the credentials", 1],
    [403, "refused access", 1],
  ] as const)
    live(`keeps pooling a server that ${what} when first tried`, () => {
      const captured = poolLogs()
      return Effect.gen(function* () {
        const refusing = { value: true, count: 0 }
        const server = yield* resourceServer({
          modern: true,
          resources: false,
          respond: () => {
            if (!refusing.value) return undefined
            refusing.count += 1
            return new Response(null, { status: code })
          },
        })
        const pool = yield* makePool
        const a = yield* open(pooled(server.url), pool, directories[0])
        expect((yield* status(a.mcp))?.status).toBe("failed")
        // The startup retries transient failures, and nothing tries a private connection afterwards.
        expect(refusing.count).toBe(attempts)
        expect(captured.logs).toEqual([])
        refusing.value = false
        yield* a.mcp.connect("resources")
        yield* open(pooled(server.url), pool, directories[1])
        expect(server.state.toolLists).toBe(1)
      }).pipe(Effect.provide(captured.layer))
    })

  live("decides a legacy server is private right after the handshake, before listing its tools", () => {
    const captured = poolLogs()
    return Effect.gen(function* () {
      const tmp = yield* tmpdirScoped()
      const wire = path.join(tmp.path, "wire")
      // Connects without roots but needs one to list its tools, which shared connections never declare.
      const config = new ConfigMCP.Local({
        type: "local",
        command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
        cwd: directories[0],
        environment: { MCP_REQUIRE_ROOTS_TO_LIST: "1", MCP_WIRE_LOG: wire },
        pool: true,
      })
      const pool = yield* makePool
      const a = yield* open(config, pool, directories[0])
      const b = yield* open(config, pool, directories[1])
      expect(yield* status(a.mcp)).toEqual({ status: "connected" })
      expect(yield* status(b.mcp)).toEqual({ status: "connected" })
      const methods = (yield* Effect.promise(() => fs.readFile(wire, "utf8"))).trim().split("\n")
      const count = (method: string) => methods.filter((item) => item === method).length
      // The shared attempt probes, falls back to initialize, then closes without listing. The private
      // connections know the server is legacy, so they skip the probe; each lists its two pages.
      expect(count("server/discover")).toBe(1)
      expect(count("initialize")).toBe(3)
      expect(count("tools/list")).toBe(4)
      expect(captured.logs).toEqual([
        ["mcp pool ignored: server speaks an older protocol", { server: "resources", version: expect.any(String) }],
      ])
    }).pipe(Effect.provide(captured.layer))
  })

  live("fails a pooled server that can't be spawned after one attempt", () =>
    Effect.gen(function* () {
      const probes: Array<ChildProcess.Command> = []
      const spawns: Array<ChildProcess.Command> = []
      const pool = yield* recordingPool(probes)
      const config = new ConfigMCP.Local({ type: "local", command: ["opencode-missing-mcp-server"], pool: true })
      const a = yield* open(config, pool, directories[0], { environment: recordingEnvironmentLayer(spawns) })
      expect((yield* status(a.mcp))?.status).toBe("failed")
      expect(probes).toHaveLength(1)
      // On Windows the spawn succeeds through the shell and the missing command shows up only as an exit
      // during the handshake, which can't be told apart from a server refusing it, so it tries once more.
      expect(spawns).toHaveLength(process.platform === "win32" ? 1 : 0)
    }),
  )

  live("connects each Location on its own without pool", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false })
      const pool = yield* makePool
      const config = new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" })
      const a = yield* open(config, pool, directories[0])
      const b = yield* open(config, pool, directories[0])
      expect((yield* a.mcp.tools())[0]?.inputSchema).not.toBe((yield* b.mcp.tools())[0]?.inputSchema)
      expect(server.state.toolLists).toBe(2)
    }),
  )

  live("closes the shared connection when its last Location lets go", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false })
      const pool = yield* makePool
      const a = yield* open(pooled(server.url), pool, directories[0])
      const b = yield* open(pooled(server.url), pool, directories[1])
      yield* a.shutdown
      yield* b.mcp.callTool({ server: "resources", name: "echo" })
      const c = yield* open(pooled(server.url), pool, directories[0])
      expect(server.state.toolLists).toBe(1)

      yield* b.shutdown
      yield* c.mcp.disconnect("resources")
      yield* open(pooled(server.url), pool, directories[0])
      expect(server.state.toolLists).toBe(2)
    }),
  )

  live("connects with the credential its key was made from, even when a sign-in lands mid-start", () =>
    Effect.gen(function* () {
      // The authorization header of every tool call, in order.
      const calls: Array<string | null> = []
      const server = yield* resourceServer({
        modern: true,
        resources: false,
        respond: async (request) => {
          if (request.method !== "POST") return undefined
          const body: unknown = await request.clone().json()
          if (isRecord(body) && body.method === "tools/call") calls.push(request.headers.get("authorization"))
          return undefined
        },
      })
      const pool = yield* makePool
      // "resources" signs in right after its first read of the credential store; "guest" never signs in.
      let signed: Integration.ID | undefined
      let reads = 0
      const row = (integrationID: Integration.ID) =>
        new Credential.Info({
          id: Credential.ID.make("cred_signed"),
          integrationID,
          label: "signed",
          value: {
            type: "oauth",
            methodID: Integration.MethodID.make("signed"),
            access: "signed-in",
            refresh: "signed-refresh",
            expires: Date.now() + 3_600_000,
            metadata: { serverUrl: server.url, tokenType: "Bearer" },
          },
        })
      const credentials: Partial<Credential.Interface> = {
        list: (integrationID) =>
          Effect.sync(() => {
            signed ??= integrationID
            if (integrationID !== signed) return []
            reads += 1
            return reads > 1 ? [row(integrationID)] : []
          }),
        get: () => Effect.sync(() => (signed ? row(signed) : undefined)),
      }
      const config = pooled(server.url, { oauth: undefined })
      const owner = yield* open(config, pool, directories[0], { credentials })
      const guest = yield* open(config, pool, directories[1], { credentials, name: "guest" })
      yield* guest.mcp.callTool({ server: "guest", name: "echo" })
      // Reconnecting picks up the sign-in under a key of its own.
      yield* owner.mcp.connect("resources")
      yield* owner.mcp.callTool({ server: "resources", name: "echo" })
      yield* guest.mcp.callTool({ server: "guest", name: "echo" })
      expect(calls).toEqual([null, "Bearer signed-in", null])
    }),
  )

  for (const shared of [true, false])
    live(
      `registers a server's prompt commands again after reconnecting${shared ? " to a connection another Location holds" : ""}`,
      () =>
        Effect.gen(function* () {
          const server = yield* resourceServer({ modern: true, resources: false })
          const config = shared
            ? pooled(server.url)
            : new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" })
          const pool = yield* makePool
          yield* open(config, pool, directories[0])
          // The second Location gets a working bus, so the command plugin hears its MCP events.
          const hub = yield* PubSub.unbounded<{ readonly type: string }>()
          const bus = Layer.mock(Bus.Service, {
            publish: (definition, data) =>
              Effect.suspend(() => {
                const event = { id: ID.create(), type: definition.type, data } as Payload<typeof definition>
                return PubSub.publish(hub, event).pipe(Effect.as(event))
              }),
            subscribe: ((definition?: { readonly type: string }) =>
              Stream.fromPubSub(hub).pipe(
                Stream.filter((event) => !definition || event.type === definition.type),
              )) as Bus.Interface["subscribe"],
          })
          const b = yield* open(config, pool, directories[1], { bus })
          const command = Context.get(yield* Layer.build(Command.layer.pipe(Layer.provide(bus))), Command.Service)
          yield* CommandPlugin.Plugin.effect(
            host({
              command: {
                list: () => Effect.die("unused command.list"),
                transform: command.transform,
                reload: command.reload,
              },
            }),
          ).pipe(
            Effect.provideService(Mcp.Service, b.mcp),
            Effect.provide(bus),
            Effect.provideService(
              Location.Service,
              Location.Service.of(location({ directory: AbsolutePath.make(directories[1]) })),
            ),
          )
          const registered = (expected: boolean) =>
            command.get("resources:greet").pipe(
              Effect.filterOrFail((found) => (found !== undefined) === expected),
              Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }),
            )
          yield* registered(true)
          yield* b.mcp.disconnect("resources")
          yield* registered(false)
          yield* b.mcp.connect("resources")
          expect(yield* status(b.mcp)).toEqual({ status: "connected" })
          yield* registered(true)
          // A shared reconnect joins the open connection instead of listing again.
          expect(server.state.toolLists).toBe(shared ? 1 : 3)
        }),
    )

  live("routes an elicitation to the Location whose call caused it", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false, urlElicitation: true })
      const pool = yield* makePool
      const created = yield* Deferred.make<CreatedForm>()
      const elsewhere: Form.Info[] = []
      const a = yield* open(pooled(server.url), pool, directories[0], {
        onFormCreated: (form) => Effect.sync(() => elsewhere.push(form)),
      })
      const b = yield* open(pooled(server.url), pool, directories[1], {
        onFormCreated: (form) => recordForm(created, form),
      })
      const call = yield* b.mcp.callTool({ server: "resources", name: "url-elicitation" }).pipe(Effect.forkScoped)
      const { form, directory } = yield* Deferred.await(created)
      expect(directory).toBe(directories[1])
      yield* b.forms.reply({ id: form.id, answer: { elicitation: true } })
      expect((yield* Fiber.join(call)).structured).toEqual({ action: "accept" })
      expect(elsewhere).toEqual([])
      expect(yield* a.forms.list()).toEqual([])
    }),
  )

  live("asks every holder about an elicitation no request claims, and takes the first answer", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false, urlElicitation: true })
      const pool = yield* makePool
      const environment = Context.get(yield* Layer.build(hostEnvironmentLayer), Environment.Service)
      const config = new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" })
      // Each holder knows the server by its own name; the connection carries the first one's.
      const holder = (directory: string, name: string) =>
        Effect.gen(function* () {
          const asked: string[] = []
          const forms = Context.get(
            yield* Layer.build(
              Form.layer.pipe(
                Layer.provide(
                  Layer.mock(Bus.Service, {
                    publish: (definition, data) =>
                      Effect.succeed({ id: ID.create(), type: definition.type, data } as Payload<typeof definition>),
                  }),
                ),
              ),
            ),
            Form.Service,
          )
          const scope = yield* Effect.scope
          const run = yield* FiberSet.makeRuntime<never, void, never>()
          const connection = pool.connection({
            name,
            config,
            directory,
            authProvider: Effect.undefined,
            elicitation: {
              create: (request) =>
                Effect.sync(() => asked.push(request.server)).pipe(
                  Effect.andThen(
                    forms.ask({ sessionID: "global", title: "Confirm", fields: [{ key: "ok", type: "boolean" }] }),
                  ),
                  Effect.map((state) => ({ action: state.status === "answered" ? "accept" : "cancel" }) as const),
                ),
              complete: () => Effect.void,
            },
            environment,
            fetch,
            scope,
            lock: Semaphore.makeUnsafe(1).withPermit,
            callbacks: run,
            fork: run,
            events: {
              connecting: Effect.void,
              connected: Effect.void,
              failed: () => Effect.void,
              settled: Effect.void,
              closed: () => Effect.void,
              changed: () => Effect.void,
            },
            select: Effect.succeed({ key: McpPool.key({ config }), authProvider: Effect.undefined }),
          })
          yield* connection.start
          return { connection, forms, asked }
        })
      const a = yield* holder(directories[0], "first")
      const b = yield* holder(directories[1], "second")
      const client = a.connection.client
      if (!client) throw new Error("expected a shared connection")
      expect(b.connection.client).toBe(client)

      // Called on the shared client directly, so no Location claims the request.
      const call = yield* client.callTool({ name: "url-elicitation" }).pipe(Effect.forkScoped)
      const pending = (forms: Form.Interface) =>
        forms.list().pipe(
          Effect.flatMap((list) => (list[0] ? Effect.succeed(list[0]) : Effect.fail("none"))),
          Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }),
        )
      const asked = yield* pending(a.forms)
      const other = yield* pending(b.forms)
      yield* a.forms.reply({ id: asked.id, answer: { ok: true } })
      expect((yield* Fiber.join(call)).structured).toEqual({ action: "accept" })
      expect((yield* b.forms.state(other.id)).status).toBe("cancelled")
      expect([a.asked, b.asked]).toEqual([["first"], ["second"]])
    }),
  )

  live("fans tool list changes out to every Location", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false, listChanged: true })
      const pool = yield* makePool
      const published = [[], []] as string[][]
      const a = yield* open(pooled(server.url), pool, directories[0], { published: published[0] })
      const b = yield* open(pooled(server.url), pool, directories[1], { published: published[1] })
      published.forEach((events) => events.splice(0))
      server.state.extraTools = [{ name: "added", inputSchema: { type: "object", properties: {} } }]
      yield* Effect.promise(server.sendToolListChanged)
      for (const mcp of [a.mcp, b.mcp])
        yield* mcp.tools().pipe(
          Effect.filterOrFail((tools) => tools.some((tool) => tool.name === "added")),
          Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }),
        )
      expect(server.state.toolLists).toBe(2)
      for (const events of published)
        yield* Effect.sync(() => events).pipe(
          Effect.filterOrFail((events) => events.includes(McpEvent.ToolsChanged.type)),
          Effect.retry({ times: 200, schedule: Schedule.spaced("10 millis") }),
        )
    }),
  )

  live("answers no roots on a shared connection", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false, roots: true })
      const pool = yield* makePool
      const own = yield* open(
        new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, protocol: "2026-07-28" }),
        pool,
        directories[0],
      )
      expect((yield* own.mcp.callTool({ server: "resources", name: "roots" })).structured).toEqual({
        kind: "roots",
        roots: [{ uri: Bun.pathToFileURL(directories[0]).href }],
      })
      const shared = yield* open(pooled(server.url), pool, directories[0])
      const error = yield* shared.mcp.callTool({ server: "resources", name: "roots" }).pipe(Effect.flip)
      expect(error.message).toContain("roots/list")
    }),
  )

  live("shares a local server across directories only when its cwd is fixed", () => {
    const captured = poolLogs()
    return Effect.gen(function* () {
      const tmp = yield* tmpdirScoped()
      const server = (pids: string, cwd?: string) =>
        new ConfigMCP.Local({
          type: "local",
          command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-modern-stdio.ts")],
          environment: { MCP_PID_FILE: path.join(tmp.path, pids) },
          cwd,
          pool: true,
        })
      const started = (pids: string) =>
        Effect.promise(() => fs.readFile(path.join(tmp.path, pids), "utf8")).pipe(
          Effect.map((text) => text.trim().split("\n")),
        )
      const pool = yield* makePool
      for (const directory of directories) yield* open(server("home", "~"), pool, directory)
      expect(yield* started("home")).toHaveLength(1)
      expect(captured.logs).toEqual([])

      for (const directory of directories) yield* open(server("project"), pool, directory)
      expect(yield* started("project")).toHaveLength(2)
      expect(captured.logs).toEqual([
        ["mcp pool ignored: local server has no fixed cwd, so each directory runs its own", { server: "resources" }],
      ])
    }).pipe(Effect.provide(captured.layer))
  })

  live("shares one local process, fails every holder when it dies, and closes it after the last", () =>
    Effect.gen(function* () {
      const tmp = yield* tmpdirScoped()
      const pids = path.join(tmp.path, "pids")
      const config = new ConfigMCP.Local({
        type: "local",
        command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-modern-stdio.ts")],
        environment: { MCP_PID_FILE: pids },
        pool: true,
      })
      const started = () =>
        Effect.promise(() => fs.readFile(pids, "utf8")).pipe(Effect.map((text) => text.trim().split("\n").map(Number)))
      const alive = (pid: number) => {
        try {
          process.kill(pid, 0)
          return true
        } catch {
          return false
        }
      }
      const until = <A>(effect: Effect.Effect<A>, check: (value: A) => boolean) =>
        effect.pipe(
          Effect.filterOrFail(check, () => "pending"),
          Effect.retry({ times: 500, schedule: Schedule.spaced("10 millis") }),
        )
      const pool = yield* makePool
      const a = yield* open(config, pool, directories[0])
      const b = yield* open(config, pool, directories[0])
      const [first] = yield* started()
      expect(yield* started()).toHaveLength(1)
      expect((yield* a.mcp.tools())[0]?.inputSchema).toBe((yield* b.mcp.tools())[0]?.inputSchema)

      process.kill(first!, "SIGKILL")
      for (const location of [a, b]) yield* until(status(location.mcp), (status) => status?.status === "failed")
      yield* a.mcp.connect("resources")
      expect(yield* status(a.mcp)).toEqual({ status: "connected" })
      expect((yield* status(b.mcp))?.status).toBe("failed")
      yield* b.mcp.connect("resources")
      expect(yield* status(b.mcp)).toEqual({ status: "connected" })
      const [, second] = yield* started()
      expect(yield* started()).toHaveLength(2)

      yield* a.shutdown
      expect(alive(second!)).toBe(true)
      yield* b.mcp.callTool({ server: "resources", name: "echo" })
      yield* b.shutdown
      yield* until(
        Effect.sync(() => alive(second!)),
        (alive) => !alive,
      )
    }),
  )

  live("keeps nothing of a Location that let go of a shared connection", () =>
    Effect.gen(function* () {
      const server = yield* resourceServer({ modern: true, resources: false, urlElicitation: true })
      const pool = yield* makePool
      // Reading a WeakRef keeps its target alive until the current job ends, so watch for finalization instead.
      const collected = new Set<string>()
      const registry = new FinalizationRegistry<string>((name) => collected.add(name))
      // The first Location opens the connection and makes a call through it, then lets go while another holds on.
      const shutdown = yield* Effect.gen(function* () {
        const created = yield* Deferred.make<CreatedForm>()
        const location = yield* open(pooled(server.url), pool, directories[0], {
          onFormCreated: (form) => recordForm(created, form),
        })
        const call = yield* location.mcp
          .callTool({ server: "resources", name: "url-elicitation" })
          .pipe(Effect.forkScoped)
        const { form } = yield* Deferred.await(created)
        yield* location.forms.reply({ id: form.id, answer: { elicitation: true } })
        yield* Fiber.join(call)
        registry.register(location.mcp, "mcp")
        registry.register(location.forms, "forms")
        return location.shutdown
      })
      const second = yield* open(pooled(server.url), pool, directories[1])
      yield* shutdown
      // The collector also scans the native stack, where a stale pointer from the shutdown can survive in
      // frames the event loop reuses. Unrelated work through the same paths overwrites those; then collect.
      for (let attempt = 0; attempt < 50 && collected.size < 2; attempt++) {
        yield* Effect.forEach(Array.from({ length: 20 }), () => Effect.promise(() => Bun.sleep(0)), {
          concurrency: "unbounded",
        })
        yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(() => resolve(Bun.gc(true)), 0)))
      }
      expect(Array.from(collected).toSorted()).toEqual(["forms", "mcp"])
      expect(yield* status(second.mcp)).toEqual({ status: "connected" })
      expect(server.state.toolLists).toBe(1)
    }),
  )
})