ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): clarify MCP config reload

Browse source
This commit is contained in:
Aiden Cline 2026-07-23 16:37:55 +00:00
commit 473d16ac12
2 changed files with 75 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

40
packages/core/src/mcp/index.ts
View file

@ -182,7 +182,7 @@ export const layer = (options?: Options) => Layer.effect(
const fork = yield* FiberSet.makeRuntime<never, void, never>()
yield* Effect.addFinalizer((exit) => Scope.close(root, exit))
const load = Effect.fnUntraced(function* () {
const loadConfig = Effect.fnUntraced(function* () {
const documents = (yield* config.entries()).filter(
(entry): entry is Config.Document => entry.type === "document",
)
@ -199,8 +199,8 @@ export const layer = (options?: Options) => Layer.effect(
}
return { timeout, servers }
})
const initial = yield* load()
const configured = { names: new Set(initial.servers.keys()), timeout: initial.timeout }
const initial = yield* loadConfig()
const configState = { names: new Set(initial.servers.keys()), timeout: initial.timeout }
// Later config files win for duplicate server names; per-server timeout overrides globals.
const runtime = new Map<ServerName, ServerEntry>()
// Serializes lifecycle operations per server. Anything taking this lock from a connection
@ -616,6 +616,22 @@ export const layer = (options?: Options) => Layer.effect(
yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
})
const reloadConfig = Effect.fnUntraced(function* () {
const next = yield* loadConfig()
configState.timeout = next.timeout
const names = new Set([...configState.names, ...next.servers.keys()])
for (const name of names) {
const updated = next.servers.get(name)
if (!updated) {
yield* removeServer(name).pipe(locks.withLock(name))
continue
}
if (isDeepStrictEqual(runtime.get(name)?.config, updated)) continue
yield* replaceServer(name, updated).pipe(locks.withLock(name))
}
configState.names = new Set(next.servers.keys())
})
// Disabled servers settle their startup immediately so queries never block on them.
for (const [name, entry] of runtime) {
if (entry.config.disabled) {
@ -653,21 +669,7 @@ export const layer = (options?: Options) => Layer.effect(
fork(
events.subscribe(Event.Updated).pipe(
Stream.runForEach(() =>
Effect.gen(function* () {
const next = yield* load()
configured.timeout = next.timeout
const names = new Set([...configured.names, ...next.servers.keys()])
for (const name of names) {
const updated = next.servers.get(name)
if (updated) {
if (isDeepStrictEqual(runtime.get(name)?.config, updated)) continue
yield* replaceServer(name, updated).pipe(locks.withLock(name))
continue
}
yield* removeServer(name).pipe(locks.withLock(name))
}
configured.names = new Set(next.servers.keys())
}).pipe(Effect.catchCause((cause) => Effect.logError("failed to reload MCP config", { cause }))),
reloadConfig().pipe(Effect.catchCause((cause) => Effect.logError("failed to reload MCP config", { cause }))),
),
Effect.ignore,
),
@ -694,7 +696,7 @@ export const layer = (options?: Options) => Layer.effect(
}),
add: Effect.fn("MCP.add")(function* (server, config) {
const name = ServerName.make(server)
yield* replaceServer(name, { ...config, timeout: { ...configured.timeout, ...config.timeout } }).pipe(
yield* replaceServer(name, { ...config, timeout: { ...configState.timeout, ...config.timeout } }).pipe(
locks.withLock(name),
)
}),

102
packages/core/test/mcp.test.ts
View file

@ -150,55 +150,54 @@ function resourceServer(
)
}
function resourceMcpLayer(
server: string | typeof ConfigMCP.Server.Type,
onFormCreated?: (form: Form.Info) => Effect.Effect<void>,
options?: MCP.Options,
overrides?: {
entries?: Config.Interface["entries"]
subscribe?: EventV2.Interface["subscribe"]
},
) {
function resourceMcpLayer(input: {
server: string | typeof ConfigMCP.Server.Type
onFormCreated?: (form: Form.Info) => Effect.Effect<void>
options?: MCP.Options
entries?: Config.Interface["entries"]
subscribe?: EventV2.Interface["subscribe"]
}) {
const directory = AbsolutePath.make(import.meta.dir)
const unusedIntegration = () => Effect.die("unused integration service")
return MCP.layer(options).pipe(
const entries =
input.entries ??
(() =>
Effect.succeed([
new Config.Document({
type: "document",
info: new Config.Info({
mcp: new ConfigMCP.Info({
servers: {
resources:
typeof input.server === "string"
? new ConfigMCP.Remote({ type: "remote", url: input.server, oauth: false })
: input.server,
},
}),
}),
}),
]))
return MCP.layer(input.options).pipe(
Layer.provideMerge(Form.layer),
Layer.provide(
Layer.mergeAll(
Layer.succeed(
Config.Service,
Config.Service.of({
entries:
overrides?.entries ??
(() =>
Effect.succeed([
new Config.Document({
type: "document",
info: new Config.Info({
mcp: new ConfigMCP.Info({
servers: {
resources:
typeof server === "string"
? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false })
: server,
},
}),
}),
}),
])),
}),
Config.Service.of({ entries }),
),
Layer.succeed(Location.Service, Location.Service.of(location({ directory }))),
Layer.mock(EventV2.Service, {
subscribe: overrides?.subscribe ?? (() => Stream.never),
subscribe: input.subscribe ?? (() => Stream.never),
publish: (definition, data) => {
const event = {
id: EventV2.ID.create(),
type: definition.type,
data,
} as EventV2.Payload<typeof definition>
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))
if (event.type !== Form.Event.Created.type || !input.onFormCreated) return Effect.succeed(event)
return input
.onFormCreated(Schema.decodeUnknownSync(Form.Event.Created.data)(data).form)
.pipe(Effect.as(event))
},
}),
Layer.mock(Integration.Service, {
@ -575,7 +574,7 @@ test("accepts empty MCP elicitations without creating forms", async () => {
const result = yield* service.callTool({ server: "resources", name: "empty-elicitation" })
expect(yield* forms.list()).toEqual([])
return result
}).pipe(Effect.provide(resourceMcpLayer(server.url)))
}).pipe(Effect.provide(resourceMcpLayer({ server: server.url })))
expect(result.structured).toEqual({ action: "accept", content: {} })
}),
@ -602,7 +601,12 @@ test("acknowledges completed MCP URL elicitations without returning internal con
expect(yield* forms.state(form.id)).toEqual({ status: "answered", answer: { elicitation: true } })
return result
}).pipe(
Effect.provide(resourceMcpLayer(server.url, (form) => Deferred.succeed(created, form).pipe(Effect.asVoid))),
Effect.provide(
resourceMcpLayer({
server: server.url,
onFormCreated: (form) => Deferred.succeed(created, form).pipe(Effect.asVoid),
}),
),
)
expect(result.structured).toEqual({ action: "accept" })
@ -655,7 +659,10 @@ test("loads and reads MCP resources", async () => {
expect(server.clientVersion()).toMatchObject({ name: "sdk", version: "1.2.3" })
}).pipe(
Effect.provide(
resourceMcpLayer(server.url, undefined, { clientInfo: { name: "sdk", version: "1.2.3" } }),
resourceMcpLayer({
server: server.url,
options: { clientInfo: { name: "sdk", version: "1.2.3" } },
}),
),
)
}),
@ -718,13 +725,13 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => {
expect(yield* service.remove("dynamic").pipe(Effect.flip)).toBeInstanceOf(MCP.NotFoundError)
}).pipe(
Effect.provide(
resourceMcpLayer(
new ConfigMCP.Local({
resourceMcpLayer({
server: new ConfigMCP.Local({
type: "local",
command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
disabled: true,
}),
),
}),
),
)
}),
@ -789,12 +796,11 @@ test("reconciles MCP servers when config updates", async () => {
expect(replaced[0]?.status).toEqual({ status: "disabled" })
}).pipe(
Effect.provide(
resourceMcpLayer(
new ConfigMCP.Local({ type: "local", command, disabled: true }),
undefined,
undefined,
{ entries, subscribe },
),
resourceMcpLayer({
server: new ConfigMCP.Local({ type: "local", command, disabled: true }),
entries,
subscribe,
}),
),
)
}),
@ -833,13 +839,13 @@ test("serializes concurrent MCP lifecycle operations", async () => {
expect((yield* service.tools()).length).toBeGreaterThan(0)
}).pipe(
Effect.provide(
resourceMcpLayer(
new ConfigMCP.Local({
resourceMcpLayer({
server: new ConfigMCP.Local({
type: "local",
command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
disabled: true,
}),
),
}),
),
)
}),