feat(sdk): support explicit HostPath bind mounts

AGENTS.md

@@ -388,6 +388,16 @@ quota or cluster default. A positive limit without an explicit scheduling
 quota also reserves that amount for scheduling. Keep this distinction in SDK
 documentation and examples; see `sdk/python/README.md` for runtime support.
 
+`Mount(host_path=...)` binds an existing path visible to the selected node's
+runtime. HostPath mounts default to read-only; writable binds require
+`read_only=False`. Image/S3 mounts remain read-only. Keep the pinned ADX SDK's
+HostPath serialization adapter under `_backends`, and preserve server-side
+permission errors. Restrict this interface to trusted management callers and
+operator-controlled source paths; client validation is not authorization to
+access the node's filesystem.
+The SDK does not create host directories, mount node storage, or replicate
+node-local data. See the SDK guide and `sdk/python/examples/hostpath_mounts.py`.
+
 Same-node failover and explicit rollback retain the logical sandbox identity:
 
 ```python

sdk/python/README.md

@@ -701,6 +701,55 @@ mount = Mount(target="/opt/tools", image_url="ubuntu:24.04")
 Firecracker mounts must instead use `type="erofs"` with an S3 object containing
 a raw EROFS image; it rejects `image_url` and directory-backed mounts.
 
+### Node HostPath bind mounts
+
+Trusted management clients can bind an existing node-visible path into a
+sandbox. For example, expose a prepared log directory on a selected runsc node:
+
+```python
+from akernel_sdk import Mount, Sandbox
+
+logs = Mount(
+    target="/apps/logs",       # Path inside the sandbox.
+    host_path="/apps/logs",    # Path visible to sandboxd on the selected node.
+    read_only=False,           # Explicitly allow application log writes.
+)
+with Sandbox(node_id="<prepared-node-id>", mounts=[logs]) as sandbox:
+    print(sandbox.commands.run("ls -ld /apps/logs").stdout)
+```
+
+`read_only=True` is the default. HostPath supports `type="bind"` only;
+`read_only=False` is rejected for Image/S3 mounts, which remain read-only.
+Exactly one of `image_url`, `s3_config`, or `host_path` is required. The source
+must be an absolute POSIX path without traversal components or control
+characters. The SDK does not inspect the client machine's filesystem or create
+the source directory; it sends `create_if_missing=false`. An absent source or
+insufficient filesystem permissions is a node-side error.
+
+In Kubernetes, the node Pod must first see the source, for example through an
+operator-managed `hostPath` volume and matching `volumeMount`. An ECS directory
+does not become visible to sandboxd merely because its path is in the SDK
+request. Directory ownership and permissions must allow the sandbox's process
+UID/GID to use it. Separate workers' local directories are separate data;
+`node_id` selects the prepared node but does not replicate storage. Data written
+through a bind can outlive the sandbox. Runtime-specific mount restrictions
+still apply; Firecracker does not support these directory-backed binds.
+
+Restrict HostPath creation to trusted management clients and source paths
+controlled by the operator. These mounts do not provide per-tenant source-path
+isolation; SDK path validation and read-only defaults do not authorize access
+to the node's filesystem. Server permission errors are returned through the
+normal `BackendOperationError` fields.
+
+The maintained example requires an explicit node and source. It lists the
+read-only mount by default; `--write` creates one uniquely named log file and
+leaves that file on the node:
+
+```bash
+python sdk/python/examples/hostpath_mounts.py \
+  --node-id <prepared-node-id> --host-path /apps/logs --write
+```
+
 ## Launch from a Dockerfile
 
 Dockerfile direct launch is a supported AKernel SDK capability and will remain
@@ -816,6 +865,7 @@ Maintained examples are under [`examples/`](./examples):
 - `dockerfile_launch.py`
 - `failover_reload.py`
 - `gpu_sandbox.py`
+- `hostpath_mounts.py`
 - `named_sandbox.py`
 - `network_policy.py`
 - `pty.py`
@@ -860,7 +910,7 @@ not part of the default test suite.
 | `SandboxInfo` | `id`, `state`, `cpu`, `memory`, `image`, `xpu`, `storage_mb` |
 | `NodeInfo` | `id`, `status`, `capacity`, `allocatable`, `labels` |
 | `S3Config` | `endpoint`, `bucket`, `object`, optional credentials |
-| `Mount` | `target`, one source, and `type` |
+| `Mount` | `target`, exactly one of `image_url` / `s3_config` / `host_path`, `type`, `read_only` (default `True`; writable only for HostPath) |
 | `HttpReverseTunnel` | `target`, `reverse_port`, `listen_port`, `connect_timeout` |
 | `PortRange` | `first`, `last` |
 | `NetworkRule` | action, direction, protocol, peer and sandbox port selectors, priority |

sdk/python/akernel_sdk/_backends/adx.py

@@ -19,6 +19,7 @@
 import logging
 import time
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any
 
 import adx_environment
@@ -29,6 +30,7 @@
     DNSPolicy,
     DNSRule,
     EntryInfo,
+    Mount,
     NetworkPolicy,
     NetworkRule,
     PortRange,
@@ -49,6 +51,60 @@
 logger = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class _NativeHostPathMount(adx_environment.Mount):
+    """Adapt node HostPath sources through the pinned native Mount serializer.
+
+    The pinned native SDK checks isinstance(Mount) before calling to_dict, but
+    its stock value type only represents Image/S3. Keep that constructor and
+    transport intact without replacing native classes or mutating global state.
+    """
+
+    host_path: str = ""
+    read_only: bool = True
+
+    def __post_init__(self) -> None:
+        self._public_mount()
+
+    def _public_mount(self) -> Mount:
+        if self.image_url is not None or self.s3_config is not None:
+            raise ValueError("HostPath adapter requires only a host_path source")
+        return Mount(
+            target=self.target,
+            type=self.type,
+            host_path=self.host_path,
+            read_only=self.read_only,
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        return self._public_mount().to_dict()
+
+
+def _native_mount(mount: Mount) -> Any:
+    if mount.host_path is not None:
+        return _NativeHostPathMount(
+            target=mount.target,
+            host_path=mount.host_path,
+            read_only=mount.read_only,
+        )
+    return adx_environment.Mount(
+        target=mount.target,
+        image_url=mount.image_url,
+        s3_config=(
+            adx_environment.S3Config(
+                endpoint=mount.s3_config.endpoint,
+                bucket=mount.s3_config.bucket,
+                object=mount.s3_config.object,
+                access_key=mount.s3_config.access_key,
+                secret_key=mount.s3_config.secret_key,
+            )
+            if mount.s3_config is not None
+            else None
+        ),
+        type=mount.type,
+    )
+
+
 def _control_pty_connection(connection: Any) -> Any:
     """PTY direct routing uses the control TLS entrypoint, not the data port."""
 
@@ -559,25 +615,7 @@ def create(self, spec: SandboxSpec) -> BackendSession:
         network = None
         if spec.network_policy is not None:
             network = _native_network_policy(spec.network_policy)
-        mounts = [
-            adx_environment.Mount(
-                target=mount.target,
-                image_url=mount.image_url,
-                s3_config=(
-                    adx_environment.S3Config(
-                        endpoint=mount.s3_config.endpoint,
-                        bucket=mount.s3_config.bucket,
-                        object=mount.s3_config.object,
-                        access_key=mount.s3_config.access_key,
-                        secret_key=mount.s3_config.secret_key,
-                    )
-                    if mount.s3_config is not None
-                    else None
-                ),
-                type=mount.type,
-            )
-            for mount in spec.mounts
-        ]
+        mounts = [_native_mount(mount) for mount in spec.mounts]
         create_timeout = spec.schedule_timeout + 60
         create_args = dict(
             image=spec.image,

sdk/python/akernel_sdk/sandbox.py

@@ -199,7 +199,9 @@ def __init__(
                 which uses the image's OCI WORKDIR when
                 ``inherit_entrypoint=True``.
             port_forwardings: Sandbox TCP ports exposed through the gateway.
-            mounts: Additional read-only OCI or S3-backed mounts.
+            mounts: Additional read-only OCI/S3 mounts or explicit node HostPath
+                bind mounts for trusted management callers. HostPath writes
+                require ``Mount(..., read_only=False)``.
             reverse_tunnel: SDK-side HTTP service exposed inside the sandbox.
             detached: Keep the sandbox alive when this client closes.
             node_id: Require placement on a specific AKernel node.

sdk/python/akernel_sdk/types.py

@@ -507,44 +507,75 @@ def to_dict(self) -> dict[str, Any]:
 
 @dataclass(frozen=True)
 class Mount:
-    """Read-only OCI image or S3 object mounted inside a sandbox.
+    """OCI/S3 read-only mount or a trusted-management node HostPath bind.
 
-    Exactly one of ``image_url`` and ``s3_config`` must be supplied. ``type``
-    selects a read-only bind mount or an EROFS image mount.
+    Supply exactly one of ``image_url``, ``s3_config``, or ``host_path``.
+    HostPath sources must already be visible to the node runtime; they are not
+    resolved on the client. HostPath supports ``type="bind"`` and defaults to
+    read-only; writes require ``read_only=False``. Image/S3 sources are always
+    read-only. This type does not authorize access to the node's filesystem.
     """
 
     target: str
     image_url: str | None = None
     s3_config: S3Config | None = None
     type: str = "bind"
+    host_path: str | None = None
+    read_only: bool = True
 
     def __post_init__(self) -> None:
         if not isinstance(self.target, str) or not self.target.startswith("/"):
             raise ValueError("target must be an absolute sandbox path")
         source_count = sum(
-            source is not None for source in (self.image_url, self.s3_config)
+            source is not None
+            for source in (self.image_url, self.s3_config, self.host_path)
         )
         if source_count != 1:
-            raise ValueError("exactly one of image_url and s3_config must be specified")
+            raise ValueError(
+                "exactly one of image_url, s3_config, host_path must be specified"
+            )
         if self.image_url is not None and (
             not isinstance(self.image_url, str) or not self.image_url.strip()
         ):
             raise ValueError("image_url must be a non-empty string")
         if self.type not in ("bind", "erofs"):
             raise ValueError("type must be 'bind' or 'erofs'")
+        if not isinstance(self.read_only, bool):
+            raise TypeError("read_only must be a boolean")
+        if self.host_path is not None:
+            if (
+                not isinstance(self.host_path, str)
+                or not self.host_path.startswith("/")
+                or any(part in (".", "..") for part in self.host_path.split("/"))
+                or any(
+                    ord(char) < 32 or 127 <= ord(char) <= 159
+                    for char in self.host_path
+                )
+            ):
+                raise ValueError(
+                    "host_path must be an absolute path without traversal "
+                    "or control characters"
+                )
+            if self.type != "bind":
+                raise ValueError("host_path supports only type='bind'")
+        elif not self.read_only:
+            raise ValueError("image and S3 mounts must be read-only")
 
     def to_dict(self) -> dict[str, Any]:
         """Return the JSON-compatible representation expected by AKernel."""
 
         value: dict[str, Any] = {
             "type": self.type,
             "target": self.target,
-            "options": ["ro"],
+            "options": ["ro" if self.read_only else "rw"],
         }
         if self.image_url is not None:
             value["image_url"] = self.image_url
         if self.s3_config is not None:
             value["s3_config"] = self.s3_config.to_dict()
+        if self.host_path is not None:
+            value["host_path"] = self.host_path
+            value["create_if_missing"] = False
         return value
 
 

sdk/python/examples/hostpath_mounts.py

@@ -0,0 +1,57 @@
+# Copyright (c) 2026 Ant Group Corporation.
+#
+# Licensed 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.
+
+"""Bind a prepared node path using trusted management credentials.
+
+The node runtime must already see the source and allow the sandbox process to
+access it. Default mode only lists the read-only mount. --write explicitly
+creates one unique log file and leaves it on the node after sandbox deletion.
+Use operator-controlled source paths; SDK validation does not authorize access
+to the node's filesystem.
+"""
+
+import argparse
+import posixpath
+import shlex
+import uuid
+
+from akernel_sdk import Mount, Sandbox
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--node-id", required=True)
+    parser.add_argument("--host-path", required=True)
+    parser.add_argument("--target", default="/apps/logs")
+    parser.add_argument("--write", action="store_true")
+    args = parser.parse_args()
+    mount = Mount(
+        target=args.target, host_path=args.host_path, read_only=not args.write
+    )
+    with Sandbox(node_id=args.node_id, mounts=[mount], cpu=1000, memory=1024) as sb:
+        command = "ls -ld -- " + shlex.quote(args.target)
+        if args.write:
+            marker = "akernel-sdk-" + uuid.uuid4().hex
+            filename = marker + ".log"
+            target = shlex.quote(posixpath.join(args.target, filename))
+            command = f"printf '%s\\n' {shlex.quote(marker)} > {target} && cat {target}"
+            print("Log on the selected node:", posixpath.join(args.host_path, filename))
+        result = sb.commands.run(command)
+        if result.exit_code != 0:
+            raise RuntimeError(result.stderr or result.stdout)
+        print(result.stdout)
+
+
+if __name__ == "__main__":
+    main()