From 35e2b8d0c969e676f1cb032b2708edbfd39bb427 Mon Sep 17 00:00:00 2001
From: anlicheng <244108715@qq.com>
Date: Thu, 7 May 2026 23:21:56 +0800
Subject: [PATCH] fix

---
 docs/container_deploy_request.md  | 761 +++++++++++++++++-------------
 docs/efka_iot_protocol.md         | 255 ++++++++++
 src/host/iot_host.erl             |  18 +-
 src/transport/tcp/ssl_channel.erl |  96 ++--
 4 files changed, 752 insertions(+), 378 deletions(-)
 create mode 100644 docs/efka_iot_protocol.md

diff --git a/docs/container_deploy_request.md b/docs/container_deploy_request.md
index 610a71b..b1f02e7 100644
--- a/docs/container_deploy_request.md
+++ b/docs/container_deploy_request.md
@@ -1,41 +1,66 @@
-# Container Deploy 请求格式说明
+# /container/deploy 容器创建请求参数说明
 
-本文档说明 HTTP 接口 `/container/deploy` 接收的 JSON 格式，以及它在服务端如何被转换为 `message.proto` 中的 `ContainerRequest.Deploy` / `ContainerDeployParams` / `ContainerSpec`。
+本文档说明 `iot` HTTP 接口 `POST /container/deploy` 当前支持的 JSON 参数、校验规则，以及服务端把 JSON 解码成 Erlang map 后如何转换成下发给 `efka` 的容器部署命令。
 
-当前实现对应代码：
+对应代码：
 
 - HTTP 入口：[src/transport/http/container_handler.erl](/usr/local/code/cloudkit/iot/src/transport/http/container_handler.erl:63)
-- 请求构造器与 deploy 配置校验：[src/host/iot_container_request_builder.erl](/usr/local/code/cloudkit/iot/src/host/iot_container_request_builder.erl:25)
-- protobuf 定义：[proto/message.proto](/usr/local/code/cloudkit/iot/proto/message.proto:37)
+- HTTP JSON 解析：[src/transport/http/http_protocol.erl](/usr/local/code/cloudkit/iot/src/transport/http/http_protocol.erl:68)
+- 参数校验与内部 map 构造：[src/docker/docker_container_builder.erl](/usr/local/code/cloudkit/iot/src/docker/docker_container_builder.erl:25)
+- 主机命令下发：[src/host/iot_host.erl](/usr/local/code/cloudkit/iot/src/host/iot_host.erl:99)
 
-## 1. HTTP 请求格式
+## 1. 接口
 
-接口：
+```http
+POST /container/deploy
+Content-Type: application/json
+```
 
-- `POST /container/deploy`
+HTTP body 必须是 JSON object。`http_protocol` 会使用 `json:decode/1` 解析请求体：
 
-请求体顶层结构：
+- JSON object -> Erlang map
+- JSON array -> Erlang list
+- JSON string -> Erlang binary
+- JSON integer -> Erlang integer
+- JSON float -> Erlang float
+- JSON boolean -> Erlang `true | false`
+
+`/container/deploy` handler 只接受顶层 map，且必须匹配：
+
+```erlang
+#{<<"uuid">> := UUID, <<"task_id">> := TaskId, <<"config">> := Config}
+```
+
+其中：
+
+- `UUID` 必须是 binary，也就是 JSON string。
+- `TaskId` 必须是 integer。
+- `Config` 必须是 map，也就是 JSON object。
+
+## 2. 完整 JSON 格式
+
+下面示例包含当前支持的所有字段。实际请求可以只传必填字段和需要的可选字段。
 
 ```json
 {
-  "uuid": "host-uuid",
+  "uuid": "qbxmjyzrkpntfgswaevodhluicqzxplkm",
   "task_id": 1001,
   "config": {
     "image": "docker.io/library/nginx:latest",
     "container_name": "my_nginx",
-    "container_dir": "/data/apps/my_nginx",
     "command": ["nginx", "-g", "daemon off;"],
+    "restart": "always",
+    "container_dir": "/data/apps/my_nginx",
     "entrypoint": ["/docker-entrypoint.sh"],
-    "envs": ["ENV1=val1", "ENV2=val2"],
-    "expose": ["80", "443/tcp"],
+    "envs": ["ENV=prod", "TZ=Asia/Shanghai"],
+    "expose": ["80", "443/tcp", "53/udp"],
     "volumes": ["/host/data:/data", "/host/log:/var/log:ro"],
     "networks": ["bridge"],
     "network_mode": "bridge",
     "labels": {
-      "role": "web",
+      "app": "nginx",
       "env": "prod"
     },
-    "restart": "always",
     "user": "www-data",
     "working_dir": "/app",
     "hostname": "myhost",
@@ -65,167 +90,311 @@
 }
 ```
 
-## 2. 顶层字段与 protobuf 的关系
+## 3. 顶层参数
 
-HTTP 顶层字段和 protobuf 请求体的关系如下：
-
-| HTTP 字段 | 类型 | 目标 protobuf 字段 | 说明 |
+| 字段 | JSON 类型 | 必填 | 说明 |
 | --- | --- | --- | --- |
-| `uuid` | `string` | 不进入 protobuf body | 仅用于在服务端定位 `iot_host` 进程 |
-| `task_id` | `integer` | `ContainerRequest.Deploy.task_id` | 必填，用于异步消息集合标识 |
-| `config` | `object` | `ContainerRequest.Deploy.params` | 必填，转换成 `ContainerDeployParams` |
+| `uuid` | string | 是 | 目标 efka 所属主机 UUID。服务端用它查找 `iot_host` 进程，不会下发到 efka。 |
+| `task_id` | integer | 是 | 部署任务 ID。会进入内部命令的 `task_id` 字段，用于关联部署结果和部署日志流。 |
+| `config` | object | 是 | 容器创建配置。会被校验并转换成内部部署参数 map。 |
 
-最终构造出来的请求体结构是：
+顶层没有 `timeout` 字段。当前 HTTP handler 等待 efka command response 的超时时间固定为 10 秒。
+
+## 4. config 参数
+
+### 必填字段
+
+| 字段 | JSON 类型 | 内部类型 | 说明 |
+| --- | --- | --- | --- |
+| `image` | string | binary | 镜像名。没有 tag 时 efka 部署逻辑会补 `:latest`。 |
+| `container_name` | string | binary | 容器名称。 |
+| `command` | string[] | binary list | 容器启动命令，对应 Docker create config 的 `cmd`。 |
+| `restart` | string | binary | 重启策略，例如 `no`、`always`、`unless-stopped`、`on-failure`、`on-failure:3`。 |
+
+### 可选字段
+
+| 字段 | JSON 类型 | 默认值 | 说明 |
+| --- | --- | --- | --- |
+| `container_dir` | string | `""` | 容器工作目录在 efka 主机上的应用目录，传给 efka 部署逻辑。 |
+| `entrypoint` | string[] | `[]` | Docker create config 的 `entrypoint`。 |
+| `envs` | string[] | `[]` | 环境变量列表，例如 `["A=1"]`，对应 Docker create config 的 `env`。 |
+| `expose` | string[] | `[]` | 容器暴露端口，只表示容器端口，不支持宿主机端口绑定。 |
+| `volumes` | string[] | `[]` | volume bind 列表，格式见后文。 |
+| `networks` | string[] | `[]` | Docker network 名称列表，用于 networking config。 |
+| `network_mode` | string | `""` | Docker host config 的 `network_mode`。 |
+| `labels` | object string:string | `{}` | 容器 labels。key/value 都必须是 string。 |
+| `user` | string | `""` | 容器运行用户。 |
+| `working_dir` | string | `""` | 容器工作目录。 |
+| `hostname` | string | `""` | 容器 hostname。 |
+| `privileged` | boolean | `false` | 是否 privileged。 |
+| `cap_add` | string[] | `[]` | 追加 Linux capability。 |
+| `cap_drop` | string[] | `[]` | 删除 Linux capability。 |
+| `devices` | string[] | `[]` | 设备映射列表，格式见后文。 |
+| `mem_limit` | string | `0` | 内存上限，解析成字节数。 |
+| `mem_reservation` | string | `0` | 内存软限制，解析成字节数。 |
+| `cpu_shares` | integer | `0` | Docker CPU shares。 |
+| `cpus` | number | `0` | CPU 数量，转换成 nano cpus。 |
+| `ulimits` | object string:string | `[]` | ulimit 配置，格式见后文。 |
+| `sysctls` | object string:string | `{}` | sysctl 配置。key/value 都必须是 string。 |
+| `tmpfs` | string[] | `{}` | tmpfs mount 配置，格式见后文。 |
+| `extra_hosts` | string[] | `[]` | 额外 hosts，例如 `["host.docker.internal:host-gateway"]`。 |
+| `healthcheck` | object | `undefined` | 健康检查配置，格式见后文。 |
+
+### 不支持字段
+
+| 字段 | 当前行为 | 说明 |
+| --- | --- | --- |
+| `ports` | 明确拒绝 | 如果传入会返回 `unsupported container config keys: ports`。当前只支持 `expose`，不支持宿主机端口绑定。 |
+| `env_file` | 忽略 | 当前校验不会识别该字段，后续构造 Docker create options 时也不会使用。 |
+| 其他未知字段 | 忽略 | 除 `ports` 外，未知字段不会报错，也不会进入内部部署参数。 |
+
+## 5. 校验规则
+
+校验分为两层。
+
+第一层在 `container_handler`：
+
+- `uuid` 必须是 binary。
+- `task_id` 必须是 integer。
+- `config` 必须是 map。
+
+第二层在 `docker_container_builder:deploy_request/2`：
+
+- 先拒绝不支持的 `ports` 字段。
+- 检查必填字段是否存在。
+- 检查已知字段类型。
+- 构造内部部署 map。
+- 对需要解析的字段做格式解析，例如端口、volume、size、duration、ulimit。
+
+类型错误会返回类似：
 
 ```text
-ContainerRequest {
-  action = {
-    deploy,
-    ContainerRequest.Deploy {
-      task_id = TaskId,
-      params = ContainerDeployParams {...}
+required parameter: <<"command">>, type must be: list of string
+optional parameter: <<"labels">>, type must be: map of string:string
+```
+
+缺少必填字段会返回类似：
+
+```text
+miss requied parameter: <<"image">>
+```
+
+注意：错误文本中的 `requied` 是当前代码里的原始拼写。
+
+## 6. 内部转换结果
+
+`config` 校验通过后，`docker_container_builder:deploy_request(TaskId, Config)` 生成：
+
+```erlang
+#{
+    action => deploy,
+    task_id => TaskId,
+    params => #{
+        container_name => ContainerName,
+        container_dir => ContainerDir,
+        create => #{
+            config => ContainerConfig,
+            host_config => HostConfig,
+            networking_config => NetworkingConfig
+        }
     }
-  }
 }
 ```
 
-## 3. `config` 到 `ContainerDeployParams` 的映射
+该 map 会通过 efka/iot 长连接协议下发：
 
-`config` 会被转换成：
-
-```text
-ContainerDeployParams {
-  container_name,
-  container_dir,
-  spec = ContainerSpec {...}
-}
+```erlang
+{command, Ref, {container, #{
+    action => deploy,
+    task_id => TaskId,
+    params => Params
+}}}
 ```
 
-字段映射如下：
+efka 返回：
 
-| `config` 字段 | 类型 | protobuf 字段 | 说明 |
-| --- | --- | --- | --- |
-| `container_name` | `string` | `ContainerDeployParams.container_name` | 必填 |
-| `container_dir` | `string` | `ContainerDeployParams.container_dir` | 可选，默认 `""` |
-| 其余部署字段 | 多种 | `ContainerDeployParams.spec` | 转成 `ContainerSpec` |
+```erlang
+{command_response, Ref, {container, Reply}}
+```
 
-## 4. `config` 到 `ContainerSpec` 的详细映射
+HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行错误通常返回 HTTP 400；找不到 host 返回业务错误 code 404。
 
-### 4.1 直接映射字段
+## 7. Docker create config 映射
 
-以下字段基本按原值写入：
+`create.config` 由 `build_docker_container_config/1` 生成：
 
-| `config` 字段 | 类型 | protobuf 字段 | 默认值 |
-| --- | --- | --- | --- |
-| `image` | `string` | `ContainerSpec.image` | 必填 |
-| `command` | `string[]` | `ContainerSpec.command` | 必填 |
-| `entrypoint` | `string[]` | `ContainerSpec.entrypoint` | `[]` |
-| `envs` | `string[]` | `ContainerSpec.env` | `[]` |
-| `networks` | `string[]` | `ContainerSpec.networks` | `[]` |
-| `network_mode` | `string` | `ContainerSpec.network_mode` | `""` |
-| `user` | `string` | `ContainerSpec.user` | `""` |
-| `working_dir` | `string` | `ContainerSpec.working_dir` | `""` |
-| `hostname` | `string` | `ContainerSpec.hostname` | `""` |
-| `privileged` | `boolean` | `ContainerSpec.privileged` | `false` |
-| `cap_add` | `string[]` | `ContainerSpec.cap_add` | `[]` |
-| `cap_drop` | `string[]` | `ContainerSpec.cap_drop` | `[]` |
-| `extra_hosts` | `string[]` | `ContainerSpec.extra_hosts` | `[]` |
+| 内部字段 | 来源 JSON 字段 | 转换规则 |
+| --- | --- | --- |
+| `image` | `image` | 原值。 |
+| `cmd` | `command` | 原值。 |
+| `entrypoint` | `entrypoint` | 默认 `[]`。 |
+| `env` | `envs` | 默认 `[]`。 |
+| `labels` | `labels` | 默认 `{}`。 |
+| `volumes` | `volumes` | 只保留 container path 列表。 |
+| `user` | `user` | 默认 `""`。 |
+| `working_dir` | `working_dir` | 默认 `""`。 |
+| `hostname` | `hostname` | 默认 `""`。 |
+| `exposed_ports` | `expose` | 转成 `#{container_port, protocol}` map 列表。 |
+| `healthcheck` | `healthcheck` | 未传时为 `undefined`。 |
 
-### 4.2 map 转换字段
+## 8. Docker host config 映射
 
-| `config` 字段 | 类型 | protobuf 字段 | 转换方式 |
-| --- | --- | --- | --- |
-| `labels` | `map<string,string>` | `ContainerSpec.labels` | 转成 key/value 列表 |
-| `sysctls` | `map<string,string>` | `ContainerSpec.sysctls` | 转成 key/value 列表 |
+`create.host_config` 由 `build_docker_host_config/1` 生成：
 
-### 4.3 结构化转换字段
+| 内部字段 | 来源 JSON 字段 | 转换规则 |
+| --- | --- | --- |
+| `binds` | `volumes` | 转成 Docker bind 字符串列表。 |
+| `network_mode` | `network_mode` | 默认 `""`。 |
+| `restart_policy` | `restart` | 转成 `#{name, maximum_retry_count}`。 |
+| `privileged` | `privileged` | 默认 `false`。 |
+| `cap_add` | `cap_add` | 默认 `[]`。 |
+| `cap_drop` | `cap_drop` | 默认 `[]`。 |
+| `devices` | `devices` | 转成设备映射 map 列表。 |
+| `memory` | `mem_limit` | 解析成字节数，未传为 `0`。 |
+| `memory_reservation` | `mem_reservation` | 解析成字节数，未传为 `0`。 |
+| `nano_cpus` | `cpus` | `cpus * 1000000000`，未传为 `0`。 |
+| `cpu_shares` | `cpu_shares` | 未传为 `0`。 |
+| `ulimits` | `ulimits` | 转成 ulimit map 列表。 |
+| `tmpfs` | `tmpfs` | 转成 map。 |
+| `sysctls` | `sysctls` | 默认 `{}`。 |
+| `extra_hosts` | `extra_hosts` | 默认 `[]`。 |
 
-#### `volumes`
+## 9. Docker networking config 映射
 
-输入类型：
+`create.networking_config` 由 `build_docker_networking_config/1` 生成：
+
+```erlang
+#{endpoints => [#{name => Network} || Network <- Networks]}
+```
+
+来源字段：
 
 ```json
-["/host/data:/data", "/host/log:/var/log:ro"]
+"networks": ["bridge", "mynet"]
 ```
 
-目标：
+转换结果：
 
-```text
-ContainerSpec.volumes = [VolumeBind...]
+```erlang
+#{endpoints => [
+    #{name => <<"bridge">>},
+    #{name => <<"mynet">>}
+]}
 ```
 
-转换规则：
+## 10. 复杂字段转换规则
 
-- `host_path:container_path`
-  - `read_only = false`
-- `host_path:container_path:ro`
-  - `read_only = true`
-- `host_path:container_path:rw`
-  - 当前实现也会被接受，但 `read_only = false`
+### restart
 
-生成结构：
+输入：
 
-```text
-VolumeBind {
-  host_path,
-  container_path,
-  read_only
-}
+```json
+"restart": "always"
 ```
 
-#### `expose`
+转换：
 
-输入类型：
+```erlang
+#{name => <<"always">>, maximum_retry_count => 0}
+```
+
+输入：
+
+```json
+"restart": "on-failure:3"
+```
+
+转换：
+
+```erlang
+#{name => <<"on-failure">>, maximum_retry_count => 3}
+```
+
+### expose
+
+输入：
 
 ```json
 ["80", "443/tcp", "53/udp"]
 ```
 
-目标：
+转换：
 
-```text
-ContainerSpec.expose = [PortExpose...]
+```erlang
+[
+    #{container_port => 80, protocol => <<"tcp">>},
+    #{container_port => 443, protocol => <<"tcp">>},
+    #{container_port => 53, protocol => <<"udp">>}
+]
 ```
 
-转换规则：
+端口必须是无符号整数，且不能超过 `4294967295`。
 
-- `"80"` -> `container_port = 80`, `protocol = "tcp"`
-- `"443/tcp"` -> `container_port = 443`, `protocol = "tcp"`
-- `"53/udp"` -> `container_port = 53`, `protocol = "udp"`
+### volumes
 
-生成结构：
+输入：
 
-```text
-PortExpose {
-  container_port,
-  protocol
-}
+```json
+["/host/data:/data", "/host/log:/var/log:ro", "/host/cache:/cache:rw"]
 ```
 
-#### `devices`
+转换为 `create.config.volumes`：
 
-输入类型：
+```erlang
+[<<"/data">>, <<"/var/log">>, <<"/cache">>]
+```
+
+转换为 `create.host_config.binds`：
+
+```erlang
+[
+    <<"/host/data:/data">>,
+    <<"/host/log:/var/log:ro">>,
+    <<"/host/cache:/cache">>
+]
+```
+
+规则：
+
+- `host_path:container_path` -> 读写挂载。
+- `host_path:container_path:ro` -> 只读挂载。
+- `host_path:container_path:rw` -> 当前实现视为读写挂载，输出时不会保留 `:rw`。
+- host path 和 container path 都不能为空。
+
+### devices
+
+输入：
 
 ```json
 ["/dev/ttyUSB0:/dev/ttyUSB0", "/dev/snd:/dev/snd:rwm"]
 ```
 
-目标：
+转换：
 
-```text
-ContainerSpec.devices = [DeviceMapping...]
+```erlang
+[
+    #{
+        path_on_host => <<"/dev/ttyUSB0">>,
+        path_in_container => <<"/dev/ttyUSB0">>,
+        cgroup_permissions => <<"rwm">>
+    },
+    #{
+        path_on_host => <<"/dev/snd">>,
+        path_in_container => <<"/dev/snd">>,
+        cgroup_permissions => <<"rwm">>
+    }
+]
 ```
 
-转换规则：
+规则：
 
-- `host_path:container_path`
-  - `cgroup_permissions = "rwm"`
-- `host_path:container_path:permissions`
-  - 使用第三段作为 `cgroup_permissions`
+- `host_path:container_path` -> 权限默认为 `rwm`。
+- `host_path:container_path:permissions` -> 使用第三段作为权限。
+- 任意路径或 permissions 为空时返回 `invalid device mapping`。
 
-#### `ulimits`
+### ulimits
 
-输入类型：
+输入：
 
 ```json
 {
@@ -234,39 +403,47 @@ ContainerSpec.devices = [DeviceMapping...]
 }
 ```
 
-目标：
+转换：
 
-```text
-ContainerSpec.ulimits = [Ulimit...]
+```erlang
+[
+    #{name => <<"nofile">>, soft => 1024, hard => 2048},
+    #{name => <<"nproc">>, soft => 4096, hard => 4096}
+]
 ```
 
-转换规则：
+规则：
 
-- `"1024:2048"` -> `soft = 1024`, `hard = 2048`
-- `"4096"` -> `soft = 4096`, `hard = 4096`
+- `"soft:hard"` -> 分别设置 soft 和 hard。
+- `"limit"` -> soft 和 hard 都等于 limit。
+- 数值必须是无符号整数。
 
-#### `tmpfs`
+### tmpfs
 
-输入类型：
+输入：
 
 ```json
 ["/tmp", "/run:rw,size=64m"]
 ```
 
-目标：
+转换：
 
-```text
-ContainerSpec.tmpfs = [TmpfsMount...]
+```erlang
+#{
+    <<"/tmp">> => <<>>,
+    <<"/run">> => <<"rw,size=64m">>
+}
 ```
 
-转换规则：
+规则：
 
-- `"path"` -> `options = ""`
-- `"path:options"` -> `options` 为第二段
+- `"path"` -> options 为 `""`。
+- `"path:options"` -> options 为第二段。
+- path 不能为空。
 
-#### `healthcheck`
+### healthcheck
 
-输入类型：
+输入：
 
 ```json
 {
@@ -277,188 +454,94 @@ ContainerSpec.tmpfs = [TmpfsMount...]
 }
 ```
 
-目标：
+转换：
 
-```text
-ContainerSpec.healthcheck = Healthcheck {
-  test,
-  interval_ns,
-  timeout_ns,
-  retries
+```erlang
+#{
+    test => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
+    interval_ns => 30000000000,
+    timeout_ns => 10000000000,
+    retries => 3
 }
 ```
 
-转换规则：
+字段规则：
 
-- `test` 直接写入 `Healthcheck.test`
-- `interval` / `timeout` 会被解析成纳秒
-- `retries` 直接写入
+| 字段 | JSON 类型 | 默认值 | 说明 |
+| --- | --- | --- | --- |
+| `test` | any | `[]` | 当前只要求 `healthcheck` 的 key 是 string，不单独校验 `test` 类型；建议传 string array。 |
+| `interval` | string 或 integer | `"0s"` | string 会解析时间单位；integer 直接视为纳秒。 |
+| `timeout` | string 或 integer | `"0s"` | string 会解析时间单位；integer 直接视为纳秒。 |
+| `retries` | integer | `0` | 当前构造阶段直接取值；建议传非负整数。 |
 
-支持的时间单位：
+时间单位：
 
-- `ns`
-- `us`
-- `ms`
-- `s`
-- `m`
-- `h`
+| 单位 | 含义 |
+| --- | --- |
+| `ns` | 纳秒 |
+| `us` | 微秒 |
+| `ms` | 毫秒 |
+| `s` | 秒 |
+| `m` | 分钟 |
+| `h` | 小时 |
+| 无单位 | 秒 |
 
 示例：
 
 - `"30s"` -> `30000000000`
 - `"10ms"` -> `10000000`
 - `"2m"` -> `120000000000`
+- `"5"` -> `5000000000`
 
-### 4.4 资源限制字段
+### mem_limit 和 mem_reservation
 
-#### `restart`
-
-输入类型：
+输入：
 
 ```json
-"always"
+"mem_limit": "512m",
+"mem_reservation": "1g"
 ```
 
-或：
+转换：
 
-```json
-"on-failure:3"
+```erlang
+memory => 536870912,
+memory_reservation => 1073741824
 ```
 
-目标：
-
-```text
-RestartPolicy {
-  name,
-  maximum_retry_count
-}
-```
-
-转换规则：
-
-- `"always"` -> `name = "always"`, `maximum_retry_count = 0`
-- `"on-failure:3"` -> `name = "on-failure"`, `maximum_retry_count = 3`
-
-#### `mem_limit` / `mem_reservation`
-
-目标字段：
-
-- `ResourceLimits.memory_bytes`
-- `ResourceLimits.memory_reservation_bytes`
-
 支持单位：
 
-- `b`
-- `k`, `kb`, `ki`, `kib`
-- `m`, `mb`, `mi`, `mib`
-- `g`, `gb`, `gi`, `gib`
-- `t`, `tb`, `ti`, `tib`
+| 单位 | 倍数 |
+| --- | --- |
+| `b` 或无单位 | 1 |
+| `k`, `kb`, `ki`, `kib` | 1024 |
+| `m`, `mb`, `mi`, `mib` | 1048576 |
+| `g`, `gb`, `gi`, `gib` | 1073741824 |
+| `t`, `tb`, `ti`, `tib` | 1099511627776 |
 
-示例：
+数值支持整数或小数，例如 `"1.5g"`。
 
-- `"512m"` -> `536870912`
-- `"1g"` -> `1073741824`
+### cpus
 
-#### `cpus`
+输入：
 
-目标字段：
-
-- `ResourceLimits.nano_cpus`
-
-转换规则：
-
-- 直接乘以 `1_000_000_000`
-
-示例：
-
-- `1` -> `1000000000`
-- `1.5` -> `1500000000`
-
-#### `cpu_shares`
-
-目标字段：
-
-- `ResourceLimits.cpu_shares`
-
-直接按整数写入。
-
-#### `resources` 对象生成规则
-
-只有在以下字段至少存在一个时，才会生成 `ContainerSpec.resources`：
-
-- `mem_limit`
-- `mem_reservation`
-- `cpus`
-- `cpu_shares`
-
-如果这些字段都不存在，则 `resources = undefined`。
-
-## 5. 参数校验规则
-
-当前实现里：
-
-- `container_handler` 只校验顶层请求结构：
-  - `uuid` 必须是 `binary`
-  - `task_id` 必须是 `integer`
-  - `config` 必须是 `map`
-- `config` 内部字段的必填项、类型检查、格式解析和不支持字段判断，全部在 `iot_container_request_builder:deploy_request/2` 中完成
-
-必填字段：
-
-- `image: string`
-- `container_name: string`
-- `command: string[]`
-- `restart: string`
-
-可选字段：
-
-- `privileged: boolean`
-- `entrypoint: string[]`
-- `envs: string[]`
-- `ports: string[]`
-- `expose: string[]`
-- `volumes: string[]`
-- `networks: string[]`
-- `labels: map<string,string>`
-- `user: string`
-- `working_dir: string`
-- `hostname: string`
-- `container_dir: string`
-- `network_mode: string`
-- `cap_add: string[]`
-- `cap_drop: string[]`
-- `devices: string[]`
-- `mem_limit: string`
-- `mem_reservation: string`
-- `cpu_shares: integer`
-- `cpus: number`
-- `ulimits: map<string,string>`
-- `sysctls: map<string,string>`
-- `tmpfs: string[]`
-- `extra_hosts: string[]`
-- `healthcheck: map<string,any>`
-
-## 6. 当前限制
-
-### `ports`
-
-虽然 HTTP 校验允许 `ports` 字段出现，但当前 builder 仍然会拒绝它：
-
-```text
-unsupported container config keys: ports
+```json
+"cpus": 1.5
 ```
 
-原因是当前 `message.proto` 里只有：
+转换：
 
-- `ContainerSpec.expose`
+```erlang
+nano_cpus => 1500000000
+```
 
-它表达的是容器端口暴露，不包含主机端口绑定信息；而 `ports` 一般是类似 `8080:80` 的 host/container 绑定语义，两者并不等价。
+规则：
 
-### `env_file`
+- integer 或 float 都可以。
+- 必须大于等于 0。
+- 转换公式：`trunc(Cpus * 1000000000)`。
 
-当前实现不再接收 `env_file`。如果传入，HTTP 层就不会通过类型校验，因为它不在 `validate_config/1` 的允许字段列表中。
-
-## 7. 推荐请求示例
+## 11. 最小请求示例
 
 ```json
 {
@@ -467,59 +550,85 @@ unsupported container config keys: ports
   "config": {
     "image": "docker.io/library/nginx:latest",
     "container_name": "my_nginx",
-    "container_dir": "/data/apps/my_nginx",
     "command": ["nginx", "-g", "daemon off;"],
-    "entrypoint": ["/docker-entrypoint.sh"],
-    "envs": ["ENV=prod", "TZ=Asia/Shanghai"],
-    "expose": ["80", "443/tcp"],
-    "volumes": ["/host/data:/data", "/host/log:/var/log:ro"],
-    "networks": ["bridge"],
-    "network_mode": "bridge",
-    "labels": {
-      "app": "nginx",
-      "env": "prod"
-    },
-    "restart": "always",
-    "user": "www-data",
-    "working_dir": "/app",
-    "hostname": "myhost",
-    "privileged": false,
-    "cap_add": ["NET_ADMIN"],
-    "cap_drop": ["MKNOD"],
-    "devices": ["/dev/ttyUSB0:/dev/ttyUSB0:rwm"],
-    "mem_limit": "512m",
-    "mem_reservation": "256m",
-    "cpu_shares": 512,
-    "cpus": 1.5,
-    "ulimits": {
-      "nofile": "1024:2048"
-    },
-    "sysctls": {
-      "net.ipv4.ip_forward": "1"
-    },
-    "tmpfs": ["/tmp", "/run:rw,size=64m"],
-    "extra_hosts": ["host.docker.internal:host-gateway"],
-    "healthcheck": {
-      "test": ["CMD-SHELL", "curl -f http://localhost || exit 1"],
-      "interval": "30s",
-      "timeout": "10s",
-      "retries": 3
-    }
+    "restart": "always"
   }
 }
 ```
 
-## 8. 后续如果要支持 `ports`
+对应内部命令示意：
 
-建议先扩展 `message.proto`，增加明确表达 host/container 端口绑定的结构，例如：
+```erlang
+{command, Ref, {container, #{
+    action => deploy,
+    task_id => 1001,
+    params => #{
+        container_name => <<"my_nginx">>,
+        container_dir => <<>>,
+        create => #{
+            config => #{
+                image => <<"docker.io/library/nginx:latest">>,
+                cmd => [<<"nginx">>, <<"-g">>, <<"daemon off;">>],
+                entrypoint => [],
+                env => [],
+                labels => #{},
+                volumes => [],
+                user => <<>>,
+                working_dir => <<>>,
+                hostname => <<>>,
+                exposed_ports => [],
+                healthcheck => undefined
+            },
+            host_config => #{
+                binds => [],
+                network_mode => <<>>,
+                restart_policy => #{name => <<"always">>, maximum_retry_count => 0},
+                privileged => false,
+                cap_add => [],
+                cap_drop => [],
+                devices => [],
+                memory => 0,
+                memory_reservation => 0,
+                nano_cpus => 0,
+                cpu_shares => 0,
+                ulimits => [],
+                tmpfs => #{},
+                sysctls => #{},
+                extra_hosts => []
+            },
+            networking_config => #{endpoints => []}
+        }
+    }
+}}}
+```
 
-```proto
-message PortBinding {
-  uint32 host_port = 1;
-  uint32 container_port = 2;
-  string protocol = 3;
-  string host_ip = 4;
+## 12. 响应
+
+成功时 HTTP body 由 `iot_util:json_data/1` 包装：
+
+```json
+{
+  "result": "ok"
 }
 ```
 
-然后再在 `ContainerSpec` 中增加类似 `repeated PortBinding ports = ...;` 的字段，再由 builder 把 `"8080:80/tcp"` 这类字符串解析进去。这样语义才完整。
+如果 efka 返回的是 JSON binary，handler 会尝试解码后再放入 `result`。
+
+错误示例：
+
+```json
+{
+  "error": {
+    "code": 400,
+    "message": "unsupported container config keys: ports"
+  }
+}
+```
+
+常见状态：
+
+| HTTP 状态 | 场景 |
+| --- | --- |
+| `200` | host not found 或构造阶段返回的业务错误也可能通过 200 包装业务错误。 |
+| `400` | 参数校验失败、efka 返回普通错误。 |
+| `504` | 等待 efka command response 超时。 |
diff --git a/docs/efka_iot_protocol.md b/docs/efka_iot_protocol.md
new file mode 100644
index 0000000..f5e3472
--- /dev/null
+++ b/docs/efka_iot_protocol.md
@@ -0,0 +1,255 @@
+# EFKA 与 IOT 交互协议
+
+本文档描述 `efka` 与 `iot` 之间的 TLS 长连接协议。当前协议由 Erlang term 直接序列化，发送端使用 `term_to_binary/1`，接收端使用 `binary_to_term(PacketBin, [safe])`。
+
+## 传输层
+
+- `efka` 作为 TLS client 连接 `iot`。
+- `iot` 作为 TLS server 接收多个 `efka` 连接，一个连接对应一个 `ssl_channel` 进程。
+- socket 使用 `{packet, 4}`，每个 Erlang term binary 作为一个完整包发送。
+- `Ref` 使用 `make_ref()` 生成，只在当前连接的 inflight 表内匹配。
+
+## 顶层帧
+
+协议顶层 tuple 用来表达交互语义：
+
+```erlang
+{request, Ref, Body}
+{response, Ref, Reply}
+{command, Ref, {Domain, Payload}}
+{command_response, Ref, {Domain, Reply}}
+{message, Body}
+```
+
+语义说明：
+
+| 帧 | 方向 | 语义 |
+| --- | --- | --- |
+| `{request, Ref, Body}` | efka -> iot | efka 发起请求，需要 iot 回复 |
+| `{response, Ref, Reply}` | iot -> efka | iot 对 efka request 的回复 |
+| `{command, Ref, {Domain, Payload}}` | iot -> efka | iot 下发命令，需要 efka 回复 |
+| `{command_response, Ref, {Domain, Reply}}` | efka -> iot | efka 对 iot command 的回复 |
+| `{message, Body}` | 双向 | 异步消息，不要求回复 |
+
+`command` 和 `command_response` 的 `Domain` 表示业务域，目前支持：
+
+- `auth`
+- `container`
+
+## 鉴权请求
+
+初始连接由 `efka` 发起鉴权 request：
+
+```erlang
+{request, Ref, {auth_request, #{
+    uuid => UUID,
+    token => Token,
+    timestamp => Timestamp
+}}}
+```
+
+`iot` 回复：
+
+```erlang
+{response, Ref, {auth_response, ok}}
+{response, Ref, {auth_response, {error, {denied, Reason}}}}
+{response, Ref, {auth_response, {error, {failed, Reason}}}}
+```
+
+处理语义：
+
+- `ok`：`efka` 进入 `activated` 状态。
+- `{error, {denied, Reason}}`：`efka` 进入 `restricted` 状态，不能正常上报数据，但仍可接收部分命令。
+- `{error, {failed, Reason}}`：鉴权失败，连接关闭后重连。
+
+## 授权控制命令
+
+`iot` 对 `efka` 的授权控制使用 command 语义：
+
+```erlang
+{command, Ref, {auth, activate}}
+{command, Ref, {auth, deactivate}}
+```
+
+`efka` 回复：
+
+```erlang
+{command_response, Ref, {auth, ok}}
+{command_response, Ref, {auth, {error, Reason}}}
+```
+
+处理语义：
+
+- `activate`：如果 `efka` 已经是 `activated`，直接回复 `ok`；否则重新发送 `auth_request`，等待鉴权结果后再回复该 command。
+- `deactivate`：`efka` 进入 `restricted` 状态，并回复 `ok`。
+
+## 容器管理命令
+
+`iot` 对 `efka` 的容器管理使用 command 语义：
+
+```erlang
+{command, Ref, {container, CommandMap}}
+```
+
+`efka` 回复：
+
+```erlang
+{command_response, Ref, {container, Reply}}
+```
+
+`Reply` 取值：
+
+```erlang
+ok
+{ok, Result}
+{error, Reason}
+```
+
+### list
+
+```erlang
+#{action => list}
+```
+
+返回当前 `efka` 主机上的容器列表。
+
+### deploy
+
+```erlang
+#{
+    action => deploy,
+    task_id => TaskId,
+    params => Params
+}
+```
+
+触发容器部署。部署过程中的流式日志不通过该 command response 返回，而是通过 `message` 的 `task_event` 上报。
+
+### start
+
+```erlang
+#{
+    action => start,
+    target => Target
+}
+```
+
+### stop
+
+```erlang
+#{
+    action => stop,
+    target => Target,
+    timeout_seconds => TimeoutSeconds
+}
+```
+
+### kill
+
+```erlang
+#{
+    action => kill,
+    target => Target,
+    signal => Signal
+}
+```
+
+### remove
+
+```erlang
+#{
+    action => remove,
+    target => Target,
+    force => Force,
+    remove_volumes => RemoveVolumes
+}
+```
+
+### config
+
+```erlang
+#{
+    action => config,
+    target => Target,
+    config => Config
+}
+```
+
+更新容器配置文件。
+
+### Target
+
+容器目标使用 map 表示：
+
+```erlang
+#{
+    name => ContainerName,
+    id => ContainerId
+}
+```
+
+`name` 和 `id` 至少一个非空；优先使用 `name`，`name` 为空时使用 `id`。
+
+## 异步消息
+
+`message` 不带 `Ref`，不要求对端回复。
+
+### efka -> iot: data
+
+```erlang
+{message, {data, #{
+    route_key => RouteKey,
+    metric => Metric
+}}}
+```
+
+用于 `efka` 上报业务指标数据。
+
+### efka -> iot: task_event
+
+```erlang
+{message, {task_event, #{
+    task_id => TaskId,
+    type => Type,
+    stream => Stream
+}}}
+```
+
+任务事件流关闭时：
+
+```erlang
+{message, {task_event, #{
+    task_id => TaskId,
+    type => <<"close">>,
+    stream => Reason
+}}}
+```
+
+### iot -> efka: pub
+
+```erlang
+{message, {pub, #{
+    topic => Topic,
+    qos => Qos,
+    content => Content
+}}}
+```
+
+用于 `iot` 向 `efka` 本地订阅系统发布 topic 消息。
+
+## 状态与超时
+
+- `efka` 鉴权超时时间：5 秒。
+- `iot` command inflight 超时时间：60 秒。
+- `iot` 管理多个 `efka` 时，每个连接有独立 `ssl_channel` 和独立 inflight 表。
+- command 超时后，`iot` 删除 inflight 记录；之后如果迟到的 `command_response` 到达，会被视为未预期响应。
+
+## 兼容性
+
+当前协议不兼容旧 tuple：
+
+- 旧容器管理：`{request, Ref, {container_request, ...}}`
+- 旧容器回复：`{response, Ref, {container_response, ...}}`
+- 旧授权控制：`{message, {auth_control, Command}}`
+
+如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。
diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl
index 49aab46..8307d8f 100644
--- a/src/host/iot_host.erl
+++ b/src/host/iot_host.erl
@@ -128,14 +128,14 @@ remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName)
     ok | {ok, Result :: term()} | {error, Reason :: term()}.
 await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) ->
     receive
-        {request_reply, Ref, ok} ->
+        {command_reply, Ref, ok} ->
             ok;
-        {request_reply, Ref, {ok, Result}} ->
+        {command_reply, Ref, {ok, Result}} ->
             {ok, Result};
-        {request_reply, Ref, {error, Reason}} ->
+        {command_reply, Ref, {error, Reason}} ->
             {error, Reason}
     after Timeout ->
-        ok = gen_statem:call(Pid, {cancel_request_call, Ref}),
+        ok = gen_statem:call(Pid, {cancel_command_call, Ref}),
         flush_reply(Ref),
         {error, timeout}
     end.
@@ -217,13 +217,13 @@ handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #s
             {keep_state, State, [{reply, From, {ok, Ref}}]};
         false ->
             logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]),
-            {keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
+            {keep_state, State, [{reply, From, {error, <<"主机离线，发送命令失败"/utf8>>}}]}
     end;
 
-handle_event({call, From}, {cancel_request_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
+handle_event({call, From}, {cancel_command_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
     case is_pid(ChannelPid) of
         true ->
-            ok = ssl_channel:cancel_request_call(ChannelPid, Ref),
+            ok = ssl_channel:cancel_command_call(ChannelPid, Ref),
             {keep_state, State, [{reply, From, ok}]};
         false ->
             {keep_state, State, [{reply, From, ok}]}
@@ -234,7 +234,7 @@ handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State =
     case HasSession andalso is_pid(ChannelPid) of
         true ->
             logger:debug("[iot_host] host: ~p, publish to topic: ~p, content: ~p", [UUID, Topic, Content]),
-            %% 通过websocket发送请求
+            %% 通过websocket发送消息
             ssl_channel:pub(ChannelPid, Topic, Qos, Content),
 
             {keep_state, State, [{reply, From, ok}]};
@@ -375,7 +375,7 @@ state_map(#state{host_id = HostId, uuid = UUID, has_session = HasSession, heartb
 
 flush_reply(Ref) ->
     receive
-        {request_reply, Ref, _Reply} ->
+        {command_reply, Ref, _Reply} ->
             ok
     after 0 ->
         ok
diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl
index 2168ede..e2e0eba 100644
--- a/src/transport/tcp/ssl_channel.erl
+++ b/src/transport/tcp/ssl_channel.erl
@@ -13,7 +13,7 @@
 -define(INFLIGHT_TIMEOUT, 60000).
 
 %% API
--export([pub/4, container_call/3, cancel_request_call/2, command/2, activate/2]).
+-export([pub/4, container_call/3, cancel_command_call/2, command/2, activate/2]).
 
 -export([start_link/3, stop/2]).
 %% gen_server callbacks
@@ -27,12 +27,12 @@
     %% 用户进程id
     host_pid = undefined,
 
-    %% 请求响应的对应关系
+    %% iot 发起的 command 与 command_response 的对应关系
     inflight = #{}
 }).
 
--record(inflight_request, {
-    receiver_pid :: pid(),
+-record(inflight_command, {
+    receiver_pid :: undefined | pid(),
     timer_ref :: reference()
 }).
 
@@ -54,12 +54,12 @@ activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) ->
 -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: map()) -> Ref :: reference().
 container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_map(Request) ->
     Ref = make_ref(),
-    gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {container_request, Request}}),
+    gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}),
     Ref.
 
--spec cancel_request_call(Pid :: pid(), Ref :: reference()) -> ok.
-cancel_request_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
-    gen_server:call(Pid, {cancel_request_call, Ref}).
+-spec cancel_command_call(Pid :: pid(), Ref :: reference()) -> ok.
+cancel_command_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
+    gen_server:call(Pid, {cancel_command_call, Ref}).
 
 %% 关闭方法
 -spec stop(Pid :: pid(), Reason :: any()) -> no_return().
@@ -83,9 +83,9 @@ init(Ref, Transport, _Opts = []) ->
     % erlang:start_timer(?PING_TICKER, self(), ping_ticker),
     gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
 
-handle_call({cancel_request_call, Ref}, _From, State = #state{inflight = Inflight}) ->
+handle_call({cancel_command_call, Ref}, _From, State = #state{inflight = Inflight}) ->
     case maps:take(Ref, Inflight) of
-        {#inflight_request{timer_ref = TimerRef}, NInflight} ->
+        {#inflight_command{timer_ref = TimerRef}, NInflight} ->
             erlang:cancel_timer(TimerRef),
             {reply, ok, State#state{inflight = NInflight}};
         error ->
@@ -101,25 +101,30 @@ handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, so
     {noreply, State};
 
 %% 发送Command消息
-handle_cast({command, Command}, State = #state{transport = Transport, socket = Socket}) ->
-    Packet = term_to_binary({message, {auth_control, Command}}),
-    Transport:send(Socket, Packet),
-    {noreply, State};
-
-%% 推送需要响应的请求
-handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
-    Packet = term_to_binary({request, Ref, Body}),
+handle_cast({command, Command}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
+    Ref = make_ref(),
+    Packet = term_to_binary({command, Ref, {auth, Command}}),
     Transport:send(Socket, Packet),
 
-    TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, Ref}),
-    RequestInfo = #inflight_request{receiver_pid = ReceiverPid, timer_ref = TimerRef},
+    TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}),
+    CommandInfo = #inflight_command{receiver_pid = undefined, timer_ref = TimerRef},
 
-    {noreply, State#state{inflight = maps:put(Ref, RequestInfo, Inflight)}}.
+    {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}};
 
-handle_info({timeout, TimerRef, {request_timeout, Ref}}, State = #state{inflight = Inflight}) ->
+%% iot 请求 efka 时使用 command/command_response 语义。
+handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
+    Packet = term_to_binary({command, Ref, Body}),
+    Transport:send(Socket, Packet),
+
+    TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}),
+    CommandInfo = #inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef},
+
+    {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}}.
+
+handle_info({timeout, TimerRef, {command_timeout, Ref}}, State = #state{inflight = Inflight}) ->
     case maps:get(Ref, Inflight, undefined) of
-        #inflight_request{timer_ref = TimerRef} ->
-            logger:warning("[ws_channel] request timeout, ref: ~p", [Ref]),
+        #inflight_command{timer_ref = TimerRef} ->
+            logger:warning("[ws_channel] command timeout, ref: ~p", [Ref]),
             {noreply, State#state{inflight = maps:remove(Ref, Inflight)}};
         _ ->
             {noreply, State}
@@ -140,8 +145,8 @@ handle_info({ssl, Socket, PacketBin}, State = #state{transport = Transport, sock
             handle_request_frame(Ref, Body, Transport, Socket, State);
         {message, Body} ->
             handle_message_frame(Body, HostPid, State);
-        {response, Ref, Response} ->
-            handle_response_frame(Ref, Response, Inflight, State);
+        {command_response, Ref, Response} ->
+            handle_command_response_frame(Ref, Response, Inflight, State);
         Other ->
             logger:warning("[ssl_channel] unsupported packet: ~p", [Other]),
             {stop, bad_packet, State}
@@ -213,8 +218,8 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timesta
             logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]),
             {stop, Reason, State}
     end;
-handle_request_frame(Ref, {container_request, ContainerRequest}, _Transport, _Socket, State) ->
-    logger:warning("[ws_channel] unsupported request message type: container_request, ref: ~p, request: ~p", [Ref, ContainerRequest]),
+handle_request_frame(Ref, {container, ContainerCommand}, _Transport, _Socket, State) ->
+    logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]),
     {stop, normal, State};
 handle_request_frame(Ref, Body, _Transport, _Socket, State) ->
     logger:warning("[ws_channel] unsupported request body, ref: ~p, body: ~p", [Ref, Body]),
@@ -239,24 +244,29 @@ handle_event_stream_frame(#{task_id := TaskId, type := Type, stream := Stream})
     logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
     iot_event_stream_observer:stream_data(TaskId, Type, Stream).
 
--spec handle_response_frame(reference(), tuple(), map(), #state{}) ->
+-spec handle_command_response_frame(reference(), tuple(), map(), #state{}) ->
     {noreply, #state{}}.
-handle_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref) ->
+handle_command_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref) ->
     case maps:take(Ref, Inflight) of
         error ->
             {noreply, State};
-        {#inflight_request{receiver_pid = ReceiverPid, timer_ref = TimerRef}, NInflight} ->
+        {#inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef}, NInflight} ->
             erlang:cancel_timer(TimerRef),
-            case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
-                true ->
-                    ReceiverPid ! {request_reply, Ref, decode_reply(Reply)};
-                false ->
-                    logger:warning("[ws_channel] get reply message: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref])
+            case ReceiverPid of
+                undefined ->
+                    ok;
+                _ when is_pid(ReceiverPid) ->
+                    case is_process_alive(ReceiverPid) of
+                        true ->
+                            ReceiverPid ! {command_reply, Ref, decode_command_response(Reply)};
+                        false ->
+                            logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref])
+                    end
             end,
             {noreply, State#state{inflight = NInflight}}
     end;
-handle_response_frame(Ref, Reply, _Inflight, State) ->
-    logger:warning("[ws_channel] unexpected response frame, ref: ~p, reply: ~p", [Ref, Reply]),
+handle_command_response_frame(Ref, Reply, _Inflight, State) ->
+    logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]),
     {noreply, State}.
 
 -spec send_reply_frame(module(), any(), reference(), tuple()) -> any().
@@ -264,15 +274,15 @@ send_reply_frame(Transport, Socket, Ref, Reply) ->
     Packet = term_to_binary({response, Ref, Reply}),
     Transport:send(Socket, Packet).
 
--spec decode_reply({container_response, ok | {ok, term()} | {error, term()}} | tuple()) ->
+-spec decode_command_response({container, ok | {ok, term()} | {error, term()}} | tuple()) ->
     ok | {ok, term()} | {error, term()}.
-decode_reply({container_response, ok}) ->
+decode_command_response({container, ok}) ->
     ok;
-decode_reply({container_response, {ok, Result}}) ->
+decode_command_response({container, {ok, Result}}) ->
     {ok, Result};
-decode_reply({container_response, {error, Reason}}) ->
+decode_command_response({container, {error, Reason}}) ->
     {error, Reason};
-decode_reply(_Reply) ->
+decode_command_response(_Reply) ->
     {error, invalid_response}.
 
 %% 检测token是否是合法值