refactor(framework): Enable HTTP-based RuntimeAPI

framework/py/flwr/supercore/cli/flower_superexec.py

@@ -33,6 +33,7 @@
 from flwr.supercore.constant import EXEC_PLUGIN_SECTION, ExecutorType
 from flwr.supercore.exit import ExitCode, flwr_exit
 from flwr.supercore.grpc_health import add_args_health
+from flwr.supercore.runtime import RuntimeHttpStub as CoreRuntimeHttpStub
 from flwr.supercore.superexec.executor.config import (
     ExecutorConfig,
     ExecutorConfigError,
@@ -49,6 +50,9 @@
 from flwr.supercore.update_check import warn_if_flwr_update_available
 from flwr.supercore.utils import disable_process_dumping
 from flwr.supercore.version import package_version
+from flwr.superlink.runtime import RuntimeHttpStub as SuperLinkRuntimeHttpStub
+
+RuntimeStubClass = type[RuntimeStub] | type[CoreRuntimeHttpStub]
 
 
 def flower_superexec() -> None:
@@ -102,7 +106,9 @@ def flower_superexec() -> None:
             ExecPluginType.SERVER_APP,
         )
 
-    plugin_class, stub_class = _get_plugin_and_stub_class(args.plugin_type)
+    plugin_class, stub_class = _get_plugin_and_stub_class(
+        args.plugin_type, args.enable_http_api
+    )
     superexec_auth_secret = None
     if args.superexec_auth_secret_file is not None:
         try:
@@ -126,7 +132,7 @@ def flower_superexec() -> None:
 
     run_superexec(
         plugin_class=plugin_class,
-        stub_class=stub_class,  # type: ignore
+        stub_class=stub_class,
         runtime_api_address=args.runtime_api_address,
         insecure=args.insecure,
         root_certificates_path=args.root_certificates,
@@ -137,6 +143,7 @@ def flower_superexec() -> None:
         runtime_dependency_install=args.runtime_dependency_install,
         executor_type=args.executor,
         executor_config=executor_config,
+        enable_http_api=args.enable_http_api,
     )
 
 
@@ -158,6 +165,12 @@ def _parse_args() -> argparse.ArgumentParser:
         required=True,
         help="Address of the Runtime API",
     )
+    parser.add_argument(
+        "--enable-http-api",
+        action="store_true",
+        default=False,
+        help="EXPERIMENTAL: Connect to the Runtime API over HTTP instead of gRPC.",
+    )
     parser.add_argument(
         "--plugin-type",
         type=str,
@@ -221,14 +234,17 @@ def _load_executor_config(
 
 def _get_plugin_and_stub_class(
     plugin_type: str,
-) -> tuple[type[ExecPlugin], type[object]]:
+    enable_http_api: bool = False,
+) -> tuple[type[ExecPlugin], RuntimeStubClass]:
     """Get the plugin class and stub class based on the plugin type."""
-    mapping: dict[str, tuple[type[ExecPlugin], type[object]]] = {
-        ExecPluginType.CLIENT_APP: (ClientAppExecPlugin, RuntimeStub),
-        ExecPluginType.SERVER_APP: (ServerAppExecPlugin, RuntimeStub),
+    clientapp_stub = CoreRuntimeHttpStub if enable_http_api else RuntimeStub
+    serverapp_stub = SuperLinkRuntimeHttpStub if enable_http_api else RuntimeStub
+    mapping: dict[str, tuple[type[ExecPlugin], RuntimeStubClass]] = {
+        ExecPluginType.CLIENT_APP: (ClientAppExecPlugin, clientapp_stub),
+        ExecPluginType.SERVER_APP: (ServerAppExecPlugin, serverapp_stub),
         ExecPluginType.SERVER_APP_EPHEMERAL: (
             ServerAppEphemeralExecPlugin,
-            RuntimeStub,
+            serverapp_stub,
         ),
     }
     if plugin_type in mapping:

framework/py/flwr/supercore/cli/flower_superexec_test.py

@@ -24,7 +24,9 @@
 from flwr.common.constant import ExecPluginType
 from flwr.proto.runtime_pb2_grpc import RuntimeStub
 from flwr.supercore.constant import ExecutorType
+from flwr.supercore.runtime import RuntimeHttpStub as CoreRuntimeHttpStub
 from flwr.supercore.version import package_version
+from flwr.superlink.runtime import RuntimeHttpStub as SuperLinkRuntimeHttpStub
 
 from .flower_superexec import _parse_args
 
@@ -63,6 +65,21 @@ def test_parse_superexec_accepts_kubernetes_executor_config() -> None:
     assert args.executor_config == "executor.yaml"
 
 
+def test_parse_superexec_accepts_http_api_flag() -> None:
+    """SuperExec should accept the temporary Runtime HTTP API flag."""
+    args = _parse_args().parse_args(
+        [
+            "--appio-api-address",
+            "127.0.0.1:8000",
+            "--plugin-type",
+            ExecPluginType.CLIENT_APP,
+            "--enable-http-api",
+        ]
+    )
+
+    assert args.enable_http_api is True
+
+
 def test_flower_superexec_checks_for_update(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -117,6 +134,7 @@ def test_flower_superexec_clientapp_allows_missing_secret(
         health_server_address=None,
         runtime_dependency_install=False,
         executor=ExecutorType.SUBPROCESS,
+        enable_http_api=False,
     )
     captured: dict[str, object] = {}
 
@@ -137,7 +155,7 @@ def _run_superexec(**kwargs: object) -> None:
     monkeypatch.setattr(
         flower_superexec_module,
         "_get_plugin_and_stub_class",
-        lambda _plugin_type: (object, RuntimeStub),
+        lambda _plugin_type, _enable_http_api: (object, RuntimeStub),
     )
     monkeypatch.setattr(flower_superexec_module, "run_superexec", _run_superexec)
 
@@ -161,6 +179,7 @@ def test_flower_superexec_serverapp_allows_missing_secret(
         health_server_address=None,
         runtime_dependency_install=False,
         executor=ExecutorType.SUBPROCESS,
+        enable_http_api=False,
     )
 
     class _Parser:
@@ -182,7 +201,7 @@ def _run_superexec(**kwargs: object) -> None:
     monkeypatch.setattr(
         flower_superexec_module,
         "_get_plugin_and_stub_class",
-        lambda _plugin_type: (object, RuntimeStub),
+        lambda _plugin_type, _enable_http_api: (object, RuntimeStub),
     )
     monkeypatch.setattr(flower_superexec_module, "run_superexec", _run_superexec)
 
@@ -206,6 +225,7 @@ def test_flower_superexec_passes_executor_to_run_superexec(
         health_server_address=None,
         runtime_dependency_install=False,
         executor=ExecutorType.SUBPROCESS,
+        enable_http_api=False,
     )
     parser = Mock()
     parser.parse_args.return_value = args
@@ -222,7 +242,7 @@ def test_flower_superexec_passes_executor_to_run_superexec(
     monkeypatch.setattr(
         flower_superexec_module,
         "_get_plugin_and_stub_class",
-        lambda _plugin_type: (object, RuntimeStub),
+        lambda _plugin_type, _enable_http_api: (object, RuntimeStub),
     )
     monkeypatch.setattr(flower_superexec_module, "run_superexec", run_superexec_mock)
 
@@ -233,3 +253,22 @@ def test_flower_superexec_passes_executor_to_run_superexec(
         run_superexec_mock.call_args.kwargs["executor_type"] == ExecutorType.SUBPROCESS
     )
     assert run_superexec_mock.call_args.kwargs["executor_config"] is None
+
+
+@pytest.mark.parametrize(
+    ("plugin_type", "expected_stub"),
+    [
+        (ExecPluginType.CLIENT_APP, CoreRuntimeHttpStub),
+        (ExecPluginType.SERVER_APP, SuperLinkRuntimeHttpStub),
+        (ExecPluginType.SERVER_APP_EPHEMERAL, SuperLinkRuntimeHttpStub),
+    ],
+)
+def test_get_plugin_and_stub_class_selects_http_stub(
+    plugin_type: str, expected_stub: type[object]
+) -> None:
+    """HTTP mode should select the Runtime stub for each plugin surface."""
+    _, stub_class = flower_superexec_module._get_plugin_and_stub_class(
+        plugin_type, enable_http_api=True
+    )
+
+    assert stub_class is expected_stub

framework/py/flwr/supercore/heartbeat.py

@@ -21,6 +21,7 @@
 from collections.abc import Callable
 
 import grpc
+import httpx
 
 from flwr.common.constant import (
     HEARTBEAT_BASE_MULTIPLIER,
@@ -33,6 +34,7 @@
 from flwr.proto.runtime_pb2 import SendTaskHeartbeatRequest
 from flwr.proto.runtime_pb2_grpc import RuntimeStub
 from flwr.supercore.retry import RetryInvoker, exponential
+from flwr.supercore.runtime import RuntimeHttpStub
 
 # pylint: enable=E0611
 
@@ -154,3 +156,22 @@ def fn() -> bool:
         return True
 
     return fn
+
+
+def make_task_heartbeat_fn_http(
+    stub: RuntimeHttpStub,
+) -> Callable[[], bool]:
+    """Get the function to send a heartbeat to an HTTP Runtime endpoint."""
+    req = SendTaskHeartbeatRequest()
+
+    def fn() -> bool:
+        try:
+            res = stub.SendTaskHeartbeat(req)
+        except httpx.TransportError:
+            return False
+
+        if not res.success:
+            signal.raise_signal(signal.SIGINT)
+        return True
+
+    return fn

framework/py/flwr/supercore/heartbeat_test.py

@@ -19,7 +19,11 @@
 import unittest
 from unittest.mock import Mock
 
-from .heartbeat import HeartbeatSender
+import httpx
+
+from flwr.proto.runtime_pb2 import SendTaskHeartbeatResponse  # pylint: disable=E0611
+
+from .heartbeat import HeartbeatSender, make_task_heartbeat_fn_http
 
 
 # pylint: disable=protected-access
@@ -90,3 +94,19 @@ def test_heartbeat_fail_and_retry(self) -> None:
     def test_thread_is_daemon(self) -> None:
         """Test that the thread is a daemon thread."""
         self.assertTrue(self.heartbeat_sender._thread.daemon)
+
+
+def test_http_heartbeat_returns_true_on_success() -> None:
+    """HTTP heartbeat should report a successful Runtime response."""
+    stub = Mock()
+    stub.SendTaskHeartbeat.return_value = SendTaskHeartbeatResponse(success=True)
+
+    assert make_task_heartbeat_fn_http(stub)() is True
+
+
+def test_http_heartbeat_retries_transport_errors() -> None:
+    """HTTP heartbeat should make transport failures retryable."""
+    stub = Mock()
+    stub.SendTaskHeartbeat.side_effect = httpx.ConnectError("connection failed")
+
+    assert make_task_heartbeat_fn_http(stub)() is False

framework/py/flwr/supercore/superexec/executor/kubernetes_executor.py

@@ -517,6 +517,8 @@ def _taskexecutor_args(
         "--token-file",
         APPIO_TOKEN_FILE_PATH,
     ]
+    if spec.enable_http_api:
+        args.append("--enable-http-api")
 
     if spec.insecure:
         args.append("--insecure")

framework/py/flwr/supercore/superexec/executor/kubernetes_executor_test.py

@@ -407,6 +407,20 @@ def test_build_taskexecutor_pod_supports_clientapp_insecure_args() -> None:
     ]
 
 
+def test_build_taskexecutor_pod_forwards_http_api_flag() -> None:
+    """Test Pod construction forwards Runtime HTTP mode."""
+    pod = _as_dict(
+        _build_taskexecutor_pod(
+            _execution_spec(enable_http_api=True),
+            _executor_config(runtime_root_certificates=None),
+            None,
+            _LAUNCH_ATTEMPT_ID,
+        )
+    )
+
+    assert "--enable-http-api" in pod["spec"]["containers"][0]["args"]
+
+
 def test_build_taskexecutor_pod_supports_secure_default_trust_store() -> None:
     """Test secure Pod args can rely on container default trust store."""
     spec = _execution_spec()

framework/py/flwr/supercore/superexec/executor/subprocess_executor.py

@@ -40,6 +40,8 @@ def launch(self, spec: ExecutionSpec) -> LaunchResult:
             "--token",
             spec.token,
         ]
+        if spec.enable_http_api:
+            args.append("--enable-http-api")
 
         if spec.insecure:
             args.append("--insecure")

framework/py/flwr/supercore/superexec/executor/subprocess_executor_test.py

@@ -93,6 +93,14 @@ def test_launch_renders_runtime_dependency_install_flag() -> None:
     assert "--allow-runtime-dependency-installation" in popen_mock.call_args.args[0]
 
 
+def test_launch_renders_http_api_flag() -> None:
+    """Test subprocess executor forwards Runtime HTTP mode."""
+    with patch.object(subprocess, "Popen") as popen_mock:
+        SubprocessExecutor().launch(_execution_spec(enable_http_api=True))
+
+    assert "--enable-http-api" in popen_mock.call_args.args[0]
+
+
 def test_launch_renders_parent_pid_flag() -> None:
     """Test subprocess executor renders subprocess parent PID flag."""
     with patch.object(subprocess, "Popen") as popen_mock:

framework/py/flwr/supercore/superexec/executor/types.py

@@ -36,6 +36,7 @@ class ExecutionSpec:  # pylint: disable=too-many-instance-attributes
     parent_pid: int | None
     suppress_output: bool
     task_id: int
+    enable_http_api: bool = False
 
     def __post_init__(self) -> None:
         """Validate fields required by all executors."""

framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin.py

@@ -57,6 +57,8 @@ def launch_task(self, token: str, task: Task) -> None:  # type: ignore[override]
         elif self.root_certificates_path:
             cmds += ["--root-certificates", self.root_certificates_path]
         cmds += [self.runtime_api_address_arg, self.runtime_api_address]
+        if self.enable_http_api:
+            cmds += ["--enable-http-api"]
         cmds += ["--token", token]
         cmds += ["--parent-pid", str(os.getpid())]
         if self.runtime_dependency_install:

framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin_test.py

@@ -42,12 +42,13 @@ class _EphemeralExecPlugin(BaseEphemeralExecPlugin):
     runtime_api_address_arg = "--serverappio-api-address"
 
 
-def _get_ephemeral_plugin() -> _EphemeralExecPlugin:
+def _get_ephemeral_plugin(*, enable_http_api: bool = False) -> _EphemeralExecPlugin:
     return _EphemeralExecPlugin(
         runtime_api_address="127.0.0.1:9091",
         get_run=_get_run,
         insecure=True,
         root_certificates_path=None,
+        enable_http_api=enable_http_api,
     )
 
 
@@ -119,3 +120,18 @@ def test_launch_task_calls_cleanup_before_launch() -> None:
 
     # Assert
     assert call_log == ["cleanup", "subprocess"]
+
+
+def test_launch_task_forwards_http_api_flag() -> None:
+    """Launch should forward Runtime HTTP mode to the app process."""
+    plugin = _get_ephemeral_plugin(enable_http_api=True)
+
+    with (
+        patch(
+            "flwr.supercore.superexec.plugin.base_ephemeral_exec_plugin.subprocess.run"
+        ) as run,
+        patch("flwr.supercore.superexec.plugin.base_ephemeral_exec_plugin.flwr_exit"),
+    ):
+        plugin.launch_task(token="token-123", task=_get_task(task_id=5))
+
+    assert "--enable-http-api" in run.call_args.args[0]

framework/py/flwr/supercore/superexec/plugin/base_exec_plugin.py

@@ -49,6 +49,7 @@ def __init__(  # pylint: disable=R0913, R0917
         runtime_dependency_install: bool = RUNTIME_DEPENDENCY_INSTALL,
         *,
         executor: Executor,
+        enable_http_api: bool = False,
     ) -> None:
         super().__init__(
             runtime_api_address=runtime_api_address,
@@ -57,6 +58,7 @@ def __init__(  # pylint: disable=R0913, R0917
             get_run=get_run,
             runtime_dependency_install=runtime_dependency_install,
             executor=executor,
+            enable_http_api=enable_http_api,
         )
         self.executor: Executor = executor
 
@@ -103,6 +105,7 @@ def _build_execution_spec(
                 self.suppress_output and task_type not in self.visible_output_task_types
             ),
             task_id=task_id,
+            enable_http_api=self.enable_http_api,
         )
 
     def _get_supported_task_type(self, task: Task) -> TaskType | None: