Address review: stable-API glue, workflow version gate, generic command tunnel

- The generated glue now calls the stable workflow.management Ballerina
  API (getWorkflowMetadata / executeManagementCommand) instead of
  importing workflow.management.rest — no module that reads as an HTTP
  server enters the user's build (companion change:
  ballerina-platform/module-ballerina-workflow#94 moved the dispatcher
  into workflow.management).
- The glue is only generated when the package's resolved
  ballerina/workflow dependency is 0.9.0 or later, where those APIs first
  shipped; older workflow versions build exactly as before.
- The command plumbing is generic (command_tunnel.bal): payload parsing,
  deadline handling, at-most-once execution with result replay, and the
  result envelope are shared infrastructure — a new tunneled command kind
  only adds a ControlAction arm selecting its executor. Types renamed
  accordingly (TunneledCommandPayload/Result, CommandIdentity), and the
  workflow-specific file keeps only the metadata provider, executor
  registration, and capability advertisement.
- Redelivered commandIds are now reserved atomically before execution, so
  concurrent heartbeat rounds cannot execute the same command twice; an
  executor result without an int httpStatus is reported FAILED with a
  diagnostic instead of a silent COMPLETED/500.

Author: hasithaa

ballerina/command_tunnel.bal

@@ -0,0 +1,158 @@
+// Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com) All Rights Reserved.
+//
+// WSO2 LLC. licenses this file to you under the Apache License,
+// Version 2.0 (the "License"); you may not use this file except
+// in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import ballerina/log;
+
+// ================================================================================
+// COMMAND TUNNEL
+// ================================================================================
+// Generic plumbing for control commands the ICP tunnels through heartbeat
+// responses to be executed in-process: payload parsing, at-most-once execution
+// with result replay for redelivered commands, and the {commandId, httpStatus,
+// body} result envelope posted back on POST /icp/commandResult. Nothing here is
+// specific to any command kind — a new tunneled ControlAction only needs an
+// executor (see handleTunneledCommand in main.bal, where actions are mapped to
+// their executors).
+
+# Executes one tunneled command's operation. Takes the operation request
+# (`{operation, params, identity}`) and returns `{httpStatus, body}` exactly as the
+# corresponding management API would have responded.
+public type TunneledCommandExecutor isolated function (map<json> command) returns map<json>|error;
+
+// Outcomes of recently executed commands, kept so a redelivered commandId (e.g. its
+// result was lost after execution) replays the stored result instead of executing the
+// operation twice — this is what makes tunneled mutations safe against duplicate
+// delivery. A commandId is reserved atomically before execution, so concurrent
+// heartbeat rounds delivering the same command cannot both execute it.
+// Insertion-ordered FIFO eviction; one record so a single lock covers all structures
+// (a lock statement may access only one isolated module-level variable).
+const int PROCESSED_COMMAND_CACHE_CAPACITY = 64;
+
+type ProcessedCommandCache record {|
+    map<TunneledCommandResult> results = {};
+    map<boolean> inFlight = {};
+    string[] insertionOrder = [];
+|};
+
+isolated ProcessedCommandCache processedCommands = {};
+
+// Atomically claims a commandId for execution. Returns the stored result when the
+// command was already executed (replay it), `true` when this caller now owns the
+// execution, and `false` when another round is executing it right now (post nothing —
+// the owning execution will).
+isolated function reserveOrReplay(string commandId) returns TunneledCommandResult|boolean {
+    lock {
+        TunneledCommandResult? cached = processedCommands.results[commandId];
+        if cached is TunneledCommandResult {
+            return cached.clone();
+        }
+        if processedCommands.inFlight.hasKey(commandId) {
+            return false;
+        }
+        processedCommands.inFlight[commandId] = true;
+        return true;
+    }
+}
+
+isolated function storeCommandResult(TunneledCommandResult result) {
+    lock {
+        _ = processedCommands.inFlight.removeIfHasKey(result.commandId);
+        if processedCommands.results.hasKey(result.commandId) {
+            return;
+        }
+        if processedCommands.insertionOrder.length() >= PROCESSED_COMMAND_CACHE_CAPACITY {
+            string evicted = processedCommands.insertionOrder.shift();
+            _ = processedCommands.results.removeIfHasKey(evicted);
+        }
+        processedCommands.insertionOrder.push(result.commandId);
+        processedCommands.results[result.commandId] = result.clone();
+    }
+}
+
+# Executes one tunneled command with at-most-once semantics. Never panics or returns
+# an error: every executed outcome — including "not accepted" and executor failures —
+# becomes a result the ICP can deliver to the waiting caller.
+#
+# + payload - The command payload from the control command
+# + executor - The executor for this command kind, or `()` when none is registered
+# + accepted - Whether this runtime currently accepts this command kind (its opt-in
+#              configuration); `false` yields a FAILED/403 result
+# + return - The result to post to `POST /icp/commandResult`, or `()` when another
+#            round is executing the same commandId right now (nothing to post)
+isolated function executeTunneledCommand(TunneledCommandPayload payload,
+        TunneledCommandExecutor? executor, boolean accepted) returns TunneledCommandResult? {
+    TunneledCommandResult|boolean reservation = reserveOrReplay(payload.commandId);
+    if reservation is TunneledCommandResult {
+        log:printInfo(string `Replaying stored result for redelivered command: ${payload.commandId}`);
+        return reservation;
+    }
+    if !reservation {
+        log:printInfo(string `Skipping command already executing in another round: ${payload.commandId}`);
+        return ();
+    }
+
+    TunneledCommandResult result;
+    if executor is () || !accepted {
+        // The matching capability is only advertised while both hold, so this is a
+        // server-side gating bug or a config change since the last heartbeat.
+        result = {
+            runtimeId: currentRuntimeId,
+            commandId: payload.commandId,
+            status: "FAILED",
+            httpStatus: 403,
+            body: {"error": {"message": "Commands of this kind are not accepted by this runtime"}}
+        };
+    } else {
+        map<json> command = {
+            operation: payload.operation,
+            params: payload.params,
+            identity: {userId: payload.identity.userId, roles: payload.identity.roles}
+        };
+        map<json>|error outcome = executor(command);
+        if outcome is error {
+            log:printError(string `Tunneled command execution failed: ${payload.commandId}`, outcome);
+            result = {
+                runtimeId: currentRuntimeId,
+                commandId: payload.commandId,
+                status: "FAILED",
+                httpStatus: 500,
+                body: {"error": {"message": outcome.message()}}
+            };
+        } else {
+            json httpStatus = outcome["httpStatus"];
+            if httpStatus is int {
+                result = {
+                    runtimeId: currentRuntimeId,
+                    commandId: payload.commandId,
+                    status: "COMPLETED",
+                    httpStatus: httpStatus,
+                    body: outcome["body"]
+                };
+            } else {
+                log:printError(string `Tunneled command returned an unexpected result shape: ${payload.commandId}`);
+                result = {
+                    runtimeId: currentRuntimeId,
+                    commandId: payload.commandId,
+                    status: "FAILED",
+                    httpStatus: 500,
+                    body: {"error": {"message": "Unexpected command result shape"}}
+                };
+            }
+        }
+    }
+    storeCommandResult(result);
+    return result;
+}

ballerina/icp_client.bal

@@ -49,9 +49,9 @@ public client class IcpClient {
         return heartbeatResponse;
     }
 
-    // Posts the result of a tunneled workflow management command. Outbound-only, like
-    // heartbeats — the ICP correlates it to the waiting request via the commandId.
-    isolated remote function sendCommandResult(WorkflowCommandResult result) returns error? {
+    // Posts the result of a tunneled command. Outbound-only, like heartbeats — the
+    // ICP correlates it to the waiting request via the commandId.
+    isolated remote function sendCommandResult(TunneledCommandResult result) returns error? {
         http:Request request = new;
         request.setHeader(http:AUTH_HEADER, string `${http:AUTH_SCHEME_BEARER} ${check generateJwtToken()}`);
         request.setPayload(result.toJson());

ballerina/main.bal

@@ -99,7 +99,7 @@ public class HeartbeatJob {
 
     # Executes the heartbeat job: one heartbeat round, plus bounded follow-up rounds
     # while the server is actively tunneling work. A follow-up happens immediately
-    # after executing a workflow command (its result may already have unblocked the
+    # after executing a tunneled command (its result may already have unblocked the
     # next queued command) or after the server's `nextHeartbeatInSeconds` boost hint
     # (sent while a user is actively working with workflow views). Follow-ups stop
     # once their accumulated delay would exceed one regular interval, so a tick never
@@ -168,8 +168,8 @@ public class HeartbeatJob {
         }
         self.supportedHeartbeatFields = newSupportedHeartbeatFields;
         log:printDebug("Heartbeat acknowledged by ICP server");
-        boolean executedWorkflowCommand = self.handleControlCommands(heartbeatResponse.commands);
-        if executedWorkflowCommand {
+        boolean processedTunneledCommand = self.handleControlCommands(heartbeatResponse.commands);
+        if processedTunneledCommand {
             // Fetch the next queued command right away — the posted result has likely
             // unblocked the ICP-side caller already.
             return 0;
@@ -184,15 +184,15 @@ public class HeartbeatJob {
     # Handles the control commands delivered in a heartbeat response.
     #
     # + commands - The commands from the response
-    # + return - `true` when at least one tunneled workflow command was processed,
-    #            so the caller can immediately fetch the next queued command
+    # + return - `true` when at least one tunneled command was processed, so the
+    #            caller can immediately fetch the next queued command
     function handleControlCommands(ControlCommand[] commands) returns boolean {
         if commands.length() == 0 {
             return false;
         }
 
         boolean artifactsChanged = false;
-        boolean workflowCommandProcessed = false;
+        boolean tunneledCommandProcessed = false;
         foreach ControlCommand command in commands {
             log:printInfo(string `Handling control command: ${command.toJsonString()}`);
             command.status = PENDING;
@@ -201,8 +201,8 @@ public class HeartbeatJob {
             error? result = ();
             match command.action {
                 WORKFLOW_MGMT => {
-                    workflowCommandProcessed = true;
-                    result = self.handleWorkflowCommand(command.payload ?: "");
+                    tunneledCommandProcessed = true;
+                    result = self.handleTunneledCommand(command);
                 }
                 START|STOP => {
                     string artifactName = command.targetArtifact.name;
@@ -263,38 +263,50 @@ public class HeartbeatJob {
             Heartbeat|error newHeartbeat = getHeartbeat(self.supportedHeartbeatFields);
             if newHeartbeat is error {
                 log:printError("Failed to create full heartbeat after control command", newHeartbeat);
-                return workflowCommandProcessed;
+                return tunneledCommandProcessed;
             }
             self.heartbeat = newHeartbeat;
         }
-        return workflowCommandProcessed;
+        return tunneledCommandProcessed;
     }
 
-    # Executes one tunneled workflow management command and posts its result to the
-    # ICP. A command past its deadline is dropped unexecuted — the ICP-side caller
-    # has already timed out, and executing (or replying) then would be wasted work
-    # or, for mutations, an unwanted late effect.
+    # Executes one tunneled command and posts its result to the ICP. A command past
+    # its deadline is dropped unexecuted — the ICP-side caller has already timed
+    # out, and executing (or replying) then would be wasted work or, for mutations,
+    # an unwanted late effect. The command's action selects the executor; adding a
+    # new tunneled command kind means adding an arm to that match.
     #
-    # + rawPayload - The command's JSON payload string
+    # + command - The tunneled control command
     # + return - An error when the payload is unusable or the result could not be
     #            delivered (the command's status is reported FAILED then)
-    function handleWorkflowCommand(string rawPayload) returns error? {
+    function handleTunneledCommand(ControlCommand command) returns error? {
+        string rawPayload = command.payload ?: "";
         if rawPayload == "" {
-            return error("Missing payload for WORKFLOW_MGMT command");
+            return error(string `Missing payload for ${command.action} command`);
         }
-        WorkflowCommandPayload payload = check rawPayload.fromJsonStringWithType();
+        TunneledCommandPayload payload = check rawPayload.fromJsonStringWithType();
 
         string? deadline = payload?.deadline;
         if deadline is string {
             time:Utc|time:Error deadlineTime = time:utcFromString(deadline);
             if deadlineTime is time:Utc && time:utcDiffSeconds(deadlineTime, time:utcNow()) < 0d {
-                log:printWarn(string `Dropping expired workflow command ${payload.commandId} ` +
+                log:printWarn(string `Dropping expired command ${payload.commandId} ` +
                         string `(deadline ${deadline})`);
                 return;
             }
         }
 
-        WorkflowCommandResult result = executeWorkflowCommand(payload);
-        check self.icpClient->sendCommandResult(result);
+        TunneledCommandExecutor? executor = ();
+        boolean accepted = false;
+        match command.action {
+            WORKFLOW_MGMT => {
+                executor = workflowExecutor();
+                accepted = enableWorkflowManagement;
+            }
+        }
+        TunneledCommandResult? result = executeTunneledCommand(payload, executor, accepted);
+        if result is TunneledCommandResult {
+            check self.icpClient->sendCommandResult(result);
+        }
     }
 }

ballerina/types.bal

@@ -157,47 +157,47 @@ public enum ControlAction {
     STOP,
     SET_LOGGER_LEVEL,
     // A tunneled workflow management operation (list/start workflows, complete human
-    // tasks, ...). The server must only send this to runtimes that advertised the
-    // "workflowCommands" capability — older bridges fail record binding on unknown
-    // actions. The command's `payload` is a WorkflowCommandPayload JSON string; the
-    // result is posted back on POST /icp/commandResult.
+    // tasks, ...), executed in-process via the command tunnel (command_tunnel.bal).
+    // The ICP only sends this to runtimes that advertised the "workflowCommands"
+    // capability. The command's `payload` is a TunneledCommandPayload JSON string;
+    // the result is posted back on POST /icp/commandResult.
     WORKFLOW_MGMT
 };
 
-# The JSON carried in a `WORKFLOW_MGMT` control command's `payload`.
+# The JSON carried in a tunneled control command's `payload`.
 #
 # + commandId - Correlation ID; the result is posted back under this ID
-# + operation - Dot-qualified management operation name (e.g. `humanTasks.complete`)
-# + params - Operation parameters, keyed like the management REST API's query/path/body
+# + operation - Dot-qualified operation name (e.g. `humanTasks.complete`)
+# + params - Operation parameters, keyed like the management API's query/path/body
 # + identity - The end user the ICP executes this on behalf of
 # + deadline - ISO-8601 instant after which the command is dropped unexecuted
-public type WorkflowCommandPayload record {|
+public type TunneledCommandPayload record {|
     string commandId;
     string operation;
     map<json> params = {};
-    WorkflowCommandIdentity identity = {};
+    CommandIdentity identity = {};
     string deadline?;
 |};
 
-# The caller identity a tunneled workflow command executes on behalf of. Same
-# semantics as the management REST API's `x-user-id` / `x-user-roles` headers.
+# The caller identity a tunneled command executes on behalf of. Same semantics as
+# the management API's `x-user-id` / `x-user-roles` headers.
 #
 # + userId - The user ID, or `()` when unknown
 # + roles - The caller's roles; empty means "no roles"
-public type WorkflowCommandIdentity record {|
+public type CommandIdentity record {|
     string? userId = ();
     string[] roles = [];
 |};
 
-# The outcome of a tunneled workflow command, posted to `POST /icp/commandResult`.
+# The outcome of a tunneled command, posted to `POST /icp/commandResult`.
 #
 # + runtimeId - This runtime's ID
 # + commandId - The command's correlation ID
 # + status - `COMPLETED` when the operation executed (regardless of its HTTP-level
 #            outcome), `FAILED` when it could not be executed at all
-# + httpStatus - The status code the management REST API would have returned
-# + body - The response body, byte-identical to the management REST API's
-public type WorkflowCommandResult record {|
+# + httpStatus - The status code the corresponding management API would have returned
+# + body - The response body, byte-identical to the management API's
+public type TunneledCommandResult record {|
     string runtimeId;
     string commandId;
     string status;