Restart only the MCP servers a config change affects

An /mcp add, remove, approve, or reject used to build a new MCP host and
restart every server. The runtime now applies the change in place through
Engine.reload: servers with the same config keep their process, connection,
sign-in, and calls; removed or changed servers stop and their calls end;
a changed or re-added server starts once its own old process has exited.

The engine copies each server's environment and compares it by content, so
reading the same config again no longer looks like a change.
This commit is contained in:
Pranit committed 2026-10-23 19:10:17 -04:00
1 parent 314c01479c
commit 333d476456
23 files changed
+1117 -318

No files matched your search

+25 -8
View File
@@ -134,15 +134,17 @@ pub fn loadHost(
try skipUnprintableNames(alloc, &loaded);
// A host with no servers still carries the notes saying why.
if (loaded.configs.items.len == 0 and loaded.diagnostics.items.len == 0) return null;
const notes = try alloc.alloc([]const u8, loaded.diagnostics.items.len);
var rendered: usize = 0;
errdefer {
for (notes[0..rendered]) |note| alloc.free(note);
alloc.free(notes);
if (extras.reuse) |h| {
// Only the servers whose config changed restart. The host owns the
// notes either way.
if (h.reload(loaded.configs.items, &inherited, try renderNotes(alloc, loaded.diagnostics.items))) |_| return h else |err| {
debug_trace.logf("mcp", "MCP host not reloaded in place, starting a new one err={s}", .{@errorName(err)});
}
}
for (loaded.diagnostics.items, notes) |diagnostic, *note| {
note.* = try project_config.renderWorkspaceDiagnostic(alloc, diagnostic);
rendered += 1;
const notes = try renderNotes(alloc, loaded.diagnostics.items);
errdefer {
for (notes) |note| alloc.free(note);
alloc.free(notes);
}
const h = try mcp_host.Host.create(alloc, io_mod.getIo(), loaded.configs.items, &inherited, .{
.client_version = build_options.app_version,
@@ -156,6 +158,21 @@ pub fn loadHost(
return h;
}
/// Each diagnostic in a sentence, owned by `alloc`.
fn renderNotes(alloc: Allocator, diagnostics: []const project_config.WorkspaceDiagnostic) ![]const []const u8 {
const notes = try alloc.alloc([]const u8, diagnostics.len);
var rendered: usize = 0;
errdefer {
for (notes[0..rendered]) |note| alloc.free(note);
alloc.free(notes);
}
for (diagnostics, notes) |diagnostic, *note| {
note.* = try project_config.renderWorkspaceDiagnostic(alloc, diagnostic);
rendered += 1;
}
return notes;
}
fn traceWorkspaceDiagnostics(diagnostics: []const project_config.WorkspaceDiagnostic) void {
for (diagnostics) |diagnostic| {
var name_buf: [256]u8 = undefined;
+11 -5
View File
@@ -118,12 +118,18 @@ pub const State = struct {
self.replaceHost(try load(alloc, workspace_root, registry, .{ .questions = questions, .notify = self.host_sink.?.notify(), .slash_commands = true }));
}
/// Loads the config again after `/mcp` changed it; calls still using the
/// old host finish on it.
/// Loads the config again after `/mcp` changed it. The host takes it in
/// place, so only a server whose config changed restarts. A new host
/// comes only when there was none, or the last server went; calls still
/// using an old one finish on it.
pub fn reloadHost(self: *State, alloc: Allocator, workspace_root: []const u8, registry: tool_dispatch.Registry) !void {
const load = self.host_load orelse return;
const sink = self.host_sink orelse return;
self.replaceHost(try load(alloc, workspace_root, registry, .{ .questions = self.host_questions, .notify = sink.notify(), .slash_commands = true }));
const lease = self.leaseHost();
defer if (lease) |l| l.deinit();
const current = if (lease) |l| l.host else null;
const next = try load(alloc, workspace_root, registry, .{ .questions = self.host_questions, .notify = sink.notify(), .slash_commands = true, .reuse = current });
if (next != current) self.replaceHost(next);
self.startHostServers();
}
@@ -197,8 +203,8 @@ pub const State = struct {
const lease = self.leaseHost() orelse return;
defer lease.deinit();
self.host_started_ms = io_mod.milliTimestamp();
for (0..lease.host.runtime.servers().len) |i| lease.host.runtime.prepare(i) catch |err| {
debug_trace.logf("mcp", "MCP servers not started in the background at={d} err={s}", .{ i, @errorName(err) });
for (lease.host.runtime.servers()) |s| lease.host.runtime.prepare(s.name) catch |err| {
debug_trace.logf("mcp", "MCP servers not started in the background server={s} err={s}", .{ s.name, @errorName(err) });
return;
};
}
+27 -4
View File
@@ -82,10 +82,9 @@ fn commandLine(arena: Allocator, command: []const u8, args: []const []const u8)
return out.items;
}
/// Allocates everything with `arena`, which must outlive the engine: the
/// engine copies each config but keeps pointing at a stdio server's
/// environment. Saved sign-in credentials are not read here: the runtime
/// hands them to the engine from its own thread.
/// Allocates everything with `arena`; the engine copies what it keeps. Saved
/// sign-in credentials are not read here: the runtime hands them to the
/// engine from its own thread.
pub fn resolve(
arena: Allocator,
configs: []const McpServerConfig,
@@ -157,6 +156,30 @@ pub fn resolve(
return .{ .servers = servers, .engine = engine.items, .engine_server = engine_server.items };
}
/// `servers` copied into `arena`, which outlives them: a server's strings
/// that equal the ones its namesake in `prior` has are shared, so taking a
/// config that changed little again costs little more than the list.
pub fn keepServers(arena: Allocator, servers: []const Server, prior: []const Server) Allocator.Error![]Server {
const out = try arena.alloc(Server, servers.len);
for (servers, out) |s, *o| {
const old: ?Server = for (prior) |p| {
if (std.mem.eql(u8, p.name, s.name)) break p;
} else null;
o.* = s;
o.name = if (old) |p| p.name else try arena.dupe(u8, s.name);
inline for (.{ "url", "command", "missing", "workspace" }) |field| {
@field(o, field) = try reuse(arena, @field(s, field), if (old) |p| @field(p, field) else null);
}
}
return out;
}
fn reuse(arena: Allocator, value: ?[]const u8, old: ?[]const u8) Allocator.Error!?[]const u8 {
const v = value orelse return null;
if (old) |o| if (std.mem.eql(u8, o, v)) return o;
return try arena.dupe(u8, v);
}
/// `url`'s scheme and authority, such as `https://mcp.slack.com`.
fn origin(url: []const u8) []const u8 {
const authority = (std.mem.find(u8, url, "://") orelse return url) + 3;
+5 -1
View File
@@ -12,7 +12,9 @@ const testing = std.testing;
/// "withdraw" asks form question Q1, withdraws it, asks Q2, and once Q2 is
/// answered, returns what reached each: "e1 " if Q1 got an answer, then
/// "e2=" and Q2's `name`. A call that mentions "confirm" asks question c1
/// and returns "c1=" and the answer's action.
/// and returns "c1=" and the answer's action. A call that mentions "pid"
/// returns the server's process id, as "pid=" and the number. Started with
/// `linger` as its first argument, it exits a second after its stdin ends.
pub const script =
\\#!/bin/sh
\\while IFS= read -r line; do
@@ -37,9 +39,11 @@ pub const script =
\\ *'"id":"e2"'*)
\\ name=$(printf '%s' "$line" | sed -n 's/.*"name":"\([^"]*\)".*/\1/p')
\\ printf '{"jsonrpc":"2.0","id":%s,"result":{"content":[{"type":"text","text":"%se2=%s"}]}}\n' "$call" "$got" "$name" ;;
\\ *'"method":"tools/call"'*pid*) printf '{"jsonrpc":"2.0","id":%s,"result":{"content":[{"type":"text","text":"pid=%s"}]}}\n' "$id" "$$" ;;
\\ *'"method":"tools/call"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"content":[{"type":"text","text":"hi"}]}}\n' "$id" ;;
\\ esac
\\done
\\if [ "$1" = linger ]; then sleep 1; fi
\\
;
+32 -3
View File
@@ -29,6 +29,9 @@ pub const Extras = struct {
notify: ?runtime_mod.Notify = null,
/// See `runtime.Options.slash_commands`.
slash_commands: bool = false,
/// The host the config changed under: it takes the new config in place
/// (`Host.reload`) and is returned, unless no server is left.
reuse: ?*Host = null,
};
/// Loads the host for a workspace, or null when there is no home folder or no
@@ -44,8 +47,11 @@ pub const Host = struct {
gpa: Allocator,
runtime: Runtime,
catalog: Catalog,
/// What loading the config skipped and why, one sentence each; owned by `gpa`.
/// What loading the config skipped and why, one sentence each; owned by
/// `gpa`. A reload replaces them under `notes_mutex`, which a thread other
/// than the reloading one holds while it reads them.
notes: []const []const u8 = &.{},
notes_mutex: std.Io.Mutex = .init,
/// For an owner that replaces the host while others still use it: the
/// uses still running, and whether it was replaced. The owner guards both.
leases: usize = 0,
@@ -65,6 +71,7 @@ pub const Host = struct {
errdefer gpa.destroy(h);
h.gpa = gpa;
h.notes = &.{};
h.notes_mutex = .init;
h.leases = 0;
h.retired = false;
try h.runtime.start(gpa, io, configs, inherited, options);
@@ -76,13 +83,35 @@ pub const Host = struct {
/// Stops every server and frees the host.
pub fn destroy(h: *Host) void {
const gpa = h.gpa;
for (h.notes) |note| gpa.free(note);
gpa.free(h.notes);
freeNotes(gpa, h.notes);
h.catalog.deinit();
h.runtime.deinit();
gpa.destroy(h);
}
/// Takes a changed config in place: only a server whose config changed
/// restarts (`Runtime.reload`), and the catalog follows. `notes` replace
/// the host's, which it then owns, also on failure. Called by the one
/// thread that reloads; others may use the host meanwhile.
pub fn reload(
h: *Host,
configs: []const mcp_contract.McpServerConfig,
inherited: *const std.process.Environ.Map,
notes: []const []const u8,
) !void {
h.notes_mutex.lockUncancelable(h.runtime.io);
const old = h.notes;
h.notes = notes;
h.notes_mutex.unlock(h.runtime.io);
freeNotes(h.gpa, old);
try h.catalog.reload(try h.runtime.reload(configs, inherited));
}
fn freeNotes(gpa: Allocator, notes: []const []const u8) void {
for (notes) |note| gpa.free(note);
gpa.free(notes);
}
/// Replaced: until this host is gone, no other host starts a server, so
/// none runs twice. Calls that began on it still finish here.
pub fn retire(h: *Host) void {
+179 -53
View File
@@ -37,18 +37,30 @@ pub const Catalog = struct {
io: std.Io,
runtime: *Runtime,
builtins: tool_dispatch.Registry,
/// Every binding carries it, so a binding from another run never matches.
generation: u64,
/// Every binding carries it, so a binding from another run, or from
/// before a reload dropped a server, never matches. Read through
/// `currentGeneration`, since `reload` changes it.
generation: std.atomic.Value(u64),
names: tool_names.Registry,
mutex: std.Io.Mutex = .init,
/// One per `runtime.servers()`; guarded by `mutex`.
/// The servers the lists are of, from `runtime.servers()`, which keep
/// their names until the runtime goes; replaced with `lists` by
/// `reload`. Guarded by `mutex`.
servers: []const config.Server,
/// One per server in `servers`; guarded by `mutex`.
lists: []List,
/// The last number `serial` gave out; guarded by `mutex`.
serials: u64 = 0,
const List = struct {
/// Which list this is, from `serial`: a list loaded for one a reload
/// replaced meanwhile is dropped.
token: u64,
arena: ?std.heap.ArenaAllocator = null,
tools: []McpTool = &.{},
digest: ?[32]u8 = null,
/// Bumped when the list changes; bindings carry it.
/// A new `serial` when the list changes, so no two lists, even of a
/// server removed and added back, share one; bindings carry it.
generation: u64 = 0,
/// The server's `toolsEpoch` when this list was asked for.
epoch: u64 = 0,
@@ -62,30 +74,88 @@ pub const Catalog = struct {
/// `runtime` must outlive the catalog. `builtins` are fx's own tools, so an
/// MCP tool's name never takes one of theirs.
pub fn init(c: *Catalog, gpa: Allocator, io: std.Io, runtime: *Runtime, builtins: tool_dispatch.Registry) Allocator.Error!void {
const lists = try gpa.alloc(List, runtime.servers().len);
for (lists) |*l| l.* = .{};
const servers = runtime.servers();
const lists = try gpa.alloc(List, servers.len);
const now: u96 = @bitCast(std.Io.Clock.real.now(io).nanoseconds);
c.* = .{
.gpa = gpa,
.io = io,
.runtime = runtime,
.builtins = builtins,
.generation = @as(u64, @truncate(now)) | 1,
.generation = .init(@as(u64, @truncate(now)) | 1),
.names = .init(gpa),
.servers = servers,
.lists = lists,
};
for (lists) |*l| l.* = .{ .token = c.serial() };
}
pub fn deinit(c: *Catalog) void {
for (c.lists) |*l| {
if (l.arena) |*arena| arena.deinit();
if (l.failure) |why| c.gpa.free(why);
}
for (c.lists) |*l| freeList(c.gpa, l);
c.gpa.free(c.lists);
c.names.deinit();
c.* = undefined;
}
/// Follows `Runtime.reload`, on the thread that called it: a server that
/// kept running keeps its list, and the rest start without one. When a
/// server the catalog knew went or was replaced, every binding and
/// subagent view made before stops matching, as with a new host.
pub fn reload(c: *Catalog, done: runtime_mod.Reloaded) Allocator.Error!void {
const servers = c.runtime.servers();
std.debug.assert(servers.len == done.kept.len);
const lists = try c.gpa.alloc(List, servers.len);
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
for (servers, done.kept, lists) |s, kept, *l| {
l.* = .{ .token = c.serial() };
if (!kept) continue;
for (c.servers, c.lists) |old, *from| if (from.token != 0 and std.mem.eql(u8, old.name, s.name)) {
l.* = from.*;
from.token = 0;
break;
};
}
for (c.lists) |*l| if (l.token != 0) freeList(c.gpa, l);
c.gpa.free(c.lists);
c.servers = servers;
c.lists = lists;
if (done.dropped) _ = c.generation.fetchAdd(2, .acq_rel);
}
fn currentGeneration(c: *const Catalog) u64 {
return c.generation.load(.acquire);
}
fn freeList(gpa: Allocator, l: *List) void {
if (l.arena) |*arena| arena.deinit();
if (l.failure) |why| gpa.free(why);
}
/// A number no list or list state had, under `mutex` once others may run.
fn serial(c: *Catalog) u64 {
c.serials += 1;
return c.serials;
}
/// The servers the catalog knows now. Their names outlive the catalog.
fn currentServers(c: *Catalog) []const config.Server {
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
return c.servers;
}
fn indexLocked(c: *const Catalog, name: []const u8) ?usize {
for (c.servers, 0..) |s, i| if (std.mem.eql(u8, s.name, name)) return i;
return null;
}
/// Server `name`'s list, if it is still list `token`.
fn listLocked(c: *Catalog, name: []const u8, token: u64) ?*List {
const l = &c.lists[c.indexLocked(name) orelse return null];
return if (l.token == token) l else null;
}
/// The has, validate, and call operations, for checks made outside a
/// tool context.
pub fn runtimeCapabilities(c: *Catalog) tool_mcp_runtime.RuntimeCapabilities {
@@ -117,7 +187,7 @@ pub const Catalog = struct {
if (err == error.OutOfMemory) return error.OutOfMemory;
return .{ .invalid = try arena.print("Invalid arguments for MCP tool {s}: {s}", .{ name, @errorName(err) }) };
};
return .{ .valid = c.generation };
return .{ .valid = c.currentGeneration() };
}
/// Null when no server has the tool.
@@ -130,8 +200,8 @@ pub const Catalog = struct {
options: tool_mcp_runtime.CallOptions,
) !?tool_mcp_runtime.CallResult {
if (!c.mayAllow(options.access)) return error.McpAccessDenied;
if (options.expected_binding) |b| if (b.runtime_generation != c.generation) return error.McpAdvertisedToolChanged;
if (options.expected_runtime_generation) |g| if (g != c.generation) return error.McpAuthorityChanged;
if (options.expected_binding) |b| if (b.runtime_generation != c.currentGeneration()) return error.McpAdvertisedToolChanged;
if (options.expected_runtime_generation) |g| if (g != c.currentGeneration()) return error.McpAuthorityChanged;
try c.ensureCurrent(name, options.cancel_flag);
const server, const tool, const transport, const how = found: {
c.mutex.lockUncancelable(c.io);
@@ -141,7 +211,7 @@ pub const Catalog = struct {
if (options.expected_binding) |b| {
if (!b.sameDefinition(c.bindingLocked(found.server, found.tool.*))) return error.McpAdvertisedToolChanged;
}
const s = c.runtime.servers()[found.server];
const s = c.servers[found.server];
break :found .{ s.name, try arena.dupe(u8, found.tool.original_name), @tagName(s.transport), signIn(c.runtime.options.slash_commands, s.source) };
};
var asking: Asking = .{ .responder = options.input_responder orelse undefined, .tool = name, .cancel = options.cancel_flag, .io = c.io };
@@ -172,15 +242,15 @@ pub const Catalog = struct {
access: Access,
cancel: Cancel,
) !tool_mcp_runtime.SearchResult {
const servers = c.runtime.servers();
const servers = c.currentServers();
if (request.server) |name| {
for (servers) |s| {
if (std.mem.eql(u8, s.name, name)) break;
} else return .{ .model_output = try tool_search.renderServerNotFound(alloc) };
}
// Every server starts at once, then each list is waited for.
for (servers, 0..) |s, i| if (searched(request, s)) c.runtime.prepare(i) catch {};
for (servers, 0..) |s, i| if (searched(request, s)) try c.refresh(i, cancel);
for (servers) |s| if (searched(request, s)) c.runtime.prepare(s.name) catch {};
for (servers) |s| if (searched(request, s)) try c.refresh(s.name, cancel);
if (try c.renderUnavailable(alloc, request)) |output| return .{ .model_output = output };
var scratch: std.heap.ArenaAllocator = .init(c.gpa);
@@ -188,12 +258,12 @@ pub const Catalog = struct {
var candidates: std.ArrayList(tool_search.Candidate) = .empty;
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
for (servers, c.lists, 0..) |s, l, i| {
for (c.servers, c.lists, 0..) |s, l, i| {
if (!searched(request, s) or l.state != .loaded) continue;
for (l.tools) |*tool| {
if (!c.allowsLocked(access, i, tool.prefixed_name)) continue;
if (!try tool_search.searchable(scratch.allocator(), permission_rules, s.name, tool)) continue;
try candidates.append(scratch.allocator(), .{ .server_name = s.name, .instructions = try c.runtime.instructions(scratch.allocator(), i), .tool = tool, .binding = c.bindingLocked(i, tool.*) });
try candidates.append(scratch.allocator(), .{ .server_name = s.name, .instructions = try c.runtime.instructions(scratch.allocator(), s.name), .tool = tool, .binding = c.bindingLocked(i, tool.*) });
}
}
// An open search that matches nothing says which servers it couldn't
@@ -214,7 +284,7 @@ pub const Catalog = struct {
for (reasons.items) |reason| alloc.free(reason);
reasons.deinit(alloc);
}
for (c.runtime.servers(), c.lists) |s, l| {
for (c.servers, c.lists) |s, l| {
try reasons.ensureUnusedCapacity(alloc, 1);
reasons.appendAssumeCapacity(try c.unavailableText(alloc, s, l) orelse continue);
}
@@ -239,7 +309,7 @@ pub const Catalog = struct {
const found = c.findLocked(name) orelse return c.whyUnavailableLocked(alloc, name);
if (!c.allowsLocked(access, found.server, name)) return null;
if (permissions.rulesDenyAllTargetsForPermission(permission_rules, name)) return null;
const instructions = try c.runtime.instructions(alloc, found.server);
const instructions = try c.runtime.instructions(alloc, c.servers[found.server].name);
defer if (instructions) |text| alloc.free(text);
var projection = try selected_schema.project(alloc, found.tool.*, instructions, limits);
if (projection == .selected) projection.selected.mcp_binding = c.bindingLocked(found.server, found.tool.*);
@@ -257,16 +327,16 @@ pub const Catalog = struct {
access: Access,
cancel: Cancel,
) !selected_schema.AlwaysLoadedTools {
const servers = c.runtime.servers();
for (servers, 0..) |s, i| if (s.always_loaded) c.runtime.prepare(i) catch {};
for (servers, 0..) |s, i| if (s.always_loaded) try c.refresh(i, cancel);
const servers = c.currentServers();
for (servers) |s| if (s.always_loaded) c.runtime.prepare(s.name) catch {};
for (servers) |s| if (s.always_loaded) try c.refresh(s.name, cancel);
var loaded: selected_schema.AlwaysLoaded = .init(limits);
errdefer loaded.deinit(alloc);
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
for (servers, c.lists, 0..) |s, l, i| {
for (c.servers, c.lists, 0..) |s, l, i| {
if (!s.always_loaded or l.state != .loaded) continue;
const instructions = try c.runtime.instructions(alloc, i);
const instructions = try c.runtime.instructions(alloc, s.name);
defer if (instructions) |text| alloc.free(text);
for (l.tools) |*tool| {
if (permissions.rulesDenyAllTargetsForPermission(permission_rules, tool.prefixed_name)) continue;
@@ -295,7 +365,7 @@ pub const Catalog = struct {
if (permissions.rulesDenyAllTargetsForPermission(permission_rules, name)) return .unavailable;
const current = c.bindingLocked(found.server, found.tool.*);
if (std.meta.eql(known_binding, current)) return .current;
const instructions = try c.runtime.instructions(alloc, found.server);
const instructions = try c.runtime.instructions(alloc, c.servers[found.server].name);
defer if (instructions) |text| alloc.free(text);
var projection = try selected_schema.project(alloc, found.tool.*, instructions, limits);
if (projection == .rejected) {
@@ -352,7 +422,10 @@ pub const Catalog = struct {
const statuses = try c.runtime.statuses(scratch.allocator());
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
for (c.runtime.servers(), c.lists, statuses) |s, l, status| {
for (c.servers, c.lists) |s, l| {
// The runtime may have moved on to a reload the catalog hasn't
// taken yet; a server it no longer has is left out.
const status = statusOf(statuses, s.name) orelse continue;
switch (status.state) {
.waiting_for_approval, .rejected, .disabled, .unsupported, .missing_env, .invalid_env => continue,
else => {},
@@ -385,7 +458,7 @@ pub const Catalog = struct {
alloc.free(server_slice);
}
return .{
.runtime_generation = c.generation,
.runtime_generation = c.currentGeneration(),
.owner_id = owner,
.parent_id = parent,
.servers = server_slice,
@@ -406,7 +479,8 @@ pub const Catalog = struct {
}
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
for (statuses, c.lists) |s, l| {
for (c.servers, c.lists) |server, l| {
const s = statusOf(statuses, server.name) orelse continue;
const availability: model_catalog.Availability = switch (s.state) {
.waiting_for_approval, .rejected => continue,
.disabled => .disabled,
@@ -450,9 +524,9 @@ pub const Catalog = struct {
}
fn load(c: *Catalog, name: []const u8, cancel: Cancel) !void {
for (c.runtime.servers(), 0..) |s, i| {
for (c.currentServers()) |s| {
if (s.held != null or !tool_names.matchesServer(name, s.name)) continue;
try c.refresh(i, cancel);
try c.refresh(s.name, cancel);
if (c.known(name)) return;
}
}
@@ -469,7 +543,7 @@ pub const Catalog = struct {
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
const found = c.findLocked(name) orelse return false;
break :found .{ found.server, c.lists[found.server].epoch };
break :found .{ c.servers[found.server].name, c.lists[found.server].epoch };
};
return epoch == c.runtime.toolsEpoch(server);
}
@@ -480,7 +554,7 @@ pub const Catalog = struct {
return switch (access) {
.unrestricted => true,
.disabled => false,
.scoped => |scope| scope.captured.runtime_generation == c.generation,
.scoped => |scope| scope.captured.runtime_generation == c.currentGeneration(),
};
}
@@ -492,7 +566,7 @@ pub const Catalog = struct {
if (!c.mayAllow(access)) return false;
return switch (access) {
.scoped => |scope| scope.captured.tool(name) != null or
scope.captured.server(c.runtime.servers()[server].name) != null,
scope.captured.server(c.servers[server].name) != null,
else => true,
};
}
@@ -503,7 +577,7 @@ pub const Catalog = struct {
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
const found = c.findLocked(name) orelse return null;
const server = try alloc.dupe(u8, c.runtime.servers()[found.server].name);
const server = try alloc.dupe(u8, c.servers[found.server].name);
errdefer alloc.free(server);
const tool = try alloc.dupe(u8, found.tool.original_name);
errdefer alloc.free(tool);
@@ -529,7 +603,7 @@ pub const Catalog = struct {
fn bindingLocked(c: *const Catalog, server: usize, tool: McpTool) Binding {
const generation = c.lists[server].generation;
return .{
.runtime_generation = c.generation,
.runtime_generation = c.currentGeneration(),
.connection_generation = generation,
.catalog_generation = generation,
.auth_generation = 0,
@@ -537,15 +611,21 @@ pub const Catalog = struct {
};
}
/// Gets server `i`'s list from the runtime and replaces the catalog's copy.
fn refresh(c: *Catalog, i: usize, cancel: Cancel) !void {
const server = c.runtime.servers()[i];
/// Gets server `name`'s list from the runtime and replaces the catalog's
/// copy, unless a reload replaced that list meanwhile.
fn refresh(c: *Catalog, name: []const u8, cancel: Cancel) !void {
const server, const token = found: {
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
const i = c.indexLocked(name) orelse return;
break :found .{ c.servers[i], c.lists[i].token };
};
if (server.held != null) return;
var scratch: std.heap.ArenaAllocator = .init(c.gpa);
defer scratch.deinit();
// Read first: a change that comes while the list is on its way
// leaves this list out of date.
const epoch = c.runtime.toolsEpoch(i);
const epoch = c.runtime.toolsEpoch(server.name);
const result = try c.runtime.tools(scratch.allocator(), server.name, cancel);
switch (result) {
.tools => |listed| {
@@ -555,8 +635,11 @@ pub const Catalog = struct {
const digest = catalog_state.digestTools(tools);
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
const l = &c.lists[i];
if (l.digest == null or !std.mem.eql(u8, &l.digest.?, &digest)) l.generation += 1;
const l = c.listLocked(server.name, token) orelse {
arena.deinit();
return;
};
if (l.digest == null or !std.mem.eql(u8, &l.digest.?, &digest)) l.generation = c.serial();
if (l.arena) |*old| old.deinit();
l.arena = arena;
l.tools = tools;
@@ -567,12 +650,12 @@ pub const Catalog = struct {
.needs_login => {
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
c.setStateLocked(&c.lists[i], .needs_login, null);
c.setStateLocked(c.listLocked(server.name, token) orelse return, .needs_login, null);
},
.failed => |why| {
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
c.setStateLocked(&c.lists[i], .failed, why);
c.setStateLocked(c.listLocked(server.name, token) orelse return, .failed, why);
},
.unknown_server => {},
}
@@ -637,7 +720,7 @@ pub const Catalog = struct {
/// Why `name` can't be selected when its server's list didn't load, so
/// the model hears more than "not found".
fn whyUnavailableLocked(c: *Catalog, alloc: Allocator, name: []const u8) !?tool_mcp_runtime.ToolSchemaResult {
for (c.runtime.servers(), c.lists) |s, l| {
for (c.servers, c.lists) |s, l| {
if (!tool_names.matchesServer(name, s.name)) continue;
const text = try c.unavailableText(alloc, s, l) orelse continue;
return .{ .rejected = .{ .model_output = text } };
@@ -651,7 +734,7 @@ pub const Catalog = struct {
const named = request.server orelse request.query.raw;
c.mutex.lockUncancelable(c.io);
defer c.mutex.unlock(c.io);
for (c.runtime.servers(), c.lists) |s, l| {
for (c.servers, c.lists) |s, l| {
if (!tool_search.queryContainsCompleteIdentity(named, s.name)) continue;
if (l.state == .needs_login) {
const step = try signInStep(alloc, s.name, signIn(c.runtime.options.slash_commands, s.source));
@@ -666,7 +749,7 @@ pub const Catalog = struct {
}
}
const name = request.server orelse return null;
for (c.runtime.servers(), c.lists) |s, l| {
for (c.servers, c.lists) |s, l| {
if (!std.mem.eql(u8, s.name, name) or l.state != .failed) continue;
return try tool_search.writeServerFailure(alloc, s.name, l.failure orelse "the server is unavailable");
}
@@ -743,6 +826,11 @@ const Asking = struct {
/// Turns the runtime's answer into the result the model sees, through fx's
/// result projection: content, images, size limits, and errors.
fn statusOf(statuses: []const runtime_mod.Status, name: []const u8) ?runtime_mod.Status {
for (statuses) |s| if (std.mem.eql(u8, s.name, name)) return s;
return null;
}
fn render(arena: Allocator, server: []const u8, name: []const u8, answer: runtime_mod.CallResult, max_bytes: usize, how: SignIn) !?tool_mcp_runtime.CallResult {
const response = switch (answer) {
.result => |r| if (r.structured) |structured|
@@ -872,7 +960,7 @@ test "a subagent made before its parent's server starts may use the server's too
try testing.expectEqual(@as(usize, 0), view.tools.len);
const child: Access = .{ .scoped = .{ .captured = &view, .admission_authority_generation = 0, .live = undefined } };
try testing.expect(c.hasTool("mcp_fake_echo", child));
try testing.expectEqual(c.generation, (try c.validateTool(a, "mcp_fake_echo", "{}", child)).valid);
try testing.expectEqual(c.currentGeneration(), (try c.validateTool(a, "mcp_fake_echo", "{}", child)).valid);
const result = (try c.callTool(a, "mcp_fake_echo", "{}", 4096, .{ .access = child })).?;
try testing.expectEqual(tool_mcp_runtime.CallStatus.success, result.status);
@@ -914,14 +1002,14 @@ test "an MCP server's tools reach the model under fx's names" {
try testing.expect(std.mem.find(u8, found.model_output, "\"name\":\"mcp_fake_echo\",\"server\":\"fake\"") != null);
try testing.expectEqual(@as(usize, 1), found.selected_tools.len);
const binding = found.selected_tools[0].mcp_binding.?;
try testing.expectEqual(c.generation, binding.runtime_generation);
try testing.expectEqual(c.currentGeneration(), binding.runtime_generation);
const schema = (try c.toolSchema(a, "mcp_fake_echo", .{}, .{}, .unrestricted, null)).?;
try testing.expect(std.meta.eql(binding, schema.selected.mcp_binding.?));
try testing.expect(std.mem.find(u8, schema.selected.model_output, "mcp_fake_echo") != null);
try testing.expect((try c.snapshotToolDefinition(a, "mcp_fake_echo", binding, .{}, .{}, .unrestricted)) == .current);
try testing.expectEqual(c.generation, (try c.validateTool(a, "mcp_fake_echo", "{}", .unrestricted)).valid);
try testing.expectEqual(c.currentGeneration(), (try c.validateTool(a, "mcp_fake_echo", "{}", .unrestricted)).valid);
try testing.expect((try c.validateTool(a, "mcp_fake_echo", "[", .unrestricted)) == .invalid);
var capture: @import("../agent/runtime/observation_testing.zig").Capture = .{ .expected_call = "mcp-call" };
@@ -942,13 +1030,51 @@ test "an MCP server's tools reach the model under fx's names" {
var view = try c.snapshotAccessView(testing.allocator, "child", "parent", .{});
defer view.deinit(testing.allocator);
try testing.expectEqual(@as(usize, 1), view.tools.len);
try testing.expect(Access.allowsTool(.{ .scoped = .{ .captured = &view, .admission_authority_generation = 0, .live = undefined } }, c.generation, "mcp_fake_echo"));
try testing.expect(Access.allowsTool(.{ .scoped = .{ .captured = &view, .admission_authority_generation = 0, .live = undefined } }, c.currentGeneration(), "mcp_fake_echo"));
var snapshot = try c.modelSnapshot(testing.allocator);
defer snapshot.deinit(testing.allocator);
try testing.expectEqual(model_catalog.Availability.ready, snapshot.servers[0].availability);
try testing.expectEqual(@as(?usize, 1), snapshot.servers[0].tool_count);
}
test "a reload keeps a kept server's list and the bindings made from it, until a server is dropped" {
var f: Fixture = undefined;
try f.init();
defer f.deinit();
const fake: mcp_contract.McpServerConfig = .{ .name = "fake", .command = f.server };
const gone: mcp_contract.McpServerConfig = .{ .name = "gone", .command = f.server };
const late: mcp_contract.McpServerConfig = .{ .name = "late", .command = f.server };
var r: Runtime = undefined;
try r.start(testing.allocator, testing.io, &.{ fake, gone }, &f.inherited, f.options());
defer r.deinit();
var c: Catalog = undefined;
try c.init(testing.allocator, testing.io, &r, .{});
defer c.deinit();
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const a = arena.allocator();
const binding = (try c.toolSchema(a, "mcp_fake_echo", .{}, .{}, .unrestricted, null)).?.selected.mcp_binding.?;
try testing.expect(c.hasTool("mcp_gone_echo", .unrestricted));
const generation = c.currentGeneration();
// Adding a server keeps what the model was shown, and the lists loaded.
try c.reload(try r.reload(&.{ fake, gone, late }, &f.inherited));
try testing.expectEqual(generation, c.currentGeneration());
try testing.expectEqual(@as(usize, 2), (try c.snapshotToolNames(a, .{})).len);
try testing.expect((try c.snapshotToolDefinition(a, "mcp_fake_echo", binding, .{}, .{}, .unrestricted)) == .current);
const result = (try c.callTool(a, "mcp_fake_echo", "{}", 4096, .{ .expected_binding = binding })).?;
try testing.expectEqual(tool_mcp_runtime.CallStatus.success, result.status);
// Dropping one ends every binding made before, as a new host would.
try c.reload(try r.reload(&.{ fake, late }, &f.inherited));
try testing.expect(generation != c.currentGeneration());
try testing.expectError(error.McpAdvertisedToolChanged, c.callTool(a, "mcp_fake_echo", "{}", 4096, .{ .expected_binding = binding }));
try testing.expect((try c.callTool(a, "mcp_gone_echo", "{}", 4096, .{})) == null);
// The kept server's list stays without asking it again.
try testing.expectEqual(@as(usize, 1), (try c.snapshotToolNames(a, .{})).len);
try testing.expect(c.hasTool("mcp_late_echo", .unrestricted));
}
test "a search says why a server it names has no tools, and calls that end say how" {
var f: Fixture = undefined;
try f.init();
+533 -69
View File
@@ -236,8 +236,9 @@ const Request = struct {
const Kind = union(enum) {
tools: []const u8,
/// Starts a server's tool list without anyone waiting for it.
need: usize,
/// Starts a server's tool list without anyone waiting for it. The
/// name is one of `servers()`, which outlive the runtime.
need: []const u8,
call: struct { server: []const u8, tool: []const u8, arguments: ?[]const u8 },
cancel: *Request,
login: struct { server: []const u8, open_browser: bool },
@@ -247,12 +248,26 @@ const Request = struct {
pick_up,
/// What the user answered, posted by the caller that asked.
answer: struct { pending: Pending, json: ?[]u8 },
/// The config changed: see `reload`.
reload: struct { configs: []const mcp_contract.McpServerConfig, inherited: *const std.process.Environ.Map },
/// Whether handling it could start a server.
fn mayStart(k: Kind) bool {
return switch (k) {
.tools, .need, .call, .login, .finish_login, .pick_up => true,
.cancel, .logout, .answer => false,
.cancel, .logout, .answer, .reload => false,
};
}
/// The server it could start, if one.
fn starts(k: Kind) ?[]const u8 {
return switch (k) {
.tools => |name| name,
.need => |name| name,
.call => |c| c.server,
.login => |l| l.server,
.finish_login => |f| f.server,
.pick_up, .cancel, .logout, .answer, .reload => null,
};
}
};
@@ -263,13 +278,19 @@ const Request = struct {
failed: Error,
tools: ToolsResult,
call: CallResult,
reloaded: Reloaded,
};
};
/// A question the owner hands a calling thread, with what it takes to
/// answer it. Owned by the runtime's allocator.
const Pending = struct {
server: usize,
/// The server's name, which outlives the runtime, so the calling thread
/// reads it without the owner.
name: []const u8,
/// Which run of that server asked (`Runtime.ids`): an answer to a server
/// a reload replaced reaches no one.
server: u64,
json: []u8,
version: ?[]u8,
reply: union(enum) {
@@ -312,7 +333,10 @@ const url_elicitation_required: i64 = -32042;
const Login = struct {
runtime: *Runtime,
/// Its server's index, which a reload moves; owner thread only.
server: usize,
/// Its server's name, which outlives the runtime, for the login task.
name: []const u8,
listener: loopback.Listener,
group: std.Io.Group = .init,
deadline_ms: i64,
@@ -325,20 +349,45 @@ const Login = struct {
url: ?[]u8 = null,
};
/// What a `reload` did.
pub const Reloaded = struct {
/// Per server in the new `servers()`: it kept running as it was, so what
/// it listed still holds.
kept: []const bool,
/// A server that ran before was removed, held back, or replaced by one
/// of the same name.
dropped: bool,
};
pub const Runtime = struct {
gpa: Allocator,
io: std.Io,
options: Options,
/// Lives until `deinit`: every server list a reload makes stays, since
/// other threads may still read the one they had. Only the list and its
/// changed strings are kept; the rest of a config goes once the engine
/// copied it.
arena: std.heap.ArenaAllocator,
/// Replaced by `reload` on the owner thread under `mutex`; other threads
/// read it under `mutex` (`servers`).
resolved: config.Resolved,
engine: engine.Engine,
/// Per server in `resolved.servers`; owner thread only.
/// Per server in `resolved.servers`; owner thread only. `gpa`'s, as are
/// `ids` and `tools_epochs`, each replaced by a reload.
flags: []Flags,
/// Per server in `resolved.servers`: which run of it this is, new for
/// each server a reload adds or replaces. Owner thread only.
ids: []u64,
next_id: u64 = 1,
/// Each server's count of `tools_stale` events, so the catalog can tell
/// a list it holds from one the server may have changed since. Apart
/// from `flags`, which a sign-in or sign-out resets, so a count never
/// comes back to a value the catalog already saw.
/// comes back to a value the catalog already saw. Replaced with
/// `resolved`.
tools_epochs: []std.atomic.Value(u64),
/// Owner thread only: the last `drain` held a request while a server a
/// reload dropped stops, so the owner looks again soon.
holding: bool = false,
thread: ?std.Thread = null,
/// Set by `retire`; read by the owner thread.
retired: std.atomic.Value(bool) = .init(false),
@@ -417,24 +466,32 @@ pub const Runtime = struct {
.resolved = undefined,
.engine = undefined,
.flags = &.{},
.ids = &.{},
.tools_epochs = &.{},
.snapshot_arena = .init(gpa),
};
errdefer r.arena.deinit();
errdefer r.snapshot_arena.deinit();
const arena = r.arena.allocator();
r.resolved = try config.resolve(arena, configs, inherited, r.options.workspace_root);
r.flags = try arena.alloc(Flags, r.resolved.servers.len);
var scratch: std.heap.ArenaAllocator = .init(gpa);
defer scratch.deinit();
const resolved = try config.resolve(scratch.allocator(), configs, inherited, r.options.workspace_root);
const n = resolved.servers.len;
r.resolved = .{ .servers = try config.keepServers(arena, resolved.servers, &.{}), .engine = &.{}, .engine_server = try arena.dupe(usize, resolved.engine_server) };
r.flags = try gpa.alloc(Flags, n);
errdefer gpa.free(r.flags);
for (r.flags) |*f| f.* = .{};
r.tools_epochs = try arena.alloc(std.atomic.Value(u64), r.resolved.servers.len);
r.ids = try gpa.alloc(u64, n);
errdefer gpa.free(r.ids);
for (r.ids) |*id| id.* = r.newId();
r.tools_epochs = try gpa.alloc(std.atomic.Value(u64), n);
errdefer gpa.free(r.tools_epochs);
for (r.tools_epochs) |*e| e.* = .init(0);
// The engine copies what it keeps, so the rest of the config goes.
try r.engine.init(io, gpa, .{ .context = r, .allow = allow, .log = log }, .{
.client = .{ .info = .{ .name = "fx", .version = options.client_version }, .capabilities = declaredCapabilities(options.questions) },
}, r.resolved.engine);
}, resolved.engine);
errdefer r.engine.deinit();
// The engine has its own copies now, and these may point into
// `configs`, which the caller may free now.
r.resolved.engine = &.{};
try r.publish();
r.thread = try std.Thread.spawn(.{}, run, .{r});
}
@@ -472,6 +529,9 @@ pub const Runtime = struct {
r.required.deinit(r.gpa);
r.held.deinit(r.gpa);
r.finished.deinit(r.gpa);
r.gpa.free(r.flags);
r.gpa.free(r.ids);
r.gpa.free(r.tools_epochs);
r.snapshot_arena.deinit();
r.arena.deinit();
const was_retired = r.retired.load(.acquire);
@@ -509,28 +569,35 @@ pub const Runtime = struct {
return out;
}
/// Server `index`'s instructions (in `servers()` order), copied into
/// `alloc`; null when it gave none.
pub fn instructions(r: *Runtime, alloc: Allocator, index: usize) Allocator.Error!?[]const u8 {
/// Server `name`'s instructions, copied into `alloc`; null when it gave
/// none or there is no such server.
pub fn instructions(r: *Runtime, alloc: Allocator, name: []const u8) Allocator.Error!?[]const u8 {
_ = r.active.fetchAdd(1, .acq_rel);
defer _ = r.active.fetchSub(1, .acq_rel);
r.mutex.lockUncancelable(r.io);
defer r.mutex.unlock(r.io);
if (index >= r.snapshot.len) return null;
const text = r.snapshot[index].instructions orelse return null;
return try alloc.dupe(u8, text);
for (r.snapshot) |s| if (std.mem.eql(u8, s.name, name)) {
const text = s.instructions orelse return null;
return try alloc.dupe(u8, text);
};
return null;
}
/// Server `index`'s count of the times its tools may have changed (in
/// `servers()` order). Read before asking for a list: a list asked for
/// Server `name`'s count of the times its tools may have changed; 0 for
/// none by that name. Read before asking for a list: a list asked for
/// under an older count is out of date.
pub fn toolsEpoch(r: *const Runtime, index: usize) u64 {
return r.tools_epochs[index].load(.acquire);
pub fn toolsEpoch(r: *Runtime, name: []const u8) u64 {
r.mutex.lockUncancelable(r.io);
defer r.mutex.unlock(r.io);
const i = r.serverIndex(name) orelse return 0;
return r.tools_epochs[i].load(.acquire);
}
/// The resolved servers, in config order; they don't change while the
/// runtime runs.
pub fn servers(r: *const Runtime) []const config.Server {
/// The resolved servers, in config order. A reload replaces the list,
/// but the one returned, and its names, stay valid until `deinit`.
pub fn servers(r: *Runtime) []const config.Server {
r.mutex.lockUncancelable(r.io);
defer r.mutex.unlock(r.io);
return r.resolved.servers;
}
@@ -546,17 +613,34 @@ pub const Runtime = struct {
};
}
/// Starts getting the tool list of `resolved.servers[index]` without
/// waiting, so several servers start at once; `tools` then waits less.
pub fn prepare(r: *Runtime, index: usize) Error!void {
/// Starts getting server `name`'s tool list without waiting, so several
/// servers start at once; `tools` then waits less. `name` is one of
/// `servers()`' names, which outlive the runtime.
pub fn prepare(r: *Runtime, name: []const u8) Error!void {
const req = try r.gpa.create(Request);
req.* = .{ .kind = .{ .need = index }, .alloc = r.gpa, .owned = true };
req.* = .{ .kind = .{ .need = name }, .alloc = r.gpa, .owned = true };
r.post(req) catch |err| {
r.gpa.destroy(req);
return err;
};
}
/// Takes the changed config in place: a server whose config is the same
/// keeps running, with its connection, calls, sign-in, and tools. The
/// rest go: a removed one stops, its calls ending cancelled, and one
/// added or changed starts fresh, once that name's old process is gone.
/// `configs` and `inherited` are read only during the call. The host's
/// lifecycle is modeled in `tests/tla/mcp_host/HostLifecycle.tla`.
pub fn reload(r: *Runtime, configs: []const mcp_contract.McpServerConfig, inherited: *const std.process.Environ.Map) Error!Reloaded {
var req: Request = .{ .kind = .{ .reload = .{ .configs = configs, .inherited = inherited } }, .alloc = r.gpa };
try r.submit(&req, null);
return switch (req.result) {
.reloaded => |done| done,
.failed => |err| err,
else => unreachable,
};
}
/// Calls `tool` with a JSON object (or null), after fx's permission
/// allowed it; blocks until it ends. Setting `cancel` cancels it. The
/// server's questions go to `asker`, and are declined without one.
@@ -672,12 +756,12 @@ pub const Runtime = struct {
var scratch: std.heap.ArenaAllocator = .init(r.gpa);
defer scratch.deinit();
const answer = asker.ask(asker.context, scratch.allocator(), .{
.server = r.resolved.servers[pending.server].name,
.server = pending.name,
.input_requests_json = pending.json,
.version = pending.version,
.url_wait = if (pending.reply == .url_wait) &req.url_signal else null,
}) catch |err| answer: {
debug_trace.logf("mcp", "MCP question not answered server={s} err={s}", .{ r.resolved.servers[pending.server].name, @errorName(err) });
debug_trace.logf("mcp", "MCP question not answered server={s} err={s}", .{ pending.name, @errorName(err) });
break :answer null;
};
const reply = r.gpa.create(Request) catch return pending.deinit(r.gpa);
@@ -727,37 +811,63 @@ pub const Runtime = struct {
/// Takes the queued requests and handles them; true once stopping.
fn drain(r: *Runtime) bool {
r.mutex.lockUncancelable(r.io);
const stopping = r.stopping;
var taken = r.queue;
r.queue = .empty;
if (!stopping and r.waitsForRetired()) {
// A retired runtime's servers are still stopping: what could
// start a server waits, in order.
var kept: usize = 0;
r.holding = false;
while (true) {
r.mutex.lockUncancelable(r.io);
const stopping = r.stopping;
var taken = r.queue;
r.queue = .empty;
var after_reload = false;
if (!stopping) {
// A reload ends its batch: what came after it waits for the
// next one, so it is held against the servers it dropped.
var batch = taken.items.len;
for (taken.items, 0..) |req, k| if (req.kind == .reload) {
batch = k + 1;
break;
};
const retiring_now = r.waitsForRetired();
var kept: usize = 0;
for (taken.items[0..batch]) |req| {
// What could start a server waits, in order, while a
// retired runtime's servers stop, or while the process a
// reload dropped under that server's name does, so a
// server never runs twice.
const dropped_stopping = if (req.kind.starts()) |name| r.engine.stopping(name) else false;
if ((retiring_now and req.kind.mayStart()) or dropped_stopping) held: {
r.queue.append(r.gpa, req) catch break :held; // out of memory: handled now
if (dropped_stopping) r.holding = true;
continue;
}
taken.items[kept] = req;
kept += 1;
}
const rest = taken.items[batch..];
if (rest.len > 0) {
if (r.queue.appendSlice(r.gpa, rest)) |_| {
after_reload = true;
} else |_| {
// Out of memory: handled now, after the reload.
@memmove(taken.items[kept..][0..rest.len], rest);
kept += rest.len;
}
}
taken.shrinkRetainingCapacity(kept);
}
r.mutex.unlock(r.io);
defer taken.deinit(r.gpa);
for (taken.items) |req| {
if (req.kind.mayStart()) held: {
r.queue.append(r.gpa, req) catch break :held; // out of memory: handled now
if (stopping) {
r.finish(req, .{ .failed = error.Stopped });
continue;
}
taken.items[kept] = req;
kept += 1;
r.handle(req) catch |err| {
debug_trace.logf("mcp", "MCP request failed kind={s} err={s}", .{ @tagName(req.kind), @errorName(err) });
r.finish(req, .{ .failed = if (err == error.OutOfMemory) error.OutOfMemory else error.Failed });
};
}
taken.shrinkRetainingCapacity(kept);
if (stopping or !after_reload) return stopping;
}
r.mutex.unlock(r.io);
defer taken.deinit(r.gpa);
for (taken.items) |req| {
if (stopping) {
r.finish(req, .{ .failed = error.Stopped });
continue;
}
r.handle(req) catch |err| {
debug_trace.logf("mcp", "MCP request failed kind={s} err={s}", .{ @tagName(req.kind), @errorName(err) });
r.finish(req, .{ .failed = if (err == error.OutOfMemory) error.OutOfMemory else error.Failed });
};
}
return stopping;
}
/// Takes `req` out of the queue; false when it isn't there.
@@ -789,12 +899,13 @@ pub const Runtime = struct {
};
r.finish(req, .ok);
},
.need => |i| {
if (r.resolved.servers[i].held == null) r.engine.needTools(r.resolved.servers[i].name) catch |err| {
debug_trace.logf("mcp", "MCP tool list not started server={s} err={s}", .{ r.resolved.servers[i].name, @errorName(err) });
.need => |name| {
if (r.engineServer(name) != null) r.engine.needTools(name) catch |err| {
debug_trace.logf("mcp", "MCP tool list not started server={s} err={s}", .{ name, @errorName(err) });
};
r.finish(req, .ok);
},
.reload => |c| r.finish(req, .{ .reloaded = try r.applyReload(c.configs, c.inherited) }),
.call => |c| {
const i = r.serverIndex(c.server) orelse return r.finish(req, .{ .call = .unknown_server });
if (r.resolved.servers[i].held != null) return r.finish(req, .{ .failed = error.Held });
@@ -846,7 +957,7 @@ pub const Runtime = struct {
.answer => |a| {
// `finish` frees the pending question and the answer.
defer r.finish(req, .ok);
const name = r.resolved.servers[a.pending.server].name;
const name = a.pending.name;
var scratch: std.heap.ArenaAllocator = .init(r.gpa);
defer scratch.deinit();
switch (a.pending.reply) {
@@ -860,6 +971,13 @@ pub const Runtime = struct {
}
},
.elicit => |id| {
// Elicitation ids are per run of a server: the one a
// reload replaced never hears it, and nor does its
// successor.
if (r.serverById(a.pending.server) == null) {
debug_trace.logf("mcp", "MCP elicitation answer dropped server={s} reason=the server was reloaded", .{name});
return;
}
const answer = elicitAnswer(scratch.allocator(), a.json) orelse mcp.elicitation.Answer{ .action = .decline };
r.engine.answerElicitation(name, id, answer) catch |err| {
debug_trace.logf("mcp", "MCP elicitation answer not sent server={s} err={s}", .{ name, @errorName(err) });
@@ -966,6 +1084,7 @@ pub const Runtime = struct {
l.* = .{
.runtime = r,
.server = i,
.name = s.name,
.listener = listener,
.deadline_ms = r.nowMs() + login_timeout_ms,
.open_browser = open_browser,
@@ -996,7 +1115,7 @@ pub const Runtime = struct {
const url = l.listener.accept(r.io, r.gpa) catch |err| switch (err) {
error.Canceled => return error.Canceled,
else => {
debug_trace.logf("mcp", "MCP login listener failed server={s} err={s}", .{ r.resolved.servers[l.server].name, @errorName(err) });
debug_trace.logf("mcp", "MCP login listener failed server={s} err={s}", .{ l.name, @errorName(err) });
return;
},
};
@@ -1005,7 +1124,7 @@ pub const Runtime = struct {
return;
};
req.* = .{
.kind = .{ .finish_login = .{ .server = r.resolved.servers[l.server].name, .url = url } },
.kind = .{ .finish_login = .{ .server = l.name, .url = url } },
.alloc = r.gpa,
.owned = true,
.owned_url = url,
@@ -1226,7 +1345,7 @@ pub const Runtime = struct {
return r.engine.answerElicitation(name, x.id, cancel);
};
const json = try r.gpa.print("{{\"elicitation\":{{\"method\":\"elicitation/create\",\"params\":{s}}}}}", .{x.params});
var pending: Pending = .{ .server = i, .json = json, .version = r.versionCopy(i), .reply = .{ .elicit = x.id } };
var pending: Pending = .{ .name = r.resolved.servers[i].name, .server = r.ids[i], .json = json, .version = r.versionCopy(i), .reply = .{ .elicit = x.id } };
if (!r.offer(req, pending)) {
pending.deinit(r.gpa);
return r.engine.answerElicitation(name, x.id, cancel);
@@ -1254,7 +1373,7 @@ pub const Runtime = struct {
defer r.mutex.unlock(r.io);
var it = r.calls.valueIterator();
while (it.next()) |req| if (req.*.question) |*q| {
if (q.server != i or q.reply != .elicit or q.reply.elicit.slot != slot) continue;
if (q.server != r.ids[i] or q.reply != .elicit or q.reply.elicit.slot != slot) continue;
debug_trace.logf("mcp", "MCP question dropped server={s} reason=withdrawn before it was shown", .{r.resolved.servers[i].name});
q.deinit(r.gpa);
req.*.question = null;
@@ -1353,7 +1472,8 @@ pub const Runtime = struct {
}
out.writer.writeByte('}') catch return error.OutOfMemory;
const pending: Pending = .{
.server = req.url_server,
.name = r.resolved.servers[req.url_server].name,
.server = r.ids[req.url_server],
.json = try out.toOwnedSlice(),
.version = r.versionCopy(req.url_server),
.reply = if (phase == .consent) .{ .url_consent = req } else .{ .url_wait = req },
@@ -1494,6 +1614,202 @@ pub const Runtime = struct {
}
}
// ---- reload ----
/// The owner's side of `reload`. What can fail before a server is touched
/// comes first; only the engine's own reload can fail after that.
fn applyReload(r: *Runtime, configs: []const mcp_contract.McpServerConfig, inherited: *const std.process.Environ.Map) !Reloaded {
var scratch: std.heap.ArenaAllocator = .init(r.gpa);
defer scratch.deinit();
const s = scratch.allocator();
const next = try config.resolve(s, configs, inherited, r.options.workspace_root);
const n = next.servers.len;
// What a caller reads after the reload: the servers, and which kept running.
const arena = r.arena.allocator();
const kept_servers = try config.keepServers(arena, next.servers, r.resolved.servers);
const engine_server = try arena.dupe(usize, next.engine_server);
const kept = try arena.alloc(bool, n);
@memset(kept, false);
for (next.engine, next.engine_server) |c, j| kept[j] = r.engine.keeps(c);
// Each new server's old index, and each old one's new index, for the
// servers that keep running as they were.
const from = try s.alloc(?usize, n);
for (next.servers, kept, from) |server, k, *f| f.* = if (k) r.serverIndex(server.name) else null;
const to = try s.alloc(?usize, r.resolved.servers.len);
@memset(to, null);
for (from, 0..) |f, j| if (f) |i| {
to[i] = j;
};
const flags = try r.gpa.alloc(Flags, n);
errdefer r.gpa.free(flags);
const ids = try r.gpa.alloc(u64, n);
errdefer r.gpa.free(ids);
const epochs = try r.gpa.alloc(std.atomic.Value(u64), n);
errdefer r.gpa.free(epochs);
for (from, flags, ids, epochs) |f, *flag, *id, *epoch| {
flag.* = if (f) |i| r.flags[i] else .{};
id.* = if (f) |i| r.ids[i] else r.newId();
epoch.* = .init(if (f) |i| r.tools_epochs[i].load(.acquire) else 0);
}
var dropped = false;
for (r.resolved.engine_server) |i| if (to[i] == null) {
dropped = true;
r.dropServer(i, next.servers);
};
// The engine copies what it keeps; the rest goes with `scratch`.
try r.engine.reload(next.engine);
r.moveServers(to);
r.mutex.lockUncancelable(r.io);
const old_flags = r.flags;
const old_ids = r.ids;
const old_epochs = r.tools_epochs;
r.resolved = .{ .servers = kept_servers, .engine = &.{}, .engine_server = engine_server };
r.flags = flags;
r.ids = ids;
r.tools_epochs = epochs;
r.mutex.unlock(r.io);
// Other threads read the epochs only under `mutex`.
r.gpa.free(old_flags);
r.gpa.free(old_ids);
r.gpa.free(old_epochs);
var added: usize = 0;
for (engine_server) |j| if (from[j] == null) {
added += 1;
r.adoptSaved(j);
};
var same: usize = 0;
for (kept) |k| same += @intFromBool(k);
debug_trace.logf("mcp", "MCP config reloaded servers={d} kept={d} fresh={d} dropped={s}", .{ n, same, added, if (dropped) "yes" else "no" });
return .{ .kept = kept, .dropped = dropped };
}
/// Old server `i` doesn't run on as it was: removed, held back, or
/// replaced. Its calls end, as they may have run; a tool list asked of it
/// is asked again of whatever has its name now; its sign-in, logout,
/// refresh, and URL questions end. `next` is the servers after the reload.
fn dropServer(r: *Runtime, i: usize, next: []const config.Server) void {
const name = r.resolved.servers[i].name;
const successor: ?config.Server = for (next) |s| {
if (std.mem.eql(u8, s.name, name)) break s;
} else null;
const why = if (successor == null) "removed" else if (successor.?.held != null) "held back" else "changed";
debug_trace.logf("mcp", "MCP server stops for a reload server={s} reason={s}", .{ name, why });
while (r.callOn(name)) |waiting| {
_ = r.calls.remove(waiting.handle);
r.finish(waiting.request, .{ .call = .{ .ended = .{
.outcome = .cancelled,
.maybe_ran = r.engine.callSent(waiting.handle),
.timed_out = false,
.why = "the MCP config changed, and fx stopped the server",
} } });
}
var k: usize = 0;
while (k < r.tool_waiters.items.len) {
const w = r.tool_waiters.items[k];
if (w.server != i) {
k += 1;
continue;
}
_ = r.tool_waiters.swapRemove(k);
r.mutex.lockUncancelable(r.io);
const queued = if (r.queue.append(r.gpa, w.request)) |_| true else |_| false;
r.mutex.unlock(r.io);
if (queued) r.engine.wake() else r.finish(w.request, .{ .failed = error.OutOfMemory });
}
if (r.loginFor(i) != null) {
var buf: [320]u8 = undefined;
const reason = if (successor != null and successor.?.held == null)
std.mem.print(&buf, "the MCP config changed before the browser came back; run /mcp login {s} again", .{name}) catch "the MCP config changed before the browser came back"
else
"the MCP config changed before the browser came back";
r.notify(.{ .sign_in_failed = .{ .server = name, .reason = reason } });
r.endLogin(i);
}
k = 0;
while (k < r.logouts.items.len) {
const l = r.logouts.items[k];
if (l.server != i) {
k += 1;
continue;
}
_ = r.logouts.swapRemove(k);
r.finish(l.request, .ok);
}
r.endRefresh(i);
k = 0;
while (k < r.held.items.len) {
const req = r.held.items[k];
if (req.url_server != i) {
k += 1;
continue;
}
r.endHeld(req, .{ .call = .{ .ended = .{ .outcome = .cancelled, .maybe_ran = false, .timed_out = false, .why = "the MCP config changed, and fx stopped the server" } } });
}
k = 0;
while (k < r.required.items.len) {
if (r.required.items[k].server != i) {
k += 1;
continue;
}
r.gpa.free(r.required.orderedRemove(k).need.params);
}
k = 0;
while (k < r.url_questions.items.len) {
if (r.url_questions.items[k].server != i) {
k += 1;
continue;
}
r.gpa.free(r.url_questions.swapRemove(k).id);
}
}
/// Points what waits on a server that kept running at its new index;
/// `dropServer` ended the rest.
fn moveServers(r: *Runtime, to: []const ?usize) void {
for (r.tool_waiters.items) |*w| w.server = to[w.server].?;
for (r.logins.items) |l| l.server = to[l.server].?;
for (r.logouts.items) |*l| l.server = to[l.server].?;
for (r.refreshes.items) |*f| f.server = to[f.server].?;
for (r.held.items) |req| req.url_server = to[req.url_server].?;
for (r.required.items) |*q| q.server = to[q.server].?;
for (r.url_questions.items) |*q| q.server = to[q.server].?;
}
/// A server a reload added signs in with the login saved for it, as
/// `restoreCredentials` does at the start.
fn adoptSaved(r: *Runtime, i: usize) void {
const s = r.resolved.servers[i];
if (s.url == null) return;
const saved = r.options.credentials.store.get(r.gpa, s.credentialKey()) catch |err| {
return debug_trace.logf("mcp", "MCP saved login not read server={s} err={s}", .{ s.name, @errorName(err) });
} orelse return;
defer secret.zeroAndFree(r.gpa, saved);
r.engine.restoreCredential(s.name, saved) catch |err| {
return debug_trace.logf("mcp", "MCP credential not loaded server={s} err={s}", .{ s.name, @errorName(err) });
};
r.flags[i].signed_in = true;
r.flags[i].tried = credentials.digest(saved);
}
/// A call still waiting on server `name`.
fn callOn(r: *Runtime, name: []const u8) ?struct { handle: engine.Handle, request: *Request } {
var it = r.calls.iterator();
while (it.next()) |entry| {
if (std.mem.eql(u8, entry.value_ptr.*.kind.call.server, name)) return .{ .handle = entry.key_ptr.*, .request = entry.value_ptr.* };
}
return null;
}
fn serverById(r: *const Runtime, id: u64) ?usize {
return std.mem.findScalar(u64, r.ids, id);
}
fn newId(r: *Runtime) u64 {
defer r.next_id += 1;
return r.next_id;
}
fn shutdown(r: *Runtime) void {
while (r.logins.items.len > 0) {
// Only a config change ends a host while it signs in: the
@@ -1580,7 +1896,7 @@ pub const Runtime = struct {
for (r.refreshes.items) |f| if (!f.sent) {
wait = @min(wait, refresh_retry_ms);
};
if (r.waitsForRetired()) wait = @min(wait, retiring_wait_ms);
if (r.waitsForRetired() or r.holding) wait = @min(wait, retiring_wait_ms);
return @intCast(wait);
}
@@ -1651,7 +1967,7 @@ pub const Runtime = struct {
}
out.writer.writeByte('}') catch return error.OutOfMemory;
const json = try out.toOwnedSlice();
return .{ .server = i, .json = json, .version = r.versionCopy(i), .reply = .{ .input = .{ .call = call_handle, .keys = keys } } };
return .{ .name = r.resolved.servers[i].name, .server = r.ids[i], .json = json, .version = r.versionCopy(i), .reply = .{ .input = .{ .call = call_handle, .keys = keys } } };
}
fn versionCopy(r: *Runtime, i: usize) ?[]u8 {
@@ -2296,6 +2612,154 @@ test "a replaced runtime still in use serves its calls, and holds the others unt
try testing.expect(done.load(.acquire));
}
/// Server `name`'s process id, as the fake server says it.
fn pidOf(r: *Runtime, a: Allocator, name: []const u8) ![]const u8 {
const result = try r.call(a, name, "echo", "{\"text\":\"pid\"}", null, null);
if (result != .result) return error.TestUnexpectedResult;
const at = std.mem.find(u8, result.result.content, "pid=") orelse return error.TestUnexpectedResult;
const digits = result.result.content[at + 4 ..];
return digits[0 .. std.mem.findNone(u8, digits, "0123456789") orelse digits.len];
}
test "a reload keeps a server whose config is the same on its process, and restarts the changed ones" {
var f: Fixture = undefined;
try f.init();
defer f.deinit();
var one = [_]mcp_contract.McpEnvVar{.{ .key = @constCast("MODE"), .value = @constCast("one") }};
var two = [_]mcp_contract.McpEnvVar{.{ .key = @constCast("MODE"), .value = @constCast("two") }};
const before = [_]mcp_contract.McpServerConfig{
.{ .name = "same", .command = f.server, .env = &one },
.{ .name = "args", .command = f.server },
.{ .name = "env", .command = f.server, .env = &one },
.{ .name = "gone", .command = f.server },
};
var r: Runtime = undefined;
try r.start(testing.allocator, testing.io, &before, &f.inherited, f.options());
defer r.deinit();
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const a = arena.allocator();
try testing.expectEqual(@as(usize, 1), (try r.tools(a, "same", null)).tools.len);
const same = try pidOf(&r, a, "same");
const args = try pidOf(&r, a, "args");
const env = try pidOf(&r, a, "env");
_ = try pidOf(&r, a, "gone");
// The config is read afresh, so each environment is a new map; the same
// variables still count as the same server.
const after = [_]mcp_contract.McpServerConfig{
.{ .name = "added", .command = f.server },
.{ .name = "same", .command = f.server, .env = &one },
.{ .name = "args", .command = f.server, .args = &.{"--new"} },
.{ .name = "env", .command = f.server, .env = &two },
};
const done = try r.reload(&after, &f.inherited);
try testing.expectEqualSlices(bool, &.{ false, true, false, false }, done.kept);
try testing.expect(done.dropped);
const statuses = try r.statuses(a);
try testing.expectEqualStrings("added", statuses[0].name);
try testing.expectEqual(State.idle, statuses[0].state);
// Still up and listed, before anything asks it again.
try testing.expectEqualStrings("same", statuses[1].name);
try testing.expectEqual(State.ready, statuses[1].state);
try testing.expectEqual(@as(usize, 1), statuses[1].tools);
try testing.expectEqualStrings(same, try pidOf(&r, a, "same"));
try testing.expect(!std.mem.eql(u8, args, try pidOf(&r, a, "args")));
try testing.expect(!std.mem.eql(u8, env, try pidOf(&r, a, "env")));
_ = try pidOf(&r, a, "added");
try testing.expectEqual(CallResult.unknown_server, try r.call(a, "gone", "echo", "{}", null, null));
// Taking only an added server drops nothing.
var more: [5]mcp_contract.McpServerConfig = undefined;
@memcpy(more[0..4], &after);
more[4] = .{ .name = "late", .command = f.server };
try testing.expect(!(try r.reload(&more, &f.inherited)).dropped);
try testing.expectEqualStrings(same, try pidOf(&r, a, "same"));
}
test "a reload ends a call on a server it drops, and a call on a server it keeps goes on" {
var f: Fixture = undefined;
try f.init();
defer f.deinit();
const before = [_]mcp_contract.McpServerConfig{
.{ .name = "kept", .command = f.server },
.{ .name = "gone", .command = f.server },
};
var r: Runtime = undefined;
try r.start(testing.allocator, testing.io, &before, &f.inherited, f.options());
defer r.deinit();
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const a = arena.allocator();
_ = try r.tools(a, "kept", null);
_ = try r.tools(a, "gone", null);
const Caller = struct {
server: []const u8,
cancel: std.atomic.Value(bool) = .init(false),
done: std.atomic.Value(bool) = .init(false),
ended: ?@FieldType(CallResult, "ended") = null,
why: [128]u8 = undefined,
why_len: usize = 0,
fn run(c: *@This(), runtime: *Runtime) void {
defer c.done.store(true, .release);
var ca: std.heap.ArenaAllocator = .init(testing.allocator);
defer ca.deinit();
const result = runtime.call(ca.allocator(), c.server, "echo", "{\"text\":\"slow\"}", &c.cancel, null) catch return;
if (result != .ended) return;
c.ended = result.ended;
if (result.ended.why) |why| {
c.why_len = @min(why.len, c.why.len);
@memcpy(c.why[0..c.why_len], why[0..c.why_len]);
}
}
};
var kept: Caller = .{ .server = "kept" };
var gone: Caller = .{ .server = "gone" };
const kept_thread = try std.Thread.spawn(.{}, Caller.run, .{ &kept, &r });
const gone_thread = try std.Thread.spawn(.{}, Caller.run, .{ &gone, &r });
std.Io.sleep(testing.io, .fromMilliseconds(300), .awake) catch {};
const after = [_]mcp_contract.McpServerConfig{.{ .name = "kept", .command = f.server }};
_ = try r.reload(&after, &f.inherited);
gone_thread.join();
try testing.expectEqual(engine.Outcome.cancelled, gone.ended.?.outcome);
// It reached the server, so it may have run there.
try testing.expect(gone.ended.?.maybe_ran);
try testing.expectEqualStrings("the MCP config changed, and fx stopped the server", gone.why[0..gone.why_len]);
// The kept server's call is still out; only its cancel ends it.
std.Io.sleep(testing.io, .fromMilliseconds(200), .awake) catch {};
try testing.expect(!kept.done.load(.acquire));
kept.cancel.store(true, .release);
kept_thread.join();
try testing.expectEqual(engine.Outcome.cancelled, kept.ended.?.outcome);
try testing.expectEqual(@as(usize, 0), kept.why_len);
}
test "a changed server's new process starts only once the old one is gone" {
var f: Fixture = undefined;
try f.init();
defer f.deinit();
const before = [_]mcp_contract.McpServerConfig{.{ .name = "slow-exit", .command = f.server, .args = &.{"linger"} }};
var r: Runtime = undefined;
try r.start(testing.allocator, testing.io, &before, &f.inherited, f.options());
defer r.deinit();
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const a = arena.allocator();
const old = try std.fmt.parseInt(std.posix.pid_t, try pidOf(&r, a, "slow-exit"), 10);
const after = [_]mcp_contract.McpServerConfig{.{ .name = "slow-exit", .command = f.server, .args = &.{ "linger", "again" } }};
const reloaded_ms = std.Io.Clock.awake.now(testing.io).toMilliseconds();
_ = try r.reload(&after, &f.inherited);
const new = try std.fmt.parseInt(std.posix.pid_t, try pidOf(&r, a, "slow-exit"), 10);
try testing.expect(new != old);
// The old one lingers a second after its stdin ends; the new one waited.
try testing.expect(std.Io.Clock.awake.now(testing.io).toMilliseconds() - reloaded_ms >= 900);
try testing.expectError(error.ProcessNotFound, std.posix.kill(old, @fromBackingInt(@intCast(0))));
}
test "an answer to a withdrawn 2025 question reaches no one, and the next question gets its own" {
var f: Fixture = undefined;
try f.init();
+7 -3
View File
@@ -460,6 +460,9 @@ fn writeStatusFields(jw: *std.json.Stringify, s: Status, p: Paths) !void {
/// What loading the config skipped, so a server never vanishes without a reason.
fn writeNotes(c: Context, host: *Host) !void {
// A reload on the UI thread may replace them meanwhile.
host.notes_mutex.lockUncancelable(host.runtime.io);
defer host.notes_mutex.unlock(host.runtime.io);
for (host.notes) |note| try c.err.print("note: {s}\n", .{note});
}
@@ -490,7 +493,8 @@ fn show(c: Context, arena: Allocator, name: []const u8, json: bool) !u8 {
.needs_login, .failed => &.{},
.unknown_server => return missing(c, arena, name),
};
const status = findStatus(try host.runtime.statuses(arena), name).?;
// A reload may have taken it out since: the host takes changes in place.
const status = findStatus(try host.runtime.statuses(arena), name) orelse return missing(c, arena, name);
const p = try paths(c, arena);
if (json) {
var jw: std.json.Stringify = .{ .writer = c.out };
@@ -690,7 +694,7 @@ fn logout(c: Context, arena: Allocator, name: []const u8) !u8 {
try c.err.print("{s} runs a command, so it has no login.\n", .{try safe(arena, name)});
return 1;
}
const was = findStatus(try host.runtime.statuses(arena), name).?;
const was = findStatus(try host.runtime.statuses(arena), name) orelse return missing(c, arena, name);
// There's a login to sign out of when one is saved, by this fx or
// another, or this session holds one. A running server needn't have one.
const saved = try credentials.Store.detect(arena, c.home).get(arena, server.credentialKey());
@@ -773,7 +777,7 @@ fn login(c: Context, arena: Allocator, name: []const u8, browser: bool) !u8 {
}
host.runtime.login(name, open) catch |err| switch (err) {
error.Held => {
const status = findStatus(try host.runtime.statuses(arena), name).?;
const status = findStatus(try host.runtime.statuses(arena), name) orelse return missing(c, arena, name);
try c.err.print("{s} is {s}", .{ try safe(arena, name), try statusText(arena, status) });
if (status.state == .waiting_for_approval) try c.err.print(": {s} approve {s}", .{ c.surface.prog(), try safe(arena, name) });
try c.err.writeAll(".\n");
+68 -3
View File
@@ -26,7 +26,7 @@ pub const Phase = core.Phase;
pub const Handle = core.Handle;
pub const Transport = union(enum) {
/// `environ_map` is the host's and must outlive the engine; `cwd` is
/// The engine keeps its own copy of `environ_map`; `cwd` is
/// where the server starts, the host's own directory when null.
stdio: struct { argv: []const []const u8, environ_map: ?*const std.process.Environ.Map = null, cwd: ?[]const u8 = null },
http: struct { url: []const u8, headers: []const std.http.Header = &.{} },
@@ -185,6 +185,19 @@ pub const Engine = struct {
e.slots = kept;
}
/// Whether `reload` keeps a server with exactly this config running.
pub fn keeps(e: *const Engine, c: Config) bool {
for (e.slots.items) |slot| if (same(slot.config, c)) return true;
return false;
}
/// Whether a server `reload` dropped under this name isn't gone yet, so a
/// new one of that name would run beside it.
pub fn stopping(e: *const Engine, name: []const u8) bool {
for (e.retired.items) |slot| if (std.mem.eql(u8, slot.config.name, name)) return true;
return false;
}
pub fn len(e: *const Engine) usize {
return e.slots.items.len;
}
@@ -256,6 +269,12 @@ pub const Engine = struct {
return handle;
}
/// Whether call `handle` went to its server, so it may have run there.
pub fn callSent(e: *Engine, handle: Handle) bool {
const slot, const i = e.findCall(handle) orelse return false;
return slot.calls.items[i].sent != null;
}
/// Cancels a call: one waiting ends here, one in flight goes to its server.
pub fn cancel(e: *Engine, handle: Handle) Error!void {
const slot, const i = e.findCall(handle) orelse return;
@@ -574,7 +593,7 @@ pub const Engine = struct {
slot.config = .{
.name = try a.dupe(u8, c.name),
.transport = switch (c.transport) {
.stdio => |p| .{ .stdio = .{ .argv = try dupeAll(a, p.argv), .environ_map = p.environ_map, .cwd = if (p.cwd) |dir| try a.dupe(u8, dir) else null } },
.stdio => |p| .{ .stdio = .{ .argv = try dupeAll(a, p.argv), .environ_map = if (p.environ_map) |m| try cloneEnv(a, m) else null, .cwd = if (p.cwd) |dir| try a.dupe(u8, dir) else null } },
.http => |h| blk: {
const headers = try a.alloc(std.http.Header, h.headers.len);
for (h.headers, headers) |from, *to| to.* = .{ .name = try a.dupe(u8, from.name), .value = try a.dupe(u8, from.value) };
@@ -643,12 +662,31 @@ fn dupeAll(a: std.mem.Allocator, items: []const []const u8) ![]const []const u8
return copies;
}
fn cloneEnv(a: std.mem.Allocator, m: *const std.process.Environ.Map) std.mem.Allocator.Error!*const std.process.Environ.Map {
const copy = try a.create(std.process.Environ.Map);
copy.* = try m.clone(a);
return copy;
}
/// The same variables with the same values, whichever map holds them.
fn sameEnv(a: ?*const std.process.Environ.Map, b: ?*const std.process.Environ.Map) bool {
const x = a orelse return b == null;
const y = b orelse return false;
if (x.count() != y.count()) return false;
var it = x.iterator();
while (it.next()) |entry| {
const value = y.get(entry.key_ptr.*) orelse return false;
if (!std.mem.eql(u8, entry.value_ptr.*, value)) return false;
}
return true;
}
fn same(a: Config, b: Config) bool {
if (!std.mem.eql(u8, a.name, b.name) or std.meta.activeTag(a.transport) != std.meta.activeTag(b.transport)) return false;
switch (a.transport) {
.stdio => |p| {
const q = b.transport.stdio;
if (p.environ_map != q.environ_map or p.argv.len != q.argv.len) return false;
if (!sameEnv(p.environ_map, q.environ_map) or p.argv.len != q.argv.len) return false;
if (!std.mem.eql(u8, p.cwd orelse "", q.cwd orelse "") or (p.cwd == null) != (q.cwd == null)) return false;
for (p.argv, q.argv) |x, y| if (!std.mem.eql(u8, x, y)) return false;
},
@@ -754,6 +792,33 @@ test "a saved credential the server can't use starts nothing, and the host hears
try std.testing.expectEqual(before, e.status(0).phase);
}
test "a stdio server's environment counts by its variables, not the map that holds them" {
const gpa = std.testing.allocator;
var first: std.process.Environ.Map = .init(gpa);
defer first.deinit();
try first.put("MODE", "one");
var e: Engine = undefined;
try TestHost.start(&e);
defer e.deinit();
const argv = [_][]const u8{"/bin/cat"};
try e.reload(&.{.{ .name = "s", .transport = .{ .stdio = .{ .argv = &argv, .environ_map = &first } } }});
// The engine has its own copy: the map it was given may go.
first.deinit();
first = .init(gpa);
var again: std.process.Environ.Map = .init(gpa);
defer again.deinit();
try again.put("MODE", "one");
try std.testing.expect(e.keeps(.{ .name = "s", .transport = .{ .stdio = .{ .argv = &argv, .environ_map = &again } } }));
try again.put("MODE", "two");
try std.testing.expect(!e.keeps(.{ .name = "s", .transport = .{ .stdio = .{ .argv = &argv, .environ_map = &again } } }));
try again.put("MODE", "one");
try again.put("EXTRA", "1");
try std.testing.expect(!e.keeps(.{ .name = "s", .transport = .{ .stdio = .{ .argv = &argv, .environ_map = &again } } }));
try std.testing.expect(!e.keeps(.{ .name = "s", .transport = .{ .stdio = .{ .argv = &argv } } }));
try std.testing.expectEqualStrings("one", e.slots.items[0].config.transport.stdio.environ_map.?.get("MODE").?);
}
test "a stdio server's directory is part of its config, so a new one restarts it" {
const argv = [_][]const u8{ "node", "s.js" };
const a: Config = .{ .name = "x", .transport = .{ .stdio = .{ .argv = &argv, .cwd = "/a" } } };
+9 -2
View File
@@ -1348,7 +1348,7 @@ describe("MCP login", () => {
expect(auth.requests.filter((r) => r.path === "/.well-known/oauth-authorization-server").length).toBe(1);
}, 90_000);
test.skipIf(!tmuxAvailable())("a config change during a sign-in ends it, clears \"Signing in\", and says how to sign in again", async () => {
test.skipIf(!tmuxAvailable())("a config change during a sign-in keeps it going, unless it removes that server: then it ends and clears \"Signing in\"", async () => {
upstream = startModernMcpHttpFixture("json");
auth = startAuthFixture(upstream.url);
const root = createRoot(auth);
@@ -1363,9 +1363,16 @@ describe("MCP login", () => {
await tui.waitForText("Sign in to fixture in your browser.", 10_000);
await tui.waitForText("Signing in to fixture", 10_000);
// Another server's change leaves this sign-in waiting for its browser.
await tui.sendText("/mcp add web https://w.example/mcp");
await tui.waitForText("Added web to ~/.fx/mcp.json.", 10_000);
await tui.waitForText("Couldn't sign in to fixture: the MCP config changed before the browser came back; run /mcp login fixture again.", 10_000);
await Bun.sleep(1_000);
expect(await tui.capturePane()).toContain("Signing in to fixture");
expect(await tui.captureFullScrollback()).not.toContain("Couldn't sign in to fixture");
// Removing the server it signs in to ends it.
await tui.sendText("/mcp remove fixture");
await tui.waitForText("Couldn't sign in to fixture: the MCP config changed before the browser came back.", 10_000);
// The status line catches up within its 250 ms refresh.
const clearBy = Date.now() + 2_000;
while ((await tui.capturePane()).includes("Signing in to fixture") && Date.now() < clearBy) await Bun.sleep(50);
+38 -20
View File
@@ -448,13 +448,16 @@ describe("/mcp in the shell", () => {
expect(JSON.parse(readFileSync(root.profilePath, "utf8")).mcpServers.web).toEqual({ type: "http", url: "https://w.example/mcp" });
}, 90_000);
test.skipIf(!tmuxAvailable())("a config change never freezes the shell, and a server runs one copy at a time", async () => {
test.skipIf(!tmuxAvailable())("a config change restarts only the server it changed, never freezes the shell, and a server runs one copy at a time", async () => {
// `s` keeps running after its input closes and ignores SIGTERM, so fx
// takes about 3 s to stop it: 1.5 s for it to exit, 1.5 s after SIGTERM.
const root = createRoot();
const pidPath = join(root.home, "s.pid");
const eofPath = join(root.home, "s.eof");
writeFileSync(root.profilePath, JSON.stringify({ mcpServers: { s: fixtureServer({ env: { FX_MCP_EXPECT_ELICITATION: "both", FX_MCP_IGNORE_EOF: "1", FX_MCP_IGNORE_TERM: "1", FX_MCP_PID_PATH: pidPath, FX_MCP_EOF_PATH: eofPath } }) } }));
const projPid = join(root.home, "proj.pid");
const env = { FX_MCP_EXPECT_ELICITATION: "both", FX_MCP_IGNORE_EOF: "1", FX_MCP_IGNORE_TERM: "1", FX_MCP_PID_PATH: pidPath, FX_MCP_EOF_PATH: eofPath };
writeFileSync(root.profilePath, JSON.stringify({ mcpServers: { s: fixtureServer({ env }) } }));
writeFileSync(root.projectPath, JSON.stringify({ mcpServers: { proj: fixtureServer({ env: { FX_MCP_PID_PATH: projPid } }) } }));
gateway = startFakeGateway([], { models: [{ id: "openai/gpt-5", type: "language", tags: ["tool-use"] }] });
tui = await TmuxSession.create({
isolated: true,
@@ -466,34 +469,53 @@ describe("/mcp in the shell", () => {
// `s` outlives fx's SIGTERM at quit, so every copy is killed here.
const pids = new Set<number>();
const killServers = () => {
if (existsSync(pidPath)) pids.add(Number(readFileSync(pidPath, "utf8")));
for (const path of [pidPath, projPid]) if (existsSync(path)) pids.add(Number(readFileSync(path, "utf8")));
for (const pid of pids) if (isRunning(pid)) process.kill(pid, "SIGKILL");
};
const pidIn = (path: string) => (existsSync(path) ? Number(readFileSync(path, "utf8").trim() || "0") : 0);
try {
await tui.waitForComposer(15_000);
// `s` starts on its own once the first frame is up and writes its pid.
// Nothing is wrong, so the status line says nothing about MCP.
const upBy = Date.now() + 10_000;
while (!(existsSync(pidPath) && readFileSync(pidPath, "utf8").trim() !== "") && Date.now() < upBy) await Bun.sleep(5);
const oldPid = Number(readFileSync(pidPath, "utf8"));
while (pidIn(pidPath) === 0 && Date.now() < upBy) await Bun.sleep(5);
const oldPid = pidIn(pidPath);
pids.add(oldPid);
// Adding a server, approving one, and removing one leave `s` as it was.
await tui.sendText("/mcp add web https://w.example/mcp");
await tui.waitForText("Added web to ~/.fx/mcp.json.", 10_000);
await tui.sendText("/mcp approve proj");
await tui.waitForText("Approved proj", 10_000);
const projBy = Date.now() + 10_000;
while (pidIn(projPid) === 0 && Date.now() < projBy) await Bun.sleep(5);
expect(pidIn(projPid)).not.toBe(0);
await tui.sendText("/mcp remove web");
await tui.waitForText("Removed web from ~/.fx/mcp.json.", 10_000);
expect(pidIn(pidPath)).toBe(oldPid);
expect(existsSync(eofPath)).toBe(false);
expect(isRunning(oldPid)).toBe(true);
// Removed, `s` stops; added back at once, the new copy waits for it.
await tui.sendText("/mcp remove s");
await tui.waitForText("Removed s from ~/.fx/mcp.json.", 10_000);
const eofBy = Date.now() + 10_000;
// The fixture creates the file, then writes its pid.
while (!(existsSync(eofPath) && readFileSync(eofPath, "utf8").trim() !== "") && Date.now() < eofBy) await Bun.sleep(5);
expect(readFileSync(eofPath, "utf8").trim()).toBe(String(oldPid));
// The old host is stopping `s` now. Typing still shows at once.
const flags = Object.entries(env).map(([k, v]) => `--env ${k}=${v}`).join(" ");
await tui.sendText(`/mcp add ${flags} s -- ${process.execPath} ${MODERN_FIXTURE}`);
await tui.waitForText("Added s to ~/.fx/mcp.json.", 10_000);
// The old copy is still stopping. Typing still shows at once.
const typedAt = Date.now();
await tui.sendLiteral("zq");
await tui.waitForText("zq", 5_000);
expect(Date.now() - typedAt).toBeLessThan(1_000);
expect(isRunning(oldPid)).toBe(true);
// The new host starts `s` only once the old copy is gone.
// The new copy starts only once the old one is gone.
const startBy = Date.now() + 15_000;
while (Number(readFileSync(pidPath, "utf8")) === oldPid && Date.now() < startBy) await Bun.sleep(5);
expect(Number(readFileSync(pidPath, "utf8"))).not.toBe(oldPid);
while (pidIn(pidPath) === oldPid && Date.now() < startBy) await Bun.sleep(5);
expect(pidIn(pidPath)).not.toBe(oldPid);
expect(isRunning(oldPid)).toBe(false);
await tui.sendKeys("C-u");
await tui.sendText("/mcp list");
@@ -506,8 +528,8 @@ describe("/mcp in the shell", () => {
}
}, 90_000);
test.skipIf(!tmuxAvailable())("a config change while a server is in use lets it finish, and the server never runs twice", async () => {
// `s` takes 3 s to start, so `/mcp show s` holds the old host while the config changes.
test.skipIf(!tmuxAvailable())("a config change while a server is in use leaves it running, and the call finishes on the same process", async () => {
// `s` takes 3 s to start, so `/mcp show s` is waiting on it while the config changes.
const root = createRoot();
const pidPath = join(root.home, "s.pid");
writeFileSync(root.profilePath, JSON.stringify({ mcpServers: { s: fixtureServer({ env: { FX_MCP_EXPECT_ELICITATION: "both", FX_MCP_DISCOVER_DELAY_MS: "3000", FX_MCP_PID_PATH: pidPath } }) } }));
@@ -533,15 +555,11 @@ describe("/mcp in the shell", () => {
await tui.waitForText("Added web to ~/.fx/mcp.json.", 10_000);
expect(await tui.capturePane()).not.toContain("Echo text through the modern MCP fixture");
// The moment a new copy starts, the old one must be gone.
const nextBy = Date.now() + 15_000;
while (Number(readFileSync(pidPath, "utf8")) === oldPid && Date.now() < nextBy) await Bun.sleep(5);
const oldAliveAtSwitch = isRunning(oldPid);
pids.add(Number(readFileSync(pidPath, "utf8")));
expect(pids.size).toBe(2);
expect(oldAliveAtSwitch).toBe(false);
// And the show finished on the old host.
// The show answers from the copy it was waiting on, which keeps running.
await tui.waitForText("Echo text through the modern MCP fixture", 10_000);
pids.add(Number(readFileSync(pidPath, "utf8")));
expect([...pids]).toEqual([oldPid]);
expect(isRunning(oldPid)).toBe(true);
} finally {
await tui?.kill();
tui = null;
@@ -1,10 +1,12 @@
// A 2025-11-25 stdio MCP server for the mcp-v2-model case. It logs every
// lifecycle event to FIXTURE_LOG as JSONL ({t, name, pid, ev, ...}):
// start, listed (answered tools/list), eof (stdin ended), sigterm, recv, sent.
// start, listed (answered tools/list), eof (stdin ended, as it exits),
// sigterm, recv, sent; stdin_end when FIXTURE_EXIT_DELAY_MS holds the exit.
// Env:
// FIXTURE_NAME the server's name in its log lines
// FIXTURE_LOG the JSONL log (appended synchronously)
// FIXTURE_INIT_DELAY_MS hold initialize's answer this long (a slow server)
// FIXTURE_EXIT_DELAY_MS once stdin ends, exit only this much later (a slow stop)
// FIXTURE_CONTROL a folder; the elicit tool waits for "go" in it
// Tools: echo (answers at once), elicit (2025 elicitations, see below).
import { appendFileSync, existsSync } from "node:fs";
@@ -13,6 +15,7 @@ import { join } from "node:path";
const name = process.env.FIXTURE_NAME ?? "fixture";
const logPath = process.env.FIXTURE_LOG;
const initDelay = Number(process.env.FIXTURE_INIT_DELAY_MS ?? "0");
const exitDelay = Number(process.env.FIXTURE_EXIT_DELAY_MS ?? "0");
const control = process.env.FIXTURE_CONTROL;
const log = (ev: string, extra: object = {}) => {
if (logPath) appendFileSync(logPath, JSON.stringify({ t: Date.now(), name, pid: process.pid, ev, ...extra }) + "\n");
@@ -75,8 +78,15 @@ process.stdin.on("data", (chunk) => {
}
});
process.stdin.on("end", () => {
log("eof");
process.exit(0);
if (exitDelay === 0) {
log("eof");
process.exit(0);
}
log("stdin_end");
setTimeout(() => {
log("eof");
process.exit(0);
}, exitDelay);
});
async function handle(m: { id?: unknown; method?: string; params?: { name?: string }; result?: unknown; error?: unknown }) {
@@ -6,34 +6,27 @@
import { expect, test } from "bun:test";
import { join } from "node:path";
import { artifacts } from "../lib/context";
import { fixture, merge, save, serverEvents, stdioServer, waitFor, writeProfile, writeProject, type Fixture } from "./driver";
import { BUN, FIXTURE, fixture, merge, save, serverEvents, stdioServer, waitFor, writeProfile, writeProject, type Fixture } from "./driver";
import { serverRecords, Shell } from "./shell";
import { checkTrace } from "./trace";
function alive(pid: number) { try { process.kill(pid, 0); return true; } catch { return false; } }
const started = (f: Fixture, name: string) => serverEvents(f).filter((e) => e.name === name && e.ev === "start");
const listed = (f: Fixture, name: string) => serverEvents(f).filter((e) => e.name === name && e.ev === "listed");
const stopped = (f: Fixture, name: string) => serverEvents(f).filter((e) => e.name === name && e.ev === "eof");
async function finish(shell: Shell, constants: Record<string, unknown>) {
const events = serverEvents(shell.f);
const quitAt = shell.records.find((r) => r.event === "quit")?.at ?? null;
const records = merge(shell.records.filter((r) => r.event !== "release"), serverRecords(events, quitAt));
// A release goes right before the exit of the process its host's end stops
// (the pane is read after the fact, and one log line can share its millisecond).
for (const release of shell.records.filter((r) => r.event === "release")) {
const pid = release.data?.beforeExitOf;
const at = records.findIndex((r) => r.event === "stop" && r.note?.startsWith(`pid ${pid} `));
if (at < 0) throw new Error(`harness: no exit of pid ${pid} to place the release before`);
records.splice(at, 0, { event: "release", at: records[at]!.at, note: release.note });
}
const records = merge(shell.records, serverRecords(events, quitAt));
const verdict = await checkTrace({ module: "HostLifecycle", records, constants: constants as never, dir: join(shell.f.base, "trace") });
save(join(shell.f.base, "verdict.json"), verdict);
return { verdict, events };
}
function setup(name: string, profile: string[], project: string[], slow: string[] = []) {
function setup(name: string, profile: string[], project: string[], slow: string[] = [], extra: Record<string, Record<string, string>> = {}) {
const f = fixture(artifacts, name);
writeProfile(f, Object.fromEntries(profile.map((n) => [n, stdioServer(f, n, slow.includes(n) ? { FIXTURE_INIT_DELAY_MS: "6000" } : {})])));
writeProfile(f, Object.fromEntries(profile.map((n) => [n, stdioServer(f, n, { ...(slow.includes(n) ? { FIXTURE_INIT_DELAY_MS: "6000" } : {}), ...extra[n] })])));
if (project.length) writeProject(f, Object.fromEntries(project.map((n) => [n, stdioServer(f, n)])));
return f;
}
@@ -48,7 +41,7 @@ test("first frame starts the profile servers, the project server waits unspawned
const pids = [...started(shell.f, "a"), ...started(shell.f, "b")].map((e) => e.pid);
const quitMs = await shell.quit();
await waitFor("every server gone", () => pids.every((pid) => !alive(pid)), 5_000);
const { verdict, events } = await finish(shell, { ObsServers: ["a", "b", "p"], ObsProject: ["p"], ObsReachable: ["a", "b", "p"], ObsNeedsAuth: [], ObsMaxHosts: 1, ObsMaxLeases: 1 });
const { verdict, events } = await finish(shell, { ObsServers: ["a", "b", "p"], ObsProject: ["p"], ObsReachable: ["a", "b", "p"], ObsNeedsAuth: [], ObsMaxReloads: 0, ObsMaxLeases: 1 });
expect(started(shell.f, "p")).toEqual([]);
expect(events.filter((e) => e.ev === "sigterm").map((e) => e.name).sort()).toEqual(["a", "b"]);
// A graceful stop would wait 1.5 s on the closed stdin before SIGTERM.
@@ -57,7 +50,7 @@ test("first frame starts the profile servers, the project server waits unspawned
} finally { await shell.stop(); }
}, 120_000);
test("approve reloads: the project server starts and the unchanged servers come back (trace ACCEPTED)", async () => {
test("approve reloads in place: the project server starts and the unchanged servers keep their processes (trace ACCEPTED)", async () => {
const shell = new Shell(setup("hl-approve", ["a", "b"], ["p"]));
try {
await shell.start();
@@ -65,18 +58,21 @@ test("approve reloads: the project server starts and the unchanged servers come
await shell.status({ k: "attn", who: "p", why: "waiting" });
await shell.type("/mcp approve p", { event: "reload", data: { server: "p", change: "approved" } });
await shell.tui.waitForText("Approved p", 10_000);
await waitFor("p, a and b listed again", () => listed(shell.f, "p").length && listed(shell.f, "a").length === 2 && listed(shell.f, "b").length === 2, 20_000);
await waitFor("p listed", () => listed(shell.f, "p").length, 20_000);
expect(await shell.status({ k: "none" })).toEqual({ k: "none", who: "", why: "", r: 0, e: 0 });
// By the servers' own log: one process each of a and b, never stopped.
expect(started(shell.f, "a").length).toBe(1);
expect(started(shell.f, "b").length).toBe(1);
expect(stopped(shell.f, "a")).toEqual([]);
expect(stopped(shell.f, "b")).toEqual([]);
await shell.quit();
await Bun.sleep(500);
const { verdict, events } = await finish(shell, { ObsServers: ["a", "b", "p"], ObsProject: ["p"], ObsReachable: ["a", "b", "p"], ObsNeedsAuth: [], ObsMaxHosts: 2, ObsMaxLeases: 1 });
// The first a and b stopped at their stdin's end before their successors started.
expect(events.filter((e) => e.ev === "eof").map((e) => e.name).sort()).toEqual(["a", "b"]);
const { verdict } = await finish(shell, { ObsServers: ["a", "b", "p"], ObsProject: ["p"], ObsReachable: ["a", "b", "p"], ObsNeedsAuth: [], ObsMaxReloads: 1, ObsMaxLeases: 1 });
expect(verdict.verdict).toBe("ACCEPTED");
} finally { await shell.stop(); }
}, 150_000);
test("reject and remove reload: the project server never starts and the removed one stops for good (trace ACCEPTED)", async () => {
test("reject and remove reload in place: the project server never starts, the removed one stops for good, and the other keeps its process (trace ACCEPTED)", async () => {
const shell = new Shell(setup("hl-reject-remove", ["a", "b"], ["p"]));
try {
await shell.start();
@@ -84,23 +80,24 @@ test("reject and remove reload: the project server never starts and the removed
await shell.status({ k: "attn", who: "p", why: "waiting" });
await shell.type("/mcp reject p", { event: "reload", data: { server: "p", change: "rejected" } });
await shell.tui.waitForText("Rejected p", 10_000);
await waitFor("a and b listed again", () => listed(shell.f, "a").length === 2 && listed(shell.f, "b").length === 2, 20_000);
expect(await shell.status({ k: "none" })).toEqual({ k: "none", who: "", why: "", r: 0, e: 0 });
await shell.type("/mcp remove b", { event: "reload", data: { server: "b", change: "removed" } });
await shell.tui.waitForText("Removed b", 10_000);
await waitFor("a listed a third time", () => listed(shell.f, "a").length === 3, 20_000);
await waitFor("b stopped", () => stopped(shell.f, "b").length, 10_000);
expect(await shell.status({ k: "none" })).toEqual({ k: "none", who: "", why: "", r: 0, e: 0 });
await shell.quit();
await Bun.sleep(500);
const { verdict } = await finish(shell, { ObsServers: ["a", "b", "p"], ObsProject: ["p"], ObsReachable: ["a", "b", "p"], ObsNeedsAuth: [], ObsMaxHosts: 3, ObsMaxLeases: 1 });
const { verdict } = await finish(shell, { ObsServers: ["a", "b", "p"], ObsProject: ["p"], ObsReachable: ["a", "b", "p"], ObsNeedsAuth: [], ObsMaxReloads: 2, ObsMaxLeases: 1 });
expect(started(shell.f, "p")).toEqual([]);
expect(started(shell.f, "b").length).toBe(2);
expect(started(shell.f, "b").length).toBe(1);
expect(started(shell.f, "a").length).toBe(1);
expect(stopped(shell.f, "a")).toEqual([]);
expect(verdict.verdict).toBe("ACCEPTED");
} finally { await shell.stop(); }
}, 150_000);
test("a reload while `/mcp show` holds the old host runs one process of a server at a time: the new host starts nothing until the old one's process stops (trace VIOLATES AtMostOneOwner only, by design)", async () => {
const shell = new Shell(setup("hl-one-process", ["s"], ["p"], ["s"]));
test("a reload while `/mcp show` waits on a server keeps that server's process, and the approved server starts without waiting for the show (trace ACCEPTED)", async () => {
const shell = new Shell(setup("hl-show-reload", ["s"], ["p"], ["s"]));
try {
await shell.start();
const [first] = await waitFor("s started", () => { const s = started(shell.f, "s"); return s.length === 1 ? s : null; }, 20_000);
@@ -109,29 +106,46 @@ test("a reload while `/mcp show` holds the old host runs one process of a server
await Bun.sleep(500);
await shell.type("/mcp approve p", { event: "reload", data: { server: "p", change: "approved" } });
await shell.tui.waitForText("Approved p", 10_000);
// The show prints once the first s answers; its lease ends and the old host goes.
const [p] = await waitFor("p started", () => { const x = started(shell.f, "p"); return x.length ? x : null; }, 10_000);
// The show prints once s answers, from the process it was waiting on.
await shell.tui.waitForText("status ready", 20_000);
const seen = Date.now();
const eof = await waitFor("the first s stopped", () => serverEvents(shell.f).find((e) => e.name === "s" && e.ev === "eof" && e.pid === first!.pid), 10_000);
// `finish` puts the release right before that s's exit.
shell.records.push({ event: "release", at: seen, note: `seen ${seen}, first s eof ${eof.t}; placed before pid ${eof.pid}'s exit`, data: { beforeExitOf: eof.pid } });
await waitFor("p and the second s listed", () => listed(shell.f, "p").length && listed(shell.f, "s").length === 2, 20_000);
const runs = started(shell.f, "s");
const p = started(shell.f, "p");
save(join(shell.f.base, "one-process.json"), { firstEof: eof.t, s: runs.map((e) => ({ pid: e.pid, t: e.t })), p: p.map((e) => ({ pid: e.pid, t: e.t })) });
shell.records.push({ event: "release", at: Date.now() });
const [answered] = await waitFor("s listed", () => { const x = listed(shell.f, "s"); return x.length ? x : null; }, 10_000);
save(join(shell.f.base, "show-reload.json"), { s: started(shell.f, "s").map((e) => ({ pid: e.pid, t: e.t })), p: { pid: p!.pid, t: p!.t }, sListed: answered!.t });
await shell.status({ k: "none" });
await shell.quit();
await Bun.sleep(500);
const { verdict } = await finish(shell, { ObsServers: ["s", "p"], ObsProject: ["p"], ObsReachable: ["s", "p"], ObsNeedsAuth: [], ObsMaxHosts: 2, ObsMaxLeases: 1 });
// By the servers' own log: two runs of s, never at once, and the new
// host started both its servers only after the old s stopped.
expect(runs.length).toBe(2);
expect(runs[1]!.t).toBeGreaterThanOrEqual(eof.t);
expect(p.length).toBe(1);
expect(p[0]!.t).toBeGreaterThanOrEqual(eof.t);
// A lease keeps the old host's owner running to finish its calls, so two
// owners are alive at once by design; one process per server holds.
expect(verdict.verdict).toBe("VIOLATES");
expect(verdict.broken).toEqual(["AtMostOneOwner"]);
const { verdict } = await finish(shell, { ObsServers: ["s", "p"], ObsProject: ["p"], ObsReachable: ["s", "p"], ObsNeedsAuth: [], ObsMaxReloads: 1, ObsMaxLeases: 1 });
// One s, never stopped by the reload; p started before s answered.
expect(started(shell.f, "s").map((e) => e.pid)).toEqual([first!.pid]);
expect(p!.t).toBeLessThan(answered!.t);
expect(verdict.verdict).toBe("ACCEPTED");
} finally { await shell.stop(); }
}, 150_000);
test("a removed server added back runs one process at a time: the new one starts only once the old one, slow to stop, has exited (trace ACCEPTED)", async () => {
const slowStop = { FIXTURE_EXIT_DELAY_MS: "1000" };
const shell = new Shell(setup("hl-add-back", ["a", "s"], [], [], { s: slowStop }));
try {
await shell.start();
await waitFor("a and s listed", () => listed(shell.f, "a").length && listed(shell.f, "s").length, 20_000);
await shell.type("/mcp remove s", { event: "reload", data: { server: "s", change: "removed" } });
await shell.tui.waitForText("Removed s", 10_000);
// Added back while the first s still lingers after its stdin ended.
const env = Object.entries({ FIXTURE_NAME: "s", FIXTURE_LOG: shell.f.serverLog, FIXTURE_CONTROL: shell.f.control, ...slowStop }).map(([k, v]) => `--env ${k}=${v}`).join(" ");
await shell.type(`/mcp add ${env} s -- ${BUN} ${FIXTURE}`, { event: "reload", data: { server: "s", change: "ok" } });
await shell.tui.waitForText("Added s", 10_000);
const runs = await waitFor("a second s listed", () => { const x = started(shell.f, "s"); return x.length === 2 && listed(shell.f, "s").length === 2 ? x : null; }, 20_000);
const firstExit = stopped(shell.f, "s").find((e) => e.pid === runs[0]!.pid);
save(join(shell.f.base, "add-back.json"), { s: runs.map((e) => ({ pid: e.pid, t: e.t })), firstExit: firstExit?.t ?? null });
expect(await shell.status({ k: "none" })).toEqual({ k: "none", who: "", why: "", r: 0, e: 0 });
await shell.quit();
await Bun.sleep(500);
const { verdict } = await finish(shell, { ObsServers: ["a", "s"], ObsProject: [], ObsReachable: ["a", "s"], ObsNeedsAuth: [], ObsMaxReloads: 2, ObsMaxLeases: 1 });
// By the servers' own log: the second s started after the first exited, and a never restarted.
expect(firstExit).toBeDefined();
expect(runs[1]!.t).toBeGreaterThanOrEqual(firstExit!.t);
expect(started(shell.f, "a").length).toBe(1);
expect(verdict.verdict).toBe("ACCEPTED");
} finally { await shell.stop(); }
}, 150_000);
+1
View File
@@ -29,6 +29,7 @@ const catalog = () => JSON.parse(readFileSync(join(tlaDir, "catalog.json"), "utf
function verdictOf(run: TlcRun, expect?: { invariant?: string; property?: string }): "PASS" | "CAUGHT" | "FAIL" {
if (!expect) return run.exit === 0 && run.completed && run.queued === 0 ? "PASS" : "FAIL";
if (expect.invariant) return run.exit === 12 && run.violated === expect.invariant ? "CAUGHT" : "FAIL";
if (run.action !== undefined) return run.exit === 13 && run.action === expect.property ? "CAUGHT" : "FAIL";
return run.exit === 13 && run.temporal && run.violated === undefined ? "CAUGHT" : "FAIL";
}
+5 -3
View File
@@ -25,8 +25,10 @@ test("TLC: every model's smoke and stress pass, every planted bug is caught, sta
// SignIn's three did too, with the hold while another host is alive: its smoke and stress
// configs now check IndicatorImpliesListener, OneSignInPerServer and
// IndicatorClears. UrlRetry's finding is the code before the held call:
// it ended the call at the -32042 error.
expect(count("finding", "FINDING")).toBe(3);
// it ended the call at the -32042 error. HostLifecycle's AtMostOneOwner
// became the code once a reload kept its host: its smoke and stress
// configs check it.
expect(count("finding", "FINDING")).toBe(2);
expect(count("repair", "PASS")).toBe(1);
}, 1_800_000);
@@ -49,7 +51,7 @@ const controls: { name: string; module: string; constants: Record<string, never>
},
{
name: "host lifecycle", module: "HostLifecycle",
constants: { ObsServers: ["a", "p"], ObsProject: ["p"], ObsReachable: ["a", "p"], ObsNeedsAuth: [], ObsMaxHosts: 1, ObsMaxLeases: 1 },
constants: { ObsServers: ["a", "p"], ObsProject: ["p"], ObsReachable: ["a", "p"], ObsNeedsAuth: [], ObsMaxReloads: 1, ObsMaxLeases: 1 },
good: [
{ event: "spawn", data: { server: "a" } },
{ event: "ready", data: { server: "a" } },
+3
View File
@@ -29,6 +29,8 @@ export interface TlcRun {
distinct?: number;
queued?: number;
violated?: string;
/** An action property, which TLC names, unlike the rest of the temporal ones. */
action?: string;
temporal: boolean;
completed: boolean;
}
@@ -64,6 +66,7 @@ export async function runTlc(opts: {
label: opts.label, module: opts.module, config: cfgName, command, exit, seconds: (Date.now() - started) / 1000, log, output,
generated: num(stats?.[1]), distinct: num(stats?.[2]), queued: num(stats?.[3]),
violated: /Invariant (\w+) is violated/.exec(output)?.[1],
action: /Action property (\w+) is violated/.exec(output)?.[1],
temporal: output.includes("Temporal properties were violated"),
completed: output.includes("Model checking completed. No error has been found."),
};
+1 -1
View File
@@ -47,7 +47,7 @@ export function traceObs(records: TraceRecord[], constants: Record<string, Value
}
const substitutions: Record<string, string[]> = {
HostLifecycle: ["Servers", "Project", "Reachable", "NeedsAuth", "MaxHosts", "MaxLeases"],
HostLifecycle: ["Servers", "Project", "Reachable", "NeedsAuth", "MaxReloads", "MaxLeases"],
Credentials: ["Names", "Faults", "InitKinds", "InitEntries", "InitV1"],
SignIn: ["MaxLogins", "MaxHosts", "MaxLeases", "Bridge"],
ElicitationRouting: ["Versions", "MaxQ", "Slots", "Guard"],
+4 -2
View File
@@ -1,10 +1,10 @@
\* Smoke: one profile server, one project server that never answers; three hosts.
\* Smoke: one profile server, one project server that never answers; one host.
CONSTANTS
Servers = {"a", "p"}
Project = {"p"}
Reachable = {"a"}
NeedsAuth = {}
MaxHosts = 3
MaxReloads = 3
MaxLeases = 1
SPECIFICATION Spec
INVARIANTS
@@ -15,6 +15,8 @@ INVARIANTS
QuitSignalsEveryServer
RemovedNotRunning
OneProcessPerServer
AtMostOneOwner
PROPERTIES
KeptServersKeepRunning
ApprovedServersSettle
EveryStopEnds
@@ -1,12 +0,0 @@
\* A stated property the code does not keep: TLC must find a counterexample.
\* A lease keeps the old host's owner running to finish its calls, by design.
CONSTANTS
Servers = {"a", "p"}
Project = {"p"}
Reachable = {"a", "p"}
NeedsAuth = {}
MaxHosts = 3
MaxLeases = 1
SPECIFICATION Spec
INVARIANTS
AtMostOneOwner
+17 -17
View File
@@ -6,21 +6,21 @@
"invariant": "HeldNeverSpawned"
},
{
"description": "A reload makes the new host but never starts its servers (reloadHost without startHostServers)",
"old": " /\\ st' = [st EXCEPT ![n] = [x \\in Servers |-> Started(Initial(nd, x))]]",
"new": " /\\ st' = [st EXCEPT ![n] = [x \\in Servers |-> Initial(nd, x)]]",
"description": "A reload never starts the server it approved or added back (reloadHost without startHostServers)",
"old": " /\\ st' = [st EXCEPT ![cur][s] = Started(Initial(nd, s))]",
"new": " /\\ st' = [st EXCEPT ![cur][s] = Initial(nd, s)]",
"invariant": "StartedAfterFirstFrame"
},
{
"description": "The status line keeps reading the first host after a reload",
"description": "A status refresh keeps the line it had",
"old": " /\\ shown' = Line(cur)",
"new": " /\\ shown' = Line(1)",
"new": " /\\ shown' = shown",
"invariant": "RefreshedLineAgrees"
},
{
"description": "Quitting signals only the current host's running servers",
"description": "Quitting signals only running servers, not one a reload let go of that still stops",
"old": " /\\ termed' = termed \\cup procs \\cup stopping",
"new": " /\\ termed' = termed \\cup {p \\in procs : p[1] = cur}",
"new": " /\\ termed' = termed \\cup procs",
"invariant": "QuitSignalsEveryServer"
},
{
@@ -30,9 +30,9 @@
"invariant": "RemovedNotRunning"
},
{
"description": "Only the first host's owner starts processes, so servers after a reload never settle",
"old": " /\\ ~quit /\\ alive[h] /\\ st[h][s] = \"starting\" /\\ <<h, s>> \\notin procs\n",
"new": " /\\ ~quit /\\ alive[h] /\\ st[h][s] = \"starting\" /\\ <<h, s>> \\notin procs /\\ h = 1\n",
"description": "A server added back stays down",
"old": " /\\ st' = [st EXCEPT ![cur][s] = Started(Initial(nd, s))]",
"new": " /\\ st' = [st EXCEPT ![cur][s] = IF d = \"ok\" THEN \"absent\" ELSE Started(Initial(nd, s))]",
"property": "ApprovedServersSettle"
},
{
@@ -42,15 +42,15 @@
"property": "EveryStopEnds"
},
{
"description": "A new host starts a server while a leased old host still runs it",
"old": " /\\ \\A p \\in Live : p[1] = h\n",
"new": " /\\ stopping = {}\n",
"description": "A server added back starts while the process its removal let go of still stops",
"old": " /\\ \\A p \\in Live : p[2] # s\n",
"new": " /\\ \\A p \\in procs : p[2] # s\n",
"invariant": "OneProcessPerServer"
},
{
"description": "A new host starts a server while the old host's copy is still stopping",
"old": " /\\ \\A p \\in Live : p[1] = h\n",
"new": " /\\ \\A p \\in procs : p[1] = h\n",
"invariant": "OneProcessPerServer"
"description": "A reload stops every running server, as a new host did",
"old": " gone == {<<cur, s>>} \\cap procs\n",
"new": " gone == Of(cur)\n",
"property": "KeptServersKeepRunning"
}
]
+5 -2
View File
@@ -1,10 +1,10 @@
\* Stress: two profile servers (one needs a login), a project server; two hosts. Safety only.
\* Stress: two profile servers (one needs a login), a project server; one host. Safety only.
CONSTANTS
Servers = {"a", "b", "p"}
Project = {"p"}
Reachable = {"a", "b", "p"}
NeedsAuth = {"b"}
MaxHosts = 2
MaxReloads = 2
MaxLeases = 1
SPECIFICATION Spec
INVARIANTS
@@ -15,3 +15,6 @@ INVARIANTS
QuitSignalsEveryServer
RemovedNotRunning
OneProcessPerServer
AtMostOneOwner
PROPERTIES
KeptServersKeepRunning
+74 -56
View File
@@ -1,24 +1,30 @@
---- MODULE HostLifecycle ----
\* fx's MCP host lifecycle, as the shell runs it.
\*
\* Code (vercel-labs/fx at 53a561bea):
\* - src/core/app/app_mcp_runtime.zig: loadHost, reloadHost (replaceHost
\* then startHostServers), leaseHost/releaseHost (a replaced host lives
\* until its last lease), destroyHost (process exit: signalServersForExit,
\* nothing waits), refreshHostStatus/hostStatusLine.
\* Code (vercel-labs/fx):
\* - src/core/app/app_mcp_runtime.zig: loadHost, reloadHost (the host takes
\* the config in place, then startHostServers), leaseHost/releaseHost,
\* destroyHost (process exit: signalServersForExit, nothing waits),
\* refreshHostStatus/hostStatusLine.
\* - src/core/app/app_render_runtime.zig:1031: the first frame calls
\* startMcpHostServers, which prepares every server.
\* - src/core/mcp_host/runtime.zig: one owner thread per Runtime (start,
\* run, deinit); `.need` skips held servers (handle, `.need`); publish
\* turns held reasons and engine phases into statuses.
\* turns held reasons and engine phases into statuses; reload
\* (applyReload, dropServer) and drain's hold while a dropped server's
\* process stops (src/mcp/engine.zig reload, keeps, stopping).
\* - src/core/mcp_host/menu.zig statusLine: the status line text.
\* - src/mcp/io/process.zig signalAllForExit: SIGTERM to every process of
\* every engine in the program.
\*
\* One host is one Runtime and its owner thread. A reload makes a new host
\* from the changed config and starts every server in it; the old host is
\* retired and goes, with its servers, when its last lease ends. A lease is a
\* `/mcp` verb, a status refresh, or a call still using that host.
\* One host is one Runtime and its owner thread. A reload takes the changed
\* config in that host: a server whose config is the same keeps its state
\* and its process, a removed one's process stops, and one approved or added
\* back starts once no process of it is left. A lease is a `/mcp` verb, a
\* status refresh, or a call using the host; a reload no longer waits for
\* one. When the config empties, or was empty, the shell swaps hosts
\* instead, and the new one also waits for every retired host's processes;
\* that only narrows what the model allows, so it is left out (MaxHosts 1).
\*
\* Server states per host (runtime.zig State, folded):
\* absent not in the config (removed)
@@ -28,8 +34,8 @@
\* starting connecting, or backing off (`connecting`, `retrying`)
\* ready, needs_login, failed
\* Processes: `procs` are live server processes the host runs, `stopping`
\* those whose host let go of them (stdin closed) that haven't exited yet,
\* and `termed` those sent SIGTERM at quit.
\* those it let go of (stdin closed) that haven't exited yet, and `termed`
\* those sent SIGTERM at quit. A server has at most one of each.
EXTENDS Naturals, FiniteSets
CONSTANTS
@@ -37,13 +43,14 @@ CONSTANTS
Project, \* those from ./.mcp.json, which start waiting for approval
Reachable, \* those whose command starts and answers
NeedsAuth, \* reachable HTTP servers that answer 401 until a sign-in
MaxHosts, \* bound on reloads: hosts 1..MaxHosts
MaxReloads, \* bound on reloads
MaxLeases \* bound on leases one host holds at once
ASSUME Project \subseteq Servers /\ Reachable \subseteq Servers /\ NeedsAuth \subseteq Reachable
ASSUME MaxHosts \in Nat \ {0} /\ MaxLeases \in Nat
ASSUME MaxReloads \in Nat /\ MaxLeases \in Nat
Hosts == 1..MaxHosts
\* One host: a reload keeps it.
Hosts == {1}
States == {"absent", "waiting", "rejected", "idle", "starting", "ready", "needs_login", "failed"}
Decisions == {"ok", "pending", "approved", "rejected", "removed"}
Changes == {"approved", "rejected", "removed", "ok"}
@@ -64,9 +71,10 @@ VARIABLES
quit, \* fx exited
shown, \* the status line's MCP text, as last refreshed
refreshed, \* the step just taken was a status refresh
heldAt \* ghost: [Hosts -> SUBSET Servers] held when the host was made
heldAt, \* ghost: [Hosts -> SUBSET Servers] held as of the last reload
reloads \* reloads so far, for the bound
vars == <<decision, cur, alive, retired, leases, st, procs, stopping, termed, framed, quit, shown, refreshed, heldAt>>
vars == <<decision, cur, alive, retired, leases, st, procs, stopping, termed, framed, quit, shown, refreshed, heldAt, reloads>>
D0 == [s \in Servers |-> IF s \in Project THEN "pending" ELSE "ok"]
@@ -107,6 +115,7 @@ TypeOK ==
/\ framed \in BOOLEAN /\ quit \in BOOLEAN /\ refreshed \in BOOLEAN
/\ shown \in Lines
/\ heldAt \in [Hosts -> SUBSET Servers]
/\ reloads \in 0..MaxReloads
Init ==
/\ decision = D0
@@ -119,6 +128,7 @@ Init ==
/\ framed = FALSE /\ quit = FALSE
/\ shown = NoLine /\ refreshed = FALSE
/\ heldAt = [h \in Hosts |-> IF h = 1 THEN Held(D0) ELSE {}]
/\ reloads = 0
\* app_render_runtime.zig:1031 -> startHostServers: prepare(i) for every server.
FirstFrame ==
@@ -126,32 +136,31 @@ FirstFrame ==
/\ framed' = TRUE
/\ st' = [st EXCEPT ![cur] = [s \in Servers |-> Started(st[cur][s])]]
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, leases, procs, stopping, termed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, procs, stopping, termed, quit, shown, heldAt, reloads>>
\* Something needs a server's tools: a tool search, `/mcp show`, a call.
Need(s) ==
/\ ~quit /\ st[cur][s] = "idle"
/\ st' = [st EXCEPT ![cur][s] = "starting"]
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, leases, procs, stopping, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, procs, stopping, termed, framed, quit, shown, heldAt, reloads>>
\* The owner thread starts the server's process. A retired host's servers
\* stop on a thread of their own, so the UI never waits for them.
\* While any other host still has a live process, in use under a lease or
\* stopping, a host starts none, so a server never runs twice.
\* The owner thread starts the server's process: only once no process of
\* that server is live, even one a reload let go of that still stops, so a
\* server never runs twice. Other servers' processes don't hold it up.
Spawn(h, s) ==
/\ ~quit /\ alive[h] /\ st[h][s] = "starting" /\ <<h, s>> \notin procs
/\ \A p \in Live : p[1] = h
/\ \A p \in Live : p[2] # s
/\ procs' = procs \cup {<<h, s>>}
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, stopping, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, stopping, termed, framed, quit, shown, heldAt, reloads>>
\* The server answered detection and its tool list.
Answer(h, s) ==
/\ ~quit /\ alive[h] /\ st[h][s] = "starting" /\ <<h, s>> \in procs /\ s \in Reachable
/\ st' = [st EXCEPT ![h][s] = IF s \in NeedsAuth THEN "needs_login" ELSE "ready"]
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, leases, procs, stopping, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, procs, stopping, termed, framed, quit, shown, heldAt, reloads>>
\* The engine gave up on a server that never answers (its attempt limit).
GiveUp(h, s) ==
@@ -159,18 +168,18 @@ GiveUp(h, s) ==
/\ st' = [st EXCEPT ![h][s] = "failed"]
/\ procs' = procs \ {<<h, s>>}
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, leases, stopping, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, stopping, termed, framed, quit, shown, heldAt, reloads>>
\* leaseHost: a `/mcp` verb, a status refresh, or a call takes the current host.
Lease ==
/\ ~quit /\ leases[cur] < MaxLeases
/\ leases' = [leases EXCEPT ![cur] = @ + 1]
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, st, procs, stopping, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, st, procs, stopping, termed, framed, quit, shown, heldAt, reloads>>
\* releaseHost: the last lease of a retired host retires it (Host.retire:
\* Runtime.deinit on a thread of its own; the engine closes its servers'
\* stdin).
\* stdin). With reloads in place, no host is retired before quit.
Release(h) ==
/\ ~quit /\ leases[h] > 0
/\ leases' = [leases EXCEPT ![h] = @ - 1]
@@ -180,10 +189,12 @@ Release(h) ==
/\ stopping' = stopping \cup Of(h)
ELSE UNCHANGED <<alive, procs, stopping>>
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, retired, st, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, retired, st, termed, framed, quit, shown, heldAt, reloads>>
\* add, remove, approve, reject: hosts.changed -> applyMcpHostReload ->
\* reloadHost: a new host from the config, then startHostServers.
\* reloadHost -> Host.reload, then startHostServers. Only `s` changes: its
\* running process, if a removal took it, stops; an approved or added-back
\* server starts. Every other server keeps its state and process.
Change(s, d) ==
\/ d = "approved" /\ decision[s] = "pending"
\/ d = "rejected" /\ decision[s] = "pending"
@@ -191,27 +202,24 @@ Change(s, d) ==
\/ d = "ok" /\ decision[s] = "removed"
Reload(s, d) ==
/\ ~quit /\ cur < MaxHosts /\ Change(s, d)
/\ LET n == cur + 1
nd == [decision EXCEPT ![s] = d]
free == leases[cur] = 0
/\ ~quit /\ reloads < MaxReloads /\ Change(s, d)
/\ reloads' = reloads + 1
/\ LET nd == [decision EXCEPT ![s] = d]
gone == {<<cur, s>>} \cap procs
IN /\ decision' = nd
/\ cur' = n
/\ st' = [st EXCEPT ![n] = [x \in Servers |-> Started(Initial(nd, x))]]
/\ heldAt' = [heldAt EXCEPT ![n] = Held(nd)]
/\ retired' = [retired EXCEPT ![cur] = TRUE]
/\ alive' = [alive EXCEPT ![n] = TRUE, ![cur] = ~free]
/\ procs' = IF free THEN procs \ Of(cur) ELSE procs
/\ stopping' = IF free THEN stopping \cup Of(cur) ELSE stopping
/\ st' = [st EXCEPT ![cur][s] = Started(Initial(nd, s))]
/\ heldAt' = [heldAt EXCEPT ![cur] = Held(nd)]
/\ procs' = procs \ gone
/\ stopping' = stopping \cup gone
/\ refreshed' = FALSE
/\ UNCHANGED <<leases, termed, framed, quit, shown>>
/\ UNCHANGED <<cur, alive, retired, leases, termed, framed, quit, shown>>
\* A server whose stdin closed exits (also when fx quit meanwhile).
Exit(h, s) ==
/\ <<h, s>> \in stopping
/\ stopping' = stopping \ {<<h, s>>}
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, procs, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, procs, termed, framed, quit, shown, heldAt, reloads>>
\* destroyHost at process exit: every server of every host gets SIGTERM and
\* nothing waits; fx is gone.
@@ -221,7 +229,7 @@ Quit ==
/\ termed' = termed \cup procs \cup stopping
/\ alive' = [h \in Hosts |-> FALSE]
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, retired, leases, st, procs, stopping, framed, shown, heldAt>>
/\ UNCHANGED <<decision, cur, retired, leases, st, procs, stopping, framed, shown, heldAt, reloads>>
\* A signalled server exits.
TermExit(h, s) ==
@@ -229,14 +237,14 @@ TermExit(h, s) ==
/\ procs' = procs \ {<<h, s>>}
/\ stopping' = stopping \ {<<h, s>>}
/\ refreshed' = FALSE
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, termed, framed, quit, shown, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, termed, framed, quit, shown, heldAt, reloads>>
\* refreshHostStatus: the line from the current host's statuses.
RefreshStatus ==
/\ ~quit
/\ shown' = Line(cur)
/\ refreshed' = TRUE
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, procs, stopping, termed, framed, quit, heldAt>>
/\ UNCHANGED <<decision, cur, alive, retired, leases, st, procs, stopping, termed, framed, quit, heldAt, reloads>>
Next ==
\/ FirstFrame
@@ -260,9 +268,9 @@ Spec == Init /\ [][Next]_vars /\ Fairness
\* ---- Safety the code keeps ----
\* A held server (waiting, rejected, or removed when its host was made) is
\* never spawned by that host.
HeldNeverSpawned == \A p \in procs \cup stopping \cup termed : p[2] \notin heldAt[p[1]]
\* A held server (waiting, rejected, or removed as of the last reload) has
\* no running process; a removed one's may still be stopping.
HeldNeverSpawned == \A p \in procs : p[2] \notin heldAt[p[1]]
\* After the first frame, and after every reload, no server the current host
\* may run is left idle: unchanged servers come back.
@@ -275,15 +283,25 @@ RefreshedLineAgrees == refreshed => shown = Line(cur)
QuitSignalsEveryServer == quit => Live \subseteq termed
\* A removed server is never running in the current host.
RemovedNotRunning == \A s \in Servers : decision[s] = "removed" => <<cur, s>> \notin Live
RemovedNotRunning == \A s \in Servers : decision[s] = "removed" => <<cur, s>> \notin procs
\* ---- Stated safety the code does not keep (HostLifecycle.findings.cfg) ----
\* At most one owner thread drives servers at a time.
\* At most one owner thread drives servers at a time: a reload keeps the host.
AtMostOneOwner == Cardinality({h \in Hosts : alive[h]}) <= 1
\* At most one live process per configured server.
OneProcessPerServer == \A s \in Servers : Cardinality({h \in Hosts : <<h, s>> \in Live}) <= 1
\* At most one live process per configured server: never one running beside
\* the one a reload let go of.
OneProcessPerServer ==
/\ procs \cap stopping = {}
/\ \A s \in Servers : Cardinality({h \in Hosts : <<h, s>> \in Live}) <= 1
\* ---- Safety over steps ----
\* Only the server a reload changed stops: a running process leaves only
\* when its own server's config changed, when the engine gives up on it, or
\* when fx quits.
KeptServersKeepRunning ==
[][\A p \in procs : p \notin procs' =>
decision'[p[2]] # decision[p[2]] \/ quit \/ st[p[1]][p[2]] = "starting"]_vars
\* ---- Liveness, under the fairness above ----
+1 -6
View File
@@ -3,12 +3,7 @@
"module": "HostLifecycle",
"smoke": "HostLifecycle.cfg",
"stress": "HostLifecycle.stress.cfg",
"findings": [
{
"config": "HostLifecycle.findings.cfg",
"invariant": "AtMostOneOwner"
}
],
"findings": [],
"repair": []
},
{