Merge branch 'develop' into claude/add-listchanged-support-epMnQ

Resolves conflicts in McpProxyLifecycleFactory.java:
- Take union of FlushEx flyweight type imports (develop adds aggregate-id
  rewrite paths for elicitComplete/progress/resumable/list-changed variants
  on top of the basic McpFlushExFW import).
- Drop duplicate mcpFlushExRW builder field created by auto-merge accepting
  both sides' independently-added declaration.

Both sides' work is complementary: branch adds initialize-time listChanged
advertisement + cache→list_changed wiring + diff-gated fan-out + ITs;
develop (PR #1791) adds multi-route aggregate event IDs. Verified all
182 binding-mcp tests pass.

https://claude.ai/code/session_01Gx5yC2CuFd54Fyoy7kL3qg

Author: claude

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpAggregateEventId.java

@@ -0,0 +1,153 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.binding.mcp.internal.config;
+
+import static java.util.function.Function.identity;
+import static java.util.stream.Collectors.toMap;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.CRC32C;
+
+import org.agrona.MutableDirectBuffer;
+import org.agrona.collections.Long2ObjectHashMap;
+
+public final class McpAggregateEventId
+{
+    static final byte PAIR_DELIMITER = (byte) ';';
+    static final byte KEY_VALUE_DELIMITER = (byte) '=';
+    static final int MAX_PREFIX_LENGTH = 6;
+
+    private McpAggregateEventId()
+    {
+    }
+
+    public static Map<String, String> computePrefixes(
+        Collection<String> toolkits)
+    {
+        if (toolkits == null || toolkits.isEmpty())
+        {
+            return Map.of();
+        }
+
+        final Set<String> distinct = new HashSet<>(toolkits);
+        final Map<String, String> encoded = distinct.stream()
+            .collect(toMap(identity(), McpAggregateEventId::encodeCrc32c));
+
+        int length = 1;
+        while (length <= MAX_PREFIX_LENGTH)
+        {
+            final Set<String> seen = new HashSet<>();
+            boolean unique = true;
+            for (String code : encoded.values())
+            {
+                if (!seen.add(code.substring(0, length)))
+                {
+                    unique = false;
+                    break;
+                }
+            }
+            if (unique)
+            {
+                break;
+            }
+            length++;
+        }
+
+        if (length > MAX_PREFIX_LENGTH)
+        {
+            throw new IllegalStateException("unable to derive unique prefixes for toolkits: " + distinct);
+        }
+
+        final int prefixLength = length;
+        return encoded.entrySet().stream()
+            .collect(toMap(Map.Entry::getKey, e -> e.getValue().substring(0, prefixLength)));
+    }
+
+    public static int encode(
+        McpAggregateRoute[] routes,
+        Long2ObjectHashMap<String> ids,
+        MutableDirectBuffer buffer,
+        int offset)
+    {
+        int progress = offset;
+        for (McpAggregateRoute route : routes)
+        {
+            final String id = ids.get(route.routedId());
+            if (id == null)
+            {
+                continue;
+            }
+            if (progress > offset)
+            {
+                buffer.putByte(progress++, PAIR_DELIMITER);
+            }
+            progress += buffer.putStringWithoutLengthUtf8(progress, route.prefix());
+            buffer.putByte(progress++, KEY_VALUE_DELIMITER);
+            progress += buffer.putStringWithoutLengthUtf8(progress, id);
+        }
+        return progress == offset ? -1 : progress - offset;
+    }
+
+    public static void decode(
+        String aggregate,
+        BiConsumer<String, String> visitor)
+    {
+        if (aggregate == null || aggregate.isEmpty())
+        {
+            return;
+        }
+
+        int start = 0;
+        final int length = aggregate.length();
+        while (start < length)
+        {
+            int end = aggregate.indexOf((char) PAIR_DELIMITER, start);
+            if (end < 0)
+            {
+                end = length;
+            }
+            final int sep = aggregate.indexOf((char) KEY_VALUE_DELIMITER, start);
+            if (sep > start && sep < end)
+            {
+                final String prefix = aggregate.substring(start, sep);
+                final String value = aggregate.substring(sep + 1, end);
+                visitor.accept(prefix, value);
+            }
+            start = end + 1;
+        }
+    }
+
+    private static String encodeCrc32c(
+        String toolkit)
+    {
+        final CRC32C crc = new CRC32C();
+        crc.update(toolkit.getBytes(StandardCharsets.UTF_8));
+        final long value = crc.getValue();
+        final byte[] bytes = new byte[]
+        {
+            (byte) (value >>> 24),
+            (byte) (value >>> 16),
+            (byte) (value >>> 8),
+            (byte) value
+        };
+        return Base64.getUrlEncoder().withoutPadding().encodeToString(bytes);
+    }
+}

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpAggregateRoute.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.binding.mcp.internal.config;
+
+public record McpAggregateRoute(
+    String prefix,
+    long routedId)
+{
+}

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfig.java

@@ -20,6 +20,7 @@
 import static io.aklivity.zilla.runtime.binding.mcp.internal.types.McpCapabilities.SERVER_TOOLS_LIST_CHANGED;
 
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -46,6 +47,8 @@ public final class McpBindingConfig
     public final GuardHandler guard;
     public final McpProxyCache cache;
     public final Map<String, McpProxySession> sessions;
+    public final Map<String, McpRouteConfig> routeByPrefix;
+    public final McpAggregateRoute[] aggregateRoutes;
 
     private final List<McpRouteConfig> routes;
 
@@ -60,6 +63,26 @@ public McpBindingConfig(
             .map(McpRouteConfig::new)
             .collect(Collectors.toList());
 
+        final Map<String, McpRouteConfig> routeByPrefix = new LinkedHashMap<>();
+        if (routes.size() > 1)
+        {
+            final List<String> toolkits = routes.stream()
+                .map(McpRouteConfig::toolkit)
+                .collect(Collectors.toList());
+            final Map<String, String> prefixesByToolkit = McpAggregateEventId.computePrefixes(toolkits);
+            for (McpRouteConfig route : routes)
+            {
+                final String prefix = prefixesByToolkit.get(route.toolkit());
+                routeByPrefix.put(prefix, route);
+            }
+        }
+        this.routeByPrefix = routeByPrefix;
+
+        this.aggregateRoutes = routeByPrefix.entrySet().stream()
+            .sorted(Map.Entry.comparingByKey())
+            .map(e -> new McpAggregateRoute(e.getKey(), e.getValue().id))
+            .toArray(McpAggregateRoute[]::new);
+
         this.guard = Optional.ofNullable(options)
             .map(o -> o.authorization)
             .map(a -> a.name)

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpRouteConfig.java

@@ -49,6 +49,7 @@ public final class McpRouteConfig
 
     private final List<ConditionMatcher> matchers;
     private final LongObjectPredicate<UnaryOperator<String>> authorized;
+    private final String toolkit;
 
     public McpRouteConfig(
         RouteConfig route)
@@ -60,6 +61,16 @@ public McpRouteConfig(
             .map(ConditionMatcher::new)
             .collect(toList());
         this.authorized = route.authorized;
+        this.toolkit = matchers.stream()
+            .map(m -> m.toolkit)
+            .filter(t -> t != null)
+            .findFirst()
+            .orElse(null);
+    }
+
+    public String toolkit()
+    {
+        return toolkit;
     }
 
     public boolean authorized(
@@ -216,6 +227,7 @@ static String identifierOf(
 
     private static final class ConditionMatcher
     {
+        private final String toolkit;
         private final String toolsPrefix;
         private final String promptsPrefix;
         private final String resourcesPrefix;
@@ -225,6 +237,7 @@ private ConditionMatcher(
         {
             final List<String> capabilities = condition.capability;
             final String toolkit = condition.toolkit;
+            this.toolkit = toolkit;
 
             final boolean anyCapability = capabilities == null;
             final boolean tools = anyCapability || capabilities.contains(CAPABILITY_TOOLS);