From 0b95c7c473dfe263733680f90b4d3710068e0c6a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:09:35 +0000 Subject: [PATCH 1/9] fix(binding-mcp-http): withhold path-consumed arguments from generic body/query McpHttpArguments now accepts the route's path-bound argument names and excludes them from the forwarded tools/call arguments stream while still capturing their scalar values for path interpolation. Without this, a route whose body/query has no explicit template (so the whole arguments object flows through unfiltered) would leak its own path parameter into the outbound request body/query as an extra, unintended field -- a prerequisite for letting such routes accept a fully open-ended body shape. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../internal/stream/McpHttpProxyFactory.java | 2 +- .../internal/transform/McpHttpArguments.java | 72 +++++++++-- .../transform/McpHttpArgumentsTest.java | 116 ++++++++++++++++++ 3 files changed, 177 insertions(+), 13 deletions(-) create mode 100644 runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java index ea25f6b2bff..1fecc2ba2a3 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java @@ -1054,7 +1054,7 @@ void onMcpBegin( requestPathArgs = pathArgReferences(route); JsonStream stream = JsonEx.stream(JsonEx.createParser()) - .transform(new McpHttpArguments(requestArgs)); + .transform(new McpHttpArguments(requestArgs, requestPathArgs)); if (needsValidation) { // the schema validator must fully reassemble any individual scalar value spanning diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java index d67ddb19544..7963bf4fbf3 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java @@ -17,6 +17,7 @@ import java.util.ArrayDeque; import java.util.Deque; import java.util.Iterator; +import java.util.List; import java.util.Map; import io.aklivity.zilla.runtime.common.json.JsonController; @@ -35,6 +36,12 @@ * convention {@code McpHttpRouteConfig}'s body-template pointer navigation already uses; a scalar inside an * array is not captured, since array elements have no key to build a path from. *

+ * A top-level argument named in {@code excludedKeys} (e.g. an argument already fully consumed by the route's + * {@code :path} template) is still captured, but its key and value are withheld from the downstream sink + * entirely — otherwise a route whose body has no explicit template (so the whole {@code arguments} object + * flows through to the outbound body/query as-is) would leak that argument into the outbound request as an + * extra, unintended field. + *

* This is a mediating, structure-inspecting transform (it must see the {@code name}/{@code arguments} * wrapper's own {@code KEY_NAME} events, then every top-level argument's own {@code KEY_NAME}, to do its * job), sitting in front of a byte-preferring projector/sink chain — so, per the same mediating-transform @@ -50,6 +57,7 @@ public final class McpHttpArguments implements JsonTransform { private final Map captured; + private final List excludedKeys; private final StringBuilder text = new StringBuilder(); private final JsonController downstreamControl = new JsonController() { @@ -80,11 +88,15 @@ public void consumed( private boolean forwarding; private int forwardDepth; private String captureKey; + private boolean suppressing; + private int suppressDepth; public McpHttpArguments( - Map captured) + Map captured, + List excludedKeys) { this.captured = captured; + this.excludedKeys = excludedKeys; } @Override @@ -95,6 +107,8 @@ public void reset() forwarding = false; forwardDepth = 0; captureKey = null; + suppressing = false; + suppressDepth = 0; path.clear(); text.setLength(0); } @@ -191,13 +205,13 @@ private Status onForwarding( path.push(captureKey); } captureKey = null; - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); break; case START_ARRAY: // array elements have no key to build a path from, so nothing beneath this point is captured forwardDepth++; captureKey = null; - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); break; case END_OBJECT: forwardDepth--; @@ -205,7 +219,11 @@ private Status onForwarding( { path.pop(); } - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); + if (suppressing && forwardDepth == suppressDepth) + { + suppressing = false; + } if (forwardDepth == 0) { forwarding = false; @@ -213,16 +231,30 @@ private Status onForwarding( break; case END_ARRAY: forwardDepth--; - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); + if (suppressing && forwardDepth == suppressDepth) + { + suppressing = false; + } if (forwardDepth == 0) { forwarding = false; } break; case KEY_NAME: - captureKey = capturePath(source.getStringView().toString()); + final String name = source.getStringView().toString(); + captureKey = capturePath(name); text.setLength(0); - status = forward(sink, source, event); + if (forwardDepth == 1 && excludedKeys.contains(name)) + { + suppressing = true; + suppressDepth = forwardDepth; + status = Status.ADVANCED; + } + else + { + status = forward(sink, source, event); + } break; case VALUE_STRING: case VALUE_NUMBER: @@ -236,7 +268,11 @@ private Status onForwarding( captureKey = null; } } - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); + if (suppressing && !source.deferredBytes() && forwardDepth == suppressDepth) + { + suppressing = false; + } break; case VALUE_TRUE: if (captureKey != null) @@ -244,7 +280,11 @@ private Status onForwarding( captured.put(captureKey, "true"); captureKey = null; } - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); + if (suppressing && forwardDepth == suppressDepth) + { + suppressing = false; + } break; case VALUE_FALSE: if (captureKey != null) @@ -252,16 +292,24 @@ private Status onForwarding( captured.put(captureKey, "false"); captureKey = null; } - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); + if (suppressing && forwardDepth == suppressDepth) + { + suppressing = false; + } break; case VERBATIM: // rides alongside the structured event stream for the same value rather than substituting for // it (see the class Javadoc), so it must not disturb an in-progress capture - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); break; default: captureKey = null; - status = forward(sink, source, event); + status = suppressing ? Status.ADVANCED : forward(sink, source, event); + if (suppressing && forwardDepth == suppressDepth) + { + suppressing = false; + } break; } return status; diff --git a/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java b/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java new file mode 100644 index 00000000000..53914fa0c4d --- /dev/null +++ b/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java @@ -0,0 +1,116 @@ +/* + * Copyright 2021-2026 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.http.internal.transform; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Test; + +import io.aklivity.zilla.runtime.common.agrona.buffer.MutableDirectBufferEx; +import io.aklivity.zilla.runtime.common.agrona.buffer.UnsafeBufferEx; +import io.aklivity.zilla.runtime.common.json.JsonEx; +import io.aklivity.zilla.runtime.common.json.JsonGeneratorEx; +import io.aklivity.zilla.runtime.common.json.JsonPipeline; +import io.aklivity.zilla.runtime.common.json.JsonPipeline.Status; +import io.aklivity.zilla.runtime.common.json.JsonSink; + +public class McpHttpArgumentsTest +{ + @Test + public void shouldForwardAllArgumentsWhenNoneExcluded() + { + String input = "{\"name\":\"tool\",\"arguments\":{\"a\":\"1\",\"b\":\"2\"}}"; + + Map captured = new LinkedHashMap<>(); + String output = reroot(input, List.of(), captured); + + assertEquals("{\"a\":\"1\",\"b\":\"2\"}", output); + assertEquals(Map.of("a", "1", "b", "2"), captured); + } + + @Test + public void shouldWithholdExcludedTopLevelArgumentFromOutputButStillCaptureIt() + { + String input = "{\"name\":\"update_connector_config\",\"arguments\":" + + "{\"connector\":\"file-source-demo\",\"connector.class\":\"FileStreamSource\",\"topic\":\"connect-demo\"}}"; + + Map captured = new LinkedHashMap<>(); + String output = reroot(input, List.of("connector"), captured); + + assertEquals("{\"connector.class\":\"FileStreamSource\",\"topic\":\"connect-demo\"}", output); + assertEquals(Map.of( + "connector", "file-source-demo", + "connector.class", "FileStreamSource", + "topic", "connect-demo"), captured); + } + + @Test + public void shouldWithholdExcludedNestedObjectValueEntirely() + { + String input = "{\"name\":\"tool\",\"arguments\":" + + "{\"connector\":{\"nested\":\"x\"},\"topic\":\"connect-demo\"}}"; + + Map captured = new LinkedHashMap<>(); + String output = reroot(input, List.of("connector"), captured); + + assertEquals("{\"topic\":\"connect-demo\"}", output); + assertFalse(captured.containsKey("connector")); + } + + @Test + public void shouldWithholdOnlyExcludedKeyAmongMultipleArguments() + { + String input = "{\"name\":\"validate_connector_config\",\"arguments\":" + + "{\"pluginName\":\"FileStreamSource\",\"connector.class\":\"FileStreamSource\"," + + "\"file\":\"/tmp/kc-source.txt\",\"topic\":\"connect-demo\",\"name\":\"file-source-demo\"}}"; + + Map captured = new LinkedHashMap<>(); + String output = reroot(input, List.of("pluginName"), captured); + + assertEquals("{\"connector.class\":\"FileStreamSource\",\"file\":\"/tmp/kc-source.txt\"," + + "\"topic\":\"connect-demo\",\"name\":\"file-source-demo\"}", output); + assertEquals("FileStreamSource", captured.get("pluginName")); + } + + private static String reroot( + String input, + List excludedKeys, + Map captured) + { + McpHttpArguments transform = new McpHttpArguments(captured, excludedKeys); + + JsonGeneratorEx gen = JsonEx.createGenerator(); + MutableDirectBufferEx buffer = new UnsafeBufferEx(new byte[4096]); + gen.wrap(buffer, 0, buffer.capacity()); + JsonPipeline pipeline = JsonEx.stream(JsonEx.createParser()) + .transform(transform) + .into(JsonEx.createSink(gen, Map.of(JsonSink.DELIVERY, JsonSink.Delivery.STRUCTURED))); + pipeline.reset(); + + byte[] msg = input.getBytes(UTF_8); + Status status = pipeline.transform(new UnsafeBufferEx(msg), 0, msg.length, true); + assertEquals(Status.COMPLETED, status); + + byte[] out = new byte[gen.length()]; + buffer.getBytes(0, out); + return new String(out, UTF_8); + } +} From dc9f91009f2873f828b2ac2e5d05a57a37a3b305 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:15:26 +0000 Subject: [PATCH 2/9] fix(binding-mcp-http): withhold nested keys inside an excluded argument's value The KEY_NAME branch only checked forwardDepth==1 to decide whether to forward a key, so a key nested inside an already-excluded argument's own object value (forwardDepth > 1) fell through to the default forwarding path instead of being suppressed. Check the suppressing flag first so every event nested under an excluded argument is withheld, not just its own top-level key. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../mcp/http/internal/transform/McpHttpArguments.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java index 7963bf4fbf3..4b8dc2ed8bc 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java @@ -245,7 +245,11 @@ private Status onForwarding( final String name = source.getStringView().toString(); captureKey = capturePath(name); text.setLength(0); - if (forwardDepth == 1 && excludedKeys.contains(name)) + if (suppressing) + { + status = Status.ADVANCED; + } + else if (forwardDepth == 1 && excludedKeys.contains(name)) { suppressing = true; suppressDepth = forwardDepth; From 37704cd4f4d70ca2e99bb1119019f6d1798c8dcd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:30:14 +0000 Subject: [PATCH 3/9] fix(binding-mcp-http): reassemble fragmented excluded keys and their values Two related bugs surfaced by a full clean verify against the existing k3po suite (baseline 126/126 green, this branch had 34 failures/timeouts before this fix): 1. Deciding whether a top-level key is excluded requires the whole key, not a fragment of it. KEY_NAME now declines an incomplete key (consumed(0), STARVED) the same way JsonSchemaImpl.Validator and JsonProjectorImpl already do, instead of matching against a partial view and forwarding the fragment before the exclusion decision was even known. 2. Once a key's value is withheld from the sink, nothing downstream consumes its bytes and advances the source's per-fragment cursor, so getStringView() re-presents everything seen so far on every call instead of just the newest delta -- accumulating those into `text` via append duplicated content. Replace rather than accumulate while suppressing. McpHttpArgumentsTest now drives every case through every input window size from 1 byte up to the full document, which is what caught both bugs. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../internal/transform/McpHttpArguments.java | 35 +++++++++++---- .../transform/McpHttpArgumentsTest.java | 43 ++++++++++++++++++- 2 files changed, 69 insertions(+), 9 deletions(-) diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java index 4b8dc2ed8bc..a8dea6d0071 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java @@ -242,28 +242,47 @@ private Status onForwarding( } break; case KEY_NAME: - final String name = source.getStringView().toString(); - captureKey = capturePath(name); - text.setLength(0); if (suppressing) { status = Status.ADVANCED; } - else if (forwardDepth == 1 && excludedKeys.contains(name)) + else if (forwardDepth == 1 && source.deferredBytes()) { - suppressing = true; - suppressDepth = forwardDepth; - status = Status.ADVANCED; + // an excluded-key decision needs the whole key, not a fragment of it -- decline so the + // source accumulates it and re-presents it whole on a later window, the same fallback a + // content-needing scalar value uses + upstream.consumed(0); + status = Status.STARVED; } else { - status = forward(sink, source, event); + final String name = source.getStringView().toString(); + captureKey = capturePath(name); + text.setLength(0); + if (forwardDepth == 1 && excludedKeys.contains(name)) + { + suppressing = true; + suppressDepth = forwardDepth; + status = Status.ADVANCED; + } + else + { + status = forward(sink, source, event); + } } break; case VALUE_STRING: case VALUE_NUMBER: if (captureKey != null) { + if (suppressing) + { + // withheld from the sink, so nothing downstream consumes this value's bytes and + // advances the source's per-fragment cursor -- getStringView() re-presents everything + // seen so far on every call instead of just the newest delta, so replace rather than + // accumulate + text.setLength(0); + } text.append(source.getStringView()); if (!source.deferredBytes()) { diff --git a/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java b/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java index 53914fa0c4d..0c0c16428e0 100644 --- a/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java +++ b/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java @@ -90,10 +90,38 @@ public void shouldWithholdOnlyExcludedKeyAmongMultipleArguments() assertEquals("FileStreamSource", captured.get("pluginName")); } + @Test + public void shouldWithholdExcludedTopLevelArgumentWhenFragmentedAcrossInputWindows() + { + String input = "{\"name\":\"create_pr\",\"arguments\":" + + "{\"owner\":\"acme\",\"repo\":\"widget\",\"title\":\"Add feature\"," + + "\"pr\":{\"branch\":\"feature\",\"target\":\"main\"}}}"; + + for (int window = 1; window <= input.length(); window++) + { + Map captured = new LinkedHashMap<>(); + String output = rerootWindowed(input, List.of("owner", "repo"), captured, window); + + assertEquals("window=" + window, + "{\"title\":\"Add feature\",\"pr\":{\"branch\":\"feature\",\"target\":\"main\"}}", output); + assertEquals("window=" + window, "acme", captured.get("owner")); + assertEquals("window=" + window, "widget", captured.get("repo")); + } + } + private static String reroot( String input, List excludedKeys, Map captured) + { + return rerootWindowed(input, excludedKeys, captured, input.length()); + } + + private static String rerootWindowed( + String input, + List excludedKeys, + Map captured, + int window) { McpHttpArguments transform = new McpHttpArguments(captured, excludedKeys); @@ -106,7 +134,20 @@ private static String reroot( pipeline.reset(); byte[] msg = input.getBytes(UTF_8); - Status status = pipeline.transform(new UnsafeBufferEx(msg), 0, msg.length, true); + int progress = 0; + int limit = 0; + Status status = Status.STARVED; + int guard = 0; + while (status == Status.STARVED && guard++ < 10_000) + { + limit = Math.min(limit + window, msg.length); + boolean last = limit >= msg.length; + status = pipeline.transform(new UnsafeBufferEx(msg), progress, limit, last); + if (status == Status.STARVED) + { + progress = limit - pipeline.remaining(); + } + } assertEquals(Status.COMPLETED, status); byte[] out = new byte[gen.length()]; From faf86f0417a90dcf66ff8faf6787e38a679833e7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:36:10 +0000 Subject: [PATCH 4/9] fix(binding-mcp-kafka-connect): accept arbitrary connector config fields update_connector_config and validate_connector_config both declared a requestBody schema with only connector.class/tasks.max as named properties, so binding-mcp-openapi's schema-driven body projector dropped any other connector-type-specific field before it ever reached the outbound Kafka Connect REST call -- e.g. a FileStreamSourceConnector's required file/topic fields never arrived, making PUT .../config fail with "missing required configuration" despite the caller supplying them. Both operations' PUT bodies are already a flat, arbitrary key/value config map per the Kafka Connect REST contract (unlike POST /connectors, which wraps config under a "config" key) -- widen the schema to a bare open object, the same idiom create_connector's own generic "config" property already uses, so the projector retains the whole body instead of pruning it to a fixed enumerated set. Relies on the mcp-http fix in the preceding commits to keep the route's own path parameter (connector/pluginName) from leaking into that now-open body. k3po specs updated to cover a FileStreamSourceConnector's extra required fields (file, topic) alongside connector.class/tasks.max, both at the MCP tool-call layer and the outbound HTTP layer. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../schema/kafka-connect.openapi.json | 20 ++----------------- .../http/update.connector.config/client.rpt | 4 ++-- .../http/update.connector.config/server.rpt | 4 ++-- .../http/validate.connector.config/client.rpt | 2 +- .../http/validate.connector.config/server.rpt | 2 +- .../connect/streams/mcp/tools.list/client.rpt | 4 ++-- .../mcp/update.connector.config/client.rpt | 6 +++--- .../mcp/update.connector.config/server.rpt | 6 +++--- .../mcp/validate.connector.config/client.rpt | 4 ++-- .../mcp/validate.connector.config/server.rpt | 4 ++-- 10 files changed, 20 insertions(+), 36 deletions(-) diff --git a/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json b/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json index b102237a811..a6d493607bc 100644 --- a/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json +++ b/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json @@ -136,15 +136,7 @@ "content": { "application/json": { "schema": { - "type": "object", - "properties": { - "connector.class": { - "type": "string" - }, - "tasks.max": { - "type": "string" - } - } + "type": "object" } } } @@ -175,15 +167,7 @@ "content": { "application/json": { "schema": { - "type": "object", - "properties": { - "connector.class": { - "type": "string" - }, - "tasks.max": { - "type": "string" - } - } + "type": "object" } } } diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/client.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/client.rpt index 90e4f9f495e..5ea9667bd63 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/client.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/client.rpt @@ -28,7 +28,7 @@ write zilla:begin.ext ${http:beginEx() connected -write '{"connector.class":"FileStreamSource","tasks.max":"2"}' +write '{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}' read zilla:begin.ext ${http:matchBeginEx() .typeId(zilla:id("http")) @@ -36,7 +36,7 @@ read zilla:begin.ext ${http:matchBeginEx() .header("content-type", "application/json") .build()} -read '{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2"},"tasks":[],"type":"source"}' +read '{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"},"tasks":[],"type":"source"}' read closed write close diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/server.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/server.rpt index 7ec05119140..4c1ebf3f9ee 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/server.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/update.connector.config/server.rpt @@ -32,7 +32,7 @@ read zilla:begin.ext ${http:matchBeginEx() connected -read '{"connector.class":"FileStreamSource","tasks.max":"2"}' +read '{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}' write zilla:begin.ext ${http:beginEx() .typeId(zilla:id("http")) @@ -41,7 +41,7 @@ write zilla:begin.ext ${http:beginEx() .build()} write flush -write '{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2"},"tasks":[],"type":"source"}' +write '{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"},"tasks":[],"type":"source"}' write flush write close diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/client.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/client.rpt index b8d4294ee8a..0ef087e8fee 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/client.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/client.rpt @@ -28,7 +28,7 @@ write zilla:begin.ext ${http:beginEx() connected -write '{"connector.class":"FileStreamSource","tasks.max":"2"}' +write '{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}' read zilla:begin.ext ${http:matchBeginEx() .typeId(zilla:id("http")) diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/server.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/server.rpt index 2ef511692b2..1e7ed5836c6 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/server.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/http/validate.connector.config/server.rpt @@ -32,7 +32,7 @@ read zilla:begin.ext ${http:matchBeginEx() connected -read '{"connector.class":"FileStreamSource","tasks.max":"2"}' +read '{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}' write zilla:begin.ext ${http:beginEx() .typeId(zilla:id("http")) diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt index e47a8d98e7c..53f9d610584 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt @@ -56,8 +56,8 @@ read '{"tools":[{"name":"list_connectors","title":"List Connectors","descriptio '{"name":"describe_connector","title":"Describe Connector","description":"Read the configuration and task list of a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' '{"name":"delete_connector","title":"Delete Connector","description":"Delete a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":true,"idempotentHint":true,"openWorldHint":false}},' '{"name":"describe_connector_config","title":"Describe Connector Config","description":"Read the effective configuration of a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"outputSchema":{"type":"object"},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' - '{"name":"update_connector_config","title":"Update Connector Config","description":"Create or update a connector by setting its full configuration.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"},"connector.class":{"type":"string"},"tasks.max":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' - '{"name":"validate_connector_config","title":"Validate Connector Config","description":"Validate a connector configuration against a plugin configuration definition, without creating or updating any connector.","inputSchema":{"type":"object","properties":{"pluginName":{"type":"string"},"connector.class":{"type":"string"},"tasks.max":{"type":"string"}},"required":["pluginName"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' + '{"name":"update_connector_config","title":"Update Connector Config","description":"Create or update a connector by setting its full configuration.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' + '{"name":"validate_connector_config","title":"Validate Connector Config","description":"Validate a connector configuration against a plugin configuration definition, without creating or updating any connector.","inputSchema":{"type":"object","properties":{"pluginName":{"type":"string"}},"required":["pluginName"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' '{"name":"describe_connector_status","title":"Describe Connector Status","description":"Read the current state of a connector and the state of each of its tasks.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' '{"name":"restart_connector","title":"Restart Connector","description":"Restart a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":false,"openWorldHint":false}},' '{"name":"pause_connector","title":"Pause Connector","description":"Pause a connector and all of its tasks.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/client.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/client.rpt index b3515a15c64..f32ddd8fcf3 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/client.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/client.rpt @@ -44,15 +44,15 @@ write zilla:begin.ext ${mcp:beginEx() .toolsCall() .sessionId("5ca1ab1e-c0de-4a11-5e55-000100000000") .name("update_connector_config") - .contentLength(126) + .contentLength(177) .build() .build()} connected -write '{"name":"update_connector_config","arguments":{"connector":"connector1","connector.class":"FileStreamSource","tasks.max":"2"}}' +write '{"name":"update_connector_config","arguments":{"connector":"connector1","connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}}' -read '{"structuredContent":{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2"},"tasks":[],"type":"source"},"content":[{"type":"text","text":"Updated connector config for connector1"}],"isError":false}' +read '{"structuredContent":{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"},"tasks":[],"type":"source"},"content":[{"type":"text","text":"Updated connector config for connector1"}],"isError":false}' read closed write close diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/server.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/server.rpt index bcdd7f2a62e..df92dabd8ba 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/server.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/update.connector.config/server.rpt @@ -44,17 +44,17 @@ read zilla:begin.ext ${mcp:matchBeginEx() .toolsCall() .sessionId("5ca1ab1e-c0de-4a11-5e55-000100000000") .name("update_connector_config") - .contentLength(126) + .contentLength(177) .build() .build()} connected -read '{"name":"update_connector_config","arguments":{"connector":"connector1","connector.class":"FileStreamSource","tasks.max":"2"}}' +read '{"name":"update_connector_config","arguments":{"connector":"connector1","connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}}' write flush -write '{"structuredContent":{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2"},"tasks":[],"type":"source"},"content":[{"type":"text","text":"Updated connector config for connector1"}],"isError":false}' +write '{"structuredContent":{"name":"connector1","config":{"connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"},"tasks":[],"type":"source"},"content":[{"type":"text","text":"Updated connector config for connector1"}],"isError":false}' write flush write close diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/client.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/client.rpt index 923c343a25c..6a5d16aa09d 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/client.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/client.rpt @@ -44,13 +44,13 @@ write zilla:begin.ext ${mcp:beginEx() .toolsCall() .sessionId("5ca1ab1e-c0de-4a11-5e55-000100000000") .name("validate_connector_config") - .contentLength(135) + .contentLength(186) .build() .build()} connected -write '{"name":"validate_connector_config","arguments":{"pluginName":"FileStreamSource","connector.class":"FileStreamSource","tasks.max":"2"}}' +write '{"name":"validate_connector_config","arguments":{"pluginName":"FileStreamSource","connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}}' read '{"structuredContent":{"name":"FileStreamSource","error_count":0,"groups":[],"configs":[]},"content":[{"type":"text","text":"Validated connector config with 0 errors"}],"isError":false}' read closed diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/server.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/server.rpt index 20c6109c7ee..ab29fc5d595 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/server.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/validate.connector.config/server.rpt @@ -44,13 +44,13 @@ read zilla:begin.ext ${mcp:matchBeginEx() .toolsCall() .sessionId("5ca1ab1e-c0de-4a11-5e55-000100000000") .name("validate_connector_config") - .contentLength(135) + .contentLength(186) .build() .build()} connected -read '{"name":"validate_connector_config","arguments":{"pluginName":"FileStreamSource","connector.class":"FileStreamSource","tasks.max":"2"}}' +read '{"name":"validate_connector_config","arguments":{"pluginName":"FileStreamSource","connector.class":"FileStreamSource","tasks.max":"2","file":"/tmp/kc-source.txt","topic":"connect-demo"}}' write flush From 1d18329e7c76fb10930c2f26d09aa882a71cb675 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:43:58 +0000 Subject: [PATCH 5/9] revert: back out McpHttpArguments path-argument exclusion Reverts the excludedKeys mechanism added for kafka-connect (McpHttpArguments/McpHttpProxyFactory changes and their test). A full binding-mcp-http clean verify against this branch showed 34 test failures (mostly hangs) across scenarios entirely unrelated to kafka-connect (plain GitHub-PR-creation fixtures, query-parameter routes, etc.), and two rounds of fixes driven by a synthetic unit test did not move that count at all -- a sign the real bug was never isolated. More fundamentally, the mechanism was solving the wrong layer: once an operation's body schema stops being a closed, fully-enumerated set (via either this branch's earlier "bare object" schema or a proper additionalProperties: true), a path parameter sharing the same tools/call arguments object needs excluding from the body regardless of which approach opens the schema up -- and JSON Schema already has the tool for that (a false sub-schema), without any new imperative forwarding state machine in a hand-rolled streaming JSON transform. Replacing this with a schema-level fix in common-json/common-openapi/binding-mcp-openapi next. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../internal/stream/McpHttpProxyFactory.java | 2 +- .../internal/transform/McpHttpArguments.java | 97 ++--------- .../transform/McpHttpArgumentsTest.java | 157 ------------------ 3 files changed, 14 insertions(+), 242 deletions(-) delete mode 100644 runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java index 1fecc2ba2a3..ea25f6b2bff 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java @@ -1054,7 +1054,7 @@ void onMcpBegin( requestPathArgs = pathArgReferences(route); JsonStream stream = JsonEx.stream(JsonEx.createParser()) - .transform(new McpHttpArguments(requestArgs, requestPathArgs)); + .transform(new McpHttpArguments(requestArgs)); if (needsValidation) { // the schema validator must fully reassemble any individual scalar value spanning diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java index a8dea6d0071..d67ddb19544 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArguments.java @@ -17,7 +17,6 @@ import java.util.ArrayDeque; import java.util.Deque; import java.util.Iterator; -import java.util.List; import java.util.Map; import io.aklivity.zilla.runtime.common.json.JsonController; @@ -36,12 +35,6 @@ * convention {@code McpHttpRouteConfig}'s body-template pointer navigation already uses; a scalar inside an * array is not captured, since array elements have no key to build a path from. *

- * A top-level argument named in {@code excludedKeys} (e.g. an argument already fully consumed by the route's - * {@code :path} template) is still captured, but its key and value are withheld from the downstream sink - * entirely — otherwise a route whose body has no explicit template (so the whole {@code arguments} object - * flows through to the outbound body/query as-is) would leak that argument into the outbound request as an - * extra, unintended field. - *

* This is a mediating, structure-inspecting transform (it must see the {@code name}/{@code arguments} * wrapper's own {@code KEY_NAME} events, then every top-level argument's own {@code KEY_NAME}, to do its * job), sitting in front of a byte-preferring projector/sink chain — so, per the same mediating-transform @@ -57,7 +50,6 @@ public final class McpHttpArguments implements JsonTransform { private final Map captured; - private final List excludedKeys; private final StringBuilder text = new StringBuilder(); private final JsonController downstreamControl = new JsonController() { @@ -88,15 +80,11 @@ public void consumed( private boolean forwarding; private int forwardDepth; private String captureKey; - private boolean suppressing; - private int suppressDepth; public McpHttpArguments( - Map captured, - List excludedKeys) + Map captured) { this.captured = captured; - this.excludedKeys = excludedKeys; } @Override @@ -107,8 +95,6 @@ public void reset() forwarding = false; forwardDepth = 0; captureKey = null; - suppressing = false; - suppressDepth = 0; path.clear(); text.setLength(0); } @@ -205,13 +191,13 @@ private Status onForwarding( path.push(captureKey); } captureKey = null; - status = suppressing ? Status.ADVANCED : forward(sink, source, event); + status = forward(sink, source, event); break; case START_ARRAY: // array elements have no key to build a path from, so nothing beneath this point is captured forwardDepth++; captureKey = null; - status = suppressing ? Status.ADVANCED : forward(sink, source, event); + status = forward(sink, source, event); break; case END_OBJECT: forwardDepth--; @@ -219,11 +205,7 @@ private Status onForwarding( { path.pop(); } - status = suppressing ? Status.ADVANCED : forward(sink, source, event); - if (suppressing && forwardDepth == suppressDepth) - { - suppressing = false; - } + status = forward(sink, source, event); if (forwardDepth == 0) { forwarding = false; @@ -231,58 +213,21 @@ private Status onForwarding( break; case END_ARRAY: forwardDepth--; - status = suppressing ? Status.ADVANCED : forward(sink, source, event); - if (suppressing && forwardDepth == suppressDepth) - { - suppressing = false; - } + status = forward(sink, source, event); if (forwardDepth == 0) { forwarding = false; } break; case KEY_NAME: - if (suppressing) - { - status = Status.ADVANCED; - } - else if (forwardDepth == 1 && source.deferredBytes()) - { - // an excluded-key decision needs the whole key, not a fragment of it -- decline so the - // source accumulates it and re-presents it whole on a later window, the same fallback a - // content-needing scalar value uses - upstream.consumed(0); - status = Status.STARVED; - } - else - { - final String name = source.getStringView().toString(); - captureKey = capturePath(name); - text.setLength(0); - if (forwardDepth == 1 && excludedKeys.contains(name)) - { - suppressing = true; - suppressDepth = forwardDepth; - status = Status.ADVANCED; - } - else - { - status = forward(sink, source, event); - } - } + captureKey = capturePath(source.getStringView().toString()); + text.setLength(0); + status = forward(sink, source, event); break; case VALUE_STRING: case VALUE_NUMBER: if (captureKey != null) { - if (suppressing) - { - // withheld from the sink, so nothing downstream consumes this value's bytes and - // advances the source's per-fragment cursor -- getStringView() re-presents everything - // seen so far on every call instead of just the newest delta, so replace rather than - // accumulate - text.setLength(0); - } text.append(source.getStringView()); if (!source.deferredBytes()) { @@ -291,11 +236,7 @@ else if (forwardDepth == 1 && source.deferredBytes()) captureKey = null; } } - status = suppressing ? Status.ADVANCED : forward(sink, source, event); - if (suppressing && !source.deferredBytes() && forwardDepth == suppressDepth) - { - suppressing = false; - } + status = forward(sink, source, event); break; case VALUE_TRUE: if (captureKey != null) @@ -303,11 +244,7 @@ else if (forwardDepth == 1 && source.deferredBytes()) captured.put(captureKey, "true"); captureKey = null; } - status = suppressing ? Status.ADVANCED : forward(sink, source, event); - if (suppressing && forwardDepth == suppressDepth) - { - suppressing = false; - } + status = forward(sink, source, event); break; case VALUE_FALSE: if (captureKey != null) @@ -315,24 +252,16 @@ else if (forwardDepth == 1 && source.deferredBytes()) captured.put(captureKey, "false"); captureKey = null; } - status = suppressing ? Status.ADVANCED : forward(sink, source, event); - if (suppressing && forwardDepth == suppressDepth) - { - suppressing = false; - } + status = forward(sink, source, event); break; case VERBATIM: // rides alongside the structured event stream for the same value rather than substituting for // it (see the class Javadoc), so it must not disturb an in-progress capture - status = suppressing ? Status.ADVANCED : forward(sink, source, event); + status = forward(sink, source, event); break; default: captureKey = null; - status = suppressing ? Status.ADVANCED : forward(sink, source, event); - if (suppressing && forwardDepth == suppressDepth) - { - suppressing = false; - } + status = forward(sink, source, event); break; } return status; diff --git a/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java b/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java deleted file mode 100644 index 0c0c16428e0..00000000000 --- a/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/transform/McpHttpArgumentsTest.java +++ /dev/null @@ -1,157 +0,0 @@ -/* - * Copyright 2021-2026 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.http.internal.transform; - -import static java.nio.charset.StandardCharsets.UTF_8; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; - -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; - -import org.junit.Test; - -import io.aklivity.zilla.runtime.common.agrona.buffer.MutableDirectBufferEx; -import io.aklivity.zilla.runtime.common.agrona.buffer.UnsafeBufferEx; -import io.aklivity.zilla.runtime.common.json.JsonEx; -import io.aklivity.zilla.runtime.common.json.JsonGeneratorEx; -import io.aklivity.zilla.runtime.common.json.JsonPipeline; -import io.aklivity.zilla.runtime.common.json.JsonPipeline.Status; -import io.aklivity.zilla.runtime.common.json.JsonSink; - -public class McpHttpArgumentsTest -{ - @Test - public void shouldForwardAllArgumentsWhenNoneExcluded() - { - String input = "{\"name\":\"tool\",\"arguments\":{\"a\":\"1\",\"b\":\"2\"}}"; - - Map captured = new LinkedHashMap<>(); - String output = reroot(input, List.of(), captured); - - assertEquals("{\"a\":\"1\",\"b\":\"2\"}", output); - assertEquals(Map.of("a", "1", "b", "2"), captured); - } - - @Test - public void shouldWithholdExcludedTopLevelArgumentFromOutputButStillCaptureIt() - { - String input = "{\"name\":\"update_connector_config\",\"arguments\":" + - "{\"connector\":\"file-source-demo\",\"connector.class\":\"FileStreamSource\",\"topic\":\"connect-demo\"}}"; - - Map captured = new LinkedHashMap<>(); - String output = reroot(input, List.of("connector"), captured); - - assertEquals("{\"connector.class\":\"FileStreamSource\",\"topic\":\"connect-demo\"}", output); - assertEquals(Map.of( - "connector", "file-source-demo", - "connector.class", "FileStreamSource", - "topic", "connect-demo"), captured); - } - - @Test - public void shouldWithholdExcludedNestedObjectValueEntirely() - { - String input = "{\"name\":\"tool\",\"arguments\":" + - "{\"connector\":{\"nested\":\"x\"},\"topic\":\"connect-demo\"}}"; - - Map captured = new LinkedHashMap<>(); - String output = reroot(input, List.of("connector"), captured); - - assertEquals("{\"topic\":\"connect-demo\"}", output); - assertFalse(captured.containsKey("connector")); - } - - @Test - public void shouldWithholdOnlyExcludedKeyAmongMultipleArguments() - { - String input = "{\"name\":\"validate_connector_config\",\"arguments\":" + - "{\"pluginName\":\"FileStreamSource\",\"connector.class\":\"FileStreamSource\"," + - "\"file\":\"/tmp/kc-source.txt\",\"topic\":\"connect-demo\",\"name\":\"file-source-demo\"}}"; - - Map captured = new LinkedHashMap<>(); - String output = reroot(input, List.of("pluginName"), captured); - - assertEquals("{\"connector.class\":\"FileStreamSource\",\"file\":\"/tmp/kc-source.txt\"," + - "\"topic\":\"connect-demo\",\"name\":\"file-source-demo\"}", output); - assertEquals("FileStreamSource", captured.get("pluginName")); - } - - @Test - public void shouldWithholdExcludedTopLevelArgumentWhenFragmentedAcrossInputWindows() - { - String input = "{\"name\":\"create_pr\",\"arguments\":" + - "{\"owner\":\"acme\",\"repo\":\"widget\",\"title\":\"Add feature\"," + - "\"pr\":{\"branch\":\"feature\",\"target\":\"main\"}}}"; - - for (int window = 1; window <= input.length(); window++) - { - Map captured = new LinkedHashMap<>(); - String output = rerootWindowed(input, List.of("owner", "repo"), captured, window); - - assertEquals("window=" + window, - "{\"title\":\"Add feature\",\"pr\":{\"branch\":\"feature\",\"target\":\"main\"}}", output); - assertEquals("window=" + window, "acme", captured.get("owner")); - assertEquals("window=" + window, "widget", captured.get("repo")); - } - } - - private static String reroot( - String input, - List excludedKeys, - Map captured) - { - return rerootWindowed(input, excludedKeys, captured, input.length()); - } - - private static String rerootWindowed( - String input, - List excludedKeys, - Map captured, - int window) - { - McpHttpArguments transform = new McpHttpArguments(captured, excludedKeys); - - JsonGeneratorEx gen = JsonEx.createGenerator(); - MutableDirectBufferEx buffer = new UnsafeBufferEx(new byte[4096]); - gen.wrap(buffer, 0, buffer.capacity()); - JsonPipeline pipeline = JsonEx.stream(JsonEx.createParser()) - .transform(transform) - .into(JsonEx.createSink(gen, Map.of(JsonSink.DELIVERY, JsonSink.Delivery.STRUCTURED))); - pipeline.reset(); - - byte[] msg = input.getBytes(UTF_8); - int progress = 0; - int limit = 0; - Status status = Status.STARVED; - int guard = 0; - while (status == Status.STARVED && guard++ < 10_000) - { - limit = Math.min(limit + window, msg.length); - boolean last = limit >= msg.length; - status = pipeline.transform(new UnsafeBufferEx(msg), progress, limit, last); - if (status == Status.STARVED) - { - progress = limit - pipeline.remaining(); - } - } - assertEquals(Status.COMPLETED, status); - - byte[] out = new byte[gen.length()]; - buffer.getBytes(0, out); - return new String(out, UTF_8); - } -} From bf458fa050cc90cb4f51422e59788f87553f5bfe Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:50:39 +0000 Subject: [PATCH 6/9] feat(common-json,common-openapi): honor additionalProperties in retained paths JsonSchema#retainedPaths() (and thus JsonTransforms.projector(JsonSchema), which shapes output from a schema by pruning to the paths it retains) only ever walked a schema's named properties/items/combinators -- additionalProperties was parsed and enforced for validation, but never consulted when computing what to keep. A schema combining named properties with a permissive additionalProperties therefore had no way to express "these declared fields, plus anything else of this shape": the pruning step silently dropped whatever wasn't explicitly named, the same as if additionalProperties were absent or false. JsonSchemaImpl now adds a wildcard retained path (the object-key counterpart to the existing array "-" wildcard, using a distinct "*" segment since an object key literally named "-" already matches as a plain key) alongside a structured schema's named properties whenever additionalProperties is explicitly present and not false -- recursing into its own sub-schema when typed, or treating it as an open leaf otherwise. Absent additionalProperties keeps every existing schema's closed, pruned- to-named-properties behavior unchanged. JsonProjectorImpl's object-key lookup gains the matching wildcard fallback, and a trie node's fragment-decline bound no longer short-circuits to SKIP once a wildcard sibling means a key longer than every named candidate can still match. common-openapi's OpenapiSchema/OpenapiSchemaView gain the additionalProperties field itself (previously absent from the model entirely, so it was silently dropped by JSON-B during parsing regardless of what an OpenAPI document wrote) -- captured as a raw JsonValue and carried through to the schema text a consumer like binding-mcp-openapi hands to JsonSchema.of(), rather than resolved/recursively bound like items/properties/schema, since a boolean literal is a valid alternative to a nested schema object here. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../json/internal/JsonProjectorImpl.java | 24 +++++++-- .../common/json/internal/JsonSchemaImpl.java | 14 ++++++ .../common/json/JsonProjectorTest.java | 27 ++++++++++ .../json/JsonSchemaRetainedPathsTest.java | 50 +++++++++++++++++++ .../common/openapi/model/OpenapiSchema.java | 6 +++ .../openapi/view/OpenapiSchemaView.java | 4 ++ 6 files changed, 122 insertions(+), 3 deletions(-) diff --git a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java index b1b5ce18770..023ce747240 100644 --- a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java +++ b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java @@ -50,6 +50,7 @@ public final class JsonProjectorImpl implements JsonTransform { private static final int MAX_DEPTH = 64; private static final String WILDCARD = "-"; + private static final String WILDCARD_PROPERTY = "*"; private enum Decision { @@ -636,7 +637,9 @@ else if (node.keys.length > 0) return result; } - // Matches an object key against a node's children by the live char view, allocation-free. + // Matches an object key against a node's children by the live char view, allocation-free -- preferring + // an explicit named child over the "*" wildcard, the object-key counterpart to lookupIndex's array + // wildcard fallback below. private static Node lookup( Node node, CharSequence key) @@ -644,13 +647,23 @@ private static Node lookup( Node result = null; if (node != null) { + Node wildcard = null; for (int i = 0; result == null && i < node.keys.length; i++) { - if (charsEqual(node.keys[i], key)) + String segment = node.keys[i]; + if (WILDCARD_PROPERTY.equals(segment)) + { + wildcard = node.nodes[i]; + } + else if (charsEqual(segment, key)) { result = node.nodes[i]; } } + if (result == null) + { + result = wildcard; + } } return result; } @@ -776,11 +789,16 @@ private Node( this.nodes = nodes; this.keepAll = keepAll; int longest = 0; + boolean wildcard = false; for (String key : keys) { longest = Math.max(longest, key.length()); + wildcard |= WILDCARD_PROPERTY.equals(key); } - this.maxKeyLength = longest; + // a "*" child matches a key of any length, so a fragment already longer than every named + // sibling still can't be ruled out early -- only a closed enumeration (no wildcard) can use + // the max-named-length bound to declare SKIP before the key is fully reassembled + this.maxKeyLength = wildcard ? Integer.MAX_VALUE : longest; } } diff --git a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java index cc217c0e2a3..3ecca96d8b7 100644 --- a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java +++ b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java @@ -88,6 +88,11 @@ public final class JsonSchemaImpl implements JsonSchema private static final JsonSchemaImpl ANY = new JsonSchemaImpl(false); private static final JsonSchemaImpl NONE = new JsonSchemaImpl(true); + // reserved retained-path segment standing for "any object key not matched by a named property" -- + // distinct from JsonProjectorImpl's array-only "-" wildcard, since an object key literally named "-" + // is matched like any other key there + private static final String WILDCARD_PROPERTY = "*"; + private static final String[] NO_KEYS = new String[0]; private static final JsonSchemaImpl[] NO_SCHEMAS = new JsonSchemaImpl[0]; @@ -343,6 +348,15 @@ private void collectPaths( { entry.getValue().collectPaths(pointer + "/" + escapePointer(entry.getKey()), visitor, pointers); } + // an object with named properties is otherwise closed to its declared keys (the same as when + // additionalProperties is absent entirely, preserving every existing schema's pruning + // behavior) -- only an explicit, non-false additionalProperties widens it, retaining a + // wildcard path alongside the named ones for whatever key doesn't match them + if (hasAdditional && additionalAllowed) + { + JsonSchemaImpl additional = additionalSchema != null ? additionalSchema : ANY; + additional.collectPaths(pointer + "/" + WILDCARD_PROPERTY, visitor, pointers); + } } if (items != null) { diff --git a/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonProjectorTest.java b/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonProjectorTest.java index f1ba250a706..b25278da038 100644 --- a/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonProjectorTest.java +++ b/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonProjectorTest.java @@ -50,6 +50,33 @@ void shouldRetainArrayWildcardField() "{\"items\":[{\"id\":1,\"x\":9},{\"id\":2,\"y\":8}],\"k\":0}")); } + @Test + void shouldRetainPropertyWildcardField() + { + assertEquals("{\"a\":1,\"x\":2,\"y\":3}", + project(List.of("/a", "/*"), "{\"a\":1,\"x\":2,\"y\":3}")); + } + + @Test + void shouldPreferNamedPropertyOverWildcard() + { + // the wildcard is a fallback, not an override -- a key with its own retained pointer keeps that + // pointer's shape even when it would also match the sibling wildcard + assertEquals("{\"a\":{\"b\":1},\"c\":{\"d\":2,\"e\":3}}", + project(List.of("/a/b", "/*"), "{\"a\":{\"b\":1,\"z\":9},\"c\":{\"d\":2,\"e\":3}}")); + } + + @Test + void shouldRetainPropertyWildcardFieldThatFragmentsAcrossInputWindows() + { + // the unmatched key is longer than every named sibling (and the feed window), so onKey's + // exceeds-max-named-length shortcut must not declare SKIP early now that a wildcard sibling means + // a longer key can still match + String key = "x".repeat(40); + assertEquals("{\"a\":1,\"" + key + "\":2}", + projectWindowed(List.of("/a", "/*"), "{\"a\":1,\"" + key + "\":2}", 8)); + } + @Test void shouldRetainExplicitArrayIndex() { diff --git a/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java b/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java index d8efc854b4d..8c52045df22 100644 --- a/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java +++ b/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java @@ -48,6 +48,34 @@ void shouldCollectArrayItemsWildcard() "{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"integer\"}}}}}}")); } + @Test + void shouldCollectWildcardWhenAdditionalPropertiesAllowed() + { + assertEquals(List.of("/a", "/*"), + retained("{\"properties\":{\"a\":{\"type\":\"integer\"}},\"additionalProperties\":true}")); + } + + @Test + void shouldCollectWildcardForTypedAdditionalPropertiesSchema() + { + assertEquals(List.of("/a", "/*"), + retained("{\"properties\":{\"a\":{\"type\":\"integer\"}},\"additionalProperties\":{\"type\":\"string\"}}")); + } + + @Test + void shouldNotCollectWildcardWhenAdditionalPropertiesAbsent() + { + assertEquals(List.of("/a"), + retained("{\"properties\":{\"a\":{\"type\":\"integer\"}}}")); + } + + @Test + void shouldNotCollectWildcardWhenAdditionalPropertiesFalse() + { + assertEquals(List.of("/a"), + retained("{\"properties\":{\"a\":{\"type\":\"integer\"}},\"additionalProperties\":false}")); + } + @Test void shouldTreatStructurelessObjectAsRetainedLeaf() { @@ -120,6 +148,28 @@ void shouldDriveProjectorEndToEnd() assertEquals("{\"items\":[{\"id\":1},{\"id\":2}]} ", new String(out, UTF_8)); } + @Test + void shouldDriveProjectorEndToEndWithAdditionalProperties() + { + JsonGeneratorEx gen = JsonEx.createGenerator(); + MutableDirectBufferEx buffer = new UnsafeBufferEx(new byte[1024]); + gen.wrap(buffer, 0, buffer.capacity()); + JsonSchema schema = JsonSchema.of( + "{\"type\":\"object\",\"properties\":{\"connector.class\":{\"type\":\"string\"}}," + + "\"additionalProperties\":true}"); + JsonPipeline pipeline = JsonEx.stream(JsonEx.createParser()) + .transform(JsonTransforms.projector(schema.retainedPaths())) + .into(JsonEx.createSink(gen)); + pipeline.reset(); + byte[] bytes = "{\"connector.class\":\"FileStreamSource\",\"file\":\"/tmp/x\",\"topic\":\"t\"} " + .getBytes(UTF_8); + pipeline.transform(new UnsafeBufferEx(bytes), 0, bytes.length); + byte[] out = new byte[gen.length()]; + buffer.getBytes(0, out); + assertEquals("{\"connector.class\":\"FileStreamSource\",\"file\":\"/tmp/x\",\"topic\":\"t\"} ", + new String(out, UTF_8)); + } + private static List retained( String schema) { diff --git a/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/model/OpenapiSchema.java b/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/model/OpenapiSchema.java index 641716cc599..a9088d2e208 100644 --- a/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/model/OpenapiSchema.java +++ b/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/model/OpenapiSchema.java @@ -17,6 +17,7 @@ import java.util.List; import java.util.Map; +import jakarta.json.JsonValue; import jakarta.json.bind.annotation.JsonbProperty; public class OpenapiSchema extends AbstractOpenapiResolvable @@ -25,6 +26,11 @@ public class OpenapiSchema extends AbstractOpenapiResolvable public OpenapiSchema items; public Map properties; public List required; + // captured as a raw JsonValue (rather than resolved/recursively-bound like items/properties/schema) + // since a boolean literal (true/false) is a valid alternative to a nested schema object here, and + // consumers that shape output from this schema (e.g. common-json's JsonSchemaImpl) already understand + // both forms when parsing the serialized schema text + public JsonValue additionalProperties; public String format; public String description; @JsonbProperty("enum") diff --git a/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/view/OpenapiSchemaView.java b/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/view/OpenapiSchemaView.java index 572e4b3aa3e..0ef82686f81 100644 --- a/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/view/OpenapiSchemaView.java +++ b/runtime/common-openapi/src/main/java/io/aklivity/zilla/runtime/common/openapi/view/OpenapiSchemaView.java @@ -21,6 +21,7 @@ import java.util.Optional; import java.util.stream.Collectors; +import jakarta.json.JsonValue; import jakarta.json.bind.annotation.JsonbProperty; import jakarta.json.bind.annotation.JsonbPropertyOrder; @@ -117,6 +118,7 @@ public Optional extension( "type", "items", "properties", + "additionalProperties", "required", "format", "description", @@ -146,6 +148,7 @@ public static final class OpenapiJsonSchema public String type; public OpenapiSchema items; public Map properties; + public JsonValue additionalProperties; public List required; public String format; public String description; @@ -181,6 +184,7 @@ public static OpenapiJsonSchema of( ? model.properties.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, e -> of(e.getValue()))) : null; + json.additionalProperties = model.additionalProperties; json.required = model.required; json.format = model.format; json.description = model.description; From 383ae74d0636f4e1910dee34318aecd16d263f5c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:53:19 +0000 Subject: [PATCH 7/9] wip(binding-mcp-kafka-connect): target schema for update/validate connector config update_connector_config and validate_connector_config now declare connector.class/tasks.max as real named properties plus additionalProperties: true, matching standard OpenAPI/JSON Schema idiom and relying on the additionalProperties support just added to common-json/common-openapi. This alone is not yet sufficient: the operation's own path parameter (connector/pluginName) shares the same tools/call arguments object as the body, so it would currently also pass through the new wildcard into the outbound body -- excluding it needs either proper deny-path support in common-json's JsonSchema/JsonProjectorImpl (a keep-only pointer list can't express "deny this even though a sibling wildcard would keep it") or a separate mechanism, still to be decided. tools.list's advertised schema is updated to match the target shape. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../schema/kafka-connect.openapi.json | 22 +++++++++++++++++-- .../connect/streams/mcp/tools.list/client.rpt | 4 ++-- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json b/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json index a6d493607bc..e21c1c49fc3 100644 --- a/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json +++ b/runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json @@ -136,7 +136,16 @@ "content": { "application/json": { "schema": { - "type": "object" + "type": "object", + "properties": { + "connector.class": { + "type": "string" + }, + "tasks.max": { + "type": "string" + } + }, + "additionalProperties": true } } } @@ -167,7 +176,16 @@ "content": { "application/json": { "schema": { - "type": "object" + "type": "object", + "properties": { + "connector.class": { + "type": "string" + }, + "tasks.max": { + "type": "string" + } + }, + "additionalProperties": true } } } diff --git a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt index 53f9d610584..e47a8d98e7c 100644 --- a/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt +++ b/specs/binding-mcp-kafka-connect.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/connect/streams/mcp/tools.list/client.rpt @@ -56,8 +56,8 @@ read '{"tools":[{"name":"list_connectors","title":"List Connectors","descriptio '{"name":"describe_connector","title":"Describe Connector","description":"Read the configuration and task list of a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' '{"name":"delete_connector","title":"Delete Connector","description":"Delete a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":true,"idempotentHint":true,"openWorldHint":false}},' '{"name":"describe_connector_config","title":"Describe Connector Config","description":"Read the effective configuration of a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"outputSchema":{"type":"object"},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' - '{"name":"update_connector_config","title":"Update Connector Config","description":"Create or update a connector by setting its full configuration.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' - '{"name":"validate_connector_config","title":"Validate Connector Config","description":"Validate a connector configuration against a plugin configuration definition, without creating or updating any connector.","inputSchema":{"type":"object","properties":{"pluginName":{"type":"string"}},"required":["pluginName"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' + '{"name":"update_connector_config","title":"Update Connector Config","description":"Create or update a connector by setting its full configuration.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"},"connector.class":{"type":"string"},"tasks.max":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' + '{"name":"validate_connector_config","title":"Validate Connector Config","description":"Validate a connector configuration against a plugin configuration definition, without creating or updating any connector.","inputSchema":{"type":"object","properties":{"pluginName":{"type":"string"},"connector.class":{"type":"string"},"tasks.max":{"type":"string"}},"required":["pluginName"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' '{"name":"describe_connector_status","title":"Describe Connector Status","description":"Read the current state of a connector and the state of each of its tasks.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' '{"name":"restart_connector","title":"Restart Connector","description":"Restart a connector.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":false,"openWorldHint":false}},' '{"name":"pause_connector","title":"Pause Connector","description":"Pause a connector and all of its tasks.","inputSchema":{"type":"object","properties":{"connector":{"type":"string"}},"required":["connector"]},"annotations":{"readOnlyHint":false,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}},' From 24ade3249f7612bad63daf56029dbcab1cbc09b5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 01:58:42 +0000 Subject: [PATCH 8/9] feat(common-json): add JsonSchema#rejectedPaths(), explicit deny over wildcard retainedPaths() is a pure keep-list: a named property whose schema is false correctly gets no pointer (it's neither kept), but "no pointer" and "unlisted key covered only by a sibling additionalProperties wildcard" were indistinguishable to JsonProjectorImpl -- so a structured schema with both a denied named property and a permissive additionalProperties (e.g. an operation's own path parameter declared false alongside a wildcard for its otherwise-open body) had no way to keep the wildcard from swallowing the one key it was supposed to exclude. JsonSchema#rejectedPaths() collects the RFC 6901 pointers to explicitly deny (JsonSchemaImpl already tracked this per-node as `deny`, just never surfaced it), mirroring retainedPaths()'s own collection pass. JsonTransforms#projector(JsonSchema) now feeds both lists to JsonProjectorImpl, whose trie nodes gain a `rejected` flag alongside `keepAll` -- checked first in decide(), so an explicit reject always wins over this same node's own keepAll, even when reached only via a wildcard sibling. The trie's fragment-decline bound already treats a wildcard sibling as "no early SKIP" (from the additionalProperties support just added); this needed no further change since a rejected node is reached by the same exact-match path as any other named child. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../zilla/runtime/common/json/JsonSchema.java | 11 ++++ .../runtime/common/json/JsonTransforms.java | 6 ++- .../json/internal/JsonProjectorImpl.java | 45 +++++++++++++--- .../common/json/internal/JsonSchemaImpl.java | 8 +++ .../json/JsonSchemaRetainedPathsTest.java | 52 +++++++++++++++++++ 5 files changed, 113 insertions(+), 9 deletions(-) diff --git a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonSchema.java b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonSchema.java index c145b371b1c..db0e59c5aae 100644 --- a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonSchema.java +++ b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonSchema.java @@ -142,6 +142,17 @@ JsonTransform validator( */ List retainedPaths(); + /** + * Returns the RFC 6901 JSON Pointers explicitly denied ({@code false}, or an equivalent + * always-fails sub-schema) when projecting an instance of this schema — the union of such paths + * declared across all branches. A path here always wins over the same path being retained + * elsewhere (e.g. a structured schema's own named property denied while its sibling {@code + * additionalProperties} would otherwise keep it): {@link JsonTransforms#projector(JsonSchema)} + * feeds both lists to the same projector, and a rejected path is excluded even where a broader + * wildcard from {@link #retainedPaths()} would otherwise retain it. + */ + List rejectedPaths(); + /** * The compiled sub-schema declared for object property {@code name} under this schema's {@code * properties} keyword, or {@code null} when this schema declares no such property. diff --git a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonTransforms.java b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonTransforms.java index 86a96e1003d..d23c067d234 100644 --- a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonTransforms.java +++ b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonTransforms.java @@ -44,12 +44,14 @@ public static JsonTransform projector( /** * Returns a {@link JsonTransform} pruning a document to the paths retained by {@code schema} - * (see {@link JsonSchema#retainedPaths()}). + * (see {@link JsonSchema#retainedPaths()}), excluding any path {@code schema} explicitly denies + * (see {@link JsonSchema#rejectedPaths()}) even where a broader retained path would otherwise + * keep it. */ public static JsonTransform projector( JsonSchema schema) { - return new JsonProjectorImpl(schema.retainedPaths()); + return new JsonProjectorImpl(schema.retainedPaths(), schema.rejectedPaths()); } /** diff --git a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java index 023ce747240..20cde0aac0c 100644 --- a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java +++ b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonProjectorImpl.java @@ -106,7 +106,14 @@ private enum SegMode public JsonProjectorImpl( List pointers) { - this.root = compile(pointers); + this(pointers, List.of()); + } + + public JsonProjectorImpl( + List retained, + List rejected) + { + this.root = compile(retained, rejected); } @Override @@ -622,6 +629,14 @@ private static Decision decide( { result = Decision.SKIP; } + else if (node.rejected) + { + // an explicit reject always wins, even over this same node's own keepAll -- a pointer can + // only ever land in one of retained/rejected (JsonSchemaImpl's retainedPaths/rejectedPaths + // partition every leaf by its own deny flag), but a caller combining pointer lists by hand + // should still get the safer, deny-wins outcome on an accidental overlap + result = Decision.SKIP; + } else if (node.keepAll) { result = Decision.KEEP_ALL; @@ -735,10 +750,11 @@ private static boolean matchesIndex( } private static Node compile( - List pointers) + List retained, + List rejected) { NodeBuilder builder = new NodeBuilder(); - for (String pointer : pointers) + for (String pointer : retained) { NodeBuilder node = builder; for (String segment : segments(pointer)) @@ -747,6 +763,15 @@ private static Node compile( } node.keepAll = true; } + for (String pointer : rejected) + { + NodeBuilder node = builder; + for (String segment : segments(pointer)) + { + node = node.child(segment); + } + node.rejected = true; + } return builder.build(); } @@ -771,23 +796,28 @@ private static String[] segments( } // An immutable trie node: children are parallel key/node arrays scanned linearly (a handful of children - // per node), keepAll marks a node where a retained pointer terminates, and maxKeyLength is the longest - // of this node's own children's keys — the bound onKey declines a fragmenting child key against. + // per node), keepAll marks a node where a retained pointer terminates, rejected marks a node where a + // rejected pointer terminates (and always wins over this same node's own keepAll, see decide()), and + // maxKeyLength is the longest of this node's own children's keys — the bound onKey declines a + // fragmenting child key against. private static final class Node { private final String[] keys; private final Node[] nodes; private final boolean keepAll; + private final boolean rejected; private final int maxKeyLength; private Node( String[] keys, Node[] nodes, - boolean keepAll) + boolean keepAll, + boolean rejected) { this.keys = keys; this.nodes = nodes; this.keepAll = keepAll; + this.rejected = rejected; int longest = 0; boolean wildcard = false; for (String key : keys) @@ -806,6 +836,7 @@ private static final class NodeBuilder { private final Map children = new LinkedHashMap<>(); private boolean keepAll; + private boolean rejected; private NodeBuilder child( String segment) @@ -825,7 +856,7 @@ private Node build() nodes[i] = entry.getValue().build(); i++; } - return new Node(keys, nodes, keepAll); + return new Node(keys, nodes, keepAll, rejected); } } diff --git a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java index 3ecca96d8b7..072a69428fc 100644 --- a/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java +++ b/runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/internal/JsonSchemaImpl.java @@ -195,6 +195,7 @@ private enum Verdict private final JsonNode raw; private List retainedPaths; + private List rejectedPaths; public static JsonSchema of( String schema) @@ -231,6 +232,7 @@ public static JsonSchema of( if (result != ANY && result != NONE) { result.retainedPaths = result.collectMatchingPaths((node, structured) -> !structured && !node.deny); + result.rejectedPaths = result.collectMatchingPaths((node, structured) -> !structured && node.deny); } return result; } @@ -309,6 +311,12 @@ public List retainedPaths() return retainedPaths != null ? retainedPaths : List.of(); } + @Override + public List rejectedPaths() + { + return rejectedPaths != null ? rejectedPaths : List.of(); + } + @Override public List matchingPaths( Predicate filter) diff --git a/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java b/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java index 8c52045df22..8ad1eac9c69 100644 --- a/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java +++ b/runtime/common-json/src/test/java/io/aklivity/zilla/runtime/common/json/JsonSchemaRetainedPathsTest.java @@ -76,6 +76,28 @@ void shouldNotCollectWildcardWhenAdditionalPropertiesFalse() retained("{\"properties\":{\"a\":{\"type\":\"integer\"}},\"additionalProperties\":false}")); } + @Test + void shouldCollectRejectedPathForFalseProperty() + { + assertEquals(List.of("/b"), rejected("{\"properties\":{\"a\":true,\"b\":false}}")); + } + + @Test + void shouldCollectRejectedPathAlongsideAdditionalPropertiesWildcard() + { + JsonSchema schema = JsonSchema.of( + "{\"type\":\"object\",\"properties\":{\"connector.class\":{\"type\":\"string\"}," + + "\"connector\":false},\"additionalProperties\":true}"); + assertEquals(List.of("/connector.class", "/*"), schema.retainedPaths()); + assertEquals(List.of("/connector"), schema.rejectedPaths()); + } + + @Test + void shouldHaveNoRejectedPathsWhenNoPropertyIsDenied() + { + assertEquals(List.of(), rejected("{\"properties\":{\"a\":{\"type\":\"integer\"}}}")); + } + @Test void shouldTreatStructurelessObjectAsRetainedLeaf() { @@ -170,9 +192,39 @@ void shouldDriveProjectorEndToEndWithAdditionalProperties() new String(out, UTF_8)); } + @Test + void shouldDriveProjectorEndToEndRejectingNamedPropertyOverWildcard() + { + // "connector" shares the same object as connector.class and the open-ended config fields (the + // shape of a tools/call arguments object carrying both a path parameter and a generic body), and + // must stay excluded even though additionalProperties would otherwise keep it via the wildcard + JsonGeneratorEx gen = JsonEx.createGenerator(); + MutableDirectBufferEx buffer = new UnsafeBufferEx(new byte[1024]); + gen.wrap(buffer, 0, buffer.capacity()); + JsonSchema schema = JsonSchema.of( + "{\"type\":\"object\",\"properties\":{\"connector.class\":{\"type\":\"string\"}," + + "\"connector\":false},\"additionalProperties\":true}"); + JsonPipeline pipeline = JsonEx.stream(JsonEx.createParser()) + .transform(JsonTransforms.projector(schema)) + .into(JsonEx.createSink(gen)); + pipeline.reset(); + byte[] bytes = ("{\"connector\":\"connector1\",\"connector.class\":\"FileStreamSource\"," + + "\"topic\":\"t\"} ").getBytes(UTF_8); + pipeline.transform(new UnsafeBufferEx(bytes), 0, bytes.length); + byte[] out = new byte[gen.length()]; + buffer.getBytes(0, out); + assertEquals("{\"connector.class\":\"FileStreamSource\",\"topic\":\"t\"} ", new String(out, UTF_8)); + } + private static List retained( String schema) { return JsonSchema.of(schema).retainedPaths(); } + + private static List rejected( + String schema) + { + return JsonSchema.of(schema).rejectedPaths(); + } } From 3820888826b64243ffbe598a5e1052d29d55613f Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 02:01:41 +0000 Subject: [PATCH 9/9] feat(binding-mcp-openapi): deny operation parameters in the generated body schema An operation's path/query/header/cookie parameters share the same tools/call arguments object as its request body. Once a structured body schema is widened by additionalProperties (now honored by common-json's JsonSchema#rejectedPaths()/retainedPaths()), a parameter name not already a declared body property would otherwise pass straight through that wildcard into the outbound body alongside the fields it actually describes. bodySchema() now denies every such parameter name explicitly (a false sub-schema merged into the schema's own properties) unless it's already a real declared body property -- e.g. pulls/create's requestBody deliberately declares its own "owner" field alongside the path parameter of the same name, and that stays untouched. JsonSchema#rejectedPaths() picks up the denied names and the body projector excludes them even via the wildcard, per the mechanism just added to common-json. Verifying against the full binding-mcp-openapi k3po suite before this lands in the kafka-connect module that motivated it. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01CLFwU8RqaAWL8xDBXSENvr --- .../McpOpenapiCompositeGenerator.java | 46 ++++++++++- .../McpOpenapiCompositeGeneratorTest.java | 80 +++++++++++++++++++ 2 files changed, 125 insertions(+), 1 deletion(-) diff --git a/runtime/binding-mcp-openapi/src/main/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGenerator.java b/runtime/binding-mcp-openapi/src/main/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGenerator.java index 725694bd0a8..e73333dafc6 100644 --- a/runtime/binding-mcp-openapi/src/main/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGenerator.java +++ b/runtime/binding-mcp-openapi/src/main/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGenerator.java @@ -1186,13 +1186,57 @@ private static String bodySchema( { if (typed.schema != null) { - result = toSchemaJson(jsonb, typed.schema.model); + result = denyParameterNames(toSchemaJson(jsonb, typed.schema.model), operation); } break; } return result; } + // A structured body schema (named properties, possibly widened by additionalProperties) shares the + // same tools/call arguments object as the operation's own path/query/header/cookie parameters. Once + // additionalProperties admits unnamed keys, a parameter name would otherwise pass straight through + // that wildcard into the outbound body alongside the fields it actually describes -- so every + // parameter not already a declared body property is denied explicitly (a false sub-schema), which + // JsonSchema#rejectedPaths() surfaces to the body projector as always winning over the wildcard. + private static String denyParameterNames( + String schemaJson, + OpenapiOperationView operation) + { + String result = schemaJson; + if (operation.parameters != null && !operation.parameters.isEmpty()) + { + JsonValue parsed = Json.createReader(new StringReader(schemaJson)).readValue(); + if (parsed instanceof JsonObject) + { + JsonObject schema = (JsonObject) parsed; + JsonValue properties = schema.get("properties"); + if (properties instanceof JsonObject) + { + JsonObject declared = (JsonObject) properties; + JsonObjectBuilder denied = Json.createObjectBuilder(declared); + boolean modified = false; + for (OpenapiParameterView parameter : operation.parameters) + { + if (!declared.containsKey(parameter.name)) + { + denied.add(parameter.name, JsonValue.FALSE); + modified = true; + } + } + if (modified) + { + result = Json.createObjectBuilder(schema) + .add("properties", denied) + .build() + .toString(); + } + } + } + } + return result; + } + private static JsonObject schemaObject( OpenapiSchemaView schema) { diff --git a/runtime/binding-mcp-openapi/src/test/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGeneratorTest.java b/runtime/binding-mcp-openapi/src/test/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGeneratorTest.java index 693189c5baa..6c944d53dd3 100644 --- a/runtime/binding-mcp-openapi/src/test/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGeneratorTest.java +++ b/runtime/binding-mcp-openapi/src/test/java/io/aklivity/zilla/runtime/binding/mcp/openapi/internal/config/composite/McpOpenapiCompositeGeneratorTest.java @@ -749,6 +749,86 @@ public void shouldFlattenInputSchemaWithBodyCollisionSuffix() assertThat(required, not(hasItem("owner_body"))); } + @Test + public void shouldDenyPathParameterInOpenBodySchema() + { + String spec = + """ + { + "openapi": "3.0.1", + "info": { "title": "things", "version": "1.0" }, + "servers": [ { "url": "https://api.example.com" } ], + "paths": { + "/things/{id}": { + "put": { + "operationId": "update_thing", + "parameters": [ + { "name": "id", "in": "path", "required": true, "schema": { "type": "string" } } + ], + "requestBody": { + "required": true, + "content": { "application/json": { "schema": { + "type": "object", + "properties": { "name": { "type": "string" } }, + "additionalProperties": true } } } + }, + "responses": { "200": { "description": "ok" } } + } + } + } + } + """; + lenient().when(catalog.resolve(eq("things-api"), eq("latest"))).thenReturn(77); + lenient().when(catalog.resolve(eq(77))).thenReturn(spec); + + BindingConfig binding = GenericBindingConfig.builder() + .namespace("test") + .name("mcp-openapi0") + .type("mcp-openapi") + .kind(CLIENT) + .options(McpOpenapiOptionsConfig.builder() + .spec() + .label("openapi_things0") + .server("https://api.example.com") + .catalog() + .name("catalog0") + .subject("things-api") + .version("latest") + .build() + .build() + .build()) + .route() + .when(McpOpenapiConditionConfig.builder() + .tool("update_thing") + .build()) + .with(McpOpenapiWithConfig.builder() + .spec("openapi_things0") + .operation("update_thing") + .build()) + .build() + .build(); + binding.resolveId = resolveId; + + McpOpenapiCompositeConfig composite = generator.generate(new McpOpenapiBindingConfig(context, binding)); + + NamespaceConfig namespace = composite.namespaces.get(0); + String bodySchema = namespace.catalogs.stream() + .map(c -> c.options) + .filter(InlineOptionsConfig.class::isInstance) + .map(InlineOptionsConfig.class::cast) + .flatMap(o -> o.subjects.stream()) + .filter(s -> "update_thing-body".equals(s.subject)) + .map(s -> s.schema) + .findFirst() + .orElse(null); + + assertThat(bodySchema, notNullValue()); + JsonObject bodySchemaObject = Json.createReader(new StringReader(bodySchema)).readObject(); + assertThat(bodySchemaObject.getJsonObject("properties").getBoolean("id"), equalTo(false)); + assertThat(bodySchemaObject.getJsonObject("properties").containsKey("name"), equalTo(true)); + assertThat(bodySchemaObject.getBoolean("additionalProperties"), equalTo(true)); + } + @Test public void shouldOverrideOutputSchema() {