fix(binding-mcp-kafka-connect): accept arbitrary connector config fields

runtime/binding-mcp-kafka-connect/src/main/resources/io/aklivity/zilla/runtime/binding/mcp/kafka/connect/internal/schema/kafka-connect.openapi.json

@@ -144,7 +144,8 @@
                                     "tasks.max": {
                                         "type": "string"
                                     }
-                                }
+                                },
+                                "additionalProperties": true
                             }
                         }
                     }
@@ -183,7 +184,8 @@
                                     "tasks.max": {
                                         "type": "string"
                                     }
-                                }
+                                },
+                                "additionalProperties": true
                             }
                         }
                     }

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)
     {

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()
     {

runtime/common-json/src/main/java/io/aklivity/zilla/runtime/common/json/JsonSchema.java

@@ -142,6 +142,17 @@ JsonTransform validator(
      */
     List<String> 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<String> 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.

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());
     }
 
     /**

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
     {
@@ -105,7 +106,14 @@ private enum SegMode
     public JsonProjectorImpl(
         List<String> pointers)
     {
-        this.root = compile(pointers);
+        this(pointers, List.of());
+    }
+
+    public JsonProjectorImpl(
+        List<String> retained,
+        List<String> rejected)
+    {
+        this.root = compile(retained, rejected);
     }
 
     @Override
@@ -621,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;
@@ -636,21 +652,33 @@ 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)
     {
         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;
     }
@@ -722,10 +750,11 @@ private static boolean matchesIndex(
     }
 
     private static Node compile(
-        List<String> pointers)
+        List<String> retained,
+        List<String> rejected)
     {
         NodeBuilder builder = new NodeBuilder();
-        for (String pointer : pointers)
+        for (String pointer : retained)
         {
             NodeBuilder node = builder;
             for (String segment : segments(pointer))
@@ -734,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();
     }
 
@@ -758,36 +796,47 @@ 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)
             {
                 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;
         }
     }
 
     private static final class NodeBuilder
     {
         private final Map<String, NodeBuilder> children = new LinkedHashMap<>();
         private boolean keepAll;
+        private boolean rejected;
 
         private NodeBuilder child(
             String segment)
@@ -807,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);
         }
     }