Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,7 @@ you are in.
- [runtime-scene-switching.md](docs/agent-rules/runtime-scene-switching.md) — the host applies a scene swap after the tick; five ordering rules, plus the `Project` mount shipped games were missing.
- [server-authoritative-networking-loop.md](docs/agent-rules/server-authoritative-networking-loop.md) — ~20 well-tested networking classes produced nothing, because the entry point had zero call sites.
- [mcp-setter-based-field-registry.md](docs/agent-rules/mcp-setter-based-field-registry.md) — the copy-then-swap MCP write path is unsound when `operator=` can't reproduce a field write's side effects.
- [mcp-protocol-eras.md](docs/agent-rules/mcp-protocol-eras.md) — the 2026-07-28 stateless core is a second transport, not a field addition; `server/discover` added on its own actively breaks working clients.

**Concurrency & memory**

Expand Down
274 changes: 258 additions & 16 deletions OloEditor/src/MCP/McpClient.cpp

Large diffs are not rendered by default.

59 changes: 50 additions & 9 deletions OloEditor/src/MCP/McpClient.h
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,10 @@
// (MakeStdioTransportFactory, McpClient.cpp) owns CreateProcessW + pipes +
// a blocking reader thread; tests substitute an in-memory fake that
// replies synchronously inside WriteLine.
// * McpClientConnection — the protocol + concurrency layer: the initialize /
// initialized / tools/list handshake, request-id multiplexing (bridged
// * McpClientConnection — the protocol + concurrency layer: era negotiation
// (see NegotiateEra — `server/discover` probe, falling back to the
// initialize / initialized handshake), tools/list, request-id
// multiplexing (bridged
// tool handlers run CONCURRENTLY on httplib worker threads, so responses
// are matched to waiters by exact id, the same id.dump() keying the
// server's cancellation registry uses), timeouts, and teardown that never
Expand Down Expand Up @@ -75,12 +77,12 @@ namespace OloEngine::MCP
class McpClientConnection
{
public:
// Spawn (via `factory`), run the MCP handshake (initialize ->
// notifications/initialized -> tools/list), and build the bridged
// ToolDefs. Returns nullptr with `outError` set on any failure (spawn,
// timeout, malformed handshake). `onDeath(alias)` is invoked at most
// once, from the transport's reader thread, when the child dies or the
// stream breaks OUTSIDE a deliberate Shutdown — the owner uses it to
// Spawn (via `factory`), negotiate an era (see NegotiateEra), run the
// matching opening sequence, list the child's tools, and build the
// bridged ToolDefs. Returns nullptr with `outError` set on any failure
// (spawn, timeout, malformed handshake). `onDeath(alias)` is invoked at
// most once, from the transport's reader thread, when the child dies or
// the stream breaks OUTSIDE a deliberate Shutdown — the owner uses it to
// unpublish the alias's tools.
[[nodiscard]] static std::shared_ptr<McpClientConnection> Connect(
const McpClientConfig& config, const McpClientTransportFactory& factory,
Expand Down Expand Up @@ -118,14 +120,48 @@ namespace OloEngine::MCP
return m_BridgedToolCount;
}

// The era this connection negotiated at Connect time, and the exact
// revision string it stamps on requests (issue #777). Fixed for the
// lifetime of the connection — the spec makes era a property of the
// server, not of an individual request — so these are plain reads.
[[nodiscard]] McpProtocolEra Era() const
{
return m_Era;
}
[[nodiscard]] const std::string& ProtocolVersion() const
{
return m_ProtocolVersion;
}

private:
explicit McpClientConnection(McpClientConfig config,
std::function<void(const std::string&)> onDeath);

// Decide whether the child speaks the 2026-07-28 stateless core or the
// legacy `initialize` handshake, and settle on a mutually supported
// revision. Implements the spec's stdio backward-compatibility rule:
// probe with `server/discover`, treat a recognized MODERN reply (a
// DiscoverResult, or an UnsupportedProtocolVersionError naming the
// versions it does support) as a modern server, and fall back to the
// legacy handshake on anything else — including a timeout. Sets m_Era
// and m_ProtocolVersion; returns false with `outError` set only when the
// child is modern but shares no revision with us (an actionable
// mismatch, not a fallback).
[[nodiscard]] bool NegotiateEra(std::string& outError);

// The per-request `_meta` block every MODERN request must carry: the
// protocol version (required), our identity (SHOULD), and our capability
// set for THIS request (required — the server may not infer it from a
// previous one). Empty in the legacy era, where the handshake carried
// all three once.
[[nodiscard]] Json RequestMeta() const;

// Send one request and block the CALLING thread (an httplib worker or
// the connecting thread) until the child's response arrives, `timeout`
// expires, or the connection shuts down. Returns the full JSON-RPC
// response object, or nullopt on timeout/write-failure/shutdown.
// response object, or nullopt on timeout/write-failure/shutdown. In the
// modern era `params._meta` is stamped on by RequestMeta() unless the
// caller already supplied one.
[[nodiscard]] std::optional<Json> SendRequest(const std::string& method, const Json& params,
std::chrono::milliseconds timeout);
void SendNotification(const std::string& method, const Json& params);
Expand Down Expand Up @@ -161,6 +197,11 @@ namespace OloEngine::MCP
std::atomic<bool> m_Alive{ false };
std::atomic<bool> m_ShuttingDown{ false };

// Written once by NegotiateEra during Connect, before the connection is
// published to any other thread; read-only thereafter.
McpProtocolEra m_Era = McpProtocolEra::Legacy;
std::string m_ProtocolVersion;

std::vector<ToolDef> m_BridgedTools;
sizet m_BridgedToolCount = 0;
};
Expand Down
162 changes: 157 additions & 5 deletions OloEditor/src/MCP/McpServer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
#include <string_view>
#include <system_error>
#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>

Expand Down Expand Up @@ -1285,6 +1286,13 @@ namespace OloEngine::MCP
std::lock_guard lock(m_SessionMutex);
m_Sessions.clear();
}
// Resource subscriptions die with the session that made them: the streams
// that would carry the updates are gone, and a later Start() must not
// inherit a subscription no live client asked for.
{
std::lock_guard lock(m_SubscriptionMutex);
m_ResourceSubscriptions.clear();
}
m_Token.clear();

// Drop the session's ephemeral capture resources so a later Start()
Expand Down Expand Up @@ -1671,8 +1679,9 @@ namespace OloEngine::MCP
res.set_chunked_content_provider(
"text/event-stream",
[this, cursor = startCursor, lastWrite = std::chrono::steady_clock::now(), greeted = false,
toolsGeneration = ToolsGeneration(),
resourcesGeneration = ResourcesGeneration()](std::size_t /*offset*/, httplib::DataSink& sink) mutable -> bool
toolsGeneration = ToolsGeneration(), resourcesGeneration = ResourcesGeneration(),
subscriptionTokens = std::unordered_map<std::string, u64>{}](
std::size_t /*offset*/, httplib::DataSink& sink) mutable -> bool
{
// Server tearing down: end the stream gracefully so the worker thread
// is free to be joined by Stop().
Expand Down Expand Up @@ -1725,6 +1734,52 @@ namespace OloEngine::MCP
lastWrite = std::chrono::steady_clock::now();
}

// Resource subscriptions (#777 — the `logging` offramp carrier).
// Same polled-generation delivery as the two blocks above, for the
// same single-writer-thread reason, but per SUBSCRIBED URI: each
// subscribable resource exposes a monotonic ChangeToken and we emit
// `notifications/resources/updated` when it advances.
//
// A URI this stream has not seen before starts from the
// SUBSCRIBE-TIME baseline, never from the token as of now. Seeding
// from "now" would swallow every change between the subscribe and
// this cycle — including everything that happened before the client
// opened its stream at all — which is precisely the
// never-fires failure the ChangeToken gate exists to prevent.
// Unsubscribing drops the URI from both maps, so a resubscribe
// re-baselines rather than replaying.
//
// Updates COALESCE by construction: the token is sampled once per
// cycle, so a burst inside one poll interval yields one
// notification. That is the correct semantics — the notification
// means "re-read this resource", not "one thing happened" — but a
// consumer must read from its own cursor, not count notifications.
{
const ResourceSnapshot resources = ResourcesSnapshot();
std::unordered_map<std::string, u64> stillSubscribed;
for (const auto& [uri, baseline] : SubscriptionBaselines())
{
const ResourceDef* resource = FindResource(*resources, uri);
if (resource == nullptr || !resource->ChangeToken)
continue; // evicted or replaced by a non-subscribable def
const u64 token = resource->ChangeToken();
const auto seen = subscriptionTokens.find(uri);
const u64 lastSeen = seen != subscriptionTokens.end() ? seen->second : baseline;
if (lastSeen != token)
{
const std::string frame = FormatSseData(
Json{ { "jsonrpc", "2.0" },
{ "method", "notifications/resources/updated" },
{ "params", Json{ { "uri", uri } } } });
if (!sink.write(frame.data(), frame.size()))
return false;
lastWrite = std::chrono::steady_clock::now();
}
stillSubscribed.emplace(uri, token);
}
subscriptionTokens = std::move(stillSubscribed);
}

if (!ServiceEventStream(sink, cursor, lastWrite))
return false;

Expand Down Expand Up @@ -1980,6 +2035,10 @@ namespace OloEngine::MCP
return HandleResourcesList(id);
if (method == "resources/read")
return HandleResourcesRead(id, request.value("params", Json::object()));
if (method == "resources/subscribe")
return HandleResourcesSubscribe(id, request.value("params", Json::object()));
if (method == "resources/unsubscribe")
return HandleResourcesUnsubscribe(id, request.value("params", Json::object()));
if (method == "prompts/list")
return HandlePromptsList(id);
if (method == "prompts/get")
Expand All @@ -2006,6 +2065,12 @@ namespace OloEngine::MCP
result["protocolVersion"] = version;
// `logging` is advertised because the GET /mcp SSE stream pushes diagnostics
// events as `notifications/message` log notifications (issue #306 item B).
// It is DEPRECATED as of spec 2026-07-28 (SEP-2577, ≥12-month offramp), so
// it now has a successor running beside it rather than replacing it: the
// `olo://events/recent` resource plus `resources.subscribe` below. Both
// carriers stay live for the whole offramp — dropping `logging` early would
// break every client that speaks a 2025-* revision, which today is all of
// them. See docs/agent-rules/mcp-protocol-eras.md.
//
// `tools.listChanged` is TRUE since the script-tool live-reload item (#607):
// olo_script_tools_reload (and the MCP panel's "Reload script tools" button)
Expand All @@ -2014,10 +2079,15 @@ namespace OloEngine::MCP
// stream. `resources.listChanged` is TRUE since the resource-link work
// (#673 Tier 1): capture tools publish ephemeral resources while serving
// (RegisterEphemeralResource), announced the same generation-polled way.
// Prompts are still fixed for a server run; per-URI `subscribe` stays
// unimplemented (nothing needs a subscription registry yet).
// Prompts are still fixed for a server run.
//
// `resources.subscribe` is TRUE since the `logging` offramp (#777): a
// client can subscribe to `olo://events/recent` and receive
// `notifications/resources/updated` whenever the diagnostics ring advances.
// Only resources carrying a ResourceDef::ChangeToken accept a subscription;
// the rest are rejected by name (HandleResourcesSubscribe).
result["capabilities"] = Json{ { "tools", { { "listChanged", true } } },
{ "resources", { { "subscribe", false }, { "listChanged", true } } },
{ "resources", { { "subscribe", true }, { "listChanged", true } } },
{ "prompts", { { "listChanged", false } } },
{ "logging", Json::object() } };
// `description` on Implementation is a 2025-11-25 addition (aligns with
Expand Down Expand Up @@ -2416,6 +2486,88 @@ namespace OloEngine::MCP
return MakeResult(id, std::move(result));
}

Json McpServer::HandleResourcesSubscribe(const Json& id, const Json& params)
{
if (!params.contains("uri") || !params["uri"].is_string())
return MakeError(id, kInvalidParams, "Invalid params: 'uri' is required");

const std::string uri = params["uri"].get<std::string>();
const ResourceSnapshot snapshot = ResourcesSnapshot();
const ResourceDef* resource = FindResource(*snapshot, uri);
if (resource == nullptr)
return MakeError(id, kInvalidParams, "Unknown resource: " + uri);

// Refuse a subscription we could never fire. Without a ChangeToken there is
// no cheap way to know the resource changed, so accepting would promise
// updates that never arrive — the exact dishonesty the diagnostics server's
// "report unknown rather than fabricate" rule forbids. Name the ones that
// do work so the client can recover in one step.
if (!resource->ChangeToken)
{
std::string subscribable;
for (const ResourceDef& candidate : *snapshot)
{
if (!candidate.ChangeToken)
continue;
if (!subscribable.empty())
subscribable += ", ";
subscribable += candidate.Uri;
}
return MakeError(id, kInvalidParams,
"Resource is not subscribable: " + uri +
(subscribable.empty() ? " (no resource on this server supports subscriptions)"
: " (subscribable resources: " + subscribable + ")"));
}

// Baseline the token HERE, not on the stream's first poll: a change landing
// between this call and the next poll cycle — or before the client even
// opens its stream — must still be reported, or the subscription silently
// eats it. `emplace` keeps an existing subscription's baseline, so a
// repeated subscribe is idempotent and cannot discard a pending update;
// unsubscribe erases, so a resubscribe deliberately re-baselines.
{
std::lock_guard lock(m_SubscriptionMutex);
m_ResourceSubscriptions.emplace(uri, resource->ChangeToken());
}
return MakeResult(id, Json::object());
}

Json McpServer::HandleResourcesUnsubscribe(const Json& id, const Json& params)
{
if (!params.contains("uri") || !params["uri"].is_string())
return MakeError(id, kInvalidParams, "Invalid params: 'uri' is required");

// Idempotent by design: unsubscribing something never subscribed is a
// no-op success, so a client tearing down after a reconnect cannot be
// tripped by a subscription the server already forgot.
{
std::lock_guard lock(m_SubscriptionMutex);
m_ResourceSubscriptions.erase(params["uri"].get<std::string>());
}
return MakeResult(id, Json::object());
}

std::vector<std::string> McpServer::SubscribedUris() const
{
std::vector<std::string> uris;
{
std::lock_guard lock(m_SubscriptionMutex);
uris.reserve(m_ResourceSubscriptions.size());
for (const auto& [uri, baseline] : m_ResourceSubscriptions)
uris.push_back(uri);
}
// Sorted so the observable is deterministic — an unordered_map's iteration
// order is not, and a test asserting on it would be a latent flake.
std::sort(uris.begin(), uris.end());
return uris;
}

std::unordered_map<std::string, u64> McpServer::SubscriptionBaselines() const
{
std::lock_guard lock(m_SubscriptionMutex);
return m_ResourceSubscriptions;
}

const ResourceDef* McpServer::FindResource(const ResourceList& resources, const std::string& uri)
{
for (const auto& resource : resources)
Expand Down
Loading
Loading